takt

English | 简体中文 | Español | 日本語 | Français

takt

An abstract completion-driven dispatch engine for non-blocking I/O stacks.

Overview

In a proactor model, I/O operations are submitted to the kernel and their completions arrive asynchronously. The application must correlate each completion back to the computation that requested it, resume that computation, and handle success, progress with a live frontier, no-progress readiness/backpressure, and failure.

takt provides this execution model as an abstract layer over the kont effect system. A Dispatcher evaluates one algebraic effect at a time, classifying the result according to the iox outcome algebra. A Backend submits operations to an asynchronous engine, for example io_uring, and polls for completions. The Loop ties them together: it submits computations, polls the backend, correlates completions by token, and resumes suspended continuations.

Two equivalent APIs are available: kont.Eff (closure-based, straightforward to compose) and kont.Expr (frame-based, with lower allocation overhead on hot paths).

The event-loop path stores one pending suspension per live token. Each token tracks a suspension produced by kont.StepExpr (or by reifying kont.Eff first), so Backend.Submit must not reuse a token while the older submission carrying it remains live in the loop.

For stream or multishot-style integrations, SubscriptionLoop provides a separate abstract route runner. It tracks Subscription handles by RouteID (Token plus generation), polls StreamCompletion values, emits StreamEvent observations, and keeps More as route-liveness evidence independent of payload value or payload error. This keeps generic Loop affine and one-shot while still giving concrete runtimes a principled place to represent same-operation successor observations.

Composition Boundary

takt owns execution movement, not context meaning or outcome vocabulary. iox classifies nil, ErrWouldBlock, ErrMore, and failure; kont owns the suspension/resumption carrier; cove may wrap a suspension with explicit context through SuspensionView; takt advances any value that satisfies SuspensionLike without interpreting that context.

Installation

go get code.hybscloud.com/takt

Requires Go 1.26 or later.

Outcome Classification

Each dispatched operation yields an iox outcome. Dispatcher.Dispatch reports that outcome as (value, error); blocking runners and the stepping API interpret it as follows:

Usage

Dispatcher

A Dispatcher maps each algebraic effect to a concrete I/O operation and returns the result together with an iox outcome.

type myDispatcher struct{ /* ... */ }

func (d *myDispatcher) Dispatch(op kont.Operation) (kont.Resumed, error) {
	// dispatch op; return (value, nil), (value, iox.ErrMore), or (nil, iox.ErrWouldBlock)
}

Blocking Evaluation

Exec and ExecExpr run a computation to completion, synchronously waiting when the dispatcher yields iox.ErrWouldBlock.

result := takt.Exec(d, computation) // kont.Eff
result := takt.ExecExpr(d, exprComputation) // kont.Expr

Stepping

For proactor event loops, for example io_uring, Step and Advance evaluate one effect at a time. When the dispatcher yields iox.ErrWouldBlock, the suspension is returned to the caller so the event loop can reschedule it.

The manual stepping outcome law is explicit: if d.Dispatch(susp.Op()) returns nil or iox.ErrMore, Advance resumes the suspension with the returned value and returns the next suspension plus the original error; if it returns iox.ErrWouldBlock or an ordinary failure, Advance returns the zero result, the original suspension, and that error. ErrMore is therefore a progress signal in manual stepping, while completion-level ErrMore remains unsupported by generic Loop because the backend operation is still live.

result, susp := takt.Step[int](exprComputation)
if susp != nil {
	var err error
	result, susp, err = takt.Advance(d, susp)
	if iox.IsWouldBlock(err) {
		return susp // yield to the event loop and reschedule when ready
	}
}
// result holds the final value

Error Handling

Compose dispatcher operations with error effects. Throw short-circuits the computation eagerly and discards the pending suspension.

either := takt.ExecError[string](d, computation)
// Right on success, Left on Throw

// Stepping with errors
either, susp := takt.StepError[string, int](exprComputation)
if susp != nil {
	var err error
	either, susp, err = takt.AdvanceError[string](d, susp)
	if iox.IsWouldBlock(err) {
		return susp // yield to the event loop and reschedule when ready
	}
}

Event Loop

A Loop drives computations through a Backend. It submits operations, polls for completions, correlates them by Token, and resumes suspended continuations. NewLoop accepts functional Options. WithMaxCompletions(n) panics when n <= 0 with takt: WithMaxCompletions requires n > 0; WithMemory(nil) panics with takt: WithMemory requires a non-nil CompletionMemory.

Backend.Poll([]Completion) (int, error) reports both the number of ready completions and any infrastructure poll failure; a backend must not return ready completions together with a non-nil poll-level error. Loop treats iox.ErrWouldBlock returned by Poll as an idle tick rather than a terminal error.

Loop is a single-owner runner. Serialize calls that share the same Loop, including SubmitExpr, Submit, Poll, Run, Drain, Pending, and Failed.

Backend.Submit must return a token that is unique among all submissions still live in the loop. Tokens are correlation keys, not sequence numbers; a concrete backend may use a kernel user_data value directly. If a backend reuses a live token, the loop records ErrLiveTokenReuse, drains every pending suspension exactly once, and every subsequent SubmitExpr / Submit / Poll / Run call returns that fatal error.

When a completion carries iox.ErrWouldBlock, the loop resubmits the same operation. If a completion carries iox.ErrMore (multishot), the loop records ErrUnsupportedMultishot, drains every pending suspension exactly once, and every subsequent SubmitExpr / Submit / Poll / Run call returns that fatal error. ErrMore means the submitted backend operation remains active after the CQE, while generic Loop has no subscription or cancel carrier for later same-token completions.

Loop.Failed() reports the recorded fatal error (or nil). Loop.Drain() forces the loop into a disposed state, discards every pending suspension exactly once, and records ErrDisposed only if no fatal was previously set; it is idempotent and preserves the original fatal error when one already exists.

loop := takt.NewLoop[*myBackend, int](backend, takt.WithMaxCompletions(64))

// Submit computations
loop.SubmitExpr(exprComputation1)
loop.SubmitExpr(exprComputation2)
loop.Submit(contComputation) // kont.Eff

// Drive all to completion
results, err := loop.Run()

Stream / Subscription Runner

SubscriptionLoop is the sibling runner for route-indexed stream observations. A SubscriptionBackend starts an operation with Subscribe, polls StreamCompletion values, and accepts Cancel requests for live routes. RouteID pairs a Token with a generation so token reuse after finalization does not alias an older live route.

StreamCompletion.More means the same route remains live after the observation. HasValue and EventErr describe payload evidence at that boundary and are independent of More: a completion can report a value and more to come, no value and more to come, or a payload error while the route remains live. A completion with More == false emits a final StreamEvent and retires the route.

Subscribe rejects the zero RouteID with ErrInvalidRouteID and live-route aliasing with ErrLiveRouteReuse; both conditions put the stream loop in a fatal state. Poll-level iox.ErrWouldBlock is an idle tick. A SubscriptionBackend must not return ready stream completions together with a non-nil poll-level error; payload errors belong in StreamCompletion.EventErr. Unknown or already retired route completions are stale observations and are ignored. Cancel requests route cancellation without retiring the route immediately; a terminal completion, Drain, or a fatal loop transition retires it.

type myStreamBackend struct{ /* ... */ }

func (b *myStreamBackend) Subscribe(op kont.Operation) (takt.RouteID, error) {
	return takt.NewRouteID(tok, generation), nil
}

func (b *myStreamBackend) Poll(out []takt.StreamCompletion[int]) (int, error) {
	// Fill out with route-indexed observations.
	return n, nil
}

func (b *myStreamBackend) Cancel(id takt.RouteID) error {
	// Request cancellation of the live route.
	return nil
}

stream := takt.NewSubscriptionLoop[*myStreamBackend, int](
	backend,
	takt.WithMaxStreamCompletions(64),
)

sub, err := stream.Subscribe(op)
events, err := stream.Poll()
_ = sub
_ = events
_ = err

NewLoop uses HeapMemory as the default completion-buffer provider. Pass BoundedMemory via WithMemory when completion buffers should come from a bounded steady-state pool of default-sized 128 KiB slabs; or supply a custom CompletionMemory implementation to control allocation strategy without widening the Backend or Completion contracts. Custom providers must return exclusive non-overlapping live slabs and may treat Release as ownership transfer back to the provider:

// Default: HeapMemory + default-sized completion slab.
loop := takt.NewLoop[*myBackend, int](backend)

// Cap the per-poll completion slab length (the provider still chooses the slab shape; Loop trims the visible length back to this cap).
loop = takt.NewLoop[*myBackend, int](backend, takt.WithMaxCompletions(64))

// Share one HeapMemory's sync.Pool across several Loops by passing the same address to WithMemory; copying a HeapMemory value would not share recycled slabs.
heap := &takt.HeapMemory{}
loopA := takt.NewLoop[*myBackend, int](backend, takt.WithMemory(heap))
loopB := takt.NewLoop[*myBackend, int](backend, takt.WithMemory(heap))

// BoundedMemory: one bounded pool of default-sized 128 KiB slabs. WithPoolCapacity tunes that pool's capacity (rounded up to the next power of two by iobuf).
bounded := takt.NewBoundedMemory(takt.WithPoolCapacity(4))
loop = takt.NewLoop[*myBackend, int](
	backend,
	takt.WithMemory(bounded),
	takt.WithMaxCompletions(64),
)

API Overview

Dispatch

• Dispatcher[D Dispatcher[D]]: non-blocking dispatch interface
• Exec[D, R](d D, m kont.Eff[R]) R: blocking evaluation of kont.Eff
• ExecExpr[D, R](d D, m kont.Expr[R]) R: blocking evaluation of kont.Expr

Stepping

• SuspensionLike[S, R]: resumable interface (Op + Resume), implemented by cove.SuspensionView
• Step[R](m kont.Expr[R]) (R, *kont.Suspension[R]): evaluate until the first suspension
• AdvanceSuspension[D, S, R](d D, susp S) (R, S, error): dispatch one operation through any SuspensionLike value
• Advance[D, R](d D, susp *kont.Suspension[R]) (R, *kont.Suspension[R], error): dispatch one operation

Error Handling

• ExecError[E, D, R](d D, m kont.Eff[R]) kont.Either[E, R]: blocking evaluation with errors
• ExecErrorExpr[E, D, R](d D, m kont.Expr[R]) kont.Either[E, R]: error-aware evaluation of kont.Expr
• StepError[E, R](m kont.Expr[R]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]]): stepping with errors
• AdvanceError[E, D, R](d D, susp *kont.Suspension[kont.Either[E, R]]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]], error): advance one step with errors

Backend and Event Loop

• Backend[B Backend[B]]: asynchronous submit/poll interface
• CompletionMemory: loop-local completion-buffer provider
• HeapMemory: default implementation (sync.Pool-backed typed slabs of the default size)
• BoundedMemory: iobuf-backed implementation with a single bounded pool of default-sized 128 KiB slabs
• Option: functional options for NewLoop (WithMemory, WithMaxCompletions)
• CompletionBufOption: functional options for CompletionMemory.CompletionBuf (WithSize)
• BoundedMemoryOption: functional options for NewBoundedMemory (WithPoolCapacity)
• Token: submission-completion correlation (uint64)
• Completion: {Token, Value kont.Resumed, Err error}
• NewLoop[B, R](b B, opts ...Option) *Loop[B, R]: create an event loop (default HeapMemory, default-sized slab)
• (*Loop[B, R]).SubmitExpr(m kont.Expr[R]) (R, bool, error): step and submit Expr
• (*Loop[B, R]).Submit(m kont.Eff[R]) (R, bool, error): step and submit Cont
• (*Loop[B, R]).Poll() ([]R, error): poll and dispatch completions
• (*Loop[B, R]).Run() ([]R, error): drive all to completion
• (*Loop[B, R]).Pending() int: count pending operations
• (*Loop[B, R]).Failed() error: terminal fatal error, or nil
• (*Loop[B, R]).Drain() int: discard pending suspensions and dispose the loop
• ErrLiveTokenReuse: backend reused a token that was still live in the loop
• ErrUnsupportedMultishot: multishot completion is unsupported by generic Loop
• ErrDisposed: loop has been disposed via Drain

Stream Routes

• RouteID: stream route identity (Token plus generation)
• NewRouteID(token Token, generation uint64) RouteID: construct a route identifier for stream backends
• RouteID.Token() Token: token component
• RouteID.Generation() uint64: generation component
• RouteID.IsZero() bool: reserved invalid-route test
• Subscription[A]: opaque live stream handle
• Subscription[A].ID() RouteID: route identity carried by the handle
• Subscription[A].IsZero() bool: zero, non-live handle test
• StreamCompletion[A]: {ID, Value, HasValue, EventErr, More}
• StreamCompletion[A].RouteOutcome() iox.Outcome: projection to iox outcome vocabulary
• StreamEvent[A]: {Subscription, Value, HasValue, Final, EventErr}
• SubscriptionBackend[B, A]: abstract stream backend interface (Subscribe, Poll, Cancel)
• SubscriptionOption: functional options for NewSubscriptionLoop
• WithMaxStreamCompletions(n): cap the per-poll stream completion buffer
• NewSubscriptionLoop[B, A](b B, opts ...SubscriptionOption) *SubscriptionLoop[B, A]: create a stream runner
• (*SubscriptionLoop[B, A]).Subscribe(op kont.Operation) (Subscription[A], error): start a route-producing operation
• (*SubscriptionLoop[B, A]).Poll() ([]StreamEvent[A], error): poll route-indexed events
• (*SubscriptionLoop[B, A]).Cancel(sub Subscription[A]) error: request route cancellation
• (*SubscriptionLoop[B, A]).Pending() int: count live stream routes
• (*SubscriptionLoop[B, A]).Failed() error: terminal fatal error, or nil
• (*SubscriptionLoop[B, A]).Drain() int: cancel or retire owned routes and dispose the stream loop
• ErrLiveRouteReuse: backend reused a route that was still live
• ErrInvalidRouteID: backend returned the reserved zero RouteID
• ErrUnknownSubscription: subscription handle is not live in the stream loop

Bridge

• Reify[A](kont.Eff[A]) kont.Expr[A]: Cont → Expr
• Reflect[A](kont.Expr[A]) kont.Eff[A]: Expr → Eff

Practical Recipes

A complete event-loop integration combines a Dispatcher (the synchronous semantics) with a Backend (the proactor) under one Loop:

// 1. Define the dispatcher: maps an effect operation to an iox outcome.
type myDispatcher struct{ /* ... */ }

func (d *myDispatcher) Dispatch(op kont.Operation) (kont.Resumed, error) {
	// Return (value, nil), (value, iox.ErrMore), or (nil, iox.ErrWouldBlock).
}

// 2. Define the backend: submits ops to the OS proactor and polls completions.
type myBackend struct{ /* ... */ }
func (b *myBackend) Submit(op kont.Operation) (takt.Token, error) { /* ... */ }
func (b *myBackend) Poll(out []takt.Completion) (int, error)      { /* ... */ }

// 3. Drive: submit one or more computations, then Run to completion.
loop := takt.NewLoop[*myBackend, int](backend, takt.WithMaxCompletions(64))
loop.SubmitExpr(prog1)
loop.SubmitExpr(prog2)
results, err := loop.Run()
_ = results; _ = err

For error-aware composition use ExecError / StepError / AdvanceError in place of their non-error counterparts; Throw short-circuits the in-flight computation while leaving sibling computations on the same loop unaffected. The fused dispatcher+backend pattern shown here is the one used by sess to attach a session endpoint to a real I/O runtime.

References

• Tarmo Uustalu and Varmo Vene. 2008. Comonadic Notions of Computation. Electronic Notes in Theoretical Computer Science 203, 5 (June 2008), 263–284. https://doi.org/10.1016/j.entcs.2008.05.029
• Gordon D. Plotkin and Matija Pretnar. 2009. Handlers of Algebraic Effects. In Proc. 18th European Symposium on Programming (ESOP '09). LNCS 5502, 80–94. https://doi.org/10.1007/978-3-642-00590-9_7
• Andrej Bauer and Matija Pretnar. 2015. Programming with Algebraic Effects and Handlers. Journal of Logical and Algebraic Methods in Programming 84, 1 (Jan. 2015), 108–123. https://arxiv.org/abs/1203.1539
• Daniel Leijen. 2017. Type Directed Compilation of Row-Typed Algebraic Effects. In Proc. 44th ACM SIGPLAN Symposium on Principles of Programming Languages (POPL '17). 486–499. https://doi.org/10.1145/3009837.3009872
• Danel Ahman and Andrej Bauer. 2020. Runners in Action. In Proc. 29th European Symposium on Programming (ESOP '20). LNCS 12075, 29–55. https://arxiv.org/abs/1910.11629
• Daniel Hillerström, Sam Lindley, and Robert Atkey. 2020. Effect Handlers via Generalised Continuations. Journal of Functional Programming 30 (2020), e5. https://bentnib.org/handlers-cps-journal.pdf

License

MIT License. See LICENSE for details.

©2026 Hayabusa Cloud Co., Ltd.