Documentation
¶
Overview ¶
Package conc provides basic primitives for concurrent programming.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Buffer ¶
type Buffer struct {
// contains filtered or unexported fields
}
Buffer is the non-generic engine behind a buffered Chan: a thread-safe FIFO of values stored by copy in a ring buffer. Each element is vsize bytes; Send copies from the source pointer into the ring and Recv copies out into the destination pointer. In most cases, using Chan is more convenient.
func NewBuffer ¶
NewBuffer creates a buffered channel holding up to size values of vsize bytes each.
func (*Buffer) Close ¶
func (ch *Buffer) Close()
Close marks the channel closed. Subsequent sends panic; receivers drain any buffered items and then return false. Closing a closed channel panics.
func (*Buffer) Free ¶
func (ch *Buffer) Free()
Free releases the channel's resources. The channel is unusable afterward. Call it once fully done; a channel may be drained after Close.
func (*Buffer) Recv ¶
Recv copies the next value from the channel into dst. Reports whether a value was received: false means the channel is closed and drained, and dst is left untouched.
dst must be a non-nil pointer to storage of at least vsize bytes.
func (*Buffer) RecvTimeout ¶
RecvTimeout copies the next value from the channel into dst, waiting up to d for one if the buffer is empty. A zero or negative d makes it non-blocking.
dst must be a non-nil pointer to storage of at least vsize bytes.
Returns Ok with dst filled, Timeout if the deadline passed while the buffer stayed empty, or Closed if the channel is closed and drained. dst is left untouched unless Ok is returned.
func (*Buffer) Send ¶
Send copies v into the channel, blocking while the channel is full (back-pressure). Panics if the channel is closed.
func (*Buffer) SendTimeout ¶
SendTimeout copies v into the channel, waiting up to d for room if the buffer is full. A zero or negative d makes it non-blocking.
Returns Ok if the value was stored, Timeout if the deadline passed while the buffer stayed full, or Closed if the channel is closed.
type Chan ¶
type Chan[T any] struct { // contains filtered or unexported fields }
Chan is a thread-safe FIFO channel, similar to Go's built-in `chan T`. It carries values by copy: Send copies its argument into the channel and Recv copies a value out, just like `chan T` in Go. To pass large payloads without copying, use a channel of pointers (Chan[*T]).
It supports buffered mode (created with n > 0) and unbuffered rendezvous mode (n == 0), where each send blocks until a receiver takes the value. Exactly one of the two backing engines is non-nil, chosen at creation time.
Example ¶
package main
import (
"solod.dev/so/conc"
"solod.dev/so/fmt"
"solod.dev/so/mem"
)
// producer carries the channel to send on and the
// number of values to produce.
type producer struct {
ch conc.Chan[int]
n int
}
// produce sends the numbers 0..n-1 on the channel, then closes it.
func produce(arg any) any {
p := arg.(*producer)
for i := 0; i < p.n; i++ {
p.ch.Send(i)
}
p.ch.Close()
return nil
}
func main() {
const n = 5
ch := conc.NewChan[int](mem.System, 2)
defer ch.Free()
// Run the producer on a single worker: it sends n
// values into the channel, then closes it.
prod := producer{ch: ch, n: n}
thr := conc.Go(produce, &prod, nil)
defer thr.Wait()
// Consume in the main thread until the channel is closed and drained.
var v int
for ch.Recv(&v) {
fmt.Printf("received %d\n", v)
}
}
Output:
func NewChan ¶
NewChan creates a channel of T backed by alloc. n is the buffer size: n > 0 makes it buffered, n == 0 makes it an unbuffered rendezvous channel. Call Chan.Free exactly once when done.
func (*Chan[T]) Close ¶
func (ch *Chan[T]) Close()
Close closes the channel. Subsequent sends panic; receivers drain remaining buffered values and then report false. Closing a closed channel panics.
Close is thread-safe and may run concurrently with Send and Recv but it must be called exactly once; a repeated or concurrent Close panics.
func (*Chan[T]) Free ¶
func (ch *Chan[T]) Free()
Free releases the channel's resources. The channel is unusable afterward. Free should only be called once; it's not thread-safe.
func (*Chan[T]) Recv ¶
Recv copies the next value into dst and reports whether one was received. It returns false when the channel is closed and no buffered values remain, in which case dst is left untouched.
Recv is thread-safe.
func (*Chan[T]) RecvTimeout ¶
RecvTimeout copies the next value into dst, waiting up to d for one. A zero or negative d makes it non-blocking. Returns Ok with dst filled, Timeout if the deadline passed first, or Closed if the channel is closed and drained. dst is left untouched unless Ok is returned.
RecvTimeout is thread-safe.
func (*Chan[T]) Send ¶
func (ch *Chan[T]) Send(v T)
Send copies v into the channel, blocking until there is room (buffered) or a receiver takes it (unbuffered). Sending on a closed channel panics.
Send is thread-safe.
func (*Chan[T]) SendTimeout ¶
SendTimeout copies v into the channel, waiting up to d for room (buffered) or a receiver (unbuffered). A zero or negative d makes it non-blocking.
Returns Ok if the value was sent, Timeout if the deadline passed first, or Closed if the channel is closed. Unlike Chan.Send, it does not panic on a closed channel.
A non-blocking send on an unbuffered channel reports Timeout unless a receiver is already parked and takes the value in the brief window it is offered.
SendTimeout is thread-safe.
type Pool ¶
type Pool struct {
// contains filtered or unexported fields
}
Pool is a bounded pool of worker threads with a wait queue which execute tasks of the form func(any).
A task is a named func(any), conventionally called with a pointer to its argument struct. The worker calls the function with that pointer, and the function reads inputs and writes results in place.
If all workers are busy, submitting a task puts it in a wait queue until a worker picks it up. If the queue is full, submitting a task blocks until a slot frees.
Tasks in a pool must be independent: a task must not block waiting on another task in the same pool, or the pool can deadlock. By convention, if a task can fail, it should report the error in an err field on its argument struct.
Example ¶
package main
import (
"solod.dev/so/conc"
"solod.dev/so/fmt"
"solod.dev/so/mem"
)
// Task carries one task's input and output.
type Task struct {
in int
out int
}
// square is a task function which squares
// its input and stores the result.
func square(arg any) {
task := arg.(*Task)
task.out = task.in * task.in
}
func main() {
const n = 10
tasks := make([]Task, n)
// Process tasks in parallel with a pool of two workers.
opts := conc.PoolOpts{NumThreads: 2}
pool := conc.NewPool(mem.System, opts)
defer pool.Free()
for i := range tasks {
tasks[i].in = i
pool.Go(square, &tasks[i])
}
pool.Wait()
// Print results after all tasks have finished.
for i := range tasks {
fmt.Printf("%d squared is %d\n", tasks[i].in, tasks[i].out)
}
}
Output:
func NewPool ¶
NewPool creates a pool with limit worker threads and starts them. limit must be >= 1. opts may be nil for default settings. Call Pool.Free exactly once when done:
p := conc.NewPool[Job](mem.System, 4, nil) defer p.Free() p.Go(work, &job1) p.Go(work, &job2) p.Wait()
func (*Pool) Free ¶
func (p *Pool) Free()
Free stops the pool and releases its resources. Any queued tasks are drained first: Free blocks until every submitted task finishes, then joins the workers. The pool is unusable afterward.
Free should only be called once; it's not thread-safe.
func (*Pool) Go ¶
Go submits a task for execution, blocking while the queue is full.
The worker invokes fn with the given arg; arg must point to storage that outlives the task. If the queue is full, Go blocks until a slot frees.
Unlike Go, a pool task returns nothing: there is no join to retrieve a value. Write results into the arg struct in place, and report errors via an err field on it.
Go is thread-safe.
type PoolOpts ¶
type PoolOpts struct {
NumThreads int // number of worker threads; must be >= 1
QueueSize int // task queue size; 0 = same as NumThreads
StackSize int // thread stack size in bytes; 0 = system default
}
PoolOpts holds the Pool settings.
type Rendezvous ¶
type Rendezvous struct {
// contains filtered or unexported fields
}
Rendezvous is the non-generic engine behind an unbuffered Chan: a thread-safe handoff with no buffer. Each send blocks until a receiver takes the value; the receiver copies vsize bytes straight from the sender's value into its destination with no staging buffer. In most cases, using Chan is more convenient.
The single handoff slot is owned by the sender for its whole lifetime: the sender sets full on publish and clears it only after observing claimed, so no other sender can enter (and reset claimed) while a sender is mid-handoff. This keeps the handshake unambiguous with any number of senders. Because the sender blocks until claimed, its value stays alive for the receiver to copy.
func NewRendezvous ¶
func NewRendezvous(alloc mem.Allocator, vsize int) *Rendezvous
NewRendezvous creates an unbuffered channel handing off values of vsize bytes.
func (*Rendezvous) Close ¶
func (ch *Rendezvous) Close()
Close marks the channel closed. A sender blocked on the handoff panics; a receiver with no pending value returns false. Closing a closed channel panics.
func (*Rendezvous) Free ¶
func (ch *Rendezvous) Free()
Free releases the channel's resources. The channel is unusable afterward.
func (*Rendezvous) Recv ¶
func (ch *Rendezvous) Recv(dst any) bool
Recv copies the published value into dst. It reports whether a value was received: false means the channel is closed with no pending value, and dst is left untouched.
dst must be a non-nil pointer to storage of at least vsize bytes.
func (*Rendezvous) RecvTimeout ¶
func (ch *Rendezvous) RecvTimeout(dst any, d time.Duration) Status
RecvTimeout copies a published value into dst, waiting up to d for a sender to publish one. A zero or negative d makes it non-blocking: it succeeds only if a sender is already parked on the handoff.
dst must be a non-nil pointer to storage of at least vsize bytes.
Returns Ok with dst filled, Timeout if the deadline passed first, or Closed if the channel is closed with no pending value. dst is left untouched unless Ok is returned.
func (*Rendezvous) Send ¶
func (ch *Rendezvous) Send(v any)
Send publishes v and blocks until a receiver takes it. Panics if the channel is closed before the handoff completes.
func (*Rendezvous) SendTimeout ¶
func (ch *Rendezvous) SendTimeout(v any, d time.Duration) Status
SendTimeout publishes v and waits up to d for a receiver to take it. A zero or negative d makes it non-blocking: the value is offered only for the instant before the deadline, so the send reports Timeout unless a receiver is already parked and takes it in that window.
Returns Ok if a receiver took the value, Timeout if the deadline passed first, or Closed if the channel is closed.
type Thread ¶
type Thread struct {
// contains filtered or unexported fields
}
Thread is a handle to a running OS thread. Start a thread with Go, then wait for completion with Thread.Wait, or hand its resources to the runtime with Thread.Detach.
func Go ¶
func Go(entry func(any) any, arg any, opts *ThreadOpts) Thread
Go launches an OS thread that runs fn(arg) and returns a handle to it. fn is a named function and arg must point to storage that outlives the thread: until Wait returns, or until fn returns for a detached thread. opts may be nil for default settings.
var job Job t := conc.Go(work, &job, nil) // ... do other work concurrently ... t.Wait() // job is complete once Wait returns
Always either Thread.Wait or Thread.Detach a thread started with Go, otherwise its resources will leak.
Unlike Go's goroutines, OS threads are not cheap to start. Prefer Pool for short-lived or numerous tasks; reserve Go for long-lived threads or a small, fixed number of threads you manage (join or detach) directly.
type ThreadOpts ¶
type ThreadOpts struct {
StackSize int // thread stack size in bytes; 0 = system default
}
ThreadOpts holds optional Thread settings. A nil *ThreadOpts selects defaults.
Source Files
Directories