silk.concurrency.thread_pool.

When to use#

1. Call ThreadPool.make to start the worker threads.

2. Call Pool.block on a normal thread to run a program on the pool.

3. In a task, start more tasks with Pool.spawn and wait for one with TaskHandle.join.

Details#

Each task is a closed, Transferable lazy effect, packaged with Execution.makeTransferable. Thus a task can start on one worker, park, and continue on a different worker.

A ready task goes on one shared ready list. An idle worker moves a batch of ready tasks into its own WorkDeque. A worker with nothing to do steals from the deques of the other workers, and then sleeps on a Parker. When a parked task wakes, it goes back on the ready list and a sleeping worker wakes. Thus no worker spins.

A Pool is a small Transferable handle that tasks capture to start more tasks. The ThreadPool owns the threads. Dropping it stops and joins each worker, and drops the tasks that are not finished.

Errors#

ThreadPool.make fails with SpawnError or OutOfMemoryError. Pool.spawn and Pool.block fail with OutOfMemoryError. A task is a closed effect, so wrap a task that can fail in Effect.result and get its failure as a value.

Gotchas#

• Task bodies and results must be Transferable and 'static.

• In a task, wait for anything other than a TaskHandle with Execution.parkTransferable. A call that blocks the native thread, such as Pool.block, holds up a whole worker.

• Call TaskHandle.join only from inside a pool task.

• The pool uses the system allocator for its own storage, not the Allocator of the caller.

Examples#

Sum two halves in parallel#

import silk.mem.allocator { Allocator, OutOfMemoryError }
import silk.core.effect { Effect }
import silk.concurrency.thread_native { SpawnError }
import silk.core.result { Result }
import silk.concurrency.thread_pool { Pool, ThreadPool }

effect fn sumRange(from: i64, to: i64) -> i64 {
  let mut total: i64 = 0
  let mut at = from
  while at < to {
    total = total + at
    at = at + 1
  }
  return total
}

// Runs the low half as a new task, and the high half in this task.
effect fn both(pool: Pool) -> i64
  ! OutOfMemoryError {
  let low = run pool.spawn(sumRange(0, 500))
  let high = run sumRange(500, 1001)
  let lowValue = run low.join()
  return lowValue + high
}

effect fn program() -> i32
  ! OutOfMemoryError | SpawnError {
  let mut allocator = Allocator.systemAllocatorProvider()
  let threads = run ThreadPool.make(4) |> Effect.provideMut<Allocator>(&mut allocator)
  let pool = threads.pool()

  // A task is a closed effect, so its own failures come back as a value.
  let result = run pool.block(Effect.result(both(pool.clone())))
  return match move result {
    Result<i64, OutOfMemoryError>.Success { value } => {
      if value != 500500 {
        return 1
      }
      return 0
    }
    Result<i64, OutOfMemoryError>.Failure { error } => 1
  }
}

effect fn failed<E>(error: E) -> i32 {
  drop error
  return 1
}

pub fn main() -> i32 {
  return run Effect.catchAll(program(), failed)
}

Import as ThreadPool with import silk.concurrency.thread_pool { ThreadPool }.

Public declarations: 3.

TaskHandle#

pub struct TaskHandle<A>

The handle of one pool task, which gives back its result.

Details#

A handle is Transferable when A is. Dropping it detaches the task, which still runs.

Implementation TaskHandle<A>#

impl<A> TaskHandle<A>

Method join#

pub effect fn join(self: Self) -> A

Parks the calling task until this task finishes, then returns its result.

Gotchas#

Only a pool task can join. A join from any other place is a fatal state trap.

Pool#

pub struct Pool

A handle to a running pool, which tasks capture to start more tasks.

Details#

A handle is Transferable and Concurrent. It does not keep the worker threads running: once the ThreadPool that made it is dropped, a spawned task is dropped unstarted and its TaskHandle never finishes.

Implementation Pool#

impl Pool

Method spawn#

pub effect<'env> fn spawn<
    'env,
    A: 'static + Transferable,
    F: once Effect<'static; A> + Intrinsic.Detached + Transferable
  >(self: &Self, task: F) -> TaskHandle<A> ! OutOfMemoryError

Starts task as a new pool task and returns the handle that joins it.

Details#

The task is ready at once. It can start on any worker before this call returns.

Gotchas#

task cannot fail. For a task that can, such as an effect fn call with ! OutOfMemoryError, spawn Effect.result(task) and handle the Result that join returns.

Method block#

pub effect<'env> fn block<
    'env,
    A: 'static + Transferable,
    F: once Effect<'static; A> + Intrinsic.Detached + Transferable
  >(self: &Self, task: F) -> A ! OutOfMemoryError

Runs task on the pool and blocks the calling thread until it finishes.

Gotchas#

Calling it from a pool task blocks that task's worker for the whole wait.

Method clone#

pub fn clone(self: &Self) -> Pool

Returns another handle to the same pool.

ThreadPool#

pub struct ThreadPool

The owner of a pool's worker threads: a work-stealing pool of native threads that runs lazy tasks in parallel.

When to use#

Call ThreadPool.make to start the worker threads, then Pool.block on a normal thread to run a program on the pool. In a task, start more tasks with Pool.spawn and wait for one with TaskHandle.join.

Details#

Each task is a closed, Transferable lazy effect, so a task can start on one worker, park, and continue on a different worker. Idle workers steal work from each other and then sleep, so no worker spins. /// Dropping it stops every worker, joins its thread, and drops the tasks still unfinished.

Gotchas#

• Task bodies and results must be Transferable and 'static.

• A task is a closed effect, so wrap a task that can fail in Effect.result and get its failure as a value.

• The pool uses the system allocator for its own storage, not the Allocator of the caller.

Implementation ThreadPool#

impl ThreadPool

Associated function make#

pub effect fn make(workers: usize) -> ThreadPool ! OutOfMemoryError | SpawnError ? &mut Allocator

Starts a pool of workers native threads, all waiting for tasks.

Details#

Zero workers means one. If a thread fails to start, the ones already started are stopped and joined before the error returns.

Method pool#

pub fn pool(self: &Self) -> Pool

Returns a handle to this pool.

Implementation Drop for ThreadPool#

impl Drop for ThreadPool

Operation drop#

fn drop(self: &mut ThreadPool) -> ()

Category: Concurrency.

Profiles: aarch64-apple-darwin, aarch64-unknown-linux-gnu, aarch64-unknown-linux-gnu-no-libc, x86_64-unknown-linux-gnu, x86_64-unknown-linux-gnu-no-libc, wasm32-unknown-unknown, wasm32-wasip1, wasm32-wasip2.