Threads

Thread runs d6 code concurrently on native OS threads. Threads can use multiple CPU cores; Fiber instead provides cooperative concurrency within one OS thread.

Creating and joining a thread

Thread.new(callable, *args) starts a thread immediately. The callable must accept the supplied arguments and must not capture local state.

def square(n)
  n * n
end

left = Thread.new(square, 6)
right = Thread.new(square, 7)

left.join()   # => 36
right.join()  # => 49

join() blocks until the thread finishes and returns its result. Joining the same thread again returns the cached result. alive?() reports whether the thread is still running without blocking.

d6 does not support detached or fire-and-forget threads. If an unjoined Thread becomes unreachable, garbage collection waits for it to finish before reclaiming its resources, and a program that ends normally waits for every unjoined thread before exiting. A thread that is never joined has no one to re-raise its uncaught exception, so d6 prints it on stderr when the thread is reclaimed, as d6: runtime error in unjoined thread: ... with its backtrace. The exit status is unaffected, and a thread stopped by its supervisor is not reported. Join every thread whose failure should change what the program does, or run it under a Supervisor.

The one exception is a program that fails: if the main program ends with an uncaught exception while any spawned thread or supervised child is still running, d6 reports the error and exits with status 70 immediately, the way exit() does, rather than waiting for workers that may never finish (for example, one blocked sending to a channel only the failed program would have drained).

Isolated heaps

Every spawned thread has its own VM and heap. Threads do not share mutable objects, class variables, globals, open files, sockets, or other runtime resources. Arguments and return values cross the boundary as deep copies.

These values may cross a thread boundary:

• nil, booleans, integers, floats, strings, symbols, and bignums;
• arrays and hashes containing transferable values;
• instances whose fields contain transferable values;
• Regexp (compiled again in the receiving thread from its pattern and options);
• Time and Tensor (copied as plain data);
• zero-capture callables; and
• Channel (see Channels below) -- the one value that crosses by reference, not by deep copy: every thread that receives one gets its own handle onto the same underlying channel, not an independent copy.

Capturing closures and native-resource values such as Fiber, File, Listener, Socket, TLS socket, ProgramBuilder, and Thread cannot cross the boundary. Passing one raises TypeError.

The process permits at most 64 live spawned threads. Exceeding the limit raises ThreadError.

Per-thread initialization

Runtime mutations made before Thread.new are not copied into the new heap. In particular, a class variable configured in the parent remains unset in the child. Initialize repositories, clients, and other mutable services separately in each worker; a lazy per-thread accessor is usually the simplest pattern.

This also applies to multi-threaded Gremlin servers: startup configuration in the calling thread does not configure every request worker. Program definitions are available in every thread, so top-level methods, classes, and modules can be called normally. It is their mutable runtime state that remains isolated.

Module-level constants are the exception that makes the above workable. A spawned thread never runs the program's top-level statements, which is what sets constants, so Thread.new hands the thread a deep copy of every constant already defined when it is called (including a Regexp, which is compiled again). A constant defined after Thread.new, or whose value cannot cross the boundary (a File, a Socket, ...), is unset in that thread, and reading it there raises uninitialized constant. The copy is the thread's own, like an argument: mutating a constant's array in one thread does not show in another. A Supervisor child gets the constants defined when add_child ran, a fresh copy for every restart.

Results and exceptions

If a worker raises an uncaught d6 exception, join() re-raises that exception in the joining thread:

def fail_work()
  raise ArgumentError.new("bad job")
end

worker = Thread.new(fail_work)

begin
  worker.join()
rescue error: ArgumentError
  puts(error.message())
end

An internal VM failure is reported as ThreadError at join().

Standard input and output are process-wide native streams. Concurrent writes are memory-safe but may interleave, so applications should serialize output when ordering matters. The OS signal disposition is process-wide, while Signal.trap closures live in their installing VM. A VM without a matching trap leaves the pending signal for one that has it. If multiple VMs trap the same signal, one eligible VM claims each pending delivery; keep signal handling in one owner and broadcast cancellation through Channels when all workers need to stop.

Channels

Channel is a bounded, thread-safe mailbox: one thread sends, another receives, both blocking as needed. Unlike a plain return value, a channel lets threads exchange values throughout their whole lifetime, not just once at join().

def producer(ch)
  ["first", "second", "third"].each() do |item|
    ch.send(item)
  end
  ch.close()
end

ch = Channel.new(4)  # capacity: 0..1_000_000 (0 is a handoff, below)
worker = Thread.new(producer, ch)

loop
  item = ch.receive()
  break if item == nil
  puts(item)
end

worker.join()

