protofsm: allow multiple internal events to be emitted

In this commit, we update the execution logic to allow multiple internal
events to be emitted. This is useful to handle potential out of order
state transitions, as they can be cached, then emitted once the relevant
pre-conditions have been met.
This commit is contained in:
Olaoluwa Osuntokun 2024-03-04 23:49:11 -06:00
parent 847c1a789d
commit 6de0615cd5
No known key found for this signature in database
GPG Key ID: 90525F7DEEE0AD86
3 changed files with 25 additions and 25 deletions

View File

@ -12,7 +12,7 @@ var log btclog.Logger
// The default amount of logging is none. // The default amount of logging is none.
func init() { func init() {
UseLogger(build.NewSubLogger("PRCL", nil)) UseLogger(build.NewSubLogger("PFSM", nil))
} }
// DisableLog disables all library log output. Logging output is disabled // DisableLog disables all library log output. Logging output is disabled

View File

@ -8,7 +8,6 @@ import (
"github.com/btcsuite/btcd/btcec/v2" "github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/chaincfg/chainhash" "github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcd/wire"
"github.com/davecgh/go-spew/spew"
"github.com/lightningnetwork/lnd/chainntnfs" "github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/fn" "github.com/lightningnetwork/lnd/fn"
"github.com/lightningnetwork/lnd/lnutils" "github.com/lightningnetwork/lnd/lnutils"
@ -28,7 +27,7 @@ type EmittedEvent[Event any] struct {
// InternalEvent is an optional internal event that is to be routed // InternalEvent is an optional internal event that is to be routed
// back to the target state. This enables state to trigger one or many // back to the target state. This enables state to trigger one or many
// state transitions without a new external event. // state transitions without a new external event.
InternalEvent fn.Option[Event] InternalEvent fn.Option[[]Event]
// ExternalEvent is an optional external event that is to be sent to // ExternalEvent is an optional external event that is to be sent to
// the daemon for dispatch. Usually, this is some form of I/O. // the daemon for dispatch. Usually, this is some form of I/O.
@ -329,7 +328,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent( //nolint:funlen
// any preconditions as well as post-send events. // any preconditions as well as post-send events.
case *SendMsgEvent[Event]: case *SendMsgEvent[Event]:
sendAndCleanUp := func() error { sendAndCleanUp := func() error {
log.Debugf("FSM(%v): sending message to target(%v): "+ log.Debugf("FSM(%v): sending message to target(%x): "+
"%v", s.cfg.Env.Name(), "%v", s.cfg.Env.Name(),
daemonEvent.TargetPeer.SerializeCompressed(), daemonEvent.TargetPeer.SerializeCompressed(),
lnutils.SpewLogClosure(daemonEvent.Msgs), lnutils.SpewLogClosure(daemonEvent.Msgs),
@ -451,7 +450,11 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent( //nolint:funlen
defer s.wg.Done() defer s.wg.Done()
for { for {
select { select {
case spend := <-spendEvent.Spend: case spend, ok := <-spendEvent.Spend:
if !ok {
return
}
// If there's a post-send event, then // If there's a post-send event, then
// we'll send that into the current // we'll send that into the current
// state now. // state now.
@ -522,11 +525,8 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
newEvent Event) (State[Event, Env], error) { newEvent Event) (State[Event, Env], error) {
log.Debugf("FSM(%v): applying new event", s.cfg.Env.Name(), log.Debugf("FSM(%v): applying new event", s.cfg.Env.Name(),
lnutils.NewLogClosure(func() string { lnutils.SpewLogClosure(newEvent),
return spew.Sdump(newEvent)
}),
) )
eventQueue := fn.NewQueue(newEvent) eventQueue := fn.NewQueue(newEvent)
// Given the next event to handle, we'll process the event, then add // Given the next event to handle, we'll process the event, then add
@ -539,9 +539,7 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
err := fn.MapOptionZ(nextEvent, func(event Event) error { err := fn.MapOptionZ(nextEvent, func(event Event) error {
log.Debugf("FSM(%v): processing event: %v", log.Debugf("FSM(%v): processing event: %v",
s.cfg.Env.Name(), s.cfg.Env.Name(),
lnutils.NewLogClosure(func() string { lnutils.SpewLogClosure(event),
return spew.Sdump(event)
}),
) )
// Apply the state transition function of the current // Apply the state transition function of the current
@ -584,19 +582,19 @@ func (s *StateMachine[Event, Env]) applyEvents(currentState State[Event, Env],
// our event queue. // our event queue.
// //
//nolint:lll //nolint:lll
events.InternalEvent.WhenSome(func(es Event) { events.InternalEvent.WhenSome(func(es []Event) {
log.Debugf("FSM(%v): adding "+ for _, inEvent := range es {
"new internal event "+ log.Debugf("FSM(%v): adding "+
"to queue: %v", "new internal event "+
s.cfg.Env.Name(), "to queue: %v",
lnutils.NewLogClosure(func() string { //nolint:lll s.cfg.Env.Name(),
return spew.Sdump( //nolint:lll lnutils.SpewLogClosure(
es, inEvent,
) ),
}), )
)
eventQueue.Enqueue(es) eventQueue.Enqueue(inEvent)
}
}) })
return nil return nil

View File

@ -80,7 +80,9 @@ func (d *dummyStateStart) ProcessEvent(event dummyEvents, env *dummyEnv,
return &StateTransition[dummyEvents, *dummyEnv]{ return &StateTransition[dummyEvents, *dummyEnv]{
NextState: &dummyStateStart{}, NextState: &dummyStateStart{},
NewEvents: fn.Some(EmittedEvent[dummyEvents]{ NewEvents: fn.Some(EmittedEvent[dummyEvents]{
InternalEvent: fn.Some(dummyEvents(&goToFin{})), InternalEvent: fn.Some(
[]dummyEvents{&goToFin{}},
),
}), }),
}, nil }, nil