moon add mizchi/process_pool@0.1.3

Author

mizchi

Version

0.1.3

License

MIT

Repository

github.com/mizchi/process_pool.mbt

Last updated

6 months ago

Downloads

Dependencies

moonbitlang/async0.16.6
moonbitlang/x0.4.40

README

#mizchi/process_pool

A Moonbit process pool for js/native

moon add mizchi/process_pool

#Features

Run multiple external processes in parallel with configurable concurrency limits
Timeout support for individual jobs
Callback support for job completion notifications
Cross-platform:
- native: moonbitlang/async
- Semaphore-based concurrency control
- js: node:child_process
- wasm, wasm-gc with WASI

inspired by moonbitlang/maria daemon

#Usage

#Basic Example

let pool = @process_pool.ProcessPool::new(max_workers=4)
let results = pool.run_all([
  @process_pool.job("echo", ["hello"]),
  @process_pool.job("echo", ["world"]),
])

assert_eq(results.length(), 2)
assert_eq(results[0].exit_code, 0)
assert_eq(results[1].exit_code, 0)

#With Timeout

let pool = @process_pool.ProcessPool::new(max_workers=2)
let results = pool.run_all([
  @process_pool.job("sleep", ["10"], timeout=100),  // 100ms timeout
])

assert_eq(results.length(), 1)
assert_true(results[0].timed_out)

#With Error Checking

let pool = @process_pool.ProcessPool::new(max_workers=2)
let mut caught = false

// Raises JobError if any job fails
ignore(pool.run_all_checked([
  @process_pool.job("echo", ["hello"]),
  @process_pool.job("false", []),  // This will fail
])) catch {
  @process_pool.JobError::ProcessFailed(..) => {
    caught = true
  }
  _ => ()
}

assert_true(caught)

#With Completion Callback

let count = @ref.new(0)
let callback = @process_pool.OnJobComplete(fn(_result) {
  count.update(fn(n) { n + 1 })
})

let pool = @process_pool.ProcessPool::new(
  max_workers=4,
  on_complete=callback,
)

let results = pool.run_all([
  @process_pool.job("echo", ["task1"]),
  @process_pool.job("echo", ["task2"]),
])

assert_eq(results.length(), 2)
assert_eq(count.val, 2)

#Map Pattern (for SSG, batch processing, etc.)

let pool = @process_pool.ProcessPool::new(max_workers=4)
let items = ["a", "b", "c"]

let results = pool.map(items, fn(item) {
  @process_pool.job("echo", [item])
})

assert_eq!(results.length(), 3)
for result in results {
  assert_eq(result.exit_code, 0)
}

#API

#Types

#Job

///|
let _a : @process_pool.Job = @process_pool.job("echo", ["hello"])

#JobResult

The result of a job execution containing exit code, stdout, stderr, and timeout status.

#JobError

Error type raised by run_all_checked when a job fails or times out.

#Functions

#job(cmd, args, cwd?, timeout?) -> Job

Helper function to create a job.

#ProcessPool::new(max_workers~ = 4, on_complete?) -> ProcessPool

Create a new process pool with the specified maximum worker count.

#ProcessPool::run_all(jobs) -> Array[JobResult]

Run all jobs in parallel and return results. Job order is preserved.

#ProcessPool::run_all_checked(jobs) -> Array[JobResult]

Run all jobs and raise JobError if any job fails or times out.

#ProcessPool::map(items, to_job) -> Array[JobResult]

Transform items into jobs and run them in parallel.

#now() -> Int64

Get current time in milliseconds (useful for timing measurements).

#Benchmark

The example/ directory contains a benchmark that demonstrates parallel speedup by word-counting multiple documents.

#Running the Benchmark

# 1. Generate test documents (16 files, ~5000 words each)
node example/scripts/generate_docs.mjs

# 2. Run the benchmark
cd example && moon run --target native .

#Results

Processing 16 documents (~80,000 words total):

Workers	Time	Speedup
1	632ms	baseline
2	315ms	2.00x
4	194ms	3.25x
8	160ms	3.95x

The benchmark shows near-linear speedup with additional workers, demonstrating effective parallel process execution.

#Platform Support

Platform	Status
Native	Full support
JS (Node.js)	Full support
WASM	Not implemented
WASM-GC	Not implemented

