Event-based state machine

lib/evtsm/cbor_gen.go

@@ -0,0 +1,238 @@
+package evtsm
+
+import (
+	"fmt"
+	"io"
+
+	cbg "github.com/whyrusleeping/cbor-gen"
+	xerrors "golang.org/x/xerrors"
+)
+
+// Code generated by github.com/whyrusleeping/cbor-gen. DO NOT EDIT.
+
+var _ = xerrors.Errorf
+
+func (t *TestState) MarshalCBOR(w io.Writer) error {
+	if t == nil {
+		_, err := w.Write(cbg.CborNull)
+		return err
+	}
+	if _, err := w.Write([]byte{162}); err != nil {
+		return err
+	}
+
+	// t.A (uint64) (uint64)
+	if len("A") > cbg.MaxLength {
+		return xerrors.Errorf("Value in field \"A\" was too long")
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("A")))); err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte("A")); err != nil {
+		return err
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.A))); err != nil {
+		return err
+	}
+
+	// t.B (uint64) (uint64)
+	if len("B") > cbg.MaxLength {
+		return xerrors.Errorf("Value in field \"B\" was too long")
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("B")))); err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte("B")); err != nil {
+		return err
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.B))); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (t *TestState) UnmarshalCBOR(r io.Reader) error {
+	br := cbg.GetPeeker(r)
+
+	maj, extra, err := cbg.CborReadHeader(br)
+	if err != nil {
+		return err
+	}
+	if maj != cbg.MajMap {
+		return fmt.Errorf("cbor input should be of type map")
+	}
+
+	if extra != 2 {
+		return fmt.Errorf("cbor input had wrong number of fields")
+	}
+
+	var name string
+
+	// t.A (uint64) (uint64)
+
+	{
+		sval, err := cbg.ReadString(br)
+		if err != nil {
+			return err
+		}
+
+		name = string(sval)
+	}
+
+	if name != "A" {
+		return fmt.Errorf("expected struct map entry %s to be A", name)
+	}
+
+	maj, extra, err = cbg.CborReadHeader(br)
+	if err != nil {
+		return err
+	}
+	if maj != cbg.MajUnsignedInt {
+		return fmt.Errorf("wrong type for uint64 field")
+	}
+	t.A = uint64(extra)
+	// t.B (uint64) (uint64)
+
+	{
+		sval, err := cbg.ReadString(br)
+		if err != nil {
+			return err
+		}
+
+		name = string(sval)
+	}
+
+	if name != "B" {
+		return fmt.Errorf("expected struct map entry %s to be B", name)
+	}
+
+	maj, extra, err = cbg.CborReadHeader(br)
+	if err != nil {
+		return err
+	}
+	if maj != cbg.MajUnsignedInt {
+		return fmt.Errorf("wrong type for uint64 field")
+	}
+	t.B = uint64(extra)
+	return nil
+}
+
+func (t *TestEvent) MarshalCBOR(w io.Writer) error {
+	if t == nil {
+		_, err := w.Write(cbg.CborNull)
+		return err
+	}
+	if _, err := w.Write([]byte{162}); err != nil {
+		return err
+	}
+
+	// t.A (string) (string)
+	if len("A") > cbg.MaxLength {
+		return xerrors.Errorf("Value in field \"A\" was too long")
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("A")))); err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte("A")); err != nil {
+		return err
+	}
+
+	if len(t.A) > cbg.MaxLength {
+		return xerrors.Errorf("Value in field t.A was too long")
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len(t.A)))); err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte(t.A)); err != nil {
+		return err
+	}
+
+	// t.Val (uint64) (uint64)
+	if len("Val") > cbg.MaxLength {
+		return xerrors.Errorf("Value in field \"Val\" was too long")
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajTextString, uint64(len("Val")))); err != nil {
+		return err
+	}
+	if _, err := w.Write([]byte("Val")); err != nil {
+		return err
+	}
+
+	if _, err := w.Write(cbg.CborEncodeMajorType(cbg.MajUnsignedInt, uint64(t.Val))); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (t *TestEvent) UnmarshalCBOR(r io.Reader) error {
+	br := cbg.GetPeeker(r)
+
+	maj, extra, err := cbg.CborReadHeader(br)
+	if err != nil {
+		return err
+	}
+	if maj != cbg.MajMap {
+		return fmt.Errorf("cbor input should be of type map")
+	}
+
+	if extra != 2 {
+		return fmt.Errorf("cbor input had wrong number of fields")
+	}
+
+	var name string
+
+	// t.A (string) (string)
+
+	{
+		sval, err := cbg.ReadString(br)
+		if err != nil {
+			return err
+		}
+
+		name = string(sval)
+	}
+
+	if name != "A" {
+		return fmt.Errorf("expected struct map entry %s to be A", name)
+	}
+
+	{
+		sval, err := cbg.ReadString(br)
+		if err != nil {
+			return err
+		}
+
+		t.A = string(sval)
+	}
+	// t.Val (uint64) (uint64)
+
+	{
+		sval, err := cbg.ReadString(br)
+		if err != nil {
+			return err
+		}
+
+		name = string(sval)
+	}
+
+	if name != "Val" {
+		return fmt.Errorf("expected struct map entry %s to be Val", name)
+	}
+
+	maj, extra, err = cbg.CborReadHeader(br)
+	if err != nil {
+		return err
+	}
+	if maj != cbg.MajUnsignedInt {
+		return fmt.Errorf("wrong type for uint64 field")
+	}
+	t.Val = uint64(extra)
+	return nil
+}

