Note
State machines communicate through states (mutations, checking, and waiting).
asyncmachine-go is a batteries-included graph control flow library implementing AOP and Actor Model through a clock-based state-machine. It features atomic transitions, transparent RPC, TUI debugger, telemetry, REPL, remote workers, and diagrams.
As a control flow library, it decides about running of predefined bits of code (transition handlers) - their order and which ones to run, according to currently active states (flags). Thanks to a novel state machine, the amount of handlers can be minimized, while maximizing scenario coverage. It's fault-tolerant by design, has rule-based mutations, and can be used to target virtually any step-in-time, in any workflow.
AM takes care of most contexts, select statements, and panics, while allowing for graph-structured concurrency
with goroutine cancellation. The history log and relations have a vector format.
It aims at creating autonomous workflows with organic control flow and stateful APIs:
- autonomous - automatic states, relations, context-based decisions
- organic - relations, negotiation, cancellation
- stateful - maintaining context, responsive, atomic
Each state represents:
- binary flag
- node in a multigraph
- AOP aspect
- metric
- trace
- subscription topic
- multiple methods
- breakpoint
Top layers depend on the bottom ones.
| . | . | . | . | . | . | PubSub | . | . | . | . | . | . |
| . | . | . | . | . | Workers | . | . | . | . | . | ||
| . | . | . | . | RPC | . | . | . | . | ||||
| . | . | . | Handlers | . | . | . | ||||||
| . | . | 🐇 Machine API | . | . | ||||||||
| . | Relations | . | ||||||||||
| States | ||||||||||||
Minimal - an untyped definition of 2 states and 1 relation, then 1 mutation and a check.
import am "github.com/pancsta/asyncmachine-go/pkg/machine"
// ...
mach := am.New(nil, am.Struct{
"Foo": {Require: am.S{"Bar"}},
"Bar": {},
}, nil)
mach.Add1("Foo", nil)
mach.Is1("Foo") // falseComplicated - wait on a multi state (event) and the Ready state with a 1s timeout, then mutate with typed args, on top of a state context.
// state ctx is an expiration ctx
ctx := client.Mach.NewStateCtx(ssC.WorkerReady)
// clock-based subscription
whenPayload := client.Mach.WhenTicks(ssC.WorkerPayload, 1, ctx)
// state mutation
client.WorkerRpc.Worker.Add1(ssW.WorkRequested, Pass(&A{
Input: 2}))
// WaitFor* wraps select statements
err := amhelp.WaitForAll(ctx, time.Second,
// post-mutation subscription
mach2.When1(ss.BasicStatesDef.Ready, nil),
// pre-mutation subscription
whenPayload)
// check cancellation
if ctx.Err() != nil {
return // state ctx expired
}
// check error
if err != nil {
// err state mutation
client.Mach.AddErr(err, nil)
return // no err required
}
// client/WorkerPayload and mach2/Ready activated
Handlers - Aspect Oriented transition handlers.
// can Foo activate?
func (h *Handlers) FooEnter(e *am.Event) bool {
return true
}
// with Foo active, can Bar activate?
func (h *Handlers) FooBar(e *am.Event) bool {
return true
}
// Foo activates
func (h *Handlers) FooState(e *am.Event) {
h.foo = NewConn()
}
// Foo de-activates
func (h *Handlers) FooEnd(e *am.Event) {
h.foo.Close()
}Schema - states of a node worker.
type WorkerStatesDef struct {
ErrWork string
ErrWorkTimeout string
ErrClient string
ErrSupervisor string
LocalRpcReady string
PublicRpcReady string
RpcReady string
SuperConnected string
ServeClient string
ClientConnected string
ClientSendPayload string
SuperSendPayload string
Idle string
WorkRequested string
Working string
WorkReady string
// inherit from rpc worker
*ssrpc.WorkerStatesDef
}All examples and benchmarks can be found in /examples.
- 🦾
/pkg/machineis the main package /pkg/nodeshows a high-level usage- examples in
/examplesare good for a general grasp /docs/manual.mdand/docs/diagrams.mdgo deeper into implementation details/tools/cmd/am-genwill bootstrap/examples/mach_templateis copy-paste ready/tools/cmd/am-dbgrecords every detail- reading tests is always a good idea...
This monorepo offers the following importable packages and runnable tools:
/pkg/graphMultigraph of interconnected state machines./pkg/helpersUseful functions when working with async state machines./pkg/historyHistory tracking and traversal.- 🦾
/pkg/machineState machine, dependency free, semver compatible. /pkg/nodeDistributed worker pools with supervisors./pkg/rpcRemote state machines, with the same API as local ones./pkg/statesReusable state definitions, handlers, and piping./pkg/telemetryTelemetry exporters for dbg, metrics, traces, and logs./pkg/pubsubDecentralized PubSub based on libp2p gossipsub./tools/cmd/am-dbgMulti-client TUI debugger./tools/cmd/am-genGenerates states files and Grafana dashboards./tools/cmd/am-visGenerates diagrams of interconnected state machines./tools/cmd/arpcNetwork-native REPL and CLI.
- SecAI Autonomous AI Agents.
- arpc REPL Cobra-based REPL.
- am-dbg TUI Debugger Single state-machine TUI app.
- libp2p PubSub Simulator Sandbox simulator for libp2p-pubsub.
- libp2p PubSub Benchmark Benchmark of libp2p-pubsub ported to asyncmachine-go.
Under development, status depends on each package. The bottom layers seem prod grade, the top ones are alpha or testing.
- before
./scripts/dep-taskfile.shtask install-deps
- after
task testtask formattask lint
- good first issue
It calls struct methods according to conventions, a schema, and currently active states (eg BarEnter, FooFoo,
FooBar, BarState). It tackles nondeterminism by embracing it - like an UDP event stream with structure.
State is a binary name as in status / switch / flag, eg "process RUNNING" or "car BROKEN".
Each state has a counter of activations & de-activations, and all state counters create "machine time". It's a logical clock.
Same event happening many times will cause only 1 state activation, until the state becomes inactive.