Merge pull request #4 from singchia/sync

Sync

Author: singchia

bench/main.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	_ "net/http/pprof"
+	"syscall"
+
+	"github.com/jumboframes/armorigo/sigaction"
+	"github.com/singchia/yafsm"
+)
+
+const (
+	INIT       = "init"
+	CONN_SENT  = "conn_sent"
+	CONN_RECV  = "conn_recv"
+	CONNED     = "conned"
+	ABNORMAL   = "abnormal"
+	CLOSE_SENT = "close_sent"
+	CLOSE_RECV = "close_recv"
+	CLOSE_HALF = "close_half"
+	CLOSED     = "closed"
+	FINI       = "fini"
+
+	ET_CONNSENT  = "connsent"
+	ET_CONNRECV  = "connrecv"
+	ET_CONNACK   = "connack"
+	ET_ERROR     = "error"
+	ET_EOF       = "eof"
+	ET_CLOSESENT = "closesent"
+	ET_CLOSERECV = "closerecv"
+	ET_CLOSEACK  = "closeack"
+	ET_FINI      = "fini"
+)
+
+func initFSM(fsm *yafsm.FSM) {
+	init := fsm.AddState(INIT)
+	connrecv := fsm.AddState(CONN_RECV)
+	conned := fsm.AddState(CONNED)
+	closesent := fsm.AddState(CLOSE_SENT)
+	closerecv := fsm.AddState(CLOSE_RECV)
+	closehalf := fsm.AddState(CLOSE_HALF)
+	closed := fsm.AddState(FINI)
+	fini := fsm.AddState(FINI)
+	fsm.SetState(INIT)
+
+	// events
+	fsm.AddEvent(ET_CONNRECV, init, connrecv)
+	fsm.AddEvent(ET_CONNACK, connrecv, conned)
+
+	fsm.AddEvent(ET_ERROR, init, closed)
+	fsm.AddEvent(ET_ERROR, connrecv, closed)
+	fsm.AddEvent(ET_ERROR, conned, closed)
+	fsm.AddEvent(ET_ERROR, closesent, closed)
+	fsm.AddEvent(ET_ERROR, closerecv, closed)
+
+	fsm.AddEvent(ET_EOF, connrecv, closed)
+	fsm.AddEvent(ET_EOF, conned, closed)
+
+	fsm.AddEvent(ET_CLOSESENT, conned, closesent)
+	fsm.AddEvent(ET_CLOSESENT, closerecv, closesent) // close and been closed at same time
+	fsm.AddEvent(ET_CLOSESENT, closehalf, closehalf) // close and been closed at same time
+
+	fsm.AddEvent(ET_CLOSERECV, conned, closerecv)
+	fsm.AddEvent(ET_CLOSERECV, closesent, closerecv) // close and been closed at same time
+	fsm.AddEvent(ET_CLOSERECV, closehalf, closehalf) // close and been closed at same time
+
+	fsm.AddEvent(ET_CLOSEACK, closesent, closehalf)
+	fsm.AddEvent(ET_CLOSEACK, closerecv, closehalf)
+	fsm.AddEvent(ET_CLOSEACK, closehalf, closed)
+	// fini
+	fsm.AddEvent(ET_FINI, init, fini)
+	fsm.AddEvent(ET_FINI, connrecv, fini)
+	fsm.AddEvent(ET_FINI, conned, fini)
+	fsm.AddEvent(ET_FINI, closesent, fini)
+	fsm.AddEvent(ET_FINI, closerecv, fini)
+	fsm.AddEvent(ET_FINI, closehalf, fini)
+	fsm.AddEvent(ET_FINI, closed, fini)
+}
+
+func main() {
+	go func() {
+		http.ListenAndServe("0.0.0.0:6061", nil)
+	}()
+
+	fsms := []*yafsm.FSM{}
+	count := 100000
+	for i := 0; i < count; i++ {
+		fsm := yafsm.NewFSM()
+		fsms = append(fsms, fsm)
+		initFSM(fsm)
+		fsm.EmitEvent(ET_CONNRECV)
+		fsm.EmitEvent(CLOSE_SENT)
+		fsm.EmitEvent(ET_ERROR)
+		fsm.EmitEvent(ET_FINI)
+		fsm.Close()
+	}
+	fmt.Println("done")
+
+	sig := sigaction.NewSignal()
+	sig.Add(syscall.SIGINT)
+	sig.Wait(context.Background())
+}

go.mod

@@ -2,4 +2,4 @@ module github.com/singchia/yafsm
 
 go 1.15
 
-require github.com/jumboframes/armorigo v0.1.0
+require github.com/jumboframes/armorigo v0.2.3

go.sum

@@ -0,0 +1,15 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/jumboframes/armorigo v0.2.3 h1:Yf/Oxc81mtHKBBL6tnpZ7jWZ50TdqfdoeByxRZD8Uzo=
+github.com/jumboframes/armorigo v0.2.3/go.mod h1:sXe0R32y6V3oJD2eXcPzMlimvZx0xIDiLedpQOy06t4=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/singchia/go-timer/v2 v2.0.3/go.mod h1:PgkEQc6io8slCUiT5rHzWKU4/P2HXHWk3WWfijZXAf4=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

pkg/prioqueue/prioqueue.go

@@ -209,6 +209,7 @@ func (pq *PrioQueue) PopSync() interface{} {
 	}
 }
 
