aboutsummaryrefslogtreecommitdiffstats
path: root/eth/fetcher/fetcher.go
diff options
context:
space:
mode:
Diffstat (limited to 'eth/fetcher/fetcher.go')
-rw-r--r--eth/fetcher/fetcher.go21
1 files changed, 21 insertions, 0 deletions
diff --git a/eth/fetcher/fetcher.go b/eth/fetcher/fetcher.go
index e8a9cc093..34d368780 100644
--- a/eth/fetcher/fetcher.go
+++ b/eth/fetcher/fetcher.go
@@ -56,6 +56,7 @@ type inject struct {
type Fetcher struct {
// Various event channels
notify chan *announce
+ insert chan *inject
filter chan chan []*types.Block
quit chan struct{}
@@ -69,6 +70,7 @@ type Fetcher struct {
func New(hasBlock hashCheckFn, importBlock blockImporterFn, chainHeight chainHeightFn) *Fetcher {
return &Fetcher{
notify: make(chan *announce),
+ insert: make(chan *inject),
filter: make(chan chan []*types.Block),
quit: make(chan struct{}),
hasBlock: hasBlock,
@@ -106,6 +108,20 @@ func (f *Fetcher) Notify(peer string, hash common.Hash, time time.Time, fetcher
}
}
+// Enqueue tries to fill gaps the the fetcher's future import queue.
+func (f *Fetcher) Enqueue(peer string, block *types.Block) error {
+ op := &inject{
+ origin: peer,
+ block: block,
+ }
+ select {
+ case f.insert <- op:
+ return nil
+ case <-f.quit:
+ return errTerminated
+ }
+}
+
// Filter extracts all the blocks that were explicitly requested by the fetcher,
// returning those that should be handled differently.
func (f *Fetcher) Filter(blocks types.Blocks) types.Blocks {
@@ -192,6 +208,11 @@ func (f *Fetcher) loop() {
}
announced[notification.hash] = append(announced[notification.hash], notification)
+ case op := <-f.insert:
+ // A direct block insertion was requested, try and fill any pending gaps
+ queued.Push(op, -float32(op.block.NumberU64()))
+ glog.V(logger.Detail).Infof("Peer %s: filled block %x, total %v", op.origin, op.block.Hash().Bytes()[:4], queued.Size())
+
case hash := <-done:
// A pending import finished, remove all traces of the notification
delete(announced, hash)