silk.concurrency.work_queue.

When to use#

Details#

WorkDeque.make returns the Worker that owns a Chase-Lev deque. The worker pushes and pops at the bottom with no contention. Its Stealers take from the top with one compare-exchange. The ring grows by doubling. Old rings stay allocated until the deque is dropped, because a stealer can still read one.

Injector is a FIFO list under a Mutex. Mailbox is the Vyukov multi-producer, single-consumer queue: a send is one swap and one store, with no lock.

Each pushed value moves into its own allocation (a parcel), and the queues move parcel addresses atomically. Only the thread that wins a value reads it. Thus no thread copies a value that it does not own.

Errors#

Each push and send allocates, so it requires Allocator and can fail with OutOfMemoryError.

Gotchas#

• Stealer.steal returns Steal.Retry when it loses a race. The caller decides whether to try again.

• MailboxReceiver.receive can return None for a short time while a send is between its two steps.

Examples#

Steal work on another thread#

import silk.mem.allocator { Allocator, OutOfMemoryError }
import silk.core.effect { Effect }
import silk.concurrency.thread_native { NativeThread, SpawnError }
import silk.core.option { Option }
import silk.concurrency.work_queue { Steal, Stealer, WorkDeque }

// Takes the oldest value. A lost race gives Retry, so try again.
fn stealOne(stealer: Stealer<i32>) -> i32 {
  while true {
    match move stealer.steal() {
      Steal<i32>.Success { value } => {
        return value
      }
      Steal<i32>.Empty => {
        return 0
      }
      Steal<i32>.Retry => {}
    }
  }
  return 0
}

effect fn program() -> i32
  ! OutOfMemoryError | SpawnError {
  let mut allocator = Allocator.systemAllocatorProvider()
  let mut worker = run WorkDeque.make<i32>() |> Effect.provideMut<Allocator>(&mut allocator)
  run worker.push(1) |> Effect.provideMut<Allocator>(&mut allocator)
  run worker.push(2) |> Effect.provideMut<Allocator>(&mut allocator)
  let stealer = worker.stealer()
  let thief = run NativeThread.spawn(
    fn() -> i32 {
      return stealOne(move stealer)
    }
  )
    |> Effect.provideMut<Allocator>(&mut allocator)

  // The thief takes from the top (oldest), the owner pops from the bottom (newest).
  let stolen = thief.join()
  let popped = match move worker.pop() {
    Option<i32>.Some { value } => value
    Option<i32>.None => 0
  }
  if stolen != 1 || popped != 2 {
    return 1
  }
  return 0
}

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

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

Send through a mailbox and an injector#

import silk.mem.allocator { Allocator, OutOfMemoryError }
import silk.core.effect { Effect }
import silk.core.option { Option }
import silk.concurrency.work_queue { Injector, Mailbox }

effect fn program() -> i32
  ! OutOfMemoryError {
  let mut allocator = Allocator.systemAllocatorProvider()

  // A mailbox: senders on any thread, one receiver.
  let mut receiver = run Mailbox.make<i32>() |> Effect.provideMut<Allocator>(&mut allocator)
  let sender = receiver.sender()
  run sender.send(7) |> Effect.provideMut<Allocator>(&mut allocator)
  let received = match move receiver.receive() {
    Option<i32>.Some { value } => value
    Option<i32>.None => 0
  }

  // An injector: a shared FIFO queue.
  let queue = run Injector<i32>.make() |> Effect.provideMut<Allocator>(&mut allocator)
  run queue.push(1) |> Effect.provideMut<Allocator>(&mut allocator)
  run queue.push(2) |> Effect.provideMut<Allocator>(&mut allocator)
  let first = match move queue.pop() {
    Option<i32>.Some { value } => value
    Option<i32>.None => 0
  }
  if received != 7 || first != 1 {
    return 1
  }
  return 0
}

effect fn outOfMemory(error: OutOfMemoryError) -> i32 {
  return 1
}

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

Import as WorkDeque with import silk.concurrency.work_queue { WorkDeque }.

Public declarations: 8.

WorkDeque#

pub struct WorkDeque

The constructor of work-stealing deques: the work of one thread, which idle threads can steal.

Details#

WorkDeque.make returns the Worker that owns a Chase-Lev deque. The worker pushes and pops at the bottom with no contention. Its Stealers take from the top with one compare-exchange. The ring grows by doubling. Old rings stay allocated until the deque is dropped, because a stealer can still read one. Each push allocates, so it can fail with OutOfMemoryError.

