Use Finite State Machines to manage instance/provisioning lifecycle (#427)

Signed-off-by: David Chung <david.chung@docker.com>

Author: David Chung

pkg/broker/client/sse.go

@@ -158,7 +158,7 @@ func Subscribe(url, topic string, opt Options) (<-chan *types.Any, <-chan error,
 				} else {
 
 					select {
-					case errCh <- fmt.Errorf("no data: %v", line):
+					case errCh <- fmt.Errorf("no data: %s", string(line)):
 					default:
 					}
 

pkg/broker/client/sse_test.go

@@ -230,9 +230,13 @@ func TestBrokerMultiSubscriberPartialMatchTopic(t *testing.T) {
 
 	opts := Options{SocketDir: filepath.Dir(socketFile)}
 
-	topic1, errs1, err := Subscribe(socket, "local/instance", opts)
-	require.NoError(t, err)
+	start := make(chan struct{})
 	go func() {
+		<-start
+
+		topic1, errs1, err := Subscribe(socket, "local/instance", opts)
+		require.NoError(t, err)
+
 		for {
 			select {
 			case e := <-errs1:
@@ -249,9 +253,12 @@ func TestBrokerMultiSubscriberPartialMatchTopic(t *testing.T) {
 		}
 	}()
 
-	topic2, errs2, err := Subscribe(socket, "local/instancetest", opts)
-	require.NoError(t, err)
 	go func() {
+		<-start
+
+		topic2, errs2, err := Subscribe(socket, "local/instancetest", opts)
+		require.NoError(t, err)
+
 		for {
 			select {
 			case e := <-errs2:
@@ -269,8 +276,9 @@ func TestBrokerMultiSubscriberPartialMatchTopic(t *testing.T) {
 	}()
 
 	go func() {
+		<-start
+
 		for {
-			<-time.After(10 * time.Millisecond)
 			now := time.Now()
 			evt := event{Time: now.UnixNano(), Message: fmt.Sprintf("Now is %v", now)}
 			require.NoError(t, broker.Publish("local/instance", evt))
@@ -279,6 +287,8 @@ func TestBrokerMultiSubscriberPartialMatchTopic(t *testing.T) {
 		}
 	}()
 
+	close(start)
+
 	// Test a few rounds to make sure all subscribers get the same messages each round.
 	for i := 0; i < 5; i++ {
 		b := <-received2

pkg/fsm/clock.go

@@ -0,0 +1,99 @@
+package fsm
+
+import (
+	"time"
+)
+
+// Clock adapts a timer tick
+type Clock struct {
+	C       <-chan Tick
+	c       chan<- Tick
+	stop    chan struct{}
+	start   chan struct{}
+	driver  func()
+	running bool
+}
+
+// NewClock returns a clock
+func NewClock() *Clock {
+	c := make(chan Tick)
+	stop := make(chan struct{})
+	clock := &Clock{
+		C:     c,
+		c:     c,
+		stop:  stop,
+		start: make(chan struct{}),
+	}
+	clock.driver = func() {
+		<-clock.start
+		clock.start = nil
+
+		for {
+			select {
+			case <-clock.stop:
+				close(clock.c)
+				return
+			}
+		}
+	}
+	return clock.run()
+}
+
+// Start starts the clock
+func (t *Clock) Start() {
+	if t.start != nil {
+		close(t.start)
+	}
+}
+
+// Tick makes one tick of the clock
+func (t *Clock) Tick() {
+	t.c <- Tick(1)
+}
+
+func (t *Clock) run() *Clock {
+	if t.driver != nil {
+		go t.driver()
+	}
+	t.running = true
+	return t
+}
+
+// Stop stops the ticks
+func (t *Clock) Stop() {
+	if t.running {
+		close(t.stop)
+		t.running = false
+	}
+}
+
+// Wall adapts a regular time.Tick to return a clock
+func Wall(tick <-chan time.Time) *Clock {
+	out := make(chan Tick)
+	stop := make(chan struct{})
+	clock := &Clock{
+		C:     out,
+		c:     out,
+		stop:  stop,
+		start: make(chan struct{}),
+	}
+
+	clock.driver = func() {
+		<-clock.start
+		clock.start = nil
+
+		for {
+			select {
+			case <-clock.stop:
+				close(clock.c)
+				return
+			case <-tick:
+				// note that golang's time ticker won't close the channel when stopped.
+				// so we will do the closing ourselves to avoid leaking the goroutine
+				clock.c <- Tick(1)
+			}
+		}
+	}
+
+	return clock.run()
+}

pkg/fsm/clock_test.go

@@ -0,0 +1,102 @@
+package fsm
+
+import (
+	"testing"
+	"time"
+
+	log "github.com/golang/glog"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClock(t *testing.T) {
+	clock := NewClock()
+
+	done := make(chan struct{})
+	start := make(chan struct{})
+	go func() {
+		defer close(done)
+		<-start
+
+		for {
+			_, open := <-clock.C
+			if !open {
+				return // we expect this to be run
+			}
+		}
+	}()
+
+	log.Infoln("Start")
+	close(start)
+	clock.Start()
+	clock.Tick()
+	clock.Tick()
+	clock.Tick()
+
+	log.Infoln("Stopping")
+	clock.Stop()
+
+	log.Infoln("waiting for done")
+	<-done
+	log.Infoln("done")
+}
+
+func TestWallClock(t *testing.T) {
+
+	ticker := time.After(100 * time.Millisecond)
+	clock := Wall(ticker)
+
+	start := make(chan struct{})
+	go func() {
+		<-start
+
+		<-clock.C
+
+		clock.Stop()
+	}()
+
+	close(start) // from here receive just 1 tick
+	clock.Start()
+	<-clock.C
+}
+
+func TestWallClock2(t *testing.T) {
+
+	ticker := time.Tick(100 * time.Millisecond)
+	clock := Wall(ticker)
+
+	start := make(chan struct{})
+
+	ticks := make(chan int, 1000)
+	go func() {
+
+		defer close(ticks)
+
+		<-start
+
+		for {
+			_, open := <-clock.C
+			if !open {
+				return
+			}
+			log.Infoln("tick")
+			ticks <- 1
+		}
+	}()
+
+	close(start)
+	clock.Start()
+	log.Infoln("starting")
+
+	time.Sleep(1 * time.Second)
+
+	log.Infoln("Stopping")
+	clock.Stop()
+	log.Infoln("Stopped")
+
+	total := 0
+	for i := range ticks {
+		total += i
+	}
+	log.Infoln("count=", total)
+	require.Equal(t, 10, total)
+}

pkg/fsm/errors.go

@@ -0,0 +1,47 @@
+package fsm
+
+import (
+	"fmt"
+)
+
+type errDuplicateState Index
+
+func (e errDuplicateState) Error() string {
+	return fmt.Sprintf("duplicated state index: %d", e)
+}
+
+type unknownState Index
+
+func (e unknownState) Error() string {
+	return fmt.Sprintf("unknown state index: %d", e)
+}
+
+type unknownTransition Signal
+
+func (e unknownTransition) Error() string {
+	return fmt.Sprintf("no transition defined for signal %d", e)
+}
+
+type unknownSignal Signal
+
+func (e unknownSignal) Error() string {
+	return fmt.Sprintf("signal in action not found in transitions %d", e)
+}
+
+type unknownInstance ID
+
+func (e unknownInstance) Error() string {
+	return fmt.Sprintf("unknown instance %d", e)
+}
+
+type nilAction Signal
+
+func (e nilAction) Error() string {
+	return fmt.Sprintf("nil action corresponding to signal %d", e)
+}
+
+type noTransitions Spec
+
+func (e noTransitions) Error() string {
+	return fmt.Sprintf("no transitions defined: count(states)=%d", len(e.states))
+}

pkg/fsm/fifo.go

@@ -0,0 +1,48 @@
+package fsm
+
+func newFifo(size int) *fifo {
+	return &fifo{
+		nodes: make([]*event, size),
+		size:  size,
+	}
+}
+
+// fifo is a basic FIFO queue of events based on a circular list that resizes as needed.
+type fifo struct {
+	nodes []*event
+	size  int
+	head  int
+	tail  int
+	count int
+}
+
+// Len returns the size of the fifo
+func (q *fifo) Len() int {
+	return q.count
+}
+
+// push adds a node to the queue.
+func (q *fifo) push(n *event) {
+	if q.head == q.tail && q.count > 0 {
+		nodes := make([]*event, len(q.nodes)+q.size)
+		copy(nodes, q.nodes[q.head:])
+		copy(nodes[len(q.nodes)-q.head:], q.nodes[:q.head])
+		q.head = 0
+		q.tail = len(q.nodes)
+		q.nodes = nodes
+	}
+	q.nodes[q.tail] = n
+	q.tail = (q.tail + 1) % len(q.nodes)
+	q.count++
+}
+
+// pop removes and returns a node from the queue in first to last order.
+func (q *fifo) pop() *event {
+	if q.count == 0 {
+		return nil
+	}
+	node := q.nodes[q.head]
+	q.head = (q.head + 1) % len(q.nodes)
+	q.count--
+	return node
+}

pkg/fsm/fifo_test.go

@@ -0,0 +1,31 @@
+package fsm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFifo(t *testing.T) {
+
+	q := newFifo(10)
+
+	for i := 0; i < 100; i++ {
+		q.push(&event{instance: ID(i)})
+	}
+
+	require.Equal(t, 100, q.Len())
+
+	ids := []ID{}
+
+	for q.Len() > 0 {
+		event := q.pop()
+		require.NotNil(t, event)
+		ids = append(ids, event.instance)
+	}
+
+	require.Equal(t, 100, len(ids))
+	for i := 0; i < 100; i++ {
+		require.Equal(t, ID(i), ids[i])
+	}
+}