perf(性能优化): 统一调度优化并行与内存
build.yaml / build (push) Successful in 47s

This commit is contained in:
xiaoqidun committed 2026-10-19 21:39:40 +08:00
1 parent 4f530eb5b8
commit eb20f66ffb
4 files changed
+212 -29

No files matched your search

+7 -6
View File
@@ -1,6 +1,6 @@
import { SVG_NAMESPACE, CanvasEditor, canEditObject, editTextValue, richTextModel, fontFaceStyle, missingGlyphMessage, objectEditReason, selectedText, lineShape, selectionBounds, pagePoint } from "./ofdgo_edit.js"; 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 { 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 MM_TO_PX = 96 / 25.4;
const COMPACT_LAYOUT = window.matchMedia("(max-width: 900px)"); 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 }; fontNames: [...batch.localFonts.keys()], fontMode: format.value === "ofd" ? Number(batchElements.FontMode.value) : 0 };
return { item, format, options }; return { item, format, options };
}); });
const capacity = nativeHost ? (await nativeHost.call("batch", {})).parallelism : Math.max(1, (navigator.hardwareConcurrency || 2) - 1); const limits = nativeHost ? await nativeHost.call("batch", {}) : engineLimits();
const automatic = nativeHost ? capacity : Math.min(capacity, Math.max(1, Math.floor((navigator.deviceMemory || 4) / 2))); const capacity = limits.parallelism;
const parallel = Math.min(jobs.length, nativeHost ? capacity : jobs.length, Number(batchElements.Parallel.value) || automatic); const automatic = nativeHost ? capacity : Math.min(capacity, Math.max(1, Math.floor(limits.memoryBytes / (512 << 20))));
const workerMemory = (navigator.deviceMemory || 4) * 1024 ** 3 / 4 / parallel; const parallel = Math.min(jobs.length, capacity, Number(batchElements.Parallel.value) || automatic);
const workerMemory = (limits.memoryBytes || 0) / parallel;
let next = 0; let next = 0;
const run = async initial => { const run = async initial => {
let slot = 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; for (const face of state.svgFonts.values()) bytes += svgFontBytes.get(face) || 0;
const budget = nativeHost?.cacheBytes || PAGE_CACHE_BYTES; const budget = nativeHost?.cacheBytes || PAGE_CACHE_BYTES;
const cost = Math.max(1 << 20, bytes / Math.max(1, state.pageCache.size)); 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) { for (let i = start; i < Math.min(state.doc?.pageCount || 1, start + state.pageLayout); i += 1) {
renderFlowPage(i, { openSeq, priority: 1 }); renderFlowPage(i, { openSeq, priority: 1 });
} }
+150 -17
View File
@@ -1,16 +1,138 @@
let engineModule = null; 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 = {}) { export function createEngineWorker(options = {}) {
if (nativeHost) return nativeHost.worker(); if (nativeHost) return nativeHost.worker();
if (options.pages) return new BrowserEngine(); if (options.pages) return new BrowserEngine();
const worker = new Worker("./ofdgo_wasm.js"); return new BrowserWorker(options);
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;
} }
export async function documentBytes(file) { export async function documentBytes(file) {
@@ -21,10 +143,11 @@ export async function documentBytes(file) {
class BrowserEngine { class BrowserEngine {
constructor() { constructor() {
this.primary = createEngineWorker(); this.primary = createEngineWorker({primary:true});
this.primary.onmessage = event => { this.primary.onmessage = event => {
const data = event.data; const data = event.data;
if (data.memory) this.memory = data.memory; 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); if (data.id != null && !data.type) this.requests.delete(data.id);
this.onmessage?.(event); this.onmessage?.(event);
}; };
@@ -36,7 +159,7 @@ class BrowserEngine {
this.revision = 0; this.revision = 0;
this.memory = 0; this.memory = 0;
this.parallelism = 1; 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.closed = false;
this.source = null; this.source = null;
} }
@@ -45,7 +168,7 @@ class BrowserEngine {
this.invalidatePages(); this.invalidatePages();
if (!source || this.closed) return; if (!source || this.closed) return;
this.source = {...source, file: source.file instanceof Blob ? source.file : new Blob([source.file])}; 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() { invalidatePages() {
@@ -64,12 +187,15 @@ class BrowserEngine {
} }
createSlot(source) { createSlot(source) {
const worker = createEngineWorker(); const worker = createEngineWorker({page:true});
const slot = {worker, pending: new Map(), busy: false, memory: Math.max(128 << 20, this.memory)}; 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) => { const ready = new Promise((resolve, reject) => {
slot.rejectReady = reject; slot.rejectReady = reject;
worker.onmessage = ({data}) => { worker.onmessage = ({data}) => {
if (data.memory) slot.memory = data.memory; if (data.memory) slot.memory = data.memory;
if (this.source) this.parallelism = engineScheduler.limit;
if (data.type === "ready") { resolve(); return; } if (data.type === "ready") { resolve(); return; }
if (data.type === "exit") { this.closeSlot(slot, new Error("页面加载失败")); return; } if (data.type === "exit") { this.closeSlot(slot, new Error("页面加载失败")); return; }
if (data.type) return; if (data.type) return;
@@ -119,13 +245,14 @@ class BrowserEngine {
"ofdgoExportFormats", "ofdgoOutputFormats", "ofdgoParsePageRange", "ofdgoExportPage", "ofdgoExportDocument", "ofdgoExportFormats", "ofdgoOutputFormats", "ofdgoParsePageRange", "ofdgoExportPage", "ofdgoExportDocument",
"ofdgoExportAttachment", "ofdgoExportResource", "ofdgoExportMedia", "ofdgoExportVersion" "ofdgoExportAttachment", "ofdgoExportResource", "ofdgoExportMedia", "ofdgoExportVersion"
].includes(data.name)) this.invalidatePages(); ].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); if (data.name === "ofdgoRenderPage") this.requests.set(data.id, null);
this.primary.postMessage(data, transfer); this.primary.postMessage(data, transfer);
return; return;
} }
let slot = [...this.slots].find(slot => !slot.busy); 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 = this.memory + [...this.slots].reduce((total, item) => total + item.memory, 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.slots.size < this.parallelism - 1 && memory + Math.max(128 << 20, this.memory) <= this.budget) slot = this.createSlot(this.source);
if (!slot) { if (!slot) {
this.requests.set(data.id, null); this.requests.set(data.id, null);
@@ -138,6 +265,10 @@ class BrowserEngine {
let fallback = false; let fallback = false;
slot.ready.then(() => this.callSlot(slot, data.name, data.args)).then(result => { slot.ready.then(() => this.callSlot(slot, data.name, data.args)).then(result => {
if (revision !== this.revision) throw new DOMException("页面加载已取消", "AbortError"); 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}}); if (!this.closed) this.onmessage?.({data: {...result, id: data.id}});
}).catch(error => { }).catch(error => {
if (!this.closed && revision === this.revision && error.name !== "AbortError") { if (!this.closed && revision === this.revision && error.name !== "AbortError") {
@@ -152,8 +283,10 @@ class BrowserEngine {
}).finally(() => { }).finally(() => {
if (!fallback) this.requests.delete(data.id); if (!fallback) this.requests.delete(data.id);
slot.busy = false; 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); } if (memory > this.budget) { this.closeSlot(slot); this.parallelism = Math.max(1, this.slots.size + 1); }
engineScheduler.drain();
}); });
} }
+3 -3
View File
@@ -128,7 +128,7 @@ async function handleBatchMessage({ id, name, args }) {
}; };
payload = await globalThis.ofdgoConvertFile(read(file), file.size, options, write, payload = await globalThis.ofdgoConvertFile(read(file), file.size, options, write,
(phase, completed, total, done) => { (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); finish(done);
}, },
(action, pageName, done) => boundary(action, pageName).then(() => finish(done), err => finish(done, err)), (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(); if (handle) output = await handle.createWritable();
payload = await globalThis.ofdgoPackFiles(entries.map(entry => ({ name: entry.name, size: entry.file.size })), payload = await globalThis.ofdgoPackFiles(entries.map(entry => ({ name: entry.name, size: entry.file.size })),
index => read(entries[index].file), write, 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); done => finish(done), compression);
} }
signal.throwIfAborted(); signal.throwIfAborted();
@@ -337,7 +337,7 @@ async function loadWASM(module) {
if (!go.exited) { if (!go.exited) {
const backends = JSON.parse(globalThis.ofdgoRenderBackends()); const backends = JSON.parse(globalThis.ofdgoRenderBackends());
if (!backends.ok) throw new Error(backends.error); 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 });
} }
} }
+52 -3
View File
@@ -38,6 +38,9 @@ import (
// editorSnapshotArchiveLimit 限制编辑器常驻的快照压缩包字节数 // editorSnapshotArchiveLimit 限制编辑器常驻的快照压缩包字节数
const editorSnapshotArchiveLimit = 32 << 20 const editorSnapshotArchiveLimit = 32 << 20
// editorSnapshotChunkSize 指定快照压缩数据的分块大小
const editorSnapshotChunkSize = 256 << 10
// ErrSaveTarget 表示编辑器尚未绑定保存目标 // ErrSaveTarget 表示编辑器尚未绑定保存目标
var ErrSaveTarget = errors.New("save target is not set") var ErrSaveTarget = errors.New("save target is not set")
@@ -76,6 +79,12 @@ type editorSnapshotArchive struct {
retained *zip.Reader retained *zip.Reader
} }
// editorSnapshotBuffer 分块保存压缩数据,写入完成后支持并发随机读取
type editorSnapshotBuffer struct {
chunks [][]byte
size int64
}
// editorProgress 编辑与保存阶段的进度回调,nil表示不回报进度 // editorProgress 编辑与保存阶段的进度回调,nil表示不回报进度
type editorProgress func(stage string, completed, total int) error 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 return archive, nil
} }
} }
var data bytes.Buffer var data editorSnapshotBuffer
var count int64 var count int64
archive := zip.NewWriter(convertWriter{ctx: ctx, writer: &data, count: &count}) archive := zip.NewWriter(convertWriter{ctx: ctx, writer: &data, count: &count})
for _, file := range source.File { 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 { if err := archive.Close(); err != nil {
return nil, err return nil, err
} }
reader, err := zip.NewReader(bytes.NewReader(data.Bytes()), int64(data.Len())) reader, err := zip.NewReader(&data, data.size)
if err != nil { if err != nil {
return nil, err return nil, err
} }
@@ -174,13 +183,53 @@ func (e *Editor) copySnapshotArchive(ctx context.Context, source *zip.Reader) (*
return nil, err return nil, err
} }
cached := &editorSnapshotArchive{files: slices.Clone(source.File), archive: weak.Make(reader)} cached := &editorSnapshotArchive{files: slices.Clone(source.File), archive: weak.Make(reader)}
if data.Cap() <= editorSnapshotArchiveLimit { if len(data.chunks) <= editorSnapshotArchiveLimit/editorSnapshotChunkSize {
cached.retained = reader cached.retained = reader
} }
e.snapshotArchive = cached e.snapshotArchive = cached
return reader, nil 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 创建本地保存目标并记录当前文件状态,不创建或修改文件 // NewFileOutput 创建本地保存目标并记录当前文件状态,不创建或修改文件
// 同一目标应复用返回值,以便后续保存检测外部修改 // 同一目标应复用返回值,以便后续保存检测外部修改
// 替换前比较文件标识、大小和修改时间,不提供跨进程排他锁 // 替换前比较文件标识、大小和修改时间,不提供跨进程排他锁