diff --git a/go.mod b/go.mod index 7032ca8..cf2c23a 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,7 @@ require ( github.com/xiaoqidun/guigo v0.0.0-20261017142315-0279420b6e12 github.com/xiaoqidun/j2kgo v1.0.2 github.com/xiaoqidun/jbig2 v1.1.8 - github.com/xiaoqidun/pdfgo v0.0.0-20261016003351-18348c283ac1 + github.com/xiaoqidun/pdfgo v0.0.0-20261017161421-0b88b77eefc9 golang.org/x/crypto v0.58.0 golang.org/x/image v0.47.0 golang.org/x/text v0.43.0 diff --git a/go.sum b/go.sum index 111bdae..ecade45 100644 --- a/go.sum +++ b/go.sum @@ -64,8 +64,8 @@ github.com/xiaoqidun/j2kgo v1.0.2 h1:OAjjEOfaBaQzBNcj9wx+eM68UI5pewR6SzNZa1R1DIw github.com/xiaoqidun/j2kgo v1.0.2/go.mod h1:DCwAfuDzCDYqUiAlZ7BDwigTjaXw7o9L1FtEma/D1aA= github.com/xiaoqidun/jbig2 v1.1.8 h1:dERPW7F8gs18a/DOD9gXHR3fP1Vl6KhCNsLnq5l3BIo= github.com/xiaoqidun/jbig2 v1.1.8/go.mod h1:9jromqVwMqlptUDEt6Yip8vqYA5I1YjhG7FgMl/Z2dA= -github.com/xiaoqidun/pdfgo v0.0.0-20261016003351-18348c283ac1 h1:PNucIHG+qid5/GCN3wUGEqfsyrykhoazBbtPaGysWHk= -github.com/xiaoqidun/pdfgo v0.0.0-20261016003351-18348c283ac1/go.mod h1:AaabzsObZ8Z76AcaWbo6swRROrJsblyIGB7HbieeJA4= +github.com/xiaoqidun/pdfgo v0.0.0-20261017161421-0b88b77eefc9 h1:DsWSdFwsDpI8k0jadqcnGbpJuOpT+tIVIX5JfDFS2GQ= +github.com/xiaoqidun/pdfgo v0.0.0-20261017161421-0b88b77eefc9/go.mod h1:AaabzsObZ8Z76AcaWbo6swRROrJsblyIGB7HbieeJA4= github.com/xyproto/randomstring v1.0.5 h1:YtlWPoRdgMu3NZtP45drfy1GKoojuR7hmRcnhZqKjWU= github.com/xyproto/randomstring v1.0.5/go.mod h1:rgmS5DeNXLivK7YprL0pY+lTuhNQW3iGxZ18UQApw/E= github.com/yuin/goldmark v1.8.5 h1:r6N5afV5qj/5S4UTch8agZHJ8UxNCMwX7WjkkJam2NA= diff --git a/internal/webui/webui.go b/internal/webui/webui.go index 6ae1eea..dfb8c2c 100644 --- a/internal/webui/webui.go +++ b/internal/webui/webui.go @@ -31,6 +31,7 @@ import ( "slices" "strconv" "strings" + "sync" "time" "github.com/xiaoqidun/ofdgo" @@ -406,6 +407,7 @@ type annotationOptions struct { // outputContext 定期让出浏览器线程,将取消请求传入库层长操作 type outputContext struct { context.Context + mu sync.Mutex cancel context.CancelFunc checkpoint func() error checked time.Time @@ -7250,6 +7252,8 @@ func (c *outputContext) Err() error { if err := c.Context.Err(); err != nil { return err } + c.mu.Lock() + defer c.mu.Unlock() if c.pending != 0 { c.pending-- return nil diff --git a/internal/webui/webui_native.go b/internal/webui/webui_native.go index cb35356..38c45da 100644 --- a/internal/webui/webui_native.go +++ b/internal/webui/webui_native.go @@ -19,10 +19,12 @@ package webui import ( "bytes" "context" + "encoding" "encoding/json" "errors" "fmt" "io" + "math" "reflect" "strings" "sync" @@ -39,6 +41,9 @@ const ( platformTypeFunction ) +// nativeFields 缓存可直接展开的JSON字段,空值使用标准编码 +var nativeFields sync.Map + // NativeResolver 将授权标识解析为随机读取源,生命周期由宿主管理 // 入参: token 资源授权标识 // 返回: io.ReaderAt 随机读取器, int64 资源大小, error 解析错误 @@ -111,6 +116,13 @@ type platformValue struct{ value any } // nativeCallback 同步宿主回调 type nativeCallback func(...any) (any, error) +// nativeField 保存字段名称、位置及空值省略规则 +type nativeField struct { + name string + index int + omitEmpty bool +} + // NewNativeEngine 创建独立业务会话,resolver不接收未经授权的文件路径 func NewNativeEngine(resolve NativeResolver) *NativeEngine { return &NativeEngine{Resolve: resolve} } @@ -322,6 +334,26 @@ func nativePlain(value any) any { return nativePlain(v.Elem().Interface()) } if v.Kind() == reflect.Struct { + if fields := nativeStructFields(v.Type()); fields != nil { + result := make(map[string]any, len(fields)) + for _, field := range fields { + item := v.Field(field.index) + if field.omitEmpty { + switch item.Kind() { + case reflect.Array, reflect.Map, reflect.Slice, reflect.String: + if item.Len() == 0 { + continue + } + default: + if item.Kind() != reflect.Struct && item.IsZero() { + continue + } + } + } + result[field.name] = nativePlain(item.Interface()) + } + return result + } data, err := json.Marshal(value) if err != nil { panic(err) @@ -360,9 +392,84 @@ func nativePlain(value any) any { } return result } + if v.Kind() == reflect.Float32 || v.Kind() == reflect.Float64 { + if number := v.Float(); math.IsNaN(number) || math.IsInf(number, 0) { + _, err := json.Marshal(value) + panic(err) + } + } return value } +// nativeStructFields 按类型复用普通结构体字段,自定义编码和嵌入字段交给JSON处理 +// 入参: value 结构体类型 +// 返回: []nativeField 字段列表,nil表示使用标准编码 +func nativeStructFields(value reflect.Type) []nativeField { + if fields, ok := nativeFields.Load(value); ok { + return fields.([]nativeField) + } + var fields []nativeField + if nativePlainType(value, make(map[reflect.Type]bool)) { + fields = make([]nativeField, 0, value.NumField()) + for i := 0; i < value.NumField(); i++ { + field := value.Field(i) + if !field.IsExported() || field.Tag.Get("json") == "-" { + continue + } + name, options, _ := strings.Cut(field.Tag.Get("json"), ",") + if name == "" { + name = field.Name + } + fields = append(fields, nativeField{name: name, index: i, omitEmpty: options == "omitempty"}) + } + } + stored, _ := nativeFields.LoadOrStore(value, fields) + return stored.([]nativeField) +} + +// nativePlainType 检查类型是否可直接展开,递归引用及复杂字段沿用标准编码 +// 入参: value 字段类型, visiting 当前类型路径 +// 返回: bool 是否可直接展开 +func nativePlainType(value reflect.Type, visiting map[reflect.Type]bool) bool { + if visiting[value] || value.Implements(reflect.TypeFor[json.Marshaler]()) || value.Implements(reflect.TypeFor[encoding.TextMarshaler]()) || reflect.PointerTo(value).Implements(reflect.TypeFor[json.Marshaler]()) || reflect.PointerTo(value).Implements(reflect.TypeFor[encoding.TextMarshaler]()) || value == reflect.TypeFor[json.Number]() { + return false + } + visiting[value] = true + defer delete(visiting, value) + switch value.Kind() { + case reflect.Bool, reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64, reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr, reflect.Float32, reflect.Float64, reflect.String: + return true + case reflect.Pointer, reflect.Slice, reflect.Array: + return nativePlainType(value.Elem(), visiting) + case reflect.Struct: + names := make(map[string]bool) + for i := 0; i < value.NumField(); i++ { + field := value.Field(i) + if field.Anonymous { + return false + } + if !field.IsExported() || field.Tag.Get("json") == "-" { + continue + } + name, options, _ := strings.Cut(field.Tag.Get("json"), ",") + if name == "" { + name = field.Name + } + if name == "-" || names[name] || options != "" && options != "omitempty" || !nativePlainType(field.Type, visiting) { + return false + } + for _, char := range name { + if char != '_' && char != '-' && (char < 'a' || char > 'z') && (char < 'A' || char > 'Z') && (char < '0' || char > '9') { + return false + } + } + names[name] = true + } + return true + } + return false +} + // nativeFailure 转换结构化业务错误 func nativeFailure(err error) map[string]any { result := nativePlain(failureResult(err)).(map[string]any) @@ -693,7 +800,8 @@ func (c *nativeCall) finish(name string, result map[string]any) error { } c.output = nil } else { - data["blob"] = append([]byte(nil), c.buffer.Bytes()...) + data["blob"] = c.buffer.Bytes() + c.buffer = bytes.Buffer{} } data["size"] = c.size c.committed = true diff --git a/ofdgo_compression_parallel.go b/ofdgo_compression_parallel.go index 4a1fef2..15abda3 100644 --- a/ofdgo_compression_parallel.go +++ b/ofdgo_compression_parallel.go @@ -21,23 +21,25 @@ import ( "image" "io" "runtime" + "strconv" "sync" baseimage "github.com/xiaoqidun/ofdgo/internal/base/image" ) -// outputImageBatchLimit 限制并行批次持有的原始编码,超限图片单独处理 +// outputImageBatchLimit 限制并行任务持有的原始编码,超限图片单独处理 const outputImageBatchLimit = 16 << 20 -// outputImageWorkingSetLimit 限制并行批次的像素工作量估算,超限图片单独处理 -const outputImageWorkingSetLimit = 64 << 20 +// outputImageWorkingSetLimit 限制64位并行任务的像素工作量总估算,32位使用64MiB,超限图片单独处理 +const outputImageWorkingSetLimit = 256 << 20 // outputImageJob 保存独立图片任务,不访问编辑文档或调用进度回调 type outputImageJob struct { - key outputImageKey - data []byte - candidate []byte - err error + key outputImageKey + data []byte + workingSet int + candidate []byte + err error } // outputImageSource 保存已读取资源的摘要和格式,不保留原始编码 @@ -46,6 +48,21 @@ type outputImageSource struct { format string } +// outputImageContext 保留调用方取消检查并支持提前结束并行任务 +type outputImageContext struct { + context.Context + parent context.Context +} + +// Err 检查调用方及本次任务的取消状态 +// 返回: error 取消错误 +func (c outputImageContext) Err() error { + if err := c.Context.Err(); err != nil { + return err + } + return c.parent.Err() +} + // outputImageOptions 统一资源压缩及像素需求,受保护图片只进行无损编码 // 入参: key 资源路径 // 返回: CompressionOptions 实际策略, image.Point 像素需求 @@ -62,28 +79,52 @@ func (e *Editor) outputImageOptions(key string) (CompressionOptions, image.Point return options, demand } -// prepareOutputImages 有界并行优化独立图片,读取、缓存及进度仍在调用线程处理 +// prepareOutputImages 有界并行优化独立图片,完成后补入任务,读取、缓存及进度保持串行 // 入参: keys 资源路径, load 串行读取方法,空值表示跳过, results 候选缓存, sources 原始资源摘要,可为空 // 返回: error 读取、进度或取消错误 func (e *Editor) prepareOutputImages(keys []string, load func(string) ([]byte, error), results map[outputImageKey][]byte, sources map[string]outputImageSource) error { - workers := min(2, runtime.GOMAXPROCS(0)) + workers := min(4, runtime.GOMAXPROCS(0)) if workers < 2 || len(keys) < 2 { return nil } - batch := make([]outputImageJob, 0, workers) - size, workingSet := 0, 0 - run := func() error { - if err := e.runOutputImages(batch, results); err != nil { + limit := outputImageWorkingSetLimit + if strconv.IntSize == 32 { + limit = 64 << 20 + } + base, cancel := context.WithCancel(e.output.ctx) + ctx := outputImageContext{Context: base, parent: e.output.ctx} + var group sync.WaitGroup + defer func() { + cancel() + group.Wait() + }() + completed := make(chan outputImageJob, workers) + pending := make(map[outputImageKey]bool) + running, size, workingSet := 0, 0, 0 + finish := func() error { + job := <-completed + running-- + size -= len(job.data) + workingSet -= job.workingSet + delete(pending, job.key) + if err := e.output.ctx.Err(); err != nil { return err } - clear(batch) - batch, size, workingSet = batch[:0], 0, 0 + results[job.key] = nil + if job.err == nil && len(job.candidate) < len(job.data) { + results[job.key] = job.candidate + } return nil } for _, key := range keys { if err := e.output.ctx.Err(); err != nil { return err } + for len(completed) != 0 || running == workers { + if err := finish(); err != nil { + return err + } + } if err := editorProgress(e.OnWriteProgress).withContext(e.output.ctx).report("compress", e.output.completed, 0); err != nil { return err } @@ -105,35 +146,43 @@ func (e *Editor) prepareOutputImages(keys []string, load func(string) ([]byte, e if _, cached := results[cacheKey]; cached { continue } - duplicate := false - for _, job := range batch { - duplicate = duplicate || job.key == cacheKey - } - if duplicate { + if pending[cacheKey] { continue } cost := outputImageWorkingSet(e.output.ctx, data) - if len(batch) != 0 && (len(data) > outputImageBatchLimit-size || cost > outputImageWorkingSetLimit-workingSet) { - if err := run(); err != nil { + for running != 0 && (len(data) > outputImageBatchLimit-size || cost > limit-workingSet) { + if err := finish(); err != nil { return err } } - batch = append(batch, outputImageJob{key: cacheKey, data: data}) + job := outputImageJob{key: cacheKey, data: data, workingSet: cost} + if len(data) >= outputImageBatchLimit || cost >= limit { + if err := e.runOutputImages([]outputImageJob{job}, results); err != nil { + return err + } + continue + } + pending[cacheKey] = true + running++ size += len(data) workingSet += cost - if len(batch) == workers || size >= outputImageBatchLimit || workingSet >= outputImageWorkingSetLimit { - if err := run(); err != nil { - return err - } + group.Go(func() { + job.candidate, job.err = optimizeImageResource(ctx, job.data, job.key.options, job.key.size) + completed <- job + }) + } + for running != 0 { + if err := finish(); err != nil { + return err } } - return run() + return e.output.ctx.Err() } // outputImageWorkingSet 按原图尺寸估算解码和重采样缓冲,不以降采样目标代替原图 // 通常预留两份16位RGBA缓冲,TIFF另计分量、瓦片及蒙版存储;格式或尺寸不明时单独处理 // 入参: ctx 取消上下文, data 原始图像编码 -// 返回: int 编码及像素缓冲的估算字节数,上限为单批预算 +// 返回: int 编码及像素缓冲的估算字节数,上限为outputImageWorkingSetLimit func outputImageWorkingSet(ctx context.Context, data []byte) int { if len(data) >= outputImageWorkingSetLimit { return outputImageWorkingSetLimit