Implementation WorkDeque#

impl WorkDeque

Associated function make#

pub effect fn make<T>() -> Worker<T> ! OutOfMemoryError ? &mut Allocator

Makes an empty deque and returns the worker that owns it.

Worker#

pub struct Worker<T>

The owner of a work-stealing deque. It pushes and pops at the bottom.

Details#

A worker is Transferable when T is, so it can move to the thread that runs the work. Push and pop take it exclusively. Worker.pop returns the newest value.

Steal#

pub union Steal<T>

What one steal attempt found.

Empty#

Empty

The deque was empty.

Retry#

Retry

Another thread took the value that this attempt saw. Try again.

Success#

Success

The oldest value in the deque, now owned by the caller.

Field value#

pub value: T

The stolen value.

Implementation Worker<T>#

impl<T> Worker<T>

Method push#

pub effect<'env> fn push<'env>(self: &mut Self, value: T) -> () ! OutOfMemoryError ? &mut Allocator

Pushes value at the bottom, growing the ring when it is full.

Method pop#

pub fn pop(self: &mut Self) -> Option<T>

Pops the most recently pushed value, or reports none when the deque is empty.

Method stealer#

pub fn stealer(self: &Self) -> Stealer<T>

Returns a stealer for this deque.

Stealer#

pub struct Stealer<T>

A handle that takes the oldest values from a Worker's deque, from any thread.

Details#

A stealer is Transferable and Concurrent when T is Transferable.

Implementation Stealer<T>#

impl<T> Stealer<T>

Method steal#

pub fn steal(self: &Self) -> Steal<T>

Takes the oldest value, or reports that the deque was empty or that the attempt lost a race.

Method clone#

pub fn clone(self: &Self) -> Stealer<T>

Adds another stealer for the same deque.

Injector#

pub struct Injector<T>

A shared FIFO queue that all threads can push to and take from.

Details#

Clones are handles to the same queue. An injector is Transferable and Concurrent when T is Transferable. Each operation holds the lock only to link or unlink one parcel.

Implementation Injector<T>#

impl<T> Injector<T>

Associated function make#

pub effect<'env> fn make<'env>() -> Injector<T> ! OutOfMemoryError ? &mut Allocator

Makes an empty queue.

Method push#

pub effect<'env> fn push<'env>(self: &Self, value: T) -> () ! OutOfMemoryError ? &mut Allocator

Appends value at the back.

Method pop#

pub fn pop(self: &Self) -> Option<T>

Takes the value at the front, or reports none when the queue is empty.

Method clone#

pub fn clone(self: &Self) -> Injector<T>

Adds another handle to the same queue.

Mailbox#

pub struct Mailbox

The constructor of multi-producer, single-consumer mailboxes: many threads that send to one receiver.

Details#

It is the Vyukov queue: a send is one swap and one store, with no lock. Each send allocates, so it can fail with OutOfMemoryError.

Gotchas#

MailboxReceiver.receive can return None for a short time while a send is between its two steps.

Implementation Mailbox#

impl Mailbox

Associated function make#

pub effect fn make<T>() -> MailboxReceiver<T> ! OutOfMemoryError ? &mut Allocator

Makes an empty mailbox and returns its receiver.

MailboxReceiver#

pub struct MailboxReceiver<T>

The one receiving end of a Mailbox.

Details#

A receiver is Transferable when T is. A receive takes it exclusively.

Implementation MailboxReceiver<T>#

impl<T> MailboxReceiver<T>

Method receive#

pub fn receive(self: &mut Self) -> Option<T>

Takes the oldest value sent, or reports none.

Gotchas#

A send that has swapped itself in but not yet linked reports none until it links, so a caller that knows a value is coming retries.

Method sender#

pub fn sender(self: &Self) -> MailboxSender<T>

Returns a sender for this mailbox.

MailboxSender#

pub struct MailboxSender<T>

A sending end of a Mailbox. Any thread can hold one.

Details#

A sender is Transferable and Concurrent when T is Transferable.

Implementation MailboxSender<T>#

impl<T> MailboxSender<T>

Method send#

pub effect<'env> fn send<'env>(self: &Self, value: T) -> () ! OutOfMemoryError ? &mut Allocator

Sends value to the receiver.

Method clone#

pub fn clone(self: &Self) -> MailboxSender<T>

Adds another sender for the same mailbox.

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.