tendermint/blockchain/v2/routines.go

173 lines
3.5 KiB
Go
Raw Normal View History

2019-07-25 16:02:47 +02:00
package v2
import (
"fmt"
)
2019-07-29 11:58:13 +02:00
// XXX: what about state here, we need per routine state
// So handle func should take an event and a reference to state and return
// events and the new state
2019-07-29 08:45:59 +02:00
type handleFunc = func(event Event) Events
2019-07-27 10:58:58 +02:00
2019-07-29 08:45:59 +02:00
// Routine
type Routine struct {
name string
2019-07-27 11:38:37 +02:00
input chan Event
2019-07-29 12:40:59 +02:00
errors chan errEvent
2019-07-27 11:38:37 +02:00
output chan Event
stopped chan struct{}
2019-07-29 08:45:59 +02:00
handle handleFunc
2019-07-25 16:02:47 +02:00
}
2019-07-29 08:45:59 +02:00
func newRoutine(name string, output chan Event, handleFunc handleFunc) *Routine {
return &Routine{
name: name,
input: make(chan Event, 1),
2019-07-29 12:40:59 +02:00
errors: make(chan errEvent, 1),
output: output,
stopped: make(chan struct{}, 1),
2019-07-29 08:45:59 +02:00
handle: handleFunc,
2019-07-27 10:58:58 +02:00
}
}
2019-07-29 08:45:59 +02:00
func (rt *Routine) run() {
fmt.Printf("%s: run\n", rt.name)
2019-07-25 16:02:47 +02:00
for {
2019-07-29 12:40:59 +02:00
select {
case iEvent, ok := <-rt.input:
if !ok {
fmt.Printf("%s: stopping\n", rt.name)
rt.stopped <- struct{}{}
2019-07-29 12:47:43 +02:00
return
2019-07-29 12:40:59 +02:00
}
oEvents := rt.handle(iEvent)
fmt.Printf("%s handled %d events\n", rt.name, len(oEvents))
for _, event := range oEvents {
rt.output <- event
}
case iEvent, ok := <-rt.errors:
if !ok {
fmt.Printf("%s: errors closed\n", rt.name)
continue
}
oEvents := rt.handle(iEvent)
fmt.Printf("%s handled %d events from errors\n", rt.name, len(oEvents))
for _, event := range oEvents {
rt.output <- event
}
2019-07-27 00:25:32 +02:00
}
2019-07-25 16:02:47 +02:00
}
}
2019-07-29 08:45:59 +02:00
func (rt *Routine) send(event Event) bool {
fmt.Printf("%s: send\n", rt.name)
2019-07-29 12:40:59 +02:00
if err, ok := event.(errEvent); ok {
select {
case rt.errors <- err:
return true
default:
fmt.Printf("%s: errors channel was full\n", rt.name)
return false
}
} else {
select {
case rt.input <- event:
return true
default:
fmt.Printf("%s: channel was full\n", rt.name)
return false
}
2019-07-27 00:25:32 +02:00
}
}
2019-07-29 08:45:59 +02:00
func (rt *Routine) stop() {
fmt.Printf("%s: stop\n", rt.name)
2019-07-29 12:40:59 +02:00
close(rt.errors)
2019-07-29 08:45:59 +02:00
close(rt.input)
<-rt.stopped
}
func schedulerHandle(event Event) Events {
2019-07-27 10:58:58 +02:00
switch event.(type) {
2019-07-27 00:25:32 +02:00
case timeCheck:
fmt.Println("scheduler handle timeCheck")
case testEvent:
fmt.Println("scheduler handle testEvent")
2019-07-29 11:58:13 +02:00
return Events{scTestEvent{}}
2019-07-27 00:25:32 +02:00
}
return Events{}
}
2019-07-29 08:45:59 +02:00
func processorHandle(event Event) Events {
2019-07-27 10:58:58 +02:00
switch event.(type) {
2019-07-27 00:25:32 +02:00
case timeCheck:
2019-07-27 10:58:58 +02:00
fmt.Println("processor handle timeCheck")
2019-07-27 00:25:32 +02:00
case testEvent:
fmt.Println("processor handle testEvent")
2019-07-29 11:58:13 +02:00
case scTestEvent:
fmt.Println("processor handle scTestEvent")
// should i stop myself?
return Events{pcFinished{}}
2019-07-27 00:25:32 +02:00
}
return Events{}
}
2019-07-29 11:58:13 +02:00
type demuxer struct {
eventbus chan Event
scheduler *Routine
processor *Routine
finished chan struct{}
stopped chan struct{}
}
func newDemuxer(scheduler *Routine, processor *Routine) *demuxer {
return &demuxer{
eventbus: make(chan Event, 10),
scheduler: scheduler,
processor: processor,
stopped: make(chan struct{}, 1),
finished: make(chan struct{}, 1),
}
}
func (dm *demuxer) run() {
fmt.Printf("demuxer: run\n")
for {
select {
case event, ok := <-dm.eventbus:
if !ok {
fmt.Printf("demuxer: stopping\n")
dm.stopped <- struct{}{}
2019-07-29 12:47:43 +02:00
return
2019-07-29 11:58:13 +02:00
}
oEvents := dm.handle(event)
for _, event := range oEvents {
dm.eventbus <- event
}
case event, ok := <-dm.scheduler.output:
if !ok {
fmt.Printf("demuxer: scheduler output closed\n")
continue
// todo: close?
}
oEvents := dm.handle(event)
for _, event := range oEvents {
dm.eventbus <- event
}
case event, ok := <-dm.processor.output:
if !ok {
fmt.Printf("demuxer: pricessor output closed\n")
continue
// todo: close?
}
oEvents := dm.handle(event)
for _, event := range oEvents {
dm.eventbus <- event
}
}
}
}
type scFull struct{}
type pcFull struct{}