lib/evtsm/ctx.go

@@ -0,0 +1,16 @@
+package evtsm
+
+import "context"
+
+type Context struct {
+	ctx  context.Context
+	send func(evt interface{}) error
+}
+
+func (ctx *Context) Context() context.Context {
+	return ctx.ctx
+}
+
+func (ctx *Context) Send(evt interface{}) error {
+	return ctx.send(evt)
+}

lib/evtsm/event.go

@@ -0,0 +1,5 @@
+package evtsm
+
+type Event struct {
+	User interface{}
+}

lib/evtsm/evtsm.go

@@ -0,0 +1,120 @@
+package evtsm
+
+import (
+	"context"
+	"reflect"
+	"sync/atomic"
+
+	"github.com/filecoin-project/lotus/lib/statestore"
+	logging "github.com/ipfs/go-log"
+)
+
+var log = logging.Logger("evtsm")
+
+// returns func(ctx Context, st <T>) (func(*<T>), error), where <T> is the typeOf(User) param
+type Planner func(events []Event, user interface{}) (interface{}, error)
+
+type ESm struct {
+	planner  Planner
+	eventsIn chan Event
+
+	name      interface{}
+	st        *statestore.StoredState
+	stateType reflect.Type
+
+	stageDone chan struct{}
+	closing   chan struct{}
+	closed    chan struct{}
+
+	busy int32
+}
+
+func (fsm *ESm) run() {
+	defer close(fsm.closed)
+
+	var pendingEvents []Event
+
+	for {
+		// NOTE: This requires at least one event to be sent to trigger a stage
+		//  This means that after restarting the state machine users of this
+		//  code must send a 'restart' event
+		select {
+		case evt := <-fsm.eventsIn:
+			pendingEvents = append(pendingEvents, evt)
+		case <-fsm.stageDone:
+			if len(pendingEvents) == 0 {
+				continue
+			}
+		case <-fsm.closing:
+			return
+		}
+
+		if atomic.CompareAndSwapInt32(&fsm.busy, 0, 1) {
+			var nextStep interface{}
+			var ustate interface{}
+
+			err := fsm.mutateUser(func(user interface{}) (err error) {
+				nextStep, err = fsm.planner(pendingEvents, user)
+				ustate = user
+				return err
+			})
+			if err != nil {
+				log.Errorf("Executing event planner failed: %+v", err)
+				return
+			}
+
+			pendingEvents = nil
+
+			if nextStep == nil {
+				continue
+			}
+
+			ctx := Context{
+				ctx: context.TODO(),
+				send: func(evt interface{}) error {
+					return fsm.send(Event{User: evt})
+				},
+			}
+
+			go func() {
+				res := reflect.ValueOf(nextStep).Call([]reflect.Value{reflect.ValueOf(ctx), reflect.ValueOf(ustate).Elem()})
+
+				if res[0].Interface() != nil {
+					log.Errorf("executing step: %+v", res[0].Interface().(error)) // TODO: propagate top level
+					return
+				}
+
+				atomic.StoreInt32(&fsm.busy, 0)
+				fsm.stageDone <- struct{}{}
+			}()
+
+		}
+	}
+}
+
+func (fsm *ESm) mutateUser(cb func(user interface{}) error) error {
+	mutt := reflect.FuncOf([]reflect.Type{reflect.PtrTo(fsm.stateType)}, []reflect.Type{reflect.TypeOf(new(error)).Elem()}, false)
+
+	mutf := reflect.MakeFunc(mutt, func(args []reflect.Value) (results []reflect.Value) {
+		err := cb(args[0].Interface())
+		return []reflect.Value{reflect.ValueOf(&err).Elem()}
+	})
+
+	return fsm.st.Mutate(mutf.Interface())
+}
+
+func (fsm *ESm) send(evt Event) error {
+	fsm.eventsIn <- evt // TODO: ctx, at least
+	return nil
+}
+
+func (fsm *ESm) stop(ctx context.Context) error {
+	close(fsm.closing)
+
+	select {
+	case <-fsm.closed:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}

lib/evtsm/evtsm_test.go

@@ -0,0 +1,106 @@
+package evtsm
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"github.com/ipfs/go-datastore"
+	logging "github.com/ipfs/go-log"
+	"gotest.tools/assert"
+)
+
+func init() {
+	logging.SetLogLevel("*", "INFO")
+}
+
+type testHandler struct {
+	t       *testing.T
+	proceed chan struct{}
+	done    chan struct{}
+}
+
+func (t *testHandler) Plan(events []Event, state interface{}) (interface{}, error) {
+	return t.plan(events, state.(*TestState))
+}
+
+func (t *testHandler) plan(events []Event, state *TestState) (interface{}, error) {
+	for _, event := range events {
+		e := event.User.(*TestEvent)
+		switch e.A {
+		case "restart":
+		case "start":
+			state.A = 1
+		case "b":
+			state.A = 2
+			state.B = e.Val
+		}
+	}
+
+	switch state.A {
+	case 1:
+		return t.step0, nil
+	case 2:
+		return t.step1, nil
+	default:
+		t.t.Fatal(state.A)
+	}
+	panic("how?")
+}
+
+func (t *testHandler) step0(ctx Context, st TestState) error {
+	ctx.Send(&TestEvent{A: "b", Val: 55})
+	<-t.proceed
+	return nil
+}
+
+func (t *testHandler) step1(ctx Context, st TestState) error {
+	assert.Equal(t.t, uint64(2), st.A)
+
+	close(t.done)
+	return nil
+}
+
+func TestBasic(t *testing.T) {
+	for i := 0; i < 1000; i++ { // run a few times to expose any races
+		ds := datastore.NewMapDatastore()
+
+		th := &testHandler{t: t, done: make(chan struct{}), proceed: make(chan struct{})}
+		close(th.proceed)
+		smm := New(ds, th, reflect.TypeOf(TestState{}))
+
+		if err := smm.Send(uint64(2), &TestEvent{A: "start"}); err != nil {
+			t.Fatalf("%+v", err)
+		}
+
+		<-th.done
+	}
+}
+
+func TestPersist(t *testing.T) {
+	for i := 0; i < 1000; i++ { // run a few times to expose any races
+		ds := datastore.NewMapDatastore()
+
+		th := &testHandler{t: t, done: make(chan struct{}), proceed: make(chan struct{})}
+		smm := New(ds, th, reflect.TypeOf(TestState{}))
+
+		if err := smm.Send(uint64(2), &TestEvent{A: "start"}); err != nil {
+			t.Fatalf("%+v", err)
+		}
+
+		if err := smm.Stop(context.Background()); err != nil {
+			t.Fatal(err)
+			return
+		}
+
+		smm = New(ds, th, reflect.TypeOf(TestState{}))
+		if err := smm.Send(uint64(2), &TestEvent{A: "restart"}); err != nil {
+			t.Fatalf("%+v", err)
+		}
+		close(th.proceed)
+
+		<-th.done
+	}
+}
+
+var _ StateHandler = &testHandler{}