+// user should release the reference to pq after Close
 func (pq *PrioQueue) Close() {
 	pq.mutex.Lock()
 	defer pq.mutex.Unlock()

yafsm.go

@@ -62,16 +62,26 @@ func (et *Event) AddHandler(handler EventHandler) {
 	et.handlers = append(et.handlers, handler)
 }
 
+type FSMOption func(*FSM)
+
+func WithAsync() FSMOption {
+	return func(fsm *FSM) {
+		fsm.async = true
+	}
+}
+
 type FSM struct {
 	state  string
 	states map[string]*State
 	events map[string]*list.List
+
+	async  bool
 	mutex  sync.RWMutex
 	pq     *prioqueue.PrioQueue
 	cancel context.CancelFunc
 }
 
-func NewFSM() *FSM {
+func NewFSM(opts ...FSMOption) *FSM {
 	pq, _ := prioqueue.NewPrioQueue()
 	ctx, cancel := context.WithCancel(context.Background())
 	fsm := &FSM{
@@ -80,7 +90,12 @@ func NewFSM() *FSM {
 		states: make(map[string]*State),
 		events: make(map[string]*list.List),
 	}
-	go fsm.emit(ctx)
+	for _, opt := range opts {
+		opt(fsm)
+	}
+	if fsm.async {
+		go fsm.emit(ctx)
+	}
 	return fsm
 }
 
@@ -90,11 +105,9 @@ func (fsm *FSM) Init(state string) *State {
 }
 
 func (fsm *FSM) Close() {
-	fsm.cancel()
-	fsm.cancel = nil
-
 	fsm.mutex.Lock()
 	defer fsm.mutex.Unlock()
+
 	for k, _ := range fsm.states {
 		delete(fsm.states, k)
 	}
@@ -106,63 +119,66 @@ func (fsm *FSM) Close() {
 		delete(fsm.events, k)
 	}
 	fsm.pq.Close()
+	fsm.cancel()
 }
 
 func (fsm *FSM) emit(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			fsm.pq = nil
-			fsm.events = nil
-			fsm.states = nil
 			return
-
 		default:
-			data := fsm.pq.PopSync()
-			if data == nil {
-				continue
-			}
-			switch ec := data.(type) {
-			case *eventchan:
-				et := (*Event)(nil)
-
-				fsm.mutex.RLock()
-				etList, ok := fsm.events[ec.event]
-				if !ok {
-					ec.ch <- ErrEventNotExist
-					close(ec.ch)
-					fsm.mutex.RUnlock()
-					continue
-				}
-				for elem := etList.Front(); elem != nil; elem = elem.Next() {
-					tmp := elem.Value.(*Event)
-					if tmp.From.State == fsm.state {
-						et = tmp
-					}
-				}
-				fsm.mutex.RUnlock()
-
-				if et == nil {
-					ec.ch <- ErrIllegalStateForEvent
-					close(ec.ch)
-					continue
-				}
-				for _, left := range et.From.lefts {
-					left(et.From)
-				}
-				for _, handler := range et.handlers {
-					handler(et)
-				}
-				for _, enter := range et.To.enters {
-					enter(et.To)
-				}
-				fsm.mutex.Lock()
-				fsm.state = et.To.State
-				fsm.mutex.Unlock()
-				ec.ch <- nil
-				close(ec.ch)
+			fsm.emitOne()
+		}
+	}
+}
+
+func (fsm *FSM) emitOne() {
+	data := fsm.pq.PopSync()
+	if data == nil {
+		return
+	}
+	switch ec := data.(type) {
+	case *eventchan:
+		et := (*Event)(nil)
+
+		fsm.mutex.RLock()
+		etList, ok := fsm.events[ec.event]
+		if !ok {
+			ec.ch <- ErrEventNotExist
+			close(ec.ch)
+			fsm.mutex.RUnlock()
+			return
+		}
+		fsm.mutex.RUnlock()
+
+		fsm.mutex.Lock()
+		for elem := etList.Front(); elem != nil; elem = elem.Next() {
+			tmp := elem.Value.(*Event)
+			if tmp.From.State == fsm.state {
+				et = tmp
 			}
 		}
+		if et == nil {
+			ec.ch <- ErrIllegalStateForEvent
+			close(ec.ch)
+			fsm.mutex.Unlock()
+			return
+		}
+		fsm.state = et.To.State
+		fsm.mutex.Unlock()
+
+		for _, left := range et.From.lefts {
+			left(et.From)
+		}
+		for _, handler := range et.handlers {
+			handler(et)
+		}
+		for _, enter := range et.To.enters {
+			enter(et.To)
+		}
+		ec.ch <- nil
+		close(ec.ch)
 	}
 }
 
@@ -390,11 +406,13 @@ func (fsm *FSM) EmitEvent(event string) error {
 		event: event,
 		ch:    ch,
 	}
-
 	err := fsm.pq.Push(eventchan)
 	if err != nil {
 		return err
 	}
+	if !fsm.async {
+		fsm.emitOne()
+	}
 	err = <-ch
 	return err
 }
@@ -419,6 +437,9 @@ func (fsm *FSM) EmitEventAsync(event string) <-chan error {
 		ch <- err
 		return ch
 	}
+	if !fsm.async {
+		fsm.emitOne()
+	}
 	return ch
 }
 
@@ -440,6 +461,9 @@ func (fsm *FSM) EmitPrioEvent(prio int, event string) error {
 	if err != nil {
 		return err
 	}
+	if !fsm.async {
+		fsm.emitOne()
+	}
 	err = <-ch
 	return err
 }
@@ -464,5 +488,8 @@ func (fsm *FSM) EmitPrioEventAsync(prio int, event string) <-chan error {
 		ch <- err
 		return ch
 	}
+	if !fsm.async {
+		fsm.emitOne()
+	}
 	return ch
 }