send(value) blocks while the channel is full; receive() blocks while it's empty. Both unblock the moment the channel is close()d: a blocked send raises IOError, and a blocked receive returns nil once every already- queued value has been drained -- the same "nothing left, ever" signal File#read's own EOF-as-nil convention already uses, distinct from "wait a bit and it might arrive" (see try_send/try_receive below).

try_send(value)/try_receive() are the non-blocking pair: instead of waiting, a full send/empty receive raises WouldBlockError immediately -- the same convention non-blocking socket reads/writes already use. Combine these with readiness waits below for timeout- or signal-responsive waiting; a plain blocking receive (like Thread#join) does not service Signal.trap handlers while parked.

wait_readable(cancellations, deadline) and wait_writable(cancellations, deadline) wait for a readiness hint without consuming a value or reserving capacity. cancellations must be an Array of Channels (duplicates allowed); closing any of them wakes the wait. deadline is a finite Float absolute Time.monotonic() timestamp in seconds, or nil for no deadline. Both methods return nil on readiness, target closure, cancellation-source closure, deadline expiry, or a spurious wakeup. Recheck cancellation/deadlines and retry the nonblocking operation after each return. A VM with installed signal handlers returns periodically to dispatch them. Most callers should use the cancellation cut instead of managing these low-level readiness hints themselves.

close() is idempotent -- closing an already-closed channel is a no-op, not an error. closed?() and size() report status without blocking.

Rendezvous channels: Channel.new(0)

A capacity of 0 makes a handoff. send(value) returns only once a receiver has taken the value, so the two threads meet and the sender knows the value was received, not just queued.

ack = Channel.new(0)
worker = Thread.new(do_work, ack)
ack.send(:ready)      # blocks until the worker's receive() has the value

• A blocked send that is interrupted (the channel is closed, a deadline or cancellation fires) withdraws its value: it was not delivered. close() raises IOError in a sender whose value nobody took.
• try_send succeeds only while a blocking receive is parked on the channel, and returns without waiting for it to pick the value up. Otherwise it raises WouldBlockError. wait_writable is likewise ready only while a receiver is parked.
• receive, try_receive and Channel.select take from a rendezvous channel like any other. Channel.select_send raises TypeError for one: choosing among channels cannot promise that the receiver has the value.
• size() is always 0; nothing is buffered.

Waiting on several channels: Channel.select

Channel.select(channels, deadline = nil) receives from whichever of several channels has a value ready, like Go's select with only receive cases. It returns [index, value] -- the position of the channel in channels and the value taken from it -- and blocks until one is ready.

orders = Channel.new(8)
cancels = Channel.new(1)

loop
  picked = Channel.select([cancels, orders])
  break if picked == nil
  index, value = picked[0], picked[1]
  break if index == 0          # a cancellation arrived
  puts("order: #{value}")
end

• When several channels are ready at once, the first one in array order wins, so list the channel that should take priority (a cancellation or shutdown channel) first. Selection is deterministic, not random as in Go.
• Closing a channel wakes a blocked select. A closed channel is still selected until its queued values are drained; after that it is skipped.
• It returns nil when nothing will arrive: the deadline (an absolute Time.monotonic() Float, as for wait_readable) passed, or every channel is closed and drained. closed? on the channels tells the two apart.
• Only one value is taken per call, under that channel's own lock, so several threads may select over the same channels without losing or duplicating values.
• channels must be a non-empty Array of Channels and deadline a finite Float or nil; anything else raises TypeError. Like a plain receive, a blocking select does not service Signal.trap handlers while parked -- pass a short deadline and loop if you need to.

Sending to whichever channel has room: Channel.select_send

Channel.select_send(channels, value, deadline = nil) is the send-side twin: it sends value to the first channel (in array order) with room and returns that channel's index, blocking until one has room.

primary = Channel.new(4)
overflow = Channel.new(64)

where = Channel.select_send([primary, overflow], job)
puts("queued on the overflow channel") if where == 1

• When several channels have room, the first one in array order wins, so list the preferred destination first.
• A closed channel is skipped, not an error (unlike send on a closed channel). Closing a channel wakes a blocked select_send.
• It returns nil when nothing can accept the value: the deadline passed, or every channel is closed. closed? on the channels tells the two apart.
• The value is copied only into the channel it was sent to, under that channel's own lock, so a value that cannot cross a thread boundary raises TypeError exactly as send does.
• channels must be a non-empty Array of Channels and deadline a finite Float or nil; anything else raises TypeError. As with select, a parked call does not service Signal.trap handlers.

A Channel itself can be sent through another channel, passed as a Thread.new argument, or stored in an array/hash/instance that crosses a thread boundary -- every recipient shares the same channel (see the note in Isolated heaps above); only the values passed through it are ever deep-copied. A value unsupported for cross-thread transfer (a capturing closure, another native-resource handle) raises TypeError on send/try_send, exactly as passing one to Thread.new already does.

Cooperative cancellation

The experimental cancellation cut provides shared cancellation tokens, monotonic deadlines, cancellation-aware channel waits, and task scopes. It uses explicit checkpoints and native channel wakeups with monotonic deadlines. It does not change blocking native methods or inject exceptions into threads. See the design and limitations and persistent job service.

Supervisors

Supervisor restarts a worker automatically when it crashes -- an uncaught exception or an internal VM failure -- instead of just ending it the way a plain Thread would. By default it restarts only the crashed worker (Erlang's one_for_one strategy); a crash in one child never affects its siblings. Supervisor.new(:one_for_all) and Supervisor.new(:rest_for_one) choose the other two strategies -- see Restart strategies.

def fetch_loop(url)
  loop do
    fetch_and_process(url)  # may raise on a transient network error
  end
end

sup = Supervisor.new()
sup.add_child(fetch_loop, "https://example.com/feed")

add_child(callable, *args) starts a supervised worker immediately and returns its index (an Int, 0, 1, 2, ...) for later use with restart_count/last_error/alive?. The callable rules are identical to Thread.new's: it must accept the supplied arguments and must not capture local state. Each worker gets its own isolated heap per attempt, exactly like a Thread -- nothing is shared between one crash and the next restart except whatever was originally passed to add_child (re-copied fresh into each attempt's own heap, so a Channel argument stays the same channel across restarts, but a plain Array argument does not accumulate mutations from a previous, crashed attempt).

Generic singleton references with no captured value cells may carry immutable type bindings from their factory. Threads copy these binding graphs along with Callable workers, arguments, and results. A supervisor keeps its worker's saved bindings across attempts rather than inferring replacements from restart arguments.

A clean, non-raising return ends that child for good -- v1 restarts on crash only, never on an ordinary return. By default there is a fixed 20ms delay between a crash and the next restart attempt, a safety valve bounding CPU use if a worker crashes immediately every time, and a child that crashes forever is restarted forever; see Restart policy to bound both. A Supervisor accepts at most 32 children, raising SupervisorError past that.

sup.restart_count(0)  # => how many times child 0 has crashed and restarted
sup.last_error(0)     # => the most recent crash's message, or nil if it
                       #    has never crashed
sup.alive?(0)         # => true while child 0 is still running/restarting;
                       #    false once it has returned cleanly or been
                       #    stopped

stop() prevents any further restarts and blocks until every child's current attempt finishes -- there is no implicit native cancellation, so a worker parked in a blocking call (a Channel#receive, a socket read, an infinite loop with no exit condition) keeps stop() waiting until that call itself returns or raises. join() is the non-stopping counterpart: it blocks until every child has finished on its own, without disabling restarts -- a child that keeps crashing and restarting forever blocks join() forever too, exactly as joining a Thread that never returns already does. With the cancellation cut, cancel a shared source first and have workers catch expected cancellation and return normally before calling stop().

stop(timeout_ms) bounds that wait. It stops restarts, gives the current attempts up to timeout_ms milliseconds (0 to 2147483647) to finish by themselves, then interrupts whatever is still running and joins. It returns true if every child finished within the timeout and false if it had to interrupt one. An interrupted child has alive? false, no restart counted and a last_error of "stopped: Supervisor#stop timed out"; stop(0) interrupts immediately.

sup.stop(5000)  # => true if everything wound down within 5 seconds

The interruption is the one a sibling restart uses: the attempt's VM stops at its next instruction and cannot be rescued, so ensure blocks do not run. It ends an infinite loop or a crash loop, and it also wakes a worker parked in a blocking native call: Channel#receive/#send/#wait_readable, IO.poll (and so a sleep built on it), Thread#join, TCPServer#accept, a read from a socket (read, gets), UDPSocket#receive, a TCP or TLS connect (including its host-name lookup), a TLS handshake (client or accept), a TLS read or write, Process.run (the child is killed) and a SQLite, PostgreSQL or MySQL/MariaDB query (cancelled with sqlite3_interrupt / PQcancel, or by shutting the connection's socket down, which leaves that connection unusable). Such a wait sleeps in slices of 20 ms, so the interruption lands within about that long. A thread the interrupted worker started is interrupted too, instead of being waited for.