lib/evtsm/sched.go

@@ -0,0 +1,99 @@
+package evtsm
+
+import (
+	"context"
+	"reflect"
+	"sync"
+
+	"github.com/ipfs/go-datastore"
+	"golang.org/x/xerrors"
+
+	"github.com/filecoin-project/lotus/lib/statestore"
+)
+
+type StateHandler interface {
+	// returns
+	Plan(events []Event, user interface{}) (interface{}, error)
+}
+
+type Sched struct {
+	sts       *statestore.StateStore
+	hnd       StateHandler
+	stateType reflect.Type
+
+	lk  sync.Mutex
+	sms map[datastore.Key]*ESm
+}
+
+// stateType: T - (reflect.TypeOf(MyStateStruct{}))
+func New(ds datastore.Datastore, hnd StateHandler, stateType reflect.Type) *Sched {
+	return &Sched{
+		sts:       statestore.New(ds),
+		hnd:       hnd,
+		stateType: stateType,
+
+		sms: map[datastore.Key]*ESm{},
+	}
+}
+
+func (s *Sched) Send(to interface{}, evt interface{}) (err error) {
+	s.lk.Lock()
+	defer s.lk.Unlock()
+
+	sm, exist := s.sms[statestore.ToKey(to)]
+	if !exist {
+		sm, err = s.loadOrCreate(to)
+		if err != nil {
+			return xerrors.Errorf("loadOrCreate state: %w", err)
+		}
+		s.sms[statestore.ToKey(to)] = sm
+	}
+
+	return sm.send(Event{User: evt})
+}
+
+func (s *Sched) loadOrCreate(name interface{}) (*ESm, error) {
+	exists, err := s.sts.Has(name)
+	if err != nil {
+		return nil, xerrors.Errorf("failed to check if state for %v exists: %w", name, err)
+	}
+
+	if !exists {
+		userState := reflect.New(s.stateType).Interface()
+
+		err = s.sts.Begin(name, userState)
+		if err != nil {
+			return nil, xerrors.Errorf("saving initial state: %w", err)
+		}
+	}
+
+	res := &ESm{
+		planner:  s.hnd.Plan,
+		eventsIn: make(chan Event),
+
+		name:      name,
+		st:        s.sts.Get(name),
+		stateType: s.stateType,
+
+		stageDone: make(chan struct{}),
+		closing:   make(chan struct{}),
+		closed:    make(chan struct{}),
+	}
+
+	go res.run()
+
+	return res, nil
+}
+
+func (s *Sched) Stop(ctx context.Context) error {
+	s.lk.Lock()
+	defer s.lk.Unlock()
+
+	for _, sm := range s.sms {
+		if err := sm.stop(ctx); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

lib/evtsm/testing.go

@@ -0,0 +1,11 @@
+package evtsm
+
+type TestState struct {
+	A uint64
+	B uint64
+}
+
+type TestEvent struct {
+	A   string
+	Val uint64
+}

lib/statestore/state.go

@@ -0,0 +1,93 @@
+package statestore
+
+import (
+	"bytes"
+	"reflect"
+
+	cborutil "github.com/filecoin-project/go-cbor-util"
+	"github.com/ipfs/go-datastore"
+	cbg "github.com/whyrusleeping/cbor-gen"
+	"golang.org/x/xerrors"
+)
+
+type StoredState struct {
+	ds   datastore.Datastore
+	name datastore.Key
+}
+
+func (st *StoredState) End() error {
+	has, err := st.ds.Has(st.name)
+	if err != nil {
+		return err
+	}
+	if !has {
+		return xerrors.Errorf("No state for %s", st.name)
+	}
+	if err := st.ds.Delete(st.name); err != nil {
+		return xerrors.Errorf("removing state from datastore: %w", err)
+	}
+	st.name = datastore.Key{}
+	st.ds = nil
+
+	return nil
+}
+
+func (st *StoredState) Get(out cbg.CBORUnmarshaler) error {
+	val, err := st.ds.Get(st.name)
+	if err != nil {
+		if xerrors.Is(err, datastore.ErrNotFound) {
+			return xerrors.Errorf("No state for %s: %w", st.name, err)
+		}
+		return err
+	}
+
+	return out.UnmarshalCBOR(bytes.NewReader(val))
+}
+
+// mutator func(*T) error
+func (st *StoredState) Mutate(mutator interface{}) error {
+	return st.mutate(cborMutator(mutator))
+}
+
+func (st *StoredState) mutate(mutator func([]byte) ([]byte, error)) error {
+	has, err := st.ds.Has(st.name)
+	if err != nil {
+		return err
+	}
+	if !has {
+		return xerrors.Errorf("No state for %s", st.name)
+	}
+
+	cur, err := st.ds.Get(st.name)
+	if err != nil {
+		return err
+	}
+
+	mutated, err := mutator(cur)
+	if err != nil {
+		return err
+	}
+
+	return st.ds.Put(st.name, mutated)
+}
+
+func cborMutator(mutator interface{}) func([]byte) ([]byte, error) {
+	rmut := reflect.ValueOf(mutator)
+
+	return func(in []byte) ([]byte, error) {
+		state := reflect.New(rmut.Type().In(0).Elem())
+
+		err := cborutil.ReadCborRPC(bytes.NewReader(in), state.Interface())
+		if err != nil {
+			return nil, err
+		}
+
+		out := rmut.Call([]reflect.Value{state})
+
+		if err := out[0].Interface(); err != nil {
+			return nil, err.(error)
+		}
+
+		return cborutil.Dump(state.Interface())
+	}
+}

lib/statestore/store.go

@@ -5,13 +5,11 @@ import (
 	"fmt"
 	"reflect"
 
+	"github.com/filecoin-project/go-cbor-util"
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/query"
-	cbg "github.com/whyrusleeping/cbor-gen"
 	"go.uber.org/multierr"
 	"golang.org/x/xerrors"
-
-	"github.com/filecoin-project/go-cbor-util"
 )
 
 type StateStore struct {
@@ -22,7 +20,7 @@ func New(ds datastore.Datastore) *StateStore {
 	return &StateStore{ds: ds}
 }
 
-func toKey(k interface{}) datastore.Key {
+func ToKey(k interface{}) datastore.Key {
 	switch t := k.(type) {
 	case uint64:
 		return datastore.NewKey(fmt.Sprint(t))
@@ -34,7 +32,7 @@ func toKey(k interface{}) datastore.Key {
 }
 
 func (st *StateStore) Begin(i interface{}, state interface{}) error {
-	k := toKey(i)
+	k := ToKey(i)
 	has, err := st.ds.Has(k)
 	if err != nil {
 		return err
@@ -51,82 +49,15 @@ func (st *StateStore) Begin(i interface{}, state interface{}) error {
 	return st.ds.Put(k, b)
 }
 
-func (st *StateStore) End(i interface{}) error {
-	k := toKey(i)
-	has, err := st.ds.Has(k)
-	if err != nil {
-		return err
+func (st *StateStore) Get(i interface{}) *StoredState {
+	return &StoredState{
+		ds:   st.ds,
+		name: ToKey(i),
 	}
-	if !has {
-		return xerrors.Errorf("No state for %s", i)
-	}
-	return st.ds.Delete(k)
-}
-
-func cborMutator(mutator interface{}) func([]byte) ([]byte, error) {
-	rmut := reflect.ValueOf(mutator)
-
-	return func(in []byte) ([]byte, error) {
-		state := reflect.New(rmut.Type().In(0).Elem())
-
-		err := cborutil.ReadCborRPC(bytes.NewReader(in), state.Interface())
-		if err != nil {
-			return nil, err
-		}
-
-		out := rmut.Call([]reflect.Value{state})
-
-		if err := out[0].Interface(); err != nil {
-			return nil, err.(error)
-		}
-
-		return cborutil.Dump(state.Interface())
-	}
-}
-
-// mutator func(*T) error
-func (st *StateStore) Mutate(i interface{}, mutator interface{}) error {
-	return st.mutate(i, cborMutator(mutator))
-}
-
-func (st *StateStore) mutate(i interface{}, mutator func([]byte) ([]byte, error)) error {
-	k := toKey(i)
-	has, err := st.ds.Has(k)
-	if err != nil {
-		return err
-	}
-	if !has {
-		return xerrors.Errorf("No state for %s", i)
-	}
-
-	cur, err := st.ds.Get(k)
-	if err != nil {
-		return err
-	}
-
-	mutated, err := mutator(cur)
-	if err != nil {
-		return err
-	}
-
-	return st.ds.Put(k, mutated)
 }
 
 func (st *StateStore) Has(i interface{}) (bool, error) {
-	return st.ds.Has(toKey(i))
-}
-
-func (st *StateStore) Get(i interface{}, out cbg.CBORUnmarshaler) error {
-	k := toKey(i)
-	val, err := st.ds.Get(k)
-	if err != nil {
-		if xerrors.Is(err, datastore.ErrNotFound) {
-			return xerrors.Errorf("No state for %s: %w", i, err)
-		}
-		return err
-	}
-
-	return out.UnmarshalCBOR(bytes.NewReader(val))
+	return st.ds.Has(ToKey(i))
 }
 
 // out: *[]T