@fideus-labs/工人池

一个有固定货币和同伴的工人池 @fideus-labs/fizarrita (法语) 加速的软件包 对工人的Zarrita密码操作。 浏览器中的 Web 工人运行 node:worker_threads 在节点中,在一个接口后.

特征

• 受约束货币 最多 poolSize 工人同时运行.
• 工人回收利用 ——工人跨任务重复使用,而不是 被重新创建.
• 春克队列接口 —— 说 add() + 键 onIdle(),与 sarrita.js 兼容 和队列图案.
• 批量接口 —— 说 runTasks() 包括进度报告和取消.
• 浏览器和节点 ——在Web Workers和上运行相同的任务代码 node:worker_threads; 见 节点.js。 。 。 .
• 零运行时间依赖关系.

安装

npm add @fideus-labs/worker-pool

使用量

任务职能合同

每个任务职能都得到一名可用的工人(或 null 当游泳池需要的时候 (创建新工人) 必须 返回与工人一起回收的对象 和结果:

type WorkerPoolTask<T> = (
  worker: WorkerLike | null
) => Promise<{ worker: WorkerLike; result: T }>

WorkerLike 是一块 Worker API 池及其任务使用—— postMessage, (中文(简体) ), terminate, (中文(简体) ), addEventListener, (中文(简体) ), removeEventListener浏览器 Worker 满足它,同样 node:worker_threads 适配器,即 让一个任务功能同时为运行时间服务.

Chunk 队列接口 (add 编号 : onIdle(中文(简体) )

import { WorkerPool } from '@fideus-labs/worker-pool'
import type { WorkerLike } from '@fideus-labs/worker-pool'

function createTask(input: number) {
  return (worker: WorkerLike | null) => {
    // Two things worth copying here:
    //  - the literal `new Worker(new URL(...), ...)` form, which is the only
    //    shape bundlers recognise as a worker entry point;
    //  - annotating the local, because `WorkerLike | Worker` is a union
    //    TypeScript resolves to the wrong `postMessage` overload.
    const w: WorkerLike =
      worker ??
      new Worker(new URL('./my-worker.js', import.meta.url), { type: 'module' })
    return new Promise<{ worker: WorkerLike; result: number }>((resolve, reject) => {
      // Handle 'error' too, or a failing worker leaves onIdle() pending forever.
      const onMessage = (e: { data: number }) => {
        detach()
        resolve({ worker: w, result: e.data })
      }
      const onError = (e: { message: string }) => {
        detach()
        reject(new Error(e.message))
      }
      const detach = () => {
        w.removeEventListener('message', onMessage)
        w.removeEventListener('error', onError)
      }
      w.addEventListener('message', onMessage)
      w.addEventListener('error', onError)
      w.postMessage(input)
    })
  }
}

const pool = new WorkerPool(4) // 4 concurrent workers

pool.add(createTask(1))
pool.add(createTask(2))
pool.add(createTask(3))

try {
  const results = await pool.onIdle<number>()
  // results: [result1, result2, result3] — in add() order
} finally {
  pool.terminateWorkers()
}

批量接口( E)runTasks(中文(简体) )

随即提交一系列任务,同时提交可选进度报告和 取消 :

const pool = new WorkerPool(2)

const tasks = inputs.map((input) => createTask(input))

const { promise, runId } = pool.runTasks(tasks, (completed, total) => {
  console.log(`${completed}/${total}`)
})

// Cancel if needed:
// pool.cancel(runId)

try {
  const results = await promise
} finally {
  pool.terminateWorkers()
}

批次也可以绑在 AbortSignal。当信号起火时,任务 承诺与信号的拒绝 理由; 任务已完成, 但其结果被丢弃 他们的工人返回游泳池:

const controller = new AbortController()

const { promise } = pool.runTasks(tasks, null, { signal: controller.signal })

// e.g. the viewport moved and these results are stale:
controller.abort(new Error('viewport moved'))

API 密码

new WorkerPool(poolSize: number)

最多创建一个池 poolSize 并行工作者.

pool.add<T>(fn: WorkerPoolTask<T>): void

执行一个任务。 任务在 onIdle() 名为.

pool.onIdle<T>(): Promise<T[]>

执行所有排队任务并等待完成 。 返回的结果是 添加了命令任务.

pool.runTasks<T>(taskFns, progressCallback?, options?): { promise, runId }

提交一批任务 。 这个 promise 以命令的结果解决。 这个 选项 progressCallback 被援引为 (completedTasks: number, totalTasks: number) => void 每次任务结束后 完成 . options.signal 接受一个 AbortSignal: 当它起火时, 任务 承诺与信号的拒绝 理由,你觉得呢? 已中止的信号在任何任务之前拒绝批次 启动.

pool.cancel(runId: number): void

取消待决事项 runTasks 批号。 承诺拒绝 'Remaining tasks canceled'。 。 。 .

pool.terminateWorkers(): void

终止所有闲工. 池子在此之后仍然可以使用——新的 将根据需要创建工人.

createWorker(url, options?): WorkerLike

为当前运行时间创建工作者:模块 Worker 如果有的话 (浏览器、德诺、邦)a NodeWorker 在节点上。 如果两者都不可用,则投出.

捆绑的浏览器代码 : 捆绑者只检测工人从 字义 new Worker(new URL('./w.js', import.meta.url), { type: 'module' }) 形式。 一通电话 createWorker 他们不透明,故在捆绑的浏览器中 应用工人脚本必须达到输出的另一种方式(a) ?worker 导入, 复制资产或文字 new Worker(...) 在浏览器分支上).

new NodeWorker(url, options?)

A 类 node:worker_threads 后面的工人们 WorkerLike 接口。 基础 线程是同步创建的 — node:worker_threads 按需进口 浏览器捆绑从不试图解决它——但构建器是同步并 在线索存在之前发布的消息按顺序排列.

options 接受 name 财务报告和审定财务报表 workerData。 。 。 .

isNodeRuntime(): boolean

当前运行时间是否报告为节点 .

节点.js

平面 节点没有全局 Worker这样 new Worker(...) 在任务抛出 ReferenceError: Worker is not defined启用 createWorker,选择 运行时间的正确执行 :

import { createWorker, WorkerPool } from '@fideus-labs/worker-pool'
import type { WorkerLike } from '@fideus-labs/worker-pool'

const workerUrl = new URL('./my-worker.mjs', import.meta.url)

function createTask(input: number) {
  return async (worker: WorkerLike | null) => {
    const w = worker ?? createWorker(workerUrl)
    const result = await new Promise<number>((resolve, reject) => {
      const onMessage = (e: { data: number }) => { detach(); resolve(e.data) }
      const onError = (e: { message: string }) => { detach(); reject(new Error(e.message)) }
      const detach = () => {
        w.removeEventListener('message', onMessage)
        w.removeEventListener('error', onError)
      }
      w.addEventListener('message', onMessage)
      w.addEventListener('error', onError)
      w.postMessage(input)
    })
    return { worker: w, result }
  }
}

const pool = new WorkerPool(4)
const { promise } = pool.runTasks(inputs.map(createTask))

try {
  const results = await promise
} finally {
  // Required, and in a finally: worker threads hold the event loop open, so a
  // failed run would otherwise leave the process unable to exit.
  pool.terminateWorkers()
}

工人剧本本身就很普通 node:worker_threads 模块 :

import { parentPort } from 'node:worker_threads'

parentPort.on('message', (input) => parentPort.postMessage(input * input))

注释:

• 总是打电话 pool.terminateWorkers()。 。 。 . 一个活工线保持了节点 事件循环存在,所以没有它,进程就不会退出.
• 事件形状. 节点向 on('message') 还有一个 Error 改为 on('error');适配器将其包装为 { data } 财务报告和审定财务报表 { message, error } 这样 听众在两个运行时间读取相同.
• SharedArrayBuffer 和转让清单 在浏览器中工作.

zarrita.js 一体化

这个 @fideus-labs/fizarrita 软件包提供 getWorker 财务报告和审定财务报表 setWorker 作为 萨里塔的空降替换 get 财务报告和审定财务报表 set代码c 通过工人池向工人提供编码/解码。 每个都装一个密码工 运行时间和在它们之间自动选择, 所以下面的例子工作 节点不变.

安装

pnpm add @fideus-labs/fizarrita @fideus-labs/worker-pool zarrita

基本用途

import { WorkerPool } from '@fideus-labs/worker-pool'
import { getWorker, setWorker } from '@fideus-labs/fizarrita'
import * as zarr from 'zarrita'

const pool = new WorkerPool(4)

// Open an array
const store = new zarr.FetchStore('https://example.com/data.zarr')
const arr = await zarr.open(store, { kind: 'array' })

try {
  // Read with codec decode offloaded to workers
  const chunk = await getWorker(arr, null, { pool })

  // Write with codec encode offloaded to workers
  await setWorker(arr, null, chunk, { pool })
} finally {
  pool.terminateWorkers()
}

共享对话框支持

两者 getWorker 财务报告和审定财务报表 setWorker 支助a useSharedArrayBuffer 选项 额外性能 :

// Read — output allocated on SharedArrayBuffer, workers decode directly
// into shared memory (eliminates one transfer + one copy per chunk)
const chunk = await getWorker(arr, null, {
  pool,
  useSharedArrayBuffer: true,
})

// chunk.data.buffer instanceof SharedArrayBuffer === true
// The chunk can be shared with other workers without copying.

// Write — intermediate buffers use SharedArrayBuffer for zero-transfer
// sharing between main thread and codec workers
await setWorker(arr, null, chunk, {
  pool,
  useSharedArrayBuffer: true,
})

getWorker 与 SAB 连接 :

• 输出键盘由 SharedArrayBuffer
• 解码器工人解码块并直接写入共享输出 通过 decode_into 信件协议
• 取消 1 个矩阵缓冲转移( 工人对主线) 和一个主线 set_from_chunk 每块复制
• 主线仍然处理填充值块

setWorker 与 SAB 连接 :

• 用于部分更新的中间块缓冲器 SharedArrayBuffer
• 减少主线程和编解码器之间的分流 在解码-修改-编码周期期间

COOP/COEP信头

SharedArrayBuffer 需要用这些 HTTP 标题服务页面 :

Cross-Origin-Opener-Policy: same-origin
Cross-Origin-Embedder-Policy: require-corp

如果这些信头不见了, useSharedArrayBuffer: true 将投出一个 描述性错误消息 .

实例 :

// vite.config.ts
export default defineConfig({
  server: {
    headers: {
      'Cross-Origin-Opener-Policy': 'same-origin',
      'Cross-Origin-Embedder-Policy': 'require-corp',
    },
  },
})

getWorker 选项

setWorker 选项

工人信件协议

解码器工人处理四种信息类型:

基准

实况基准 -试试看 你的浏览器.

库中包含一个比较香草沙瑞塔的基准应用 get编号 :set 与 getWorker编号 :setWorker (有和没有共享箭塔):

pnpm bench
# Opens at http://localhost:5174

该基准既支持合成内膜阵列,也支持远程 OME-Zarr 数据集来自AWS S3. 实时版本部署在 GitHub Pages 上 推动到 main。 。 。 .

发展

pnpm install
pnpm dev           # Start test app dev server (port 5173)
pnpm bench         # Start benchmark app (port 5174)
pnpm test          # Run Playwright browser tests
pnpm test:ui       # Interactive Playwright UI
pnpm test:node     # Build both packages, then run the Node tests

许可证

麻省理工学院