Stopping & Termination:

+ Modify handleFunc signature to return an error
    + if an error is returned, terminate the routine
    + finished channel will then return a termination reason

blockchain/v2/demuxer.go

@@ -6,7 +6,7 @@ type demuxer struct {
 	eventbus  chan Event
 	scheduler *Routine
 	processor *Routine
-	finished  chan struct{}
+	finished  chan error
 	stopped   chan struct{}
 }
 
@@ -16,10 +16,12 @@ func newDemuxer(scheduler *Routine, processor *Routine) *demuxer {
 		scheduler: scheduler,
 		processor: processor,
 		stopped:   make(chan struct{}, 1),
-		finished:  make(chan struct{}, 1),
+		finished:  make(chan error, 1),
 	}
 }
 
+// What should the termination clause be?
+// Is any of the subroutines finishe, the demuxer finishes
 func (dm *demuxer) run() {
 	fmt.Printf("demuxer: run\n")
 	for {
@@ -30,7 +32,11 @@ func (dm *demuxer) run() {
 				dm.stopped <- struct{}{}
 				return
 			}
-			oEvents := dm.handle(event)
+			oEvents, err := dm.handle(event)
+			if err != nil {
+				// TODO Termination time
+				return
+			}
 			for _, event := range oEvents {
 				dm.eventbus <- event
 			}
@@ -38,9 +44,12 @@ func (dm *demuxer) run() {
 			if !ok {
 				fmt.Printf("demuxer: scheduler output closed\n")
 				continue
-				// todo: close?
 			}
-			oEvents := dm.handle(event)
+			oEvents, err := dm.handle(event)
+			if err != nil {
+				// TODO tTermination time
+				return
+			}
 			for _, event := range oEvents {
 				dm.eventbus <- event
 			}
@@ -48,39 +57,35 @@ func (dm *demuxer) run() {
 			if !ok {
 				fmt.Printf("demuxer: processor output closed\n")
 				continue
-				// todo: close?
 			}
-			oEvents := dm.handle(event)
+			oEvents, err := dm.handle(event)
+			if err != nil {
+				// TODO tTermination time
+				return
+			}
 			for _, event := range oEvents {
 				dm.eventbus <- event
 			}
+		case err := <-dm.scheduler.finished:
+			dm.finished <- err
+		case err := <-dm.processor.finished:
+			dm.finished <- err
 		}
 	}
 }
 
-// XXX: What is the corerct behaviour here?
-// onPcFinish, process no further events
-// OR onPcFinish, process all queued events and then close
-func (dm *demuxer) handle(event Event) Events {
-	switch event.(type) {
-	case pcFinished:
-		// dm.stop()
-		fmt.Println("demuxer received pcFinished")
-		dm.finished <- struct{}{}
-	default:
-		received := dm.scheduler.send(event)
-		if !received {
-			return Events{scFull{}} // backpressure
-		}
-
-		received = dm.processor.send(event)
-		if !received {
-			return Events{pcFull{}} // backpressure
-		}
-
-		return Events{}
+func (dm *demuxer) handle(event Event) (Events, error) {
+	received := dm.scheduler.send(event)
+	if !received {
+		return Events{scFull{}}, nil // backpressure
 	}
-	return Events{}
+
+	received = dm.processor.send(event)
+	if !received {
+		return Events{pcFull{}}, nil // backpressure
+	}
+
+	return Events{}, nil
 }
 
 func (dm *demuxer) send(event Event) bool {
@@ -98,4 +103,13 @@ func (dm *demuxer) stop() {
 	fmt.Printf("demuxer stop\n")
 	close(dm.eventbus)
 	<-dm.stopped
+	dm.terminate(fmt.Errorf("stopped"))
+}
+
+func (dm *demuxer) terminate(reason error) {
+	dm.finished <- reason
+}
+
+func (dm *demuxer) wait() error {
+	return <-dm.finished
 }

blockchain/v2/reactor.go

@@ -5,18 +5,18 @@ import (
 	"time"
 )
 
-func schedulerHandle(event Event) Events {
+func schedulerHandle(event Event) (Events, error) {
 	switch event.(type) {
 	case timeCheck:
 		fmt.Println("scheduler handle timeCheck")
 	case testEvent:
 		fmt.Println("scheduler handle testEvent")
-		return Events{scTestEvent{}}
+		return Events{scTestEvent{}}, nil
 	}
-	return Events{}
+	return Events{}, nil
 }
 
-func processorHandle(event Event) Events {
+func processorHandle(event Event) (Events, error) {
 	switch event.(type) {
 	case timeCheck:
 		fmt.Println("processor handle timeCheck")
@@ -24,10 +24,9 @@ func processorHandle(event Event) Events {
 		fmt.Println("processor handle testEvent")
 	case scTestEvent:
 		fmt.Println("processor handle scTestEvent")
-		// should i stop myself?
-		return Events{pcFinished{}}
+		return Events{}, fmt.Errorf("processor done")
 	}
-	return Events{}
+	return Events{}, nil
 }
 
 // reactor
@@ -69,7 +68,6 @@ func (r *Reactor) Start() {
 }
 
 func (r *Reactor) Wait() {
-	<-r.demuxer.finished // maybe put this in a wait method
 	fmt.Println("completed routines")
 	r.Stop()
 }

blockchain/v2/routine.go

@@ -2,15 +2,16 @@ package v2
 
 import (
 	"fmt"
+	"sync/atomic"
 )
 
 // TODO
-// break out routines
 // logging
 // metrics
 
-type handleFunc = func(event Event) Events
+type handleFunc = func(event Event) (Events, error)
 
+// XXX: This should comply with the BaseService interface
 // Routine
 type Routine struct {
 	name     string
@@ -18,7 +19,8 @@ type Routine struct {
 	errors   chan error
 	output   chan Event
 	stopped  chan struct{}
-	finished chan struct{}
+	finished chan error
+	running  *uint32
 	handle   handleFunc
 }
 
@@ -29,42 +31,56 @@ func newRoutine(name string, output chan Event, handleFunc handleFunc) *Routine
 		errors:   make(chan error, 1),
 		output:   output,
 		stopped:  make(chan struct{}, 1),
-		finished: make(chan struct{}, 1),
+		finished: make(chan error, 1),
+		running:  new(uint32),
 		handle:   handleFunc,
 	}
 }
 
-// TODO: refactor the handle to return an second variable, error which can signal
-//to the run looop when the handler is done
 func (rt *Routine) run() {
 	fmt.Printf("%s: run\n", rt.name)
+	starting := atomic.CompareAndSwapUint32(rt.running, uint32(0), uint32(1))
+	if !starting {
+		panic("Routine has already started")
+	}
+	errorsDrained := false
 	for {
+		if !rt.isRunning() {
+			break
+		}
 		select {
 		case iEvent, ok := <-rt.input:
 			if !ok {
+				if !errorsDrained {
+					continue // wait for errors to be drainned
+				}
 				fmt.Printf("%s: stopping\n", rt.name)
 				rt.stopped <- struct{}{}
 				return
 			}
-			oEvents := rt.handle(iEvent)
-			// XXX: this should check for error and exit if error
+			oEvents, err := rt.handle(iEvent)
+			if err != nil {
+				rt.terminate(err)
+				return
+			}
+
 			fmt.Printf("%s handled %d events\n", rt.name, len(oEvents))
 			for _, event := range oEvents {
-				// check for finished
-				if _, ok := event.(routineFinished); ok {
-					fmt.Printf("%s: finished\n", rt.name)
-					rt.finished <- struct{}{}
-					return
-				}
 				fmt.Println("writting back to output")
 				rt.output <- event
 			}
 		case iEvent, ok := <-rt.errors:
 			if !ok {
 				fmt.Printf("%s: errors closed\n", rt.name)
+				errorsDrained = true
 				continue
 			}
-			oEvents := rt.handle(iEvent)
+			// rename flush?
+			oEvents, err := rt.handle(iEvent)
+			if err != nil {
+				rt.terminate(err)
+				return
+			}
 			fmt.Printf("%s handled %d events from errors\n", rt.name, len(oEvents))
 			for _, event := range oEvents {
 				rt.output <- event
@@ -80,6 +96,7 @@ func (rt *Routine) feedback() {
 
 func (rt *Routine) send(event Event) bool {
 	fmt.Printf("%s: send\n", rt.name)
+	// XXX: What if we arn't running? this will panic
 	if err, ok := event.(error); ok {
 		select {
 		case rt.errors <- err:
@@ -99,15 +116,29 @@ func (rt *Routine) send(event Event) bool {
 	}
 }
 
+func (rt *Routine) isRunning() bool {
+	return atomic.LoadUint32(rt.running) == 1
+}
+
+// rename flush?
 func (rt *Routine) stop() {
 	fmt.Printf("%s: stop\n", rt.name)
-	close(rt.errors)
 	close(rt.input)
-	<-rt.stopped
+	close(rt.errors)
+	<-rt.stopped // stuck here
+	rt.terminate(fmt.Errorf("routine stopped"))
+}
+
+func (rt *Routine) terminate(reason error) {
+	stopped := atomic.CompareAndSwapUint32(rt.running, uint32(1), uint32(0))
+	if !stopped {
+		panic("called stop but already stopped")
+	}
+	rt.finished <- reason
 }
 
 // XXX: this should probably produced the finished
 // channel and let the caller deicde how long to wait
-func (rt *Routine) wait() {
-	<-rt.finished
+func (rt *Routine) wait() error {
+	return <-rt.finished
 }

blockchain/v2/routine_test.go

@@ -1,6 +1,7 @@
 package v2
 
 import (
+	"fmt"
 	"testing"
 	"time"
 )
@@ -8,14 +9,16 @@ import (
 type eventA struct{}
 type eventB struct{}
 
-func simpleHandler(event Event) Events {
+var done = fmt.Errorf("done")
+
+func simpleHandler(event Event) (Events, error) {
 	switch event.(type) {
 	case eventA:
-		return Events{eventB{}}
+		return Events{eventB{}}, nil
 	case eventB:
-		return Events{routineFinished{}}
+		return Events{routineFinished{}}, done
 	}
-	return Events{}
+	return Events{}, nil
 }
 
 func TestRoutine(t *testing.T) {
@@ -32,17 +35,17 @@ func TestRoutine(t *testing.T) {
 
 func genStatefulHandler(maxCount int) handleFunc {
 	counter := 0
-	return func(event Event) Events {
+	return func(event Event) (Events, error) {
 		switch event.(type) {
 		case eventA:
 			counter += 1
 			if counter >= maxCount {
-				return Events{routineFinished{}}
+				return Events{}, done
 			}
 
-			return Events{eventA{}}
+			return Events{eventA{}}, nil
 		}
-		return Events{}
+		return Events{}, nil
 	}
 }
 
@@ -59,14 +62,14 @@ func TestStatefulRoutine(t *testing.T) {
 	routine.wait()
 }
 
-func handleWithErrors(event Event) Events {
+func handleWithErrors(event Event) (Events, error) {
 	switch event.(type) {
 	case eventA:
-		return Events{}
+		return Events{}, nil
 	case errEvent:
-		return Events{routineFinished{}}
+		return Events{}, done
 	}
-	return Events{}
+	return Events{}, nil
 }
 
 func TestErrorSaturation(t *testing.T) {