Skip to content

Declarative control flow library implementing AOP and Actor Model through a clock-based state machine

License

Notifications You must be signed in to change notification settings

pancsta/asyncmachine-go

Repository files navigation

asyncmachine-go

Note

State machines communicate through states (mutations, checking, and waiting).

asyncmachine-go is a declarative control flow library implementing AOP and Actor Model through a clock-based state machine. It has atomic transitions, relations, transparent RPC, TUI debugger, telemetry, workers, and soon diagrams.

Its main purpose is workflows (in-process or distributed), although it can be used for a wide range of stateful applications - daemons, UIs, bots, agents, firewalls, consensus algos, etc. asyncmachine can precisely (and transparently) target a specific point in a scenario and easily bring structure to event-based systems. It takes care of most contexts, select statements, and panics.

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 - maintaing context, responsive, atomic

Stack

Top layers depend on the bottom ones.

            PubSub            
          Workers          
        RPC        
      Handlers      
    🦾 Machine API    
  Relations  
States

Samples

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") // false

Complicated - wait on a multi state (event) with 1s timeout, and mutate with typed args, on top of a state context.

// state ctx is a non-err ctx
ctx := client.Mach.NewStateCtx(ssC.WorkerReady)
// clock-based subscription
whenPayload := client.Mach.WhenTicks(ssC.WorkerPayload, 1, ctx)
// 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 {
    // 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.

Getting Started

🦾 /pkg/machine is the main package, while /pkg/node is the high-level usage. Examples in /examples are good for a general grasp, while /docs/manual.md and /docs/diagrams.md go deeper into implementation details. Reading tests is always a good idea.

This monorepo offers the following importable packages and runnable tools:

  • /pkg/helpers Useful functions when working with async state machines.
  • /pkg/history History tracking and traversal.
  • 🦾 /pkg/machine State machine, dependency free, semver compatible.
  • /pkg/node Distributed worker pools with supervisors.
  • /pkg/rpc Remote state machines, with the same API as local ones.
  • /pkg/states Reusable state definitions, handlers, and piping.
  • /pkg/telemetry Telemetry exporters for metrics, traces, and logs.
  • /pkg/pubsub Planned.
  • /tools/cmd/am-dbg Multi-client TUI debugger.
  • /tools/cmd/am-gen Generates states files and Grafana dashboards.
  • /tools/cmd/am-vis Planned.

dashboard

Case Studies

Documentation

Community

Status

Under development, status depends on each package. The bottom layers seem prod grade, the top ones are alpha or testing.

Development

  • all PRs welcome
  • before
    • ./scripts/dep-taskfile.sh
    • task install-deps
  • after
    • task test
    • task format
    • task lint
TUI Debugger

How does asyncmachine work?

It calls struct methods according to conventions and currently active states.

What is a "state" in asyncmachine?

State as in status / switch / flag, eg "process RUNNING" or "car BROKEN".

What does "clock-based" mean?

Each state has a counter of activations & de-activations, and all state counters create "machine time".

What's the difference between states and events?

Same event happening many times will cause only 1 state activation, until the state becomes inactive.

Changes