#JS Semaphore Behavior Note

The JS implementation uses a cooperative multitasking model (async/await), which behaves differently from the native implementation's OS-level synchronization. Due to JavaScript's event loop scheduling, the semaphore may not strictly enforce concurrency limits in all scenarios. See lib_js.mbt for details.

#License

MIT

#
JobError

source

pub suberror JobError {
  ProcessFailed(Job, Int, String, String)
  ProcessTimedOut(Job, Int, String, String)
}

Job execution error

#
Job

source

pub struct Job {
  cmd : String
  args : Array[String]
  cwd : String?
  timeout : Int?
}

Job definition

#
JobResult

source

pub struct JobResult {
  job : Job
  exit_code : Int
  stdout : String
  stderr : String
  timed_out : Bool
}

Job execution result

#
OnJobComplete

source

pub(all) type OnJobComplete (JobResult) -> Unit

Callback invoked on job completion

#
OnJobComplete::inner

source

#deprecated("Use `struct T(A)` to declare a newtype and use `.0` access the underlying type instead.")
fn OnJobComplete::inner(self : OnJobComplete) -> ((JobResult) -> Unit)

Convert newtype to its underlying type, automatically derived.

#
ProcessPool

source

pub struct ProcessPool {
  max_workers : Int
  semaphore : @moonbitlang/async/semaphore.Semaphore
Semaphore
  on_complete : OnJobComplete?
}

Process pool

#
ProcessPool::map

source

async fn[T] ProcessPool::map(self : ProcessPool, items : Array[T], to_job : (T) -> Job) -> Array[JobResult]

Transform input data into jobs and run them in parallel

Useful for SSG and similar use cases where jobs are generated from file lists and executed in parallel.

#
ProcessPool::new

source

fn ProcessPool::new(max_workers? : Int, on_complete? : OnJobComplete) -> ProcessPool

Create a process pool

Parameters:

max_workers: Maximum number of concurrent processes (default: 4)
on_complete: Callback invoked on job completion (optional)

#
ProcessPool::run_all

source

async fn ProcessPool::run_all(self : ProcessPool, jobs : Array[Job]) -> Array[JobResult]

Run multiple jobs in parallel and return all results

Waits for all jobs to complete and returns the results array. Job order is preserved matching the input order.

#
ProcessPool::run_all_checked

source

async fn ProcessPool::run_all_checked(self : ProcessPool, jobs : Array[Job]) -> Array[JobResult]

Run multiple jobs in parallel and raise an exception on error

If any job fails (exit_code != 0) or times out, raises JobError.

#
job

source

fn job(cmd : String, args : Array[String], cwd? : String, timeout? : Int) -> Job

Helper function to create a job

Parameters:

cmd: Command to execute
args: Command arguments
cwd: Working directory (optional)
timeout: Timeout in milliseconds (optional)

#
now

source

fn now() -> Int64

Get current time in milliseconds

Source Files

Site source Report issue Packages Build queue Skills Statistics

process_pool

#mizchi/process_pool

#Features

#Usage

#Basic Example

#With Timeout

#With Error Checking

#With Completion Callback

#Map Pattern (for SSG, batch processing, etc.)

#API

#Types

#Job

#JobResult

#JobError

#Functions

#job(cmd, args, cwd?, timeout?) -> Job

#ProcessPool::new(max_workers~ = 4, on_complete?) -> ProcessPool

#ProcessPool::run_all(jobs) -> Array[JobResult]

#ProcessPool::run_all_checked(jobs) -> Array[JobResult]

#ProcessPool::map(items, to_job) -> Array[JobResult]

#now() -> Int64

#Benchmark

#Running the Benchmark

#Results

#Platform Support

#JS Semaphore Behavior Note

#License

#JobError

#Job

#JobResult

#OnJobComplete

#OnJobComplete::inner

#ProcessPool

#ProcessPool::map

#ProcessPool::new

#ProcessPool::run_all

#ProcessPool::run_all_checked

#job

#now

Source Files

#
JobError

#
Job

#
JobResult

#
OnJobComplete

#
OnJobComplete::inner

#
ProcessPool

#
ProcessPool::map

#
ProcessPool::new

#
ProcessPool::run_all

#
ProcessPool::run_all_checked

#
job

#
now