diff --git a/blockchain/v2/demuxer.go b/blockchain/v2/demuxer.go index 32165735e..9ae4c3f38 100644 --- a/blockchain/v2/demuxer.go +++ b/blockchain/v2/demuxer.go @@ -9,7 +9,7 @@ type demuxer struct { input chan Event scheduler *Routine processor *Routine - finished chan error + fin chan error stopped chan struct{} running *uint32 } @@ -26,12 +26,12 @@ func newDemuxer(scheduler *Routine, processor *Routine) *demuxer { scheduler: scheduler, processor: processor, stopped: make(chan struct{}, 1), - finished: make(chan error, 1), + fin: make(chan error, 1), running: new(uint32), } } -func (dm *demuxer) run() { +func (dm *demuxer) start() { starting := atomic.CompareAndSwapUint32(dm.running, uint32(0), uint32(1)) if !starting { panic("Routine has already started") @@ -57,7 +57,7 @@ func (dm *demuxer) run() { for _, event := range oEvents { dm.input <- event } - case event, ok := <-dm.scheduler.output(): + case event, ok := <-dm.scheduler.next(): if !ok { fmt.Printf("demuxer: scheduler output closed\n") continue @@ -70,7 +70,7 @@ func (dm *demuxer) run() { for _, event := range oEvents { dm.input <- event } - case event, ok := <-dm.processor.output(): + case event, ok := <-dm.processor.next(): if !ok { fmt.Printf("demuxer: processor output closed\n") continue @@ -88,12 +88,12 @@ func (dm *demuxer) run() { } func (dm *demuxer) handle(event Event) (Events, error) { - received := dm.scheduler.send(event) + received := dm.scheduler.trySend(event) if !received { return Events{scFull{}}, nil // backpressure } - received = dm.processor.send(event) + received = dm.processor.trySend(event) if !received { return Events{pcFull{}}, nil // backpressure } @@ -101,7 +101,7 @@ func (dm *demuxer) handle(event Event) (Events, error) { return Events{}, nil } -func (dm *demuxer) send(event Event) bool { +func (dm *demuxer) trySend(event Event) bool { if !dm.isRunning() { fmt.Println("dummuxer isn't running") return false @@ -133,9 +133,9 @@ func (dm *demuxer) terminate(reason error) { if !stopped { panic("called terminate but already terminated") } - dm.finished <- reason + dm.fin <- reason } -func (dm *demuxer) wait() error { - return <-dm.finished +func (dm *demuxer) final() chan error { + return dm.fin } diff --git a/blockchain/v2/reactor.go b/blockchain/v2/reactor.go index e8e10ae2d..85f669739 100644 --- a/blockchain/v2/reactor.go +++ b/blockchain/v2/reactor.go @@ -47,9 +47,9 @@ func (r *Reactor) Start() { r.demuxer = newDemuxer(r.scheduler, r.processor) r.tickerStopped = make(chan struct{}) - go r.scheduler.run() - go r.processor.run() - go r.demuxer.run() + go r.scheduler.start() + go r.processor.start() + go r.demuxer.start() for { if r.scheduler.isRunning() && r.processor.isRunning() && r.demuxer.isRunning() { @@ -57,7 +57,7 @@ func (r *Reactor) Start() { break } fmt.Println("waiting") - time.Sleep(1 * time.Second) + time.Sleep(10 * time.Millisecond) } go func() { @@ -65,7 +65,7 @@ func (r *Reactor) Start() { for { select { case <-ticker.C: - r.demuxer.send(timeCheck{}) + r.demuxer.trySend(timeCheck{}) case <-r.tickerStopped: fmt.Println("ticker stopped") return @@ -94,7 +94,7 @@ func (r *Reactor) Stop() { func (r *Reactor) Receive(event Event) { fmt.Println("receive event") - sent := r.demuxer.send(event) + sent := r.demuxer.trySend(event) if !sent { fmt.Println("demuxer is full") } diff --git a/blockchain/v2/routine.go b/blockchain/v2/routine.go index e70c76679..1addff468 100644 --- a/blockchain/v2/routine.go +++ b/blockchain/v2/routine.go @@ -11,7 +11,6 @@ import ( // * revisit panic conditions // * audit log levels // * Convert routine to an interface with concrete implmentation -// * determine the public interface type handleFunc = func(event Event) (Events, error) @@ -21,7 +20,8 @@ type Routine struct { errors chan error out chan Event stopped chan struct{} - finished chan error + rdy chan struct{} + fin chan error running *uint32 handle handleFunc logger log.Logger @@ -37,7 +37,8 @@ func newRoutine(name string, handleFunc handleFunc) *Routine { errors: make(chan error, 1), out: make(chan Event, 1), stopped: make(chan struct{}, 1), - finished: make(chan error, 1), + rdy: make(chan struct{}, 1), + fin: make(chan error, 1), running: new(uint32), stopping: new(uint32), logger: log.NewNopLogger(), @@ -53,12 +54,13 @@ func (rt *Routine) setMetrics(metrics *Metrics) { rt.metrics = metrics } -func (rt *Routine) run() { +func (rt *Routine) start() { rt.logger.Info(fmt.Sprintf("%s: run\n", rt.name)) starting := atomic.CompareAndSwapUint32(rt.running, uint32(0), uint32(1)) if !starting { panic("Routine has already started") } + rt.rdy <- struct{}{} errorsDrained := false for { if !rt.isRunning() { @@ -113,11 +115,11 @@ func (rt *Routine) run() { } func (rt *Routine) feedback() { for event := range rt.out { - rt.send(event) + rt.trySend(event) } } -func (rt *Routine) send(event Event) bool { +func (rt *Routine) trySend(event Event) bool { if !rt.isRunning() || rt.isStopping() { return false } @@ -154,7 +156,11 @@ func (rt *Routine) isStopping() bool { return atomic.LoadUint32(rt.stopping) == 1 } -func (rt *Routine) output() chan Event { +func (rt *Routine) ready() chan struct{} { + return rt.rdy +} + +func (rt *Routine) next() chan Event { return rt.out } @@ -174,21 +180,14 @@ func (rt *Routine) stop() { <-rt.stopped } +func (rt *Routine) final() chan error { + return rt.fin +} + 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 + rt.fin <- reason } - -// XXX: this should probably produced the finished -// channel and let the caller deicde how long to wait -func (rt *Routine) wait() error { - return <-rt.finished -} - -/* - Problem: - We can't write to channels from one thread and close channels from another thread -*/ diff --git a/blockchain/v2/routine_test.go b/blockchain/v2/routine_test.go index 7ef8da4d7..cbf4768a8 100644 --- a/blockchain/v2/routine_test.go +++ b/blockchain/v2/routine_test.go @@ -19,7 +19,7 @@ func simpleHandler(event Event) (Events, error) { case eventA: return Events{eventB{}}, nil case eventB: - return Events{routineFinished{}}, done + return Events{}, done } return Events{}, nil } @@ -29,53 +29,54 @@ func TestRoutine(t *testing.T) { assert.False(t, routine.isRunning(), "expected an initialized routine to not be running") - go routine.run() + go routine.start() go routine.feedback() - for { - if routine.isRunning() { - break - } - time.Sleep(10 * time.Millisecond) - } + <-routine.ready() - routine.send(eventA{}) + assert.True(t, routine.trySend(eventA{}), + "expected sending to a ready routine to succeed") - routine.stop() + assert.Equal(t, done, <-routine.final(), + "expected the final event to be done") } func TesRoutineSend(t *testing.T) { routine := newRoutine("simpleRoutine", simpleHandler) - assert.False(t, routine.send(eventA{}), + assert.False(t, routine.trySend(eventA{}), "expected sending to an unstarted routine to fail") - go routine.run() + go routine.start() go routine.feedback() - for { - if routine.isRunning() { - break - } - time.Sleep(10 * time.Millisecond) - } + <-routine.ready() - assert.True(t, routine.send(eventA{}), + assert.True(t, routine.trySend(eventA{}), "expected sending to a running routine to succeed") routine.stop() - assert.False(t, routine.send(eventA{}), + assert.False(t, routine.trySend(eventA{}), "expected sending to a stopped routine to fail") } +type finalCount struct { + count int +} + +func (f finalCount) Error() string { + return "end" +} + func genStatefulHandler(maxCount int) handleFunc { counter := 0 return func(event Event) (Events, error) { + // golint fixme switch event.(type) { case eventA: counter += 1 if counter >= maxCount { - return Events{}, done + return Events{}, finalCount{counter} } return Events{eventA{}}, nil @@ -85,23 +86,27 @@ func genStatefulHandler(maxCount int) handleFunc { } func TestStatefulRoutine(t *testing.T) { - handler := genStatefulHandler(10) + count := 10 + handler := genStatefulHandler(count) routine := newRoutine("statefulRoutine", handler) routine.setLogger(log.TestingLogger()) - go routine.run() + go routine.start() go routine.feedback() - for { - if routine.isRunning() { - break - } - time.Sleep(10 * time.Millisecond) + <-routine.ready() + + assert.True(t, routine.trySend(eventA{}), + "expected sending to a started routine to succeed") + + final := <-routine.final() + fnl, ok := final.(finalCount) + if ok { + assert.Equal(t, count, fnl.count, + "expected the routine to count to 10") + } else { + t.Fail() } - - routine.send(eventA{}) - - routine.stop() } func handleWithErrors(event Event) (Events, error) { @@ -116,22 +121,17 @@ func handleWithErrors(event Event) (Events, error) { func TestErrorSaturation(t *testing.T) { routine := newRoutine("errorRoutine", handleWithErrors) - go routine.run() + go routine.start() + <-routine.ready() go func() { for { - routine.send(eventA{}) + routine.trySend(eventA{}) time.Sleep(10 * time.Millisecond) } }() - for { - if routine.isRunning() { - break - } - time.Sleep(10 * time.Millisecond) - } - assert.True(t, routine.send(errEvent{}), + assert.True(t, routine.trySend(errEvent{}), "expected send to succeed even when saturated") - routine.wait() + assert.Equal(t, done, <-routine.final()) }