diff --git a/assets/webui/index.html b/assets/webui/index.html index a653812..cddd09c 100644 --- a/assets/webui/index.html +++ b/assets/webui/index.html @@ -199,6 +199,7 @@
+
diff --git a/assets/webui/ofdgo.css b/assets/webui/ofdgo.css index d3888e3..216e3a6 100644 --- a/assets/webui/ofdgo.css +++ b/assets/webui/ofdgo.css @@ -803,7 +803,7 @@ button:not(:disabled):hover { .convert-options { flex: none; display: grid; - grid-template-columns: repeat(4, minmax(0, 1fr)); + grid-template-columns: repeat(5, minmax(0, 1fr)); grid-column: 1 / -1; grid-row: 1; width: 100%; @@ -812,11 +812,11 @@ button:not(:disabled):hover { } .convert-options:has(#convertDPIRow[hidden]):has(#convertFontRow[hidden]) { - grid-template-columns: repeat(3, minmax(0, 1fr)); + grid-template-columns: repeat(4, minmax(0, 1fr)); } .convert-options:has(#convertDPIRow:not([hidden])):has(#convertFontRow:not([hidden])) { - grid-template-columns: repeat(5, minmax(0, 1fr)); + grid-template-columns: repeat(6, minmax(0, 1fr)); } .convert-options .form-row { @@ -1059,12 +1059,13 @@ button:not(:disabled):hover { margin: 0; } .convert-options:has(#convertDPIRow[hidden]):has(#convertFontRow[hidden]) { - grid-template-columns: repeat(3, minmax(0, 1fr)); + grid-template-columns: repeat(2, minmax(0, 1fr)); } .convert-options:has(#convertDPIRow:not([hidden])):has(#convertFontRow:not([hidden])) { grid-template-columns: repeat(2, minmax(0, 1fr)); } - .convert-options:has(#convertDPIRow:not([hidden])):has(#convertFontRow:not([hidden])) .form-row:last-child { + .convert-options:has(#convertDPIRow[hidden]):has(#convertFontRow:not([hidden])) .form-row:last-child, + .convert-options:has(#convertDPIRow:not([hidden])):has(#convertFontRow[hidden]) .form-row:last-child { grid-column: 1 / -1; } .convert-options .form-row { diff --git a/assets/webui/ofdgo.js b/assets/webui/ofdgo.js index 2532f38..b74e107 100644 --- a/assets/webui/ofdgo.js +++ b/assets/webui/ofdgo.js @@ -20,8 +20,8 @@ const STATUS = { }; const wasmRequests = new Map(); const metaContents = new WeakMap(); -const batch = { items: [], formats: [], running: false, canceled: false, worker: null, requests: new Map(), sequence: 0, activeID: 0, progressTime: 0, progress: null }; -const batchElements = Object.fromEntries(["Button", "Panel", "Form", "Input", "Add", "Clear", "Count", "Format", "Compression", "FontRow", "FontMode", "Destination", "DPIRow", "DPI", "List", "Empty", "Progress", "Status", "Close", "Cancel", "Start"].map(name => [name, document.querySelector(`#convert${name}`)])); +const batch = { items: [], formats: [], running: false, canceled: false, workers: new Set(), requests: new Map(), sequence: 0, progress: null, jobs: [], credentials: Promise.resolve() }; +const batchElements = Object.fromEntries(["Button", "Panel", "Form", "Input", "Add", "Clear", "Count", "Format", "Compression", "FontRow", "FontMode", "Destination", "DPIRow", "DPI", "Parallel", "List", "Empty", "Progress", "Status", "Close", "Cancel", "Start"].map(name => [name, document.querySelector(`#convert${name}`)])); const fontBinding = { font: null, choices: [], sequence: 0 }; const fontBindingElements = Object.fromEntries(["Panel", "Form", "Title", "Name", "SourceRow", "Source", "File", "Hint", "Status", "Cancel", "Submit"].map(name => [name, document.querySelector(`#fontBinding${name}`)])); const resourceElements = Object.fromEntries(["ManageButton", "Panel", "Form", "Kind", "List", "Empty", "Reference", "ReferenceGroup", "Actions", "Add", "Preview", "Export", "Rename", "Replace", "Delete", "Locate", "Close", "Binding", "BindingActions", "Page", "Object", "Pick", "Bind", "Unbind", "Viewer", "Status", "File"].map(name => [name, document.querySelector(`#resource${name}`)])); @@ -8128,7 +8128,7 @@ function renderBatchList() { } function updateBatchControls() { - for (const name of ["Add", "Clear", "Format", "Compression", "FontMode", "Destination", "DPI"]) batchElements[name].disabled = batch.running; + for (const name of ["Add", "Clear", "Format", "Compression", "FontMode", "Destination", "DPI", "Parallel"]) batchElements[name].disabled = batch.running; batchElements.FontRow.hidden = !(batch.items.length ? batch.items : [null]).some(item => (item?.format || batchElements.Format.value) === "ofd"); batchElements.Compression.disabled ||= (batch.items.length ? batch.items : [null]).every(item => (item?.format || batchElements.Format.value) === "txt"); batchElements.Clear.disabled ||= !batch.items.length; @@ -8159,16 +8159,31 @@ function batchStatus(item, status, text) { function batchProgress(item, position, count, progress) { if (batch.canceled) return; const now = performance.now(); - const previous = batch.progress; + const previous = item ? item.progress : batch.progress; const sameStage = previous?.item === item && previous?.progress.phase === progress.phase; - if (sameStage && now - batch.progressTime < 80 && progress.completed !== progress.total) return; - batch.progressTime = now; - batch.progress = { item, position, count, progress, started: sameStage ? previous.started : now }; + if (sameStage && now - previous.updated < 80 && progress.completed !== progress.total) return; + const current = { item, position, count, progress, started: sameStage ? previous.started : now, updated: now }; + if (item) item.progress = current; + else batch.progress = current; updateBatchProgress(); } function updateBatchProgress() { - if (!batch.progress) return; + if (batch.canceled) return; + for (const item of batch.jobs) { + if (item.status !== "running" || !item.progress) continue; + const progress = item.progress.progress; + const stage = progress.phase?.includes("compress") ? "压缩" : progress.phase?.startsWith("write") ? "写入" : { open: "读取", pages: "解析", convert: "转换", prepare: "准备", fonts: "字体", export: "导出" }[progress.phase] || "处理"; + batchStatus(item, "running", `${stage}${progress.total > 0 ? ` ${progress.completed}/${progress.total}` : ""}`); + } + if (!batch.progress) { + const completed = batch.jobs.filter(item => ["done", "ready", "unchanged", "failed", "canceled"].includes(item.status)).length; + const running = batch.jobs.filter(item => item.status === "running").length; + batchElements.Status.textContent = `完成 ${completed}/${batch.jobs.length} · 运行 ${running}`; + batchElements.Progress.max = batch.jobs.length || 1; + batchElements.Progress.value = completed; + return; + } const { item, position, count, progress, started } = batch.progress; const compressing = progress.phase === "compress" || progress.phase === "write.compress"; const fontMode = (item?.format || batchElements.Format.value) === "ofd" ? batchElements.FontMode.value : "0"; @@ -8182,29 +8197,29 @@ function updateBatchProgress() { else batchElements.Progress.removeAttribute("value"); } -function batchCall(name, args, onProgress) { +function batchCall(slot, name, args, onProgress) { return new Promise((resolve, reject) => { - if (batch.workerError) { reject(batch.workerError); return; } + if (slot.error) { reject(slot.error); return; } + if (batch.canceled) { reject(new DOMException("已取消转换", "AbortError")); return; } const id = ++batch.sequence; - batch.activeID = id; - batch.requests.set(id, { resolve, reject, onProgress, destination: name === "ofdgoConvertFile" ? args[3] : null }); - try { batch.worker.postMessage({ id, name, args }); } - catch (err) { batch.requests.delete(id); batch.activeID = 0; reject(err); } + batch.requests.set(id, { slot, resolve, reject, onProgress, destination: name === "ofdgoConvertFile" ? args[3] : null }); + try { slot.worker.postMessage({ id, name, args }); } + catch (err) { batch.requests.delete(id); reject(err); } }); } function startBatchWorker() { - batch.workerError = null; return new Promise((resolve, reject) => { const worker = createEngineWorker(); - batch.worker = worker; + const slot = { worker, error: null }; + batch.workers.add(slot); const fail = async error => { - batch.workerError = error; + slot.error = error; worker.terminate(); reject(error); - const requests = [...batch.requests.values()]; - batch.requests.clear(); - for (const request of requests) { + for (const [id, request] of batch.requests) { + if (request.slot !== slot) continue; + batch.requests.delete(id); if (request.destination && request.outputName) { try { await request.destination.removeEntry(request.outputName, { recursive: true }); } catch (err) { error = new Error(`${error.message}；清理失败：${err.message}`); } @@ -8215,7 +8230,7 @@ function startBatchWorker() { worker.onerror = event => fail(new Error(event.message || "转换引擎异常")); worker.onmessageerror = () => fail(new Error("转换数据传递失败")); worker.onmessage = async ({ data }) => { - if (data.type === "ready") { resolve(); return; } + if (data.type === "ready") { resolve(slot); return; } if (data.type === "exit") { fail(new Error(data.error)); return; } if (data.type === "progress") { if (!batch.canceled) batchElements.Status.textContent = { download: "正在下载", compile: "正在编译", start: "正在启动" }[data.phase] || "正在准备"; @@ -8237,8 +8252,8 @@ function startBatchWorker() { } if (data.type === "output") { request.outputName = data.name; return; } if (data.type === "conversion") { request.onProgress?.(data); return; } + if (data.memory) slot.memory = data.memory; batch.requests.delete(data.id); - if (batch.activeID === data.id) batch.activeID = 0; if (data.ok) request.resolve(data.data); else request.reject(Object.assign(new Error(data.error), { code: data.code, name: data.canceled ? "AbortError" : "Error" })); }; @@ -8250,7 +8265,8 @@ function cancelBatch() { batchElements.Cancel.disabled = true; batch.progress = null; batchElements.Status.textContent = "正在取消"; - if (batch.activeID) batch.worker.postMessage({ type: "cancel", id: batch.activeID }); + if (batch.unlocking) finishCredentials(null); + for (const [id, request] of batch.requests) request.slot.worker.postMessage({ type: "cancel", id }); } function warnBatch(event) { @@ -8286,6 +8302,8 @@ async function runBatch() { batch.running = true; batch.canceled = false; batch.progress = null; + batch.jobs = jobs; + batch.credentials = Promise.resolve(); const progressTimer = window.setInterval(updateBatchProgress, 1000); renderBatchList(); window.addEventListener("beforeunload", warnBatch); @@ -8294,7 +8312,7 @@ async function runBatch() { let storage = null; let stagingName = ""; let packed = false; - const entries = []; + const results = new Array(jobs.length); try { if (archive) { zipFile = await saveFilePicker({ suggestedName: "转换结果.zip", types: [{ description: "ZIP", accept: { "application/zip": [".zip"] } }] }); @@ -8313,59 +8331,93 @@ async function runBatch() { batch.localFonts = fontManager.localEntries(); batch.fontData = new Map(); batchElements.Status.textContent = "正在准备"; - await startBatchWorker(); + const first = await startBatchWorker(); const fonts = await fontManager.files(fontManager.records()); const names = new Set(); - for (const [position, item] of jobs.entries()) { - if (batch.canceled) break; + const plans = jobs.map(item => { item.details = []; const format = batch.formats.find(format => format.value === (item.format || defaultFormat.value)); const base = item.file.name.replace(/\.[^.]+$/, "").replace(/[<>:"/\\|?*\x00-\x1f]/g, "_").replace(/[. ]+$/g, "") || "document"; let unique = base; const outputName = () => (format.paged ? unique : `${unique}.${format.extension}`).toLowerCase(); - for (let suffix = 2; archive && names.has(outputName()); suffix++) unique = `${base} (${suffix})`; + for (let suffix = 2; names.has(outputName()); suffix++) unique = `${base} (${suffix})`; names.add(outputName()); const options = { format: format.value, pages: pageRange(item), dpi, backend, archive, compression, base: unique, credentials: null, fontNames: [...batch.localFonts.keys()], fontMode: format.value === "ofd" ? Number(batchElements.FontMode.value) : 0 }; - try { - for (;;) { - try { - const result = await batchCall("ofdgoConvertFile", [item.file, options, fonts, destination], progress => batchProgress(item, position, jobs.length, progress)); - for (const file of result.files) entries.push(file); - item.details = (result.pdf?.Warnings || []).map(warning => `${warning.Page ? `第${warning.Page}页：` : ""}${warning.Message}`); - for (const font of result.fonts || []) item.details.push(`${font.name}：${font.changed ? options.fontMode === 1 ? "已嵌入" : "已外置" : `未处理 · ${fontBindingError(font.reason)}`}`); - if (result.unchanged && !result.bytes) { - batchStatus(item, "unchanged", "原样"); - item.details = ["无需转换，源文件未改动"]; - } else batchStatus(item, archive ? "ready" : "done", archive ? "待存" : "完成"); - break; - } catch (err) { - if (batch.canceled || !["credentialsRequired", "invalidCredentials", "pdfPassword"].includes(err.code)) throw err; - const code = err.code === "pdfPassword" ? options.credentials ? "invalidCredentials" : "credentialsRequired" : err.code; - options.credentials?.password?.fill(0); - options.credentials?.key?.fill(0); - options.credentials?.keyPassword?.fill(0); - options.credentials = await requestCredentials(code, state.openSeq, err.code === "pdfPassword"); - if (!options.credentials) throw new DOMException("已取消解锁", "AbortError"); - const preservesEncryption = err.code === "pdfPassword" && format.value === "pdf" && compression.mode !== 0 && !options.pages; - if (!preservesEncryption && !window.confirm("转换结果将不保留原加密，是否继续？")) throw new DOMException("已取消转换", "AbortError"); + 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; + let next = 0; + const run = async initial => { + let slot = initial; + while (!batch.canceled && next < plans.length) { + const position = next++; + const { item, format, options } = plans[position]; + batchStatus(item, "running", "准备"); + updateBatchProgress(); + try { + if (!slot || slot.error) slot = await startBatchWorker(); + for (;;) { + try { + const result = await batchCall(slot, "ofdgoConvertFile", [item.file, options, fonts, destination], progress => batchProgress(item, position, jobs.length, progress)); + results[position] = result.files; + item.details = (result.pdf?.Warnings || []).map(warning => `${warning.Page ? `第${warning.Page}页：` : ""}${warning.Message}`); + for (const font of result.fonts || []) item.details.push(`${font.name}：${font.changed ? options.fontMode === 1 ? "已嵌入" : "已外置" : `未处理 · ${fontBindingError(font.reason)}`}`); + if (result.unchanged && !result.bytes) { + batchStatus(item, "unchanged", "原样"); + item.details = ["无需转换，源文件未改动"]; + } else batchStatus(item, archive ? "ready" : "done", archive ? "待存" : "完成"); + break; + } catch (err) { + if (batch.canceled || !["credentialsRequired", "invalidCredentials", "pdfPassword"].includes(err.code)) throw err; + const code = err.code === "pdfPassword" ? options.credentials ? "invalidCredentials" : "credentialsRequired" : err.code; + options.credentials?.password?.fill(0); + options.credentials?.key?.fill(0); + options.credentials?.keyPassword?.fill(0); + const unlock = batch.credentials.then(async () => { + if (batch.canceled) throw new DOMException("已取消转换", "AbortError"); + batch.unlocking = true; + try { return await requestCredentials(code, state.openSeq, err.code === "pdfPassword"); } + finally { batch.unlocking = false; } + }); + batch.credentials = unlock.then(() => {}, () => {}); + options.credentials = await unlock; + if (!options.credentials) throw new DOMException("已取消解锁", "AbortError"); + const preservesEncryption = err.code === "pdfPassword" && format.value === "pdf" && compression.mode !== 0 && !options.pages; + if (!preservesEncryption && !window.confirm("转换结果将不保留原加密，是否继续？")) throw new DOMException("已取消转换", "AbortError"); + } + } + } catch (err) { + batchStatus(item, err.name === "AbortError" ? "canceled" : "failed", err.name === "AbortError" ? "取消" : "失败"); + item.details = [err.message]; + } finally { + item.progress = null; + options.credentials?.password?.fill(0); + options.credentials?.key?.fill(0); + options.credentials?.keyPassword?.fill(0); + if (!nativeHost && slot && (next >= plans.length || slot.memory >= workerMemory)) { + slot.worker.terminate(); + batch.workers.delete(slot); + slot = null; } } - } catch (err) { - batchStatus(item, err.name === "AbortError" ? "canceled" : "failed", err.name === "AbortError" ? "取消" : "失败"); - item.details = [err.message]; - if (batch.workerError) throw err; - } finally { - batch.progress = null; - options.credentials?.password?.fill(0); - options.credentials?.key?.fill(0); - options.credentials?.keyPassword?.fill(0); + updateBatchProgress(); } - batchElements.Progress.max = jobs.length; - batchElements.Progress.value = position + 1; - } + }; + const running = Array.from({ length: parallel }, (_, index) => (async () => { + if (index) await new Promise(resolve => window.setTimeout(resolve, index * 200)); + await run(index === 0 ? first : null); + })()); + const settled = await Promise.allSettled(running); + const failure = settled.find(result => result.status === "rejected"); + if (failure) throw failure.reason; + const entries = results.flat().filter(Boolean); if (archive && entries.length && !batch.canceled) { - const result = await batchCall("ofdgoPackFiles", [entries, zipFile, compression], progress => batchProgress(null, 0, 1, progress)); + const slot = [...batch.workers].find(slot => !slot.error) || await startBatchWorker(); + const result = await batchCall(slot, "ofdgoPackFiles", [entries, zipFile, compression], progress => batchProgress(null, 0, 1, progress)); if (result.blob) downloadBytes(result.blob, "application/zip", "转换结果.zip"); packed = true; for (const item of jobs) if (item.status === "ready") batchStatus(item, "done", "完成"); @@ -8381,10 +8433,10 @@ async function runBatch() { batch.progress = null; if (!batchElements.Progress.hasAttribute("value")) batchElements.Progress.value = 0; if (archive && !packed) for (const item of jobs) if (item.status === "ready") { item.status = "pending"; item.text = "待转"; } - batch.worker?.terminate(); - batch.worker = null; + for (const slot of batch.workers) slot.worker.terminate(); + batch.workers.clear(); batch.requests.clear(); - batch.activeID = 0; + if (nativeHost) await nativeHost.release(results); batch.fontData?.clear(); batch.localFonts?.clear(); if (storage && stagingName) { diff --git a/assets/webui/ofdgo_host.js b/assets/webui/ofdgo_host.js index bcd8932..b54a7de 100644 --- a/assets/webui/ofdgo_host.js +++ b/assets/webui/ofdgo_host.js @@ -1,5 +1,15 @@ +let engineModule = null; + export function createEngineWorker() { - return nativeHost ? nativeHost.worker() : new Worker("./ofdgo_wasm.js"); + if (nativeHost) return nativeHost.worker(); + 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; } export async function documentBytes(file) { @@ -124,7 +134,11 @@ class NativeHost { let unsubscribe; let pending = Promise.resolve(); const requests = new Map(); - const emit = data => { if (!closed) worker.onmessage?.({data}); }; + const completions = new Map(); + const emit = data => { + if (data.type === "result") { completions.get(data.id)?.(data.result); return; } + if (!closed) worker.onmessage?.({data}); + }; const ready = host.call("engine.open", {}).then(({engine}) => { if (closed) return engine; unsubscribe = host.bridge.on(`ofdgo.engine.${engine}`, emit); @@ -147,7 +161,21 @@ class NativeHost { const engine = await ready; if (closed) return; controller.signal.throwIfAborted(); - const result = await host.call("engine.call", {engine, id:data.id, name:data.name, args:data.args}, controller.signal); + const args = {engine, id:data.id, name:data.name, args:data.args}; + let result; + if (data.name === "ofdgoConvertFile" || data.name === "ofdgoPackFiles") { + const completed = new Promise(resolve => completions.set(data.id, resolve)); + const cancel = () => host.call("engine.cancel", {engine, id:data.id}).catch(() => {}); + controller.signal.addEventListener("abort", cancel, {once:true}); + try { + await host.call("engine.start", args, controller.signal); + result = await completed; + await host.call("engine.finish", {engine, id:data.id}); + } finally { + controller.signal.removeEventListener("abort", cancel); + completions.delete(data.id); + } + } else result = await host.call("engine.call", args, controller.signal); if (closed) { await host.release(result); return; } let decoded; try { @@ -163,9 +191,12 @@ class NativeHost { terminate() { if (closed) return; closed = true; - unsubscribe?.(); for (const controller of requests.values()) controller.abort(); - ready.then(engine => host.call("engine.close", {engine})).catch(() => {}); + ready.then(engine => host.call("engine.close", {engine})).catch(() => {}).finally(() => { + unsubscribe?.(); + for (const resolve of completions.values()) resolve({ok:false, canceled:true, error:"会话已关闭"}); + completions.clear(); + }); }, }; ready.catch(() => {}); diff --git a/assets/webui/ofdgo_wasm.js b/assets/webui/ofdgo_wasm.js index e3fbcf5..7be2113 100644 --- a/assets/webui/ofdgo_wasm.js +++ b/assets/webui/ofdgo_wasm.js @@ -4,6 +4,7 @@ let pending = Promise.resolve(); const operations = new Map(); const fontRequests = new Map(); let fontSequence = 0; +let engineMemory = null; function requestFont(id, signal, name, done) { if (signal.aborted) { done(null, "", true); return; } @@ -20,6 +21,11 @@ function requestFont(id, signal, name, done) { self.postMessage({ id, type: "font", request, name }); } self.onmessage = ({ data }) => { + if (data.type === "module") { + if (data.error) exitWASM(new Error(data.error)); + else loadWASM(data.module).catch(exitWASM); + return; + } if (data.type === "font") { fontRequests.get(data.request)?.(data); fontRequests.delete(data.request); @@ -84,12 +90,15 @@ async function handleBatchMessage({ id, name, args }) { if (!format) throw new Error("不支持此格式"); root = destination; const base = options.base.replace(/[<>:"/\\|?*\x00-\x1f]/g, "_").replace(/[. ]+$/g, "") || "document"; - const target = root ? await availableOutput(root, base, format) : { stem: base, name: format.paged ? base : `${base}.${format.extension}` }; - if (root && format.paged) { - directory = await root.getDirectoryHandle(target.name, { create: true }); + const target = root ? await navigator.locks.request("ofdgo-convert-output", { signal }, async () => { + signal.throwIfAborted(); + const target = await availableOutput(root, base, format); + if (format.paged) directory = await root.getDirectoryHandle(target.name, { create: true }); + else { await root.getFileHandle(target.name, { create: true }); directory = root; } created = target.name; self.postMessage({ id, type: "output", name: created }); - } else directory = root; + return target; + }) : { stem: base, name: format.paged ? base : `${base}.${format.extension}` }; const boundary = async (action, pageName) => { signal.throwIfAborted(); if (action === "open") { @@ -144,7 +153,7 @@ async function handleBatchMessage({ id, name, args }) { } result.data.size = size; operations.delete(id); - self.postMessage({ id, ...result }); + self.postMessage({ id, ...result, memory: engineMemory.buffer.byteLength }); } catch (err) { await output?.abort().catch(() => {}); let message = signal.aborted ? "转换已取消" : err.message; @@ -152,7 +161,7 @@ async function handleBatchMessage({ id, name, args }) { try { await root.removeEntry(created, { recursive: true }); } catch (cleanupError) { message += `；清理失败：${cleanupError.message}`; } } - self.postMessage({ id, ok: false, error: message, code: err.code, canceled: signal.aborted }); + self.postMessage({ id, ok: false, error: message, code: err.code, canceled: signal.aborted, memory: engineMemory.buffer.byteLength }); } finally { credentials?.password?.fill(0); credentials?.key?.fill(0); @@ -301,12 +310,10 @@ function exitWASM(err) { self.close(); } -async function loadWASM() { +async function loadWASM(module) { const go = new Go(); - self.postMessage({ type: "progress", phase: "download", text: "正在下载引擎", percent: 16 }); - const response = await fetch("./ofdgo.wasm"); - self.postMessage({ type: "progress", phase: "compile", text: "正在编译引擎", percent: 35 }); - const { instance } = await WebAssembly.instantiateStreaming(response, go.importObject); + const instance = await WebAssembly.instantiate(module, go.importObject); + engineMemory = instance.exports.mem; self.postMessage({ type: "progress", phase: "start", text: "正在启动引擎", percent: 58 }); go.run(instance).then(exitWASM, exitWASM); if (!go.exited) { @@ -316,4 +323,4 @@ async function loadWASM() { } } -loadWASM().catch(exitWASM); +self.postMessage({ type: "module" }); diff --git a/internal/webui/webui.go b/internal/webui/webui.go index dfb8c2c..c5ba69f 100644 --- a/internal/webui/webui.go +++ b/internal/webui/webui.go @@ -424,6 +424,7 @@ type businessEngine struct { copiedStyle *ofdgo.GraphicObject ctx context.Context resolveSource func(context.Context, any) (io.ReaderAt, int64, error) + conversionPool *ofdgo.ConversionPool } // Open 打开浏览器内存中的OFD文档 @@ -1391,6 +1392,7 @@ func (e *businessEngine) convertFile(args []platformValue) (any, error) { } options := ofdgo.ConvertOptions{ Format: settings.Get("format").String(), PageRange: settings.Get("pages").String(), + Pool: e.conversionPool, SkipUnchanged: true, Backends: &backends, RendererOptions: []ofdgo.RendererOption{ofdgo.WithDPI(settings.Get("dpi").Float())}, diff --git a/internal/webui/webui_native.go b/internal/webui/webui_native.go index 38c45da..4f29ecf 100644 --- a/internal/webui/webui_native.go +++ b/internal/webui/webui_native.go @@ -67,6 +67,8 @@ type NativeOutputTarget struct { // hooks应在首次Call前配置，Resolve返回的资源需保持有效直到相应会话关闭 type NativeEngine struct { businessEngine + // ConversionPool 协调不同会话的批量转换资源 + ConversionPool *ofdgo.ConversionPool // Resolve 按授权标识提供资源读取源 Resolve NativeResolver // LoadFont 按名称读取系统字体数据 @@ -134,7 +136,11 @@ func NewNativeEngine(resolve NativeResolver) *NativeEngine { return &NativeEngin // 批量输出到非nil目标时必须同时提供ReserveOutput和CleanupOutput，避免误清理既有文件 // 取消为协作式检查，不能抢占宿主阻塞的ReadAt、字体读取或加密库内部计算 func (e *NativeEngine) Call(ctx context.Context, name string, args []any, emit func(any)) (result map[string]any) { - e.once.Do(func() { e.gate = make(chan struct{}, 1); e.resolveSource = e.source }) + e.once.Do(func() { + e.gate = make(chan struct{}, 1) + e.resolveSource = e.source + e.conversionPool = e.ConversionPool + }) if ctx == nil { ctx = context.Background() } diff --git a/internal/webview/webview.go b/internal/webview/webview.go index e0f17ad..e998558 100644 --- a/internal/webview/webview.go +++ b/internal/webview/webview.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -//go:build windows && amd64 +//go:build !js || !wasm package webview @@ -44,17 +44,19 @@ const applicationID = "me.aite.ofdgo" // Host 连接桌面系统能力和共用文档业务 type Host struct { - mu sync.Mutex - window *guigo.Window - files *fileStore - engines map[string]*webui.NativeEngine - engineStops map[string]func() bool - cancels map[string]context.CancelFunc - fontOnce sync.Once - fonts []map[string]any - fontFiles map[string]string - fontError error - pendingPaths []string + mu sync.Mutex + window *guigo.Window + files *fileStore + engines map[string]*webui.NativeEngine + engineStops map[string]func() bool + conversionPool *ofdgo.ConversionPool + cancels map[string]context.CancelFunc + batchResults map[string]any + fontOnce sync.Once + fonts []map[string]any + fontFiles map[string]string + fontError error + pendingPaths []string } // resourceReader 保存资源读取位置及关闭操作 @@ -76,9 +78,14 @@ func Run(ctx context.Context, paths []string, dataDir string) error { return err } h := &Host{files: files, engines: make(map[string]*webui.NativeEngine), cancels: make(map[string]context.CancelFunc), fontFiles: make(map[string]string), pendingPaths: append([]string(nil), paths...)} + h.conversionPool, err = ofdgo.NewConversionPool(ofdgo.ConversionPoolOptions{AvailableMemory: conversionMemory}) + if err != nil { + files.close() + return err + } defer h.close() bindings := make(map[string]guigo.HandlerFunc) - for _, name := range []string{"engine.open", "engine.close", "engine.call", "open", "save", "directory", "file", "release", "child", "remove", "write.begin", "write.data", "write.commit", "write.abort", "fonts", "ready", "title"} { + for _, name := range []string{"engine.open", "engine.close", "engine.call", "engine.start", "engine.cancel", "engine.finish", "batch", "open", "save", "directory", "file", "release", "child", "remove", "write.begin", "write.data", "write.commit", "write.abort", "fonts", "ready", "title"} { bindings["ofdgo.host."+name] = func(ctx context.Context, args any) (any, error) { return h.call(ctx, name, args) } } static := http.FileServerFS(webuiassets.NativeFS) @@ -260,6 +267,7 @@ func (h *Host) call(ctx context.Context, name string, value any) (any, error) { switch name { case "engine.open": engine := webui.NewNativeEngine(h.files.resolve) + engine.ConversionPool = h.conversionPool engine.OpenWriter = h.openWriter engine.ReserveOutput = h.reserveOutput engine.CleanupOutput = h.cleanupOutput @@ -273,8 +281,69 @@ func (h *Host) call(ctx context.Context, name string, value any) (any, error) { h.engineStops[id] = context.AfterFunc(guigo.PageContext(ctx), func() { _ = h.closeEngine(id) }) h.mu.Unlock() return guigo.Result{Value: map[string]any{"engine": id}, Discard: func() { _ = h.closeEngine(id) }}, nil + case "batch": + return map[string]any{"parallelism": h.conversionPool.Parallelism()}, nil case "engine.close": return nil, h.closeEngine(text("engine")) + case "engine.finish": + key := text("engine") + "/" + fmt.Sprint(args["id"]) + h.mu.Lock() + delete(h.batchResults, key) + h.mu.Unlock() + return nil, nil + case "engine.cancel": + key := text("engine") + "/" + fmt.Sprint(args["id"]) + h.mu.Lock() + cancel := h.cancels[key] + h.mu.Unlock() + if cancel != nil { + cancel() + } + return nil, nil + case "engine.start": + operationName := text("name") + if operationName != "ofdgoConvertFile" && operationName != "ofdgoPackFiles" { + return nil, fmt.Errorf("unsupported batch operation") + } + id := text("engine") + key := id + "/" + fmt.Sprint(args["id"]) + page := guigo.PageContext(ctx) + operation, cancel := context.WithCancel(page) + h.mu.Lock() + engine := h.engines[id] + _, duplicate := h.cancels[key] + if engine == nil || duplicate { + h.mu.Unlock() + cancel() + return nil, fmt.Errorf("invalid batch operation") + } + h.cancels[key] = cancel + h.mu.Unlock() + values, _ := args["args"].([]any) + go func() { + defer cancel() + defer func() { h.mu.Lock(); delete(h.cancels, key); h.mu.Unlock() }() + result := engine.Call(operation, operationName, values, func(event any) { + if value, ok := event.(map[string]any); ok { + value["id"] = args["id"] + h.emit("ofdgo.engine."+id, value) + } + }) + encoded := h.encode(page, result) + h.mu.Lock() + if page.Err() != nil || h.engines[id] != engine { + h.mu.Unlock() + h.discard(encoded) + return + } + if h.batchResults == nil { + h.batchResults = make(map[string]any) + } + h.batchResults[key] = encoded + h.mu.Unlock() + h.emit("ofdgo.engine."+id, map[string]any{"id": args["id"], "type": "result", "result": encoded}) + }() + return guigo.Result{Value: true, Discard: cancel}, nil case "engine.call": id := text("engine") key := id + "/" + fmt.Sprint(args["id"]) @@ -670,6 +739,7 @@ func (h *Host) loadFont(ctx context.Context, name string) ([]byte, error) { // 入参: id 会话标识 // 返回: error 关闭错误 func (h *Host) closeEngine(id string) error { + var results []any h.mu.Lock() engine := h.engines[id] delete(h.engines, id) @@ -682,7 +752,16 @@ func (h *Host) closeEngine(id string) error { cancel() } } + for key, result := range h.batchResults { + if strings.HasPrefix(key, id+"/") { + results = append(results, result) + delete(h.batchResults, key) + } + } h.mu.Unlock() + for _, result := range results { + h.discard(result) + } if engine != nil { return engine.Close() } diff --git a/internal/webview/webview_memory_other.go b/internal/webview/webview_memory_other.go new file mode 100644 index 0000000..eb4ba83 --- /dev/null +++ b/internal/webview/webview_memory_other.go @@ -0,0 +1,20 @@ +// Copyright 2025-2026 肖其顿 (XIAO QI DUN) +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build !windows + +package webview + +// conversionMemory 未提供系统内存探测时仅按并行数调度 +var conversionMemory func() int64 diff --git a/internal/webview/webview_memory_windows.go b/internal/webview/webview_memory_windows.go new file mode 100644 index 0000000..9da11bd --- /dev/null +++ b/internal/webview/webview_memory_windows.go @@ -0,0 +1,45 @@ +// Copyright 2025-2026 肖其顿 (XIAO QI DUN) +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package webview + +import ( + "syscall" + "unsafe" +) + +// memoryStatusProc 查询系统物理内存和提交额度 +var memoryStatusProc = syscall.NewLazyDLL("kernel32.dll").NewProc("GlobalMemoryStatusEx") + +// conversionMemory 预留系统内存后返回可用于新增转换的额度 +// 返回: int64 可用字节数，查询失败时暂停扩容 +func conversionMemory() int64 { + status := struct { + length uint32 + load uint32 + total uint64 + available uint64 + totalPage uint64 + availablePage uint64 + totalVirtual uint64 + availableVirtual uint64 + extended uint64 + }{} + status.length = uint32(unsafe.Sizeof(status)) + ok, _, _ := memoryStatusProc.Call(uintptr(unsafe.Pointer(&status))) + if ok == 0 { + return 0 + } + return max(0, int64(min(status.available, status.availablePage))-int64(status.total/8)) +} diff --git a/ofdgo_compression_parallel.go b/ofdgo_compression_parallel.go index 15abda3..d1cb4ef 100644 --- a/ofdgo_compression_parallel.go +++ b/ofdgo_compression_parallel.go @@ -20,7 +20,6 @@ import ( "crypto/sha256" "image" "io" - "runtime" "strconv" "sync" @@ -83,11 +82,17 @@ func (e *Editor) outputImageOptions(key string) (CompressionOptions, image.Point // 入参: 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(4, runtime.GOMAXPROCS(0)) + workers := conversionWorkers(e.output.ctx, 4) if workers < 2 || len(keys) < 2 { return nil } - limit := outputImageWorkingSetLimit + limit, encodedLimit := outputImageWorkingSetLimit, outputImageBatchLimit + if pool, _ := e.output.ctx.Value(conversionPoolKey{}).(*ConversionPool); pool != nil { + if available := pool.memoryLimit(); available > 0 { + limit = int(min(available, int64(int(^uint(0)>>1)))) + encodedLimit = limit + } + } if strconv.IntSize == 32 { limit = 64 << 20 } @@ -149,14 +154,14 @@ func (e *Editor) prepareOutputImages(keys []string, load func(string) ([]byte, e if pending[cacheKey] { continue } - cost := outputImageWorkingSet(e.output.ctx, data) - for running != 0 && (len(data) > outputImageBatchLimit-size || cost > limit-workingSet) { + cost := imageWorkingSet(e.output.ctx, data, limit) + for running != 0 && (len(data) > encodedLimit-size || cost > limit-workingSet) { if err := finish(); err != nil { return err } } job := outputImageJob{key: cacheKey, data: data, workingSet: cost} - if len(data) >= outputImageBatchLimit || cost >= limit { + if len(data) >= encodedLimit || cost >= limit { if err := e.runOutputImages([]outputImageJob{job}, results); err != nil { return err } @@ -167,7 +172,7 @@ func (e *Editor) prepareOutputImages(keys []string, load func(string) ([]byte, e size += len(data) workingSet += cost group.Go(func() { - job.candidate, job.err = optimizeImageResource(ctx, job.data, job.key.options, job.key.size) + job.candidate, job.err = optimizeOutputImage(ctx, job) completed <- job }) } @@ -184,24 +189,31 @@ func (e *Editor) prepareOutputImages(keys []string, load func(string) ([]byte, e // 入参: ctx 取消上下文, data 原始图像编码 // 返回: int 编码及像素缓冲的估算字节数，上限为outputImageWorkingSetLimit func outputImageWorkingSet(ctx context.Context, data []byte) int { - if len(data) >= outputImageWorkingSetLimit { - return outputImageWorkingSetLimit + return imageWorkingSet(ctx, data, outputImageWorkingSetLimit) +} + +// imageWorkingSet 按当前预算估算图像解码工作量 +// 入参: ctx 取消上下文, data 图像编码, limit 估算上限 +// 返回: int 工作量字节数 +func imageWorkingSet(ctx context.Context, data []byte, limit int) int { + if len(data) >= limit { + return limit } input := bytes.NewReader(data) - available := outputImageWorkingSetLimit - len(data) + available := limit - len(data) if baseimage.IsTIFF(data) { layout, err := baseimage.ReadTIFFLayoutContext(ctx, input) if err != nil { - return outputImageWorkingSetLimit + return limit } return len(data) + tiffImageWorkingSet(layout, available) } config, _, err := baseimage.DecodeConfig(imageInput{ReadCloser: io.NopCloser(input), context: ctx}) if err != nil || config.Width <= 0 || config.Height <= 0 { - return outputImageWorkingSetLimit + return limit } if config.Width > available/16/config.Height { - return outputImageWorkingSetLimit + return limit } return len(data) + config.Width*config.Height*16 } @@ -239,7 +251,7 @@ func tiffImageWorkingSet(layout baseimage.TIFFLayout, limit int) int { func (e *Editor) runOutputImages(jobs []outputImageJob, results map[outputImageKey][]byte) error { ctx := e.output.ctx optimize := func(job *outputImageJob) { - job.candidate, job.err = optimizeImageResource(ctx, job.data, job.key.options, job.key.size) + job.candidate, job.err = optimizeOutputImage(ctx, *job) } if len(jobs) == 1 { optimize(&jobs[0]) @@ -261,3 +273,23 @@ func (e *Editor) runOutputImages(jobs []outputImageJob, results map[outputImageK } return nil } + +// optimizeOutputImage 在共享预算内优化独立图像 +// 入参: ctx 转换上下文, job 图像任务 +// 返回: []byte 候选编码, error 转换错误 +func optimizeOutputImage(ctx context.Context, job outputImageJob) ([]byte, error) { + if pool, _ := ctx.Value(conversionPoolKey{}).(*ConversionPool); pool != nil { + if job.workingSet == 0 { + limit := outputImageWorkingSetLimit + if available := pool.memoryLimit(); available > 0 { + limit = int(min(available, int64(int(^uint(0)>>1)))) + } + job.workingSet = imageWorkingSet(ctx, job.data, limit) + } + if err := pool.acquire(ctx, true, int64(job.workingSet)); err != nil { + return nil, err + } + defer pool.release(true, int64(job.workingSet)) + } + return optimizeImageResource(ctx, job.data, job.key.options, job.key.size) +} diff --git a/ofdgo_compression_resources.go b/ofdgo_compression_resources.go index 78bda7a..ab75a67 100644 --- a/ofdgo_compression_resources.go +++ b/ofdgo_compression_resources.go @@ -294,7 +294,7 @@ func (e *Editor) compressResourceReferences(parts map[string][]byte, reader *Rea candidate, cached := imageResults[cacheKey] var err error if !cached { - candidate, err = optimizeImageResource(e.output.ctx, data, options, demand) + candidate, err = optimizeOutputImage(e.output.ctx, outputImageJob{data: data, key: outputImageKey{options: options, size: demand}}) if err := e.output.ctx.Err(); err != nil { return err } diff --git a/ofdgo_convert.go b/ofdgo_convert.go index e221a7e..38c0073 100644 --- a/ofdgo_convert.go +++ b/ofdgo_convert.go @@ -45,6 +45,7 @@ type ConvertProgress struct { // Input为空时识别源格式，Format为输出格式，Pages使用零基索引，nil表示全部页面 // PageRange使用一基页码表达式，与Pages互斥，空字符串表示全部页面 // SkipUnchanged仅在未压缩、源格式与目标相同且未选页时不写出数据 +// Pool可由多个独立转换共享，nil时不限制文件间并发 // 整份PDF同格式压缩直接重写PDF对象，其余PDF先转换为OFD对象再输出 // RendererOptions同时用于PDF局部合成和输出，在PDF.RendererOptions之后应用 // Backends统一配置转换与输出，优先于PDF和RendererOptions中的后端设置 @@ -58,6 +59,7 @@ type ConvertOptions struct { Pages []int PageRange string SkipUnchanged bool + Pool *ConversionPool Compression CompressionOptions Fonts FontMode Backends *RenderBackends @@ -114,6 +116,13 @@ func OutputFormats() []OutputFormat { // 入参: ctx 取消上下文, source 源数据, size 源字节数, output 输出流, options 转换选项 // 返回: ConvertReport 转换报告, error 错误信息 func Convert(ctx context.Context, source io.ReaderAt, size int64, output io.Writer, options ConvertOptions) (report ConvertReport, err error) { + if options.Pool != nil { + if err = options.Pool.acquire(ctx, false, 0); err != nil { + return report, err + } + defer options.Pool.release(false, 0) + ctx = context.WithValue(ctx, conversionPoolKey{}, options.Pool) + } if err = options.Compression.Validate(); err != nil { return report, err } diff --git a/ofdgo_convert_pool.go b/ofdgo_convert_pool.go new file mode 100644 index 0000000..bc4432e --- /dev/null +++ b/ofdgo_convert_pool.go @@ -0,0 +1,146 @@ +// Copyright 2025-2026 肖其顿 (XIAO QI DUN) +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package ofdgo + +import ( + "context" + "fmt" + "runtime" + "sync" + "time" +) + +// ConversionPoolOptions 设置共享转换并行度及图像工作集预算 +// Parallelism为0时使用GOMAXPROCS，MemoryLimit为0时不指定固定预算 +// AvailableMemory返回可用于新增任务的字节数，nil表示不探测，回调须支持并发调用 +// 内存预算用于调度，不限制解析器及运行时的实际分配 +type ConversionPoolOptions struct { + Parallelism int + MemoryLimit int64 + AvailableMemory func() int64 +} + +// ConversionPool 协调独立转换和图像任务，配置在创建后不变，可跨会话复用 +type ConversionPool struct { + options ConversionPoolOptions + mu sync.Mutex + changed chan struct{} + active int + images int + memory int64 +} + +// conversionPoolKey 标识单次转换使用的共享预算 +type conversionPoolKey struct{} + +// NewConversionPool 创建共享转换调度器，不启动后台任务 +// 入参: options 并行及内存选项 +// 返回: *ConversionPool 调度器, error 配置错误 +func NewConversionPool(options ConversionPoolOptions) (*ConversionPool, error) { + if options.Parallelism < 0 || options.MemoryLimit < 0 { + return nil, fmt.Errorf("invalid conversion pool options") + } + if options.Parallelism == 0 { + options.Parallelism = runtime.GOMAXPROCS(0) + } + return &ConversionPool{options: options, changed: make(chan struct{})}, nil +} + +// Parallelism 返回同时运行的转换数量上限 +// 返回: int 并行上限 +func (p *ConversionPool) Parallelism() int { return p.options.Parallelism } + +// memoryLimit 读取当前可分配的图像工作量预算 +// 返回: int64 可用字节数，负值表示未设置预算 +func (p *ConversionPool) memoryLimit() int64 { + limit := int64(-1) + if p.options.MemoryLimit > 0 { + limit = p.options.MemoryLimit + } + if p.options.AvailableMemory != nil { + available := max(0, p.options.AvailableMemory()) + if limit < 0 || available < limit { + limit = available + } + } + return limit +} + +// acquire 等待转换或图像预算，超预算图像可独占运行 +// 入参: ctx 取消上下文, image 是否为图像任务, size 图像工作量 +// 返回: error 取消错误 +func (p *ConversionPool) acquire(ctx context.Context, image bool, size int64) error { + for { + if err := ctx.Err(); err != nil { + return err + } + limit := p.memoryLimit() + p.mu.Lock() + allowed := p.active < p.options.Parallelism && (p.active == 0 || limit != 0) + if image { + allowed = p.images < runtime.GOMAXPROCS(0) && (p.images == 0 || limit < 0 || size <= limit-p.memory) + } + if allowed { + if image { + p.images++ + p.memory += size + } else { + p.active++ + } + p.mu.Unlock() + return nil + } + changed := p.changed + p.mu.Unlock() + timer := time.NewTimer(250 * time.Millisecond) + select { + case <-ctx.Done(): + timer.Stop() + return ctx.Err() + case <-changed: + case <-timer.C: + } + timer.Stop() + } +} + +// release 归还任务预算并唤醒等待者 +// 入参: image 是否为图像任务, size 图像工作量 +func (p *ConversionPool) release(image bool, size int64) { + p.mu.Lock() + defer p.mu.Unlock() + if image { + p.images-- + p.memory -= size + } else { + p.active-- + } + close(p.changed) + p.changed = make(chan struct{}) +} + +// conversionWorkers 按当前文件数量分配内部计算任务 +// 入参: ctx 转换上下文, fallback 非共享模式并行上限 +// 返回: int 内部并行数 +func conversionWorkers(ctx context.Context, fallback int) int { + p, _ := ctx.Value(conversionPoolKey{}).(*ConversionPool) + if p == nil { + return min(fallback, runtime.GOMAXPROCS(0)) + } + p.mu.Lock() + active := p.active + p.mu.Unlock() + return max(1, runtime.GOMAXPROCS(0)/max(1, active)) +} diff --git a/ofdgo_pdf_mesh.go b/ofdgo_pdf_mesh.go index ded4600..9436fea 100644 --- a/ofdgo_pdf_mesh.go +++ b/ofdgo_pdf_mesh.go @@ -17,7 +17,6 @@ package ofdgo import ( "fmt" "math" - "runtime" "slices" "sync" @@ -438,7 +437,7 @@ func (c *pdfCompositor) mesh(mesh *pdfgo.MeshGradient, space *pdfgo.ColorSpace, } return c.importer.ctx.Err() } - workers := min(runtime.GOMAXPROCS(0), 4, max(1, c.height/32)) + workers := min(conversionWorkers(c.importer.ctx, 4), max(1, c.height/32)) if workers == 1 { if err := sampleRows(0, c.height); err != nil { return nil, err