diff options
Diffstat (limited to 'core/nonblocking-application.go')
-rw-r--r-- | core/nonblocking-application.go | 126 |
1 files changed, 126 insertions, 0 deletions
diff --git a/core/nonblocking-application.go b/core/nonblocking-application.go new file mode 100644 index 0000000..4c06ab9 --- /dev/null +++ b/core/nonblocking-application.go @@ -0,0 +1,126 @@ +// Copyright 2018 The dexon-consensus-core Authors +// This file is part of the dexon-consensus-core library. +// +// The dexon-consensus-core library is free software: you can redistribute it +// and/or modify it under the terms of the GNU Lesser General Public License as +// published by the Free Software Foundation, either version 3 of the License, +// or (at your option) any later version. +// +// The dexon-consensus-core library is distributed in the hope that it will be +// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser +// General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the dexon-consensus-core library. If not, see +// <http://www.gnu.org/licenses/>. + +package core + +import ( + "fmt" + "sync" + "time" + + "github.com/dexon-foundation/dexon-consensus-core/common" +) + +type stronglyAckedEvent struct { + blockHash common.Hash +} + +type totalOrderingDeliverEvent struct { + blockHashes common.Hashes + early bool +} + +type deliverBlockEvent struct { + blockHash common.Hash + timestamp time.Time +} + +// nonBlockingApplication implements Application and is a decorator for +// Application that makes the methods to be non-blocking. +type nonBlockingApplication struct { + app Application + eventChan chan interface{} + events []interface{} + eventsChange *sync.Cond + running sync.WaitGroup +} + +func newNonBlockingApplication(app Application) *nonBlockingApplication { + nonBlockingApp := &nonBlockingApplication{ + app: app, + eventChan: make(chan interface{}, 6), + events: make([]interface{}, 0, 100), + eventsChange: sync.NewCond(&sync.Mutex{}), + } + go nonBlockingApp.run() + return nonBlockingApp +} + +func (app *nonBlockingApplication) addEvent(event interface{}) { + app.eventsChange.L.Lock() + defer app.eventsChange.L.Unlock() + app.events = append(app.events, event) + app.eventsChange.Broadcast() +} + +func (app *nonBlockingApplication) run() { + // This go routine consume the first event from events and call the + // corresponding method of app. + for { + var event interface{} + func() { + app.eventsChange.L.Lock() + defer app.eventsChange.L.Unlock() + for len(app.events) == 0 { + app.eventsChange.Wait() + } + event = app.events[0] + app.events = app.events[1:] + app.running.Add(1) + }() + switch e := event.(type) { + case stronglyAckedEvent: + app.app.StronglyAcked(e.blockHash) + case totalOrderingDeliverEvent: + app.app.TotalOrderingDeliver(e.blockHashes, e.early) + case deliverBlockEvent: + app.app.DeliverBlock(e.blockHash, e.timestamp) + default: + fmt.Printf("Unknown event %v.", e) + } + app.running.Done() + app.eventsChange.Broadcast() + } +} + +// wait will wait for all event in events finishes. +func (app *nonBlockingApplication) wait() { + app.eventsChange.L.Lock() + defer app.eventsChange.L.Unlock() + for len(app.events) > 0 { + app.eventsChange.Wait() + } + app.running.Wait() +} + +// StronglyAcked is called when a block is strongly acked. +func (app *nonBlockingApplication) StronglyAcked(blockHash common.Hash) { + app.addEvent(stronglyAckedEvent{blockHash}) +} + +// TotalOrderingDeliver is called when the total ordering algorithm deliver +// a set of block. +func (app *nonBlockingApplication) TotalOrderingDeliver( + blockHashes common.Hashes, early bool) { + app.addEvent(totalOrderingDeliverEvent{blockHashes, early}) +} + +// DeliverBlock is called when a block is add to the compaction chain. +func (app *nonBlockingApplication) DeliverBlock( + blockHash common.Hash, timestamp time.Time) { + app.addEvent(deliverBlockEvent{blockHash, timestamp}) +} |