docs(runtime): explain digest scheduling and streaming boundaries
This commit is contained in:
@@ -5,7 +5,10 @@
|
||||
import {fault} from '../canvas-core/values.ts';
|
||||
import {createDigestPool} from './digest-pool.ts';
|
||||
/** Browser capability stays at the adapter; WorkspaceRuntime owns the pool's lifetime. */
|
||||
/** 构建 workspace 可共享的 DigestPort;任务 ID/长度/版本共同验证响应归属。 */
|
||||
/** 在构造时选择能力:Worker 不存在就使用池的默认分块执行器,否则每个 execute 新建专用 Worker。
|
||||
* 此选择不随后续 Worker 全局变化自动重算;创建/运行失败也不降级重试。
|
||||
* 使用默认池预算,响应按实例内递增 ID/输入长度/版本和摘要格式核对,不独立重算摘要。
|
||||
* 没有超时,线程无响应需外部取消/close;Worker 内原生哈希阈值由 digestWorkerSource 控制。 */
|
||||
export function createBrowserDigests(){
|
||||
let next=0;
|
||||
return createDigestPool({execute:typeof Worker==='undefined'?undefined:(bytes,signal)=>new Promise<string>((resolve,reject)=>{
|
||||
@@ -15,6 +18,8 @@ export function createBrowserDigests(){
|
||||
const id=++next;let done=false;
|
||||
/** 幂等完成:解绑所有消息/错误/abort 监听并终止 Worker,防止迟到响应泄漏或重复 resolve。 */
|
||||
const finish=(error?:unknown,sha256?:string)=>{if(done)return;done=true;signal.removeEventListener('abort',abort);worker.onmessage=null;worker.onerror=null;worker.onmessageerror=null;worker.terminate();if(error)reject(error);else resolve(sha256!);},abort=()=>finish(fault('import_aborted'));
|
||||
// 稳定 digest_worker_failed 只有身份匹配才采纳;其它响应走摘要字段校验,不把任意 error 字段当服务错误。
|
||||
// 格式正确只表明消息可解释,真实性来自受信的专用 Worker 算法;不在 UI 线程重复计算。
|
||||
// 消息须精确匹配版本、任务 ID 和输入长度,再验64位十六进制摘要;已知错误也必须匹配身份。
|
||||
signal.addEventListener('abort',abort,{once:true});worker.onmessage=event=>{const value=event.data;if(value?.version===1&&value.id===id&&value.byteLength===bytes.size&&value.error==='digest_worker_failed'){finish(fault('digest_worker_failed'));return;}if(!value||value.version!==1||value.id!==id||value.byteLength!==bytes.size||typeof value.sha256!=='string'||!(/^[a-f0-9]{64}$/.test(value.sha256)))finish(fault('invalid_digest_response'));else finish(undefined,value.sha256);};worker.onerror=event=>{event.preventDefault();finish(fault('digest_worker_failed'));};worker.onmessageerror=()=>finish(fault('invalid_digest_response'));
|
||||
// Blob 通过结构化克隆传递,不在 UI 主线程先读取整份 arrayBuffer。
|
||||
|
||||
@@ -8,7 +8,9 @@ import {importSourceDigest} from './import-integrity.ts';
|
||||
/** 可替换摘要端口,输出完整 Blob 的小写 SHA256;close 终止本池,不清业务数据或外部回执。 */
|
||||
export type DigestPort={digest(bytes:Blob,signal?:AbortSignal):Promise<string>;close():void};
|
||||
/** Read-only compute scheduling: bounded active slots + FIFO Map, never durable operation facts. */
|
||||
/** 默认2并发/64等待,上限8并发;可注入 Worker/native/测试执行器,默认使用分块摘要。 */
|
||||
/** 默认 2 并发/64 等待,上限 8 并发;可注入 execute,默认 importSourceDigest 分块计算。
|
||||
* maxQueued 只计算 waiting,不含 active;不限制 Blob 大小/MIME,不做相同文件去重或结果缓存。
|
||||
* execute 负责计算真实完整摘要,池仅用正则检查输出格式,不能证明输出确实对应输入字节。 */
|
||||
export function createDigestPool(options:{execute?:(bytes:Blob,signal:AbortSignal)=>Promise<string>;concurrency?:number;maxQueued?:number}={}):DigestPort{
|
||||
const concurrency=options.concurrency??2,maxQueued=options.maxQueued??64,execute=options.execute??importSourceDigest;
|
||||
if(!Number.isSafeInteger(concurrency)||concurrency<1||concurrency>8||!Number.isSafeInteger(maxQueued)||maxQueued<1)throw fault('invalid_digest_budget');
|
||||
@@ -16,9 +18,13 @@ export function createDigestPool(options:{execute?:(bytes:Blob,signal:AbortSigna
|
||||
type Task={id:number;bytes:Blob;controller:AbortController;resolve:(value:string)=>void;reject:(error:unknown)=>void;cleanup:()=>void;done:boolean};
|
||||
// Map 既保留入队次序,又可在等待期取消时 O(1) 删除;active 保留真实未结束执行的容量占用。
|
||||
const waiting=new Map<number,Task>(),active=new Map<number,Task>();let next=0,closed=false;
|
||||
/** 只 settle 一次并移除等待项/监听;故意不在这里 active.delete,迟到结果不能二次采纳。 */
|
||||
/** 只 settle 一次并移除 waiting/外部监听,不在这里 active.delete,迟到结果不能二次采纳。
|
||||
* error 用真值分支,execute 应以 Error 等真值拒绝;falsy rejection 不保证被当成失败处理。 */
|
||||
function finish(task:Task,error?:unknown,value?:string){if(task.done)return;task.done=true;task.cleanup();waiting.delete(task.id);if(error)task.reject(error);else task.resolve(value!);}
|
||||
/** 有空槽就取最早等待项,先登记 active 再微任务执行;结果校验/取消后 finally 归还真实槽。 */
|
||||
/** 从 Map 首项 FIFO 取任务,先占 active 再微任务调用 execute;启动前、结果返回后均复查 abort。
|
||||
* execute 的拒绝由第二个 then 回调处理,结果校验抛错再由 catch 处理,最终统一归还槽位并泵下一项。
|
||||
* 正则会按 JS 规则强制转字符串,未另做 typeof 检查;注入执行器应遵守 Promise<string> 合同。
|
||||
* 无视 abort 且不 settle 的执行器会一直占槽;池没有超时或强制终止通用执行器的能力。 */
|
||||
function pump(){
|
||||
if(closed)return;
|
||||
while(active.size<concurrency&&waiting.size){const task=waiting.values().next().value!;waiting.delete(task.id);active.set(task.id,task);
|
||||
@@ -31,7 +37,8 @@ export function createDigestPool(options:{execute?:(bytes:Blob,signal:AbortSigna
|
||||
if(closed)return Promise.reject(fault('digest_pool_closed'));if(signal?.aborted)return Promise.reject(fault('import_aborted'));if(!(bytes instanceof Blob))return Promise.reject(fault('invalid_digest_input'));if(active.size>=concurrency&&waiting.size>=maxQueued)return Promise.reject(fault('digest_capacity'));
|
||||
return new Promise((resolve,reject)=>{const task:Task={id:++next,bytes,controller:new AbortController(),resolve,reject,cleanup:()=>signal?.removeEventListener('abort',abort),done:false},abort=()=>{task.controller.abort();finish(task,fault('import_aborted'));pump();};signal?.addEventListener('abort',abort,{once:true});waiting.set(task.id,task);pump();});
|
||||
},
|
||||
/** 幂等关闭,取消等待/活跃任务的调用结果;无视 signal 的执行仍由其 Promise 收尾后释放 active。 */
|
||||
/** 幂等关闭并取消 waiting/active 的调用结果,停止接受新任务;不返回 drain Promise。
|
||||
* 无视 signal 的执行仍由自身 Promise 收尾后释放 active,close 返回不证明底层计算已停止。 */
|
||||
close(){if(closed)return;closed=true;for(const task of [...waiting.values(),...active.values()]){task.controller.abort();finish(task,fault('import_aborted'));}},
|
||||
});
|
||||
}
|
||||
|
||||
@@ -11,7 +11,10 @@ export const nativeDigestByteLimit=64*1024*1024;
|
||||
|
||||
/** Dedicated Worker only: whole-buffer native hashing never moves onto the UI thread.
|
||||
* The host still owns cancellation by terminating the request's Worker. */
|
||||
/** 验证 Blob,按能力/大小选 native 或增量实现,均返回相同的完整原字节摘要。 */
|
||||
/** 验证 Blob,SubtleCrypto 存在且大小 ≤64MiB 时读取整个 ArrayBuffer 做 SHA-256,否则流式计算。
|
||||
* 原生读/算失败直接传播,不自动切到流式重试;注入 subtle 应遵守原生 digest 的 ArrayBuffer 合同。
|
||||
* 本函数无 signal,取消靠宿主终止 Worker;代码本身不检测当前线程,Worker 限制由装配调用方遵守。
|
||||
* 阈值只限制此输入的整缓冲分配,未计 Blob/原生算法内部内存及其它并发 Worker。 */
|
||||
export async function digestWorkerSource(bytes:Blob,subtle:Pick<SubtleCrypto,'digest'>|null|undefined=globalThis.crypto?.subtle):Promise<string>{
|
||||
if(!(bytes instanceof Blob))throw fault('invalid_digest_input');
|
||||
// native digest 必须返回32字节;输出固定64字符十六进制,不接受变长响应。
|
||||
@@ -20,6 +23,7 @@ export async function digestWorkerSource(bytes:Blob,subtle:Pick<SubtleCrypto,'di
|
||||
if(digest.byteLength!==32)throw fault('invalid_digest_response');
|
||||
return [...new Uint8Array(digest)].map(value=>value.toString(16).padStart(2,'0')).join('');
|
||||
}
|
||||
// Worker fallback 每块1MiB,只做微任务让出;UI 让出策略由 UI fallback 的端口另行决定。
|
||||
// 每块 1MiB,块间 yieldTask 仅完成一个 Promise 微任务,不保证 Worker 消息队列获得宏任务机会。
|
||||
// UI fallback 用 importSourceDigest 默认定时器让出;不要把这里的微任务策略移作 UI 响应性承诺。
|
||||
return importSourceDigest(bytes,undefined,{chunkBytes:1024*1024,yieldTask:()=>Promise.resolve()});
|
||||
}
|
||||
|
||||
@@ -8,7 +8,9 @@ import {digestWorkerSource} from './digest-worker-source.ts';
|
||||
const scope=globalThis as unknown as {onmessage:((event:MessageEvent)=>void)|null;postMessage(value:unknown):void};
|
||||
/** 首条有效请求被接受后保持 true,直到整个 Worker 被宿主终止。 */
|
||||
let accepted=false;
|
||||
/** 校验协议和请求后启动只读摘要;成功/失败都回显输入 ID 与字节数,供宿主核对。 */
|
||||
/** 首条有效 version=1/正安全整数 ID/Blob 设置 accepted,之后包括失败后的重复消息都忽略。
|
||||
* 无效请求直接忽略,不返回错误;这里不检查 MIME/大小或资源身份,不拥有持久任务。
|
||||
* 算法拒绝会尝试回稳定错误;成功回信自身抛错不由同一个 then 的失败回调捕获,宿主仍需错误/取消收尾。 */
|
||||
scope.onmessage=event=>{
|
||||
const request=event.data;if(accepted||!request||request.version!==1||!Number.isSafeInteger(request.id)||request.id<1||!(request.bytes instanceof Blob))return;accepted=true;
|
||||
void digestWorkerSource(request.bytes).then(sha256=>scope.postMessage({version:1,id:request.id,byteLength:request.bytes.size,sha256}),()=>scope.postMessage({version:1,id:request.id,byteLength:request.bytes.size,error:'digest_worker_failed'}));
|
||||
|
||||
@@ -5,13 +5,17 @@
|
||||
import { fault } from '../canvas-core/values.ts';
|
||||
import {createSha256} from './sha256.ts';
|
||||
/** Complete-source SHA-256, with bounded chunk memory and cooperative fallback scheduling. */
|
||||
/** 默认64KiB分块(配置上限4MiB),每块读前/读后检查取消,并在块间合作让出。 */
|
||||
/** 顺序 slice 读取完整 Blob,默认 64KiB 分块(配置上限 4MiB),空文件也返回合法摘要。
|
||||
* 每块读前/读后及 yield 后检查取消;原生 arrayBuffer/yield 本身不被信号强制终止。
|
||||
* 默认块间 setTimeout(0) 让出,只有仍有下一块才 yield;自定义 yieldTask 决定实际调度公平性。
|
||||
* 块内 hash.update 为同步 CPU 工作,不会中途取消;读/让出错误直接传播,不返回部分摘要。 */
|
||||
export async function importSourceDigest(bytes:Blob,signal?:AbortSignal,options:{chunkBytes?:number;yieldTask?:()=>Promise<void>}={}):Promise<string>{
|
||||
const chunk=options.chunkBytes??64*1024,yieldTask=options.yieldTask??(()=>new Promise<void>(resolve=>setTimeout(resolve,0)));
|
||||
if(!(bytes instanceof Blob)||!Number.isSafeInteger(chunk)||chunk<1||chunk>4*1024*1024)throw fault('invalid_digest_input');
|
||||
/** 读/计算/让出边界统一检查 signal;异步读取即使无法中止,迟到字节也不再参与有效结果。 */
|
||||
const check=()=>{if(signal?.aborted)throw fault('import_aborted');};check();const hash=createSha256();
|
||||
// 更新增量状态后释放对本块的局部引用;空文件直接 hex 得到标准空输入摘要。
|
||||
// 只顺序持有当前块,hash 自身不累计输入;part 在本次迭代/yield 期间仍可存在,实际释放时机由 GC 决定。
|
||||
// 空文件跳过读取/让出,最后 check 后直接 hex;有界分块不是整个应用进程的内存硬上限。
|
||||
for(let cursor=0;cursor<bytes.size;cursor+=chunk){check();const part=await bytes.slice(cursor,cursor+chunk).arrayBuffer();check();hash.update(new Uint8Array(part));if(cursor+chunk<bytes.size){await yieldTask();check();}}
|
||||
check();return hash.hex();
|
||||
}
|
||||
|
||||
@@ -20,7 +20,9 @@ const K=new Uint32Array([
|
||||
/** 32位循环右移;区别于简单逻辑右移,低位从左侧回填。 */
|
||||
const rotate=(n:number,bits:number)=>(n>>>bits)|(n<<(32-bits));
|
||||
/** Incremental bytes only; owns at most a partial block, state and message schedule. */
|
||||
/** 新建单次摘要状态;允许多次 update,hex 终结后只可重复读结果,不能继续追加。 */
|
||||
/** 新建单次摘要状态,闭包拥有可变 state/block/words;冻结 API 不代表内部计算状态不变。
|
||||
* update 可多次追加,hex 首次终结后缓存结果,不能继续追加;不支持复制状态或导出恢复游标。
|
||||
* 纯同步计算,无异步让出/取消;分块大小与调度由读取 Blob 的调用方控制。 */
|
||||
export function createSha256(){
|
||||
// length 计原输入字节;pending 计尾块已占字节;result 缓存最终十六进制并标记终结。
|
||||
const state=new Uint32Array([0x6a09e667,0xbb67ae85,0x3c6ef372,0xa54ff53a,0x510e527f,0x9b05688c,0x1f83d9ab,0x5be0cd19]),block=new Uint8Array(64),words=new Uint32Array(64);let pending=0,length=0,result:string|undefined;
|
||||
@@ -35,7 +37,9 @@ export function createSha256(){
|
||||
const values=[a,b,c,d,e,f,g,h];for(let i=0;i<8;i++)state[i]=(state[i]+values[i])>>>0;
|
||||
}
|
||||
return Object.freeze({
|
||||
/** 先补上次尾块,再就地压缩输入完整块,最后只复制不足64字节的尾部。 */
|
||||
/** 先补尾块,再直接读取输入完整块压缩,最后只复制不足 64 字节的尾部,不保留输入引用。
|
||||
* 不修改调用方字节;累计长度必须保持安全整数,终结后即使追加空数组也拒绝。
|
||||
* 这里的就地指同步消费输入,不代表每次 update 只处理固定工作量或自动让出主线程。 */
|
||||
update(bytes:Uint8Array){
|
||||
if(result!==undefined)throw fault('digest_finalized');if(!(bytes instanceof Uint8Array)||!Number.isSafeInteger(length+bytes.byteLength))throw fault('invalid_digest_input');length+=bytes.byteLength;let offset=0;
|
||||
if(pending){const count=Math.min(64-pending,bytes.byteLength);block.set(bytes.subarray(0,count),pending);pending+=count;offset=count;if(pending===64){compress(block,0);pending=0;}}
|
||||
|
||||
Reference in New Issue
Block a user