MySQL.open's host-name lookup, connect and handshake are interrupted too: the connector runs on a helper thread, and an abandoned attempt cleans itself up when the connector returns. Reads from a pipe or device the worker opened itself (a named pipe, say) are interrupted, as is gets() on standard input. Native stdin reads serialize access and wait for each byte without changing the inherited descriptor's flags. If an interrupted reader consumed part of a line, the next gets() finishes that line. A worker waiting behind another stdin reader can also be interrupted. No input helper is left running after the worker stops. Code running in the opt-in JIT is interrupted at the next loop iteration (a backward jump or a function's own tail call). Native Array block loops and C-driven iterators also check before entering each compiled block, even if the block itself has no loop. A compiled function with no such polling point and no blocking call runs to its end. Use cancellation tokens for those waits.

Restart policy

restart_policy(options) bounds how hard a supervisor retries. It takes a Hash with String keys, all optional, and must be called before the first add_child (a later call raises SupervisorError):

sup = Supervisor.new()
sup.restart_policy({"max_restarts": 5, "within_ms": 10000,
                    "backoff_ms": 50, "max_backoff_ms": 2000})

• max_restarts (0 to 100) and within_ms (default 5000): the supervisor gives up when more than max_restarts crashes, counted across all of its children, fall inside a sliding within_ms window. Without max_restarts there is no limit, as before. Restarts a sibling's crash caused do not count.
• backoff_ms (default 20) and max_backoff_ms (default backoff_ms, so no growth): the pause before a crashed child restarts starts at backoff_ms and doubles with each consecutive crash of that child up to max_backoff_ms. A child whose attempt stays up longer than max_backoff_ms starts over at backoff_ms.

When the supervisor gives up it stops restarting anything and interrupts its other children the way a sibling restart does (cooperatively, so a worker parked in a blocking call is stopped once that call returns). Those children get alive? false and a last_error of "stopped: the supervisor's restart intensity was exceeded". failed?() is then true, add_child raises SupervisorError, and join() raises SupervisorError with the crash count, the window and the last error, which is how a program waiting on a supervisor learns it failed. stop() never raises. restart_count does not count the crash that made it give up, since nothing was restarted after it.

Invalid options raise TypeError: an unknown key, a value that is not an Int in range, or a max_backoff_ms below backoff_ms.

Restart strategies

Supervisor.new(strategy) takes an optional strategy symbol:

• :one_for_one (the default, same as Supervisor.new()): only the crashed child restarts.
• :one_for_all: when any child crashes, every other child that is still running restarts too.
• :rest_for_one: when a child crashes, the children added after it restart too; earlier children are untouched.

Any other value raises TypeError. Children that already finished with a clean return stay finished under every strategy.

A sibling is restarted by interrupting its current attempt: its VM stops at the next instruction and the attempt is discarded (the interruption is not an exception, so a rescue cannot swallow it). Like stop(), this is cooperative -- a sibling parked in one of the native waits listed under stop(timeout_ms) is woken within about 20 ms, one parked in a call that list leaves out is restarted only once that call returns, and code running in the opt-in JIT is interrupted at its next loop iteration. A restarted sibling's restart_count goes up and its last_error becomes "restarted: a sibling crashed"; it restarts without the 20ms crash delay, since the attempt was not failing on its own.

A Supervisor cannot cross a Thread.new/Channel boundary -- passing one raises TypeError, same as Thread, File, and the other native- resource handles in Isolated heaps above. Combined with add_child's own no-capture rule, this means a supervised child can never obtain a handle back to the Supervisor that spawned it.

Genuine supervision trees -- a supervisor whose own children are themselves supervised -- need no special nesting support: a supervised child is just an ordinary closure running on its own thread, and that closure is free to create and manage its own Supervisor internally:

def managed_subsystem(config)
  inner = Supervisor.new()
  inner.add_child(fetch_loop, config)
  inner.add_child(process_loop, config)
  inner.join()  # this outer worker "crashes" only if the whole subsystem does
end

sup = Supervisor.new()
sup.add_child(managed_subsystem, options)

Implementation details, copy semantics, and lifecycle invariants are documented in Concurrency internals.

Testing threaded code

Threaded bugs are timing bugs, and an idle many-core machine hides most of them. Three checks exist for that:

• make test-tsan builds with ThreadSanitizer (it cannot combine with ASan/UBSan, so it is its own build) and runs the Thread, Channel and Supervisor cases once each.
• make test-stress-threads re-runs every case that uses Thread, Channel or Supervisor many times, each run pinned to two CPUs under a timeout. Pinning makes the scheduler interleave threads far more unevenly than an idle machine does; it found the Supervisor#add_child race that hung CI about once in 40 runs. STRESS_RUNS, STRESS_CPUS, STRESS_TIMEOUT and STRESS_FILTER tune it, and a failure prints the expected and actual output.
• make test-stress-threads-tsan is the same loop against the TSan build.

When a test of your own waits for threads, wait for the effect you assert on, not a neighbouring one: a supervisor counts a restart before the new thread runs its first line, so a test that waits for one sibling's restart can see another's still pending. CI runs all three, the stress loop on arm64 as well.