takt

README

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

takt

Abstract completion event 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 the full range of outcomes — success, partial progress, backpressure, and failure.

takt provides this dispatch algebra 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 (e.g., io_uring) and polls for completions. The Loop event loop ties them together: it submits computations, polls the backend, correlates completions by token, and resumes the suspended continuations.

Two equivalent APIs: Cont (closure-based, straightforward composition) and Expr (frame-based, lower allocation overhead in hot paths).

Installation

go get code.hybscloud.com/takt

Requires Go 1.26+.

Outcome Classification

Every dispatched operation returns an iox outcome. The dispatcher and stepping API handle each case:

Outcome	Meaning	Dispatcher	Stepping API
nil	completed	resume	resume, return nil
ErrMore	progress, more expected	resume	resume, return ErrMore
ErrWouldBlock	no progress	wait	return suspension to caller
other	infrastructure failure	panic	return error to caller

Usage

Dispatcher

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

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

func (d *myDispatcher) Dispatch(op kont.Operation) (kont.Resumed, error) {
    // dispatch op, return (value, nil) 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)         // Cont-world
result := takt.ExecExpr(d, exprComputation) // Expr-world

Stepping

For proactor event loops (e.g., io_uring), Step and Advance evaluate one effect at a time. When the dispatcher yields iox.ErrWouldBlock, the suspension is returned to the caller, letting the event loop reschedule.

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 event loop, reschedule when ready
    }
}
// result is the final value

Error Handling

Compose dispatcher effects with error effects. Throw eagerly short-circuits the computation 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 event loop, reschedule when ready
    }
}

Event Loop

A Loop drives computations through a Backend. It submits operations, polls for completions, correlates them by Token, and resumes the suspended continuations. maxCompletions in NewLoop must be greater than zero.

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

// Submit computations
loop.SubmitExpr(exprComputation1)
loop.SubmitExpr(exprComputation2)
loop.Submit(contComputation) // Cont-world

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

API Overview

Dispatch

• Dispatcher[D Dispatcher[D]] — F-bounded dispatch interface
• Exec[D, R](d D, m kont.Eff[R]) R — blocking Cont-world evaluation
• ExecExpr[D, R](d D, m kont.Expr[R]) R — blocking Expr-world evaluation

Stepping

• Step[R](m kont.Expr[R]) (R, *kont.Suspension[R]) — evaluate to first suspension
• 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 with errors
• ExecErrorExpr[E, D, R](d D, m kont.Expr[R]) kont.Either[E, R] — Expr-world with errors
• StepError[E, R](m kont.Expr[R]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]]) — step 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 with errors

Backend and Event Loop

• Backend[B Backend[B]] — F-bounded async submit/poll interface
• Token — submission-completion correlation (uint64)
• Completion — {Token, Value kont.Resumed, Err error}
• NewLoop[B, R](b B, maxCompletions int) *Loop[B, R] — create event loop (maxCompletions > 0)
• (*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

Bridge

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

References

• G. D. Plotkin and M. Pretnar. "Handlers of Algebraic Effects." In Proc. ESOP, 2009.
• T. Uustalu and V. Vene. "Comonadic Notions of Computation." In ENTCS 203(5), 2008.
• D. Ahman and A. Bauer. "Runners in Action." In Proc. ESOP, 2020.

License

MIT License. See LICENSE for details.

©2026 Hayabusa Cloud Co., Ltd.

Documentation

Overview

Package takt provides the abstract proactor execution model via algebraic effects on code.hybscloud.com/kont.

takt dispatches effect operations through an F-bounded Dispatcher with full iox outcome classification.

Execution Strategies

• Blocking: Exec/ExecExpr wait on ErrWouldBlock
• Stepping: Step and Advance evaluate one effect at a time for proactor integration
• Event loop: Loop drives computations through a Backend (submit/poll)

iox Classification

• nil: completed
• code.hybscloud.com/iox.ErrMore: progress, more completions expected
• code.hybscloud.com/iox.ErrWouldBlock: no progress, retry later
• failure: infrastructure error

Error Handling

ExecError/ExecErrorExpr/StepError/AdvanceError compose Dispatcher and kont Error effects. Dispatch order: Error → Dispatcher. Results are code.hybscloud.com/kont.Either — Right on success, Left on Throw.

Bridge

• Reify: Cont → Expr
• Reflect: Expr → Cont

Index

• func Advance[D Dispatcher[D], R any](d D, susp *kont.Suspension[R]) (R, *kont.Suspension[R], error)
• func AdvanceError[E any, D Dispatcher[D], R any](d D, susp *kont.Suspension[kont.Either[E, R]]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]], error)
• func Exec[D Dispatcher[D], R any](d D, m kont.Eff[R]) R
• func ExecError[E any, D Dispatcher[D], R any](d D, m kont.Eff[R]) kont.Either[E, R]
• func ExecErrorExpr[E any, D Dispatcher[D], R any](d D, m kont.Expr[R]) kont.Either[E, R]
• func ExecExpr[D Dispatcher[D], R any](d D, m kont.Expr[R]) R
• func Reflect[A any](m kont.Expr[A]) kont.Eff[A]
• func Reify[A any](m kont.Eff[A]) kont.Expr[A]
• func Step[R any](m kont.Expr[R]) (R, *kont.Suspension[R])
• func StepError[E, R any](m kont.Expr[R]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]])
• type Backend
• type Completion
• type Dispatcher
• type Loop
  • func NewLoop[B Backend[B], R any](b B, maxCompletions int) *Loop[B, R]
  • func (l *Loop[B, R]) Pending() int
  • func (l *Loop[B, R]) Poll() ([]R, error)
  • func (l *Loop[B, R]) Run() ([]R, error)
  • func (l *Loop[B, R]) Submit(m kont.Eff[R]) (R, bool, error)
  • func (l *Loop[B, R]) SubmitExpr(m kont.Expr[R]) (R, bool, error)
