From eb20f66ffb37547f5ad7ed84b81fdcf7b257733f Mon Sep 17 00:00:00 2001 From: xiaoqidun Date: Mon, 19 Oct 2026 21:39:40 +0800 Subject: [PATCH] =?UTF-8?q?perf(=E6=80=A7=E8=83=BD=E4=BC=98=E5=8C=96):=20?= =?UTF-8?q?=E7=BB=9F=E4=B8=80=E8=B0=83=E5=BA=A6=E4=BC=98=E5=8C=96=E5=B9=B6?= =?UTF-8?q?=E8=A1=8C=E4=B8=8E=E5=86=85=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- assets/webui/ofdgo.js | 13 +-- assets/webui/ofdgo_host.js | 167 +++++++++++++++++++++++++++++++++---- assets/webui/ofdgo_wasm.js | 6 +- ofdgo_writer.go | 55 +++++++++++- 4 files changed, 212 insertions(+), 29 deletions(-) diff --git a/assets/webui/ofdgo.js b/assets/webui/ofdgo.js index 21afe66..7badf6b 100644 --- a/assets/webui/ofdgo.js +++ b/assets/webui/ofdgo.js @@ -1,6 +1,6 @@ import { SVG_NAMESPACE, CanvasEditor, canEditObject, editTextValue, richTextModel, fontFaceStyle, missingGlyphMessage, objectEditReason, selectedText, lineShape, selectionBounds, pagePoint } from "./ofdgo_edit.js"; import { FontManager, FontPicker } from "./ofdgo_font.js"; -import { nativeHost, createEngineWorker, documentBytes, saveFilePicker, directoryPicker } from "./ofdgo_host.js"; +import { nativeHost, createEngineWorker, engineLimits, documentBytes, saveFilePicker, directoryPicker } from "./ofdgo_host.js"; const MM_TO_PX = 96 / 25.4; const COMPACT_LAYOUT = window.matchMedia("(max-width: 900px)"); @@ -8546,10 +8546,11 @@ async function runBatch() { fontNames: [...batch.localFonts.keys()], fontMode: format.value === "ofd" ? Number(batchElements.FontMode.value) : 0 }; return { item, format, options }; }); - const capacity = nativeHost ? (await nativeHost.call("batch", {})).parallelism : Math.max(1, (navigator.hardwareConcurrency || 2) - 1); - const automatic = nativeHost ? capacity : Math.min(capacity, Math.max(1, Math.floor((navigator.deviceMemory || 4) / 2))); - const parallel = Math.min(jobs.length, nativeHost ? capacity : jobs.length, Number(batchElements.Parallel.value) || automatic); - const workerMemory = (navigator.deviceMemory || 4) * 1024 ** 3 / 4 / parallel; + const limits = nativeHost ? await nativeHost.call("batch", {}) : engineLimits(); + const capacity = limits.parallelism; + const automatic = nativeHost ? capacity : Math.min(capacity, Math.max(1, Math.floor(limits.memoryBytes / (512 << 20)))); + const parallel = Math.min(jobs.length, capacity, Number(batchElements.Parallel.value) || automatic); + const workerMemory = (limits.memoryBytes || 0) / parallel; let next = 0; const run = async initial => { let slot = initial; @@ -9922,7 +9923,7 @@ function queueNearbyPages(index, openSeq = state.openSeq) { for (const face of state.svgFonts.values()) bytes += svgFontBytes.get(face) || 0; const budget = nativeHost?.cacheBytes || PAGE_CACHE_BYTES; const cost = Math.max(1 << 20, bytes / Math.max(1, state.pageCache.size)); - const count = Math.max(0, Math.min(nativeHost?.parallelism || 1, state.pageLayout * 2, Math.floor((budget - bytes) / cost))); + const count = Math.max(0, Math.min(nativeHost?.parallelism || wasmWorker?.parallelism || 1, state.pageLayout * 2, Math.floor((budget - bytes) / cost))); for (let i = start; i < Math.min(state.doc?.pageCount || 1, start + state.pageLayout); i += 1) { renderFlowPage(i, { openSeq, priority: 1 }); } diff --git a/assets/webui/ofdgo_host.js b/assets/webui/ofdgo_host.js index fef57ef..455f77e 100644 --- a/assets/webui/ofdgo_host.js +++ b/assets/webui/ofdgo_host.js @@ -1,16 +1,138 @@ let engineModule = null; +let engineScheduler = null; + +export function engineLimits() { + engineScheduler ||= new BrowserScheduler(); + return {parallelism: engineScheduler.capacity, memoryBytes: engineScheduler.budget}; +} + +class BrowserScheduler { + constructor() { + this.budget = Math.max(256, Math.min(2048, (navigator.deviceMemory || 2) * 256)) * 1024 * 1024; + this.capacity = Math.max(1, Math.min((navigator.hardwareConcurrency || 2) - 1, Math.floor(this.budget / (128 << 20)))); + this.limit = Math.min(this.capacity, Math.max(1, Math.floor(this.budget / (512 << 20)))); + this.workers = new Set(); + this.queue = []; + } + + get memory() { + return [...this.workers].reduce((total, worker) => total + Math.max(worker.memory, worker.reservation), 0); + } + + active() { + return [...this.workers].filter(worker => worker.requests.size).length; + } + + submit(worker, data, transfer) { + if (worker.primary) { this.send({worker, data, transfer}); return; } + this.queue.push({worker, data, transfer, queued:performance.now()}); + this.drain(); + } + + send(task) { + const {worker, data, transfer} = task; + worker.requests.set(data.id, performance.now()); + try { worker.worker.postMessage(data, transfer); } + catch (error) { + worker.requests.delete(data.id); + worker.onmessage?.({data:{id:data.id, ok:false, error:error.message}}); + } + } + + complete(worker, data) { + if (data.memory) worker.memory = data.memory; + if (data.id == null || data.type) return; + const started = worker.requests.get(data.id); + worker.requests.delete(data.id); + if (this.memory > this.budget) this.limit = Math.max(1, this.limit - 1); + else if (started !== undefined && data.ok && performance.now() - started > 100 && this.memory + Math.max(128 << 20, worker.memory) < this.budget) this.limit = Math.min(this.capacity, this.limit + 1); + } + + drain() { + if (this.draining) return; + this.draining = true; + try { + if (this.queue.length) for (const worker of [...this.workers].sort((a,b) => b.memory-a.memory)) { + if (this.memory + (64 << 20) <= this.budget) break; + if (!worker.requests.size) worker.reclaim?.(); + } + this.queue.sort((a, b) => Number(a.worker.batch) - Number(b.worker.batch) + (a.queued - b.queued) / 2000); + for (let i = 0; i < this.queue.length;) { + const task = this.queue[i]; + if (task.worker.closed) { this.queue.splice(i, 1); continue; } + const active = this.active(); + if (task.worker.requests.size || active >= this.limit || active && this.memory + (64 << 20) > this.budget) { i++; continue; } + this.queue.splice(i, 1); + this.send(task); + } + } finally { this.draining = false; } + } + + cancel(worker, id) { + const index = this.queue.findIndex(task => task.worker === worker && task.data.id === id); + if (index < 0) return false; + this.queue.splice(index, 1); + worker.onmessage?.({data:{id, ok:false, canceled:true, error:"操作已取消"}}); + this.drain(); + return true; + } + + remove(worker) { + this.workers.delete(worker); + this.queue = this.queue.filter(task => task.worker !== worker); + worker.requests.clear(); + this.drain(); + } +} + +class BrowserWorker { + constructor(options) { + engineScheduler ||= new BrowserScheduler(); + this.primary = !!options.primary; + this.batch = !options.primary && !options.page; + this.requests = new Map(); + this.memory = 64 << 20; + this.reservation = 0; + this.closed = false; + this.worker = new Worker("./ofdgo_wasm.js"); + engineScheduler.workers.add(this); + this.worker.onmessage = event => { + if (this.closed || event.data.type === "module") return; + engineScheduler.complete(this, event.data); + if (event.data.type === "exit") this.terminate(); + try { this.onmessage?.(event); } + finally { engineScheduler.drain(); } + }; + this.worker.onerror = event => { this.terminate(); this.onerror?.(event); }; + this.worker.onmessageerror = event => { this.terminate(); this.onmessageerror?.(event); }; + this.worker.addEventListener("message", event => { + if (event.data.type !== "module") return; + event.stopImmediatePropagation(); + engineModule ||= WebAssembly.compileStreaming(fetch("./ofdgo.wasm")).catch(error => { engineModule = null; throw error; }); + engineModule.then(module => { if (!this.closed) this.worker.postMessage({ type:"module", module }); }, error => { if (!this.closed) this.worker.postMessage({ type:"module", error:error.message }); }); + }); + } + + postMessage(data, transfer) { + if (this.closed) throw new DOMException("引擎已关闭", "AbortError"); + if (data.type) { + if (data.type === "cancel" && engineScheduler.cancel(this, data.id)) return; + this.worker.postMessage(data, transfer); + } else engineScheduler.submit(this, data, transfer); + } + + terminate() { + if (this.closed) return; + this.closed = true; + this.worker.terminate(); + engineScheduler.remove(this); + } +} export function createEngineWorker(options = {}) { if (nativeHost) return nativeHost.worker(); if (options.pages) return new BrowserEngine(); - const worker = new Worker("./ofdgo_wasm.js"); - worker.addEventListener("message", event => { - if (event.data.type !== "module") return; - event.stopImmediatePropagation(); - engineModule ||= WebAssembly.compileStreaming(fetch("./ofdgo.wasm")).catch(error => { engineModule = null; throw error; }); - engineModule.then(module => worker.postMessage({ type: "module", module }), error => worker.postMessage({ type: "module", error: error.message })); - }); - return worker; + return new BrowserWorker(options); } export async function documentBytes(file) { @@ -21,10 +143,11 @@ export async function documentBytes(file) { class BrowserEngine { constructor() { - this.primary = createEngineWorker(); + this.primary = createEngineWorker({primary:true}); this.primary.onmessage = event => { const data = event.data; if (data.memory) this.memory = data.memory; + if (this.source) this.parallelism = engineScheduler.limit; if (data.id != null && !data.type) this.requests.delete(data.id); this.onmessage?.(event); }; @@ -36,7 +159,7 @@ class BrowserEngine { this.revision = 0; this.memory = 0; this.parallelism = 1; - this.budget = Math.max(128, Math.min(1024, (navigator.deviceMemory || 2) * 128)) * 1024 * 1024; + this.budget = engineScheduler.budget; this.closed = false; this.source = null; } @@ -45,7 +168,7 @@ class BrowserEngine { this.invalidatePages(); if (!source || this.closed) return; this.source = {...source, file: source.file instanceof Blob ? source.file : new Blob([source.file])}; - this.parallelism = Math.max(1, Math.min(2, (navigator.hardwareConcurrency || 2) - 1, Math.floor(this.budget / (256 * 1024 * 1024)))); + this.parallelism = engineScheduler.limit; } invalidatePages() { @@ -64,12 +187,15 @@ class BrowserEngine { } createSlot(source) { - const worker = createEngineWorker(); - const slot = {worker, pending: new Map(), busy: false, memory: Math.max(128 << 20, this.memory)}; + const worker = createEngineWorker({page:true}); + const slot = {worker, pending: new Map(), busy: false, memory: Math.max(128 << 20, this.memory), pages:new Set(), lastPage:0}; + worker.reservation = slot.memory; + worker.reclaim = () => { if (!slot.busy) this.closeSlot(slot); }; const ready = new Promise((resolve, reject) => { slot.rejectReady = reject; worker.onmessage = ({data}) => { if (data.memory) slot.memory = data.memory; + if (this.source) this.parallelism = engineScheduler.limit; if (data.type === "ready") { resolve(); return; } if (data.type === "exit") { this.closeSlot(slot, new Error("页面加载失败")); return; } if (data.type) return; @@ -119,13 +245,14 @@ class BrowserEngine { "ofdgoExportFormats", "ofdgoOutputFormats", "ofdgoParsePageRange", "ofdgoExportPage", "ofdgoExportDocument", "ofdgoExportAttachment", "ofdgoExportResource", "ofdgoExportMedia", "ofdgoExportVersion" ].includes(data.name)) this.invalidatePages(); - if (data.name !== "ofdgoRenderPage" || !this.source || ![...this.requests.values()].includes(null)) { + const preferred = data.name === "ofdgoRenderPage" ? [...this.slots].find(slot => !slot.busy && slot.pages.has(data.args[0])) : null; + if (data.name !== "ofdgoRenderPage" || !this.source || !preferred && ![...this.requests.values()].includes(null)) { if (data.name === "ofdgoRenderPage") this.requests.set(data.id, null); this.primary.postMessage(data, transfer); return; } - let slot = [...this.slots].find(slot => !slot.busy); - const memory = this.memory + [...this.slots].reduce((total, item) => total + item.memory, 0); + let slot = preferred || [...this.slots].filter(slot => !slot.busy).sort((a,b) => Math.abs(a.lastPage-data.args[0])-Math.abs(b.lastPage-data.args[0]))[0]; + const memory = engineScheduler.memory; if (!slot && this.slots.size < this.parallelism - 1 && memory + Math.max(128 << 20, this.memory) <= this.budget) slot = this.createSlot(this.source); if (!slot) { this.requests.set(data.id, null); @@ -138,6 +265,10 @@ class BrowserEngine { let fallback = false; slot.ready.then(() => this.callSlot(slot, data.name, data.args)).then(result => { if (revision !== this.revision) throw new DOMException("页面加载已取消", "AbortError"); + slot.lastPage = data.args[0]; + slot.pages.delete(slot.lastPage); + slot.pages.add(slot.lastPage); + if (slot.pages.size > 16) slot.pages.delete(slot.pages.values().next().value); if (!this.closed) this.onmessage?.({data: {...result, id: data.id}}); }).catch(error => { if (!this.closed && revision === this.revision && error.name !== "AbortError") { @@ -152,8 +283,10 @@ class BrowserEngine { }).finally(() => { if (!fallback) this.requests.delete(data.id); slot.busy = false; - const memory = this.memory + [...this.slots].reduce((total, item) => total + item.memory, 0); + slot.worker.reservation = 0; + const memory = engineScheduler.memory; if (memory > this.budget) { this.closeSlot(slot); this.parallelism = Math.max(1, this.slots.size + 1); } + engineScheduler.drain(); }); } diff --git a/assets/webui/ofdgo_wasm.js b/assets/webui/ofdgo_wasm.js index c33f133..c4229b9 100644 --- a/assets/webui/ofdgo_wasm.js +++ b/assets/webui/ofdgo_wasm.js @@ -128,7 +128,7 @@ async function handleBatchMessage({ id, name, args }) { }; payload = await globalThis.ofdgoConvertFile(read(file), file.size, options, write, (phase, completed, total, done) => { - self.postMessage({ id, type: "conversion", phase, completed, total }); + self.postMessage({ id, type: "conversion", phase, completed, total, memory: engineMemory.buffer.byteLength }); finish(done); }, (action, pageName, done) => boundary(action, pageName).then(() => finish(done), err => finish(done, err)), @@ -138,7 +138,7 @@ async function handleBatchMessage({ id, name, args }) { if (handle) output = await handle.createWritable(); payload = await globalThis.ofdgoPackFiles(entries.map(entry => ({ name: entry.name, size: entry.file.size })), index => read(entries[index].file), write, - (completed, total, done) => { self.postMessage({ id, type: "conversion", phase: "pack", completed, total }); finish(done); }, + (completed, total, done) => { self.postMessage({ id, type: "conversion", phase: "pack", completed, total, memory: engineMemory.buffer.byteLength }); finish(done); }, done => finish(done), compression); } signal.throwIfAborted(); @@ -337,7 +337,7 @@ async function loadWASM(module) { if (!go.exited) { const backends = JSON.parse(globalThis.ofdgoRenderBackends()); if (!backends.ok) throw new Error(backends.error); - self.postMessage({ type: "ready", backends: backends.data }); + self.postMessage({ type: "ready", backends: backends.data, memory: engineMemory.buffer.byteLength }); } } diff --git a/ofdgo_writer.go b/ofdgo_writer.go index bcb9ad6..fee6659 100644 --- a/ofdgo_writer.go +++ b/ofdgo_writer.go @@ -38,6 +38,9 @@ import ( // editorSnapshotArchiveLimit 限制编辑器常驻的快照压缩包字节数 const editorSnapshotArchiveLimit = 32 << 20 +// editorSnapshotChunkSize 指定快照压缩数据的分块大小 +const editorSnapshotChunkSize = 256 << 10 + // ErrSaveTarget 表示编辑器尚未绑定保存目标 var ErrSaveTarget = errors.New("save target is not set") @@ -76,6 +79,12 @@ type editorSnapshotArchive struct { retained *zip.Reader } +// editorSnapshotBuffer 分块保存压缩数据，写入完成后支持并发随机读取 +type editorSnapshotBuffer struct { + chunks [][]byte + size int64 +} + // editorProgress 编辑与保存阶段的进度回调，nil表示不回报进度 type editorProgress func(stage string, completed, total int) error @@ -152,7 +161,7 @@ func (e *Editor) copySnapshotArchive(ctx context.Context, source *zip.Reader) (* return archive, nil } } - var data bytes.Buffer + var data editorSnapshotBuffer var count int64 archive := zip.NewWriter(convertWriter{ctx: ctx, writer: &data, count: &count}) for _, file := range source.File { @@ -166,7 +175,7 @@ func (e *Editor) copySnapshotArchive(ctx context.Context, source *zip.Reader) (* if err := archive.Close(); err != nil { return nil, err } - reader, err := zip.NewReader(bytes.NewReader(data.Bytes()), int64(data.Len())) + reader, err := zip.NewReader(&data, data.size) if err != nil { return nil, err } @@ -174,13 +183,53 @@ func (e *Editor) copySnapshotArchive(ctx context.Context, source *zip.Reader) (* return nil, err } cached := &editorSnapshotArchive{files: slices.Clone(source.File), archive: weak.Make(reader)} - if data.Cap() <= editorSnapshotArchiveLimit { + if len(data.chunks) <= editorSnapshotArchiveLimit/editorSnapshotChunkSize { cached.retained = reader } e.snapshotArchive = cached return reader, nil } +// Write 追加快照数据，不扩容或复制已有块 +// 入参: data 压缩数据 +// 返回: int 写入字节数, error 写入错误 +func (b *editorSnapshotBuffer) Write(data []byte) (int, error) { + size := len(data) + for len(data) > 0 { + offset := int(b.size % editorSnapshotChunkSize) + if offset == 0 { + b.chunks = append(b.chunks, make([]byte, editorSnapshotChunkSize)) + } + n := copy(b.chunks[len(b.chunks)-1][offset:], data) + b.size += int64(n) + data = data[n:] + } + return size, nil +} + +// ReadAt 跨块读取快照，不复制或合并底层存储 +// 入参: data 目标缓冲, offset 数据偏移 +// 返回: int 读取字节数, error 偏移或结束错误 +func (b *editorSnapshotBuffer) ReadAt(data []byte, offset int64) (int, error) { + if offset < 0 { + return 0, fmt.Errorf("negative snapshot offset") + } + read := 0 + for len(data) > 0 { + if offset >= b.size { + return read, io.EOF + } + block := b.chunks[offset/editorSnapshotChunkSize] + start := int(offset % editorSnapshotChunkSize) + size := min(int64(len(block)-start), b.size-offset, int64(len(data))) + n := copy(data, block[start:start+int(size)]) + read += n + offset += int64(n) + data = data[n:] + } + return read, nil +} + // NewFileOutput 创建本地保存目标并记录当前文件状态，不创建或修改文件 // 同一目标应复用返回值，以便后续保存检测外部修改 // 替换前比较文件标识、大小和修改时间，不提供跨进程排他锁