• type Token

Functions

func Advance

func Advance[D Dispatcher[D], R any](d D, susp *kont.Suspension[R]) (R, *kont.Suspension[R], error)

Advance dispatches one suspended operation via a Dispatcher. IsProgress: suspension consumed. WouldBlock/failure: unconsumed.

func AdvanceError

func AdvanceError[E any, D Dispatcher[D], R any](d D, susp *kont.Suspension[kont.Either[E, R]]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]], error)

AdvanceError dispatches the suspended operation. Error ops: eager (Throw → Discard + Left). Dispatcher ops: full iox classification.

func Exec

func Exec[D Dispatcher[D], R any](d D, m kont.Eff[R]) R

Exec runs a Cont-world computation to completion via a Dispatcher.

func ExecError

func ExecError[E any, D Dispatcher[D], R any](d D, m kont.Eff[R]) kont.Either[E, R]

ExecError runs a Cont-world computation with error handling. Returns Right on success, Left on Throw.

func ExecErrorExpr

func ExecErrorExpr[E any, D Dispatcher[D], R any](d D, m kont.Expr[R]) kont.Either[E, R]

ExecErrorExpr is the Expr-world ExecError. Blocks on iox.ErrWouldBlock via adaptive backoff (iox.Backoff).

func ExecExpr

func ExecExpr[D Dispatcher[D], R any](d D, m kont.Expr[R]) R

ExecExpr runs an Expr-world computation to completion via a Dispatcher.

func Reflect

func Reflect[A any](m kont.Expr[A]) kont.Eff[A]

Reflect converts Expr-world to Cont-world.

func Reify

func Reify[A any](m kont.Eff[A]) kont.Expr[A]

Reify converts Cont-world to Expr-world.

func Step

func Step[R any](m kont.Expr[R]) (R, *kont.Suspension[R])

Step evaluates until the first effect suspension.

func StepError

func StepError[E, R any](m kont.Expr[R]) (kont.Either[E, R], *kont.Suspension[kont.Either[E, R]])

StepError evaluates a computation with error support until the first suspension. Returns (Either[E, R], nil) on completion or error, or (zero, suspension) if pending.

Types

type Backend

type Backend[B Backend[B]] interface {
	// Submit sends an operation. Returns a correlation token.
	Submit(op kont.Operation) (Token, error)

	// Poll writes ready completions into completions. Returns count.
	Poll(completions []Completion) int
}

Backend is the F-bounded interface for async submit/poll.

type Completion

type Completion struct {
	Token Token
	Value kont.Resumed
	Err   error
}

Completion carries a backend result with iox outcome.

type Dispatcher

type Dispatcher[D Dispatcher[D]] interface {
	Dispatch(op kont.Operation) (kont.Resumed, error)
}

Dispatcher is the F-bounded interface for non-blocking operation dispatch. Returns (value, nil) on success, or (nil, error) at the iox boundary.

type Loop

type Loop[B Backend[B], R any] struct {
	// contains filtered or unexported fields
}

Loop drives Expr computations through a Backend.

func NewLoop

func NewLoop[B Backend[B], R any](b B, maxCompletions int) *Loop[B, R]

NewLoop creates an event loop with the given backend and maximum completions polled per tick.

func (*Loop[B, R]) Pending

func (l *Loop[B, R]) Pending() int

Pending returns the count of in-flight operations.

func (*Loop[B, R]) Poll

func (l *Loop[B, R]) Poll() ([]R, error)

Poll dispatches ready completions. Resumed suspensions are resubmitted.

func (*Loop[B, R]) Run

func (l *Loop[B, R]) Run() ([]R, error)

Run polls until all pending operations complete.

func (*Loop[B, R]) Submit

func (l *Loop[B, R]) Submit(m kont.Eff[R]) (R, bool, error)

Submit reifies a Cont computation and submits it.

func (*Loop[B, R]) SubmitExpr

func (l *Loop[B, R]) SubmitExpr(m kont.Expr[R]) (R, bool, error)

SubmitExpr steps an Expr computation and submits its first operation.

type Token

type Token uint64

Token correlates a submitted operation with its completion.