Merge branch 'multi-coin'

1 files changed, 181 insertions, 0 deletions

diff --git a/core/bloombits/scheduler.go b/core/bloombits/scheduler.go
new file mode 100644
index 0000000..6449c74
--- /dev/null
+++ b/core/bloombits/scheduler.go
@@ -0,0 +1,181 @@
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum 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 go-ethereum 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 go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package bloombits
+
+import (
+	"sync"
+)
+
+// request represents a bloom retrieval task to prioritize and pull from the local
+// database or remotely from the network.
+type request struct {
+	section uint64 // Section index to retrieve the a bit-vector from
+	bit     uint   // Bit index within the section to retrieve the vector of
+}
+
+// response represents the state of a requested bit-vector through a scheduler.
+type response struct {
+	cached []byte        // Cached bits to dedup multiple requests
+	done   chan struct{} // Channel to allow waiting for completion
+}
+
+// scheduler handles the scheduling of bloom-filter retrieval operations for
+// entire section-batches belonging to a single bloom bit. Beside scheduling the
+// retrieval operations, this struct also deduplicates the requests and caches
+// the results to minimize network/database overhead even in complex filtering
+// scenarios.
+type scheduler struct {
+	bit       uint                 // Index of the bit in the bloom filter this scheduler is responsible for
+	responses map[uint64]*response // Currently pending retrieval requests or already cached responses
+	lock      sync.Mutex           // Lock protecting the responses from concurrent access
+}
+
+// newScheduler creates a new bloom-filter retrieval scheduler for a specific
+// bit index.
+func newScheduler(idx uint) *scheduler {
+	return &scheduler{
+		bit:       idx,
+		responses: make(map[uint64]*response),
+	}
+}
+
+// run creates a retrieval pipeline, receiving section indexes from sections and
+// returning the results in the same order through the done channel. Concurrent
+// runs of the same scheduler are allowed, leading to retrieval task deduplication.
+func (s *scheduler) run(sections chan uint64, dist chan *request, done chan []byte, quit chan struct{}, wg *sync.WaitGroup) {
+	// Create a forwarder channel between requests and responses of the same size as
+	// the distribution channel (since that will block the pipeline anyway).
+	pend := make(chan uint64, cap(dist))
+
+	// Start the pipeline schedulers to forward between user -> distributor -> user
+	wg.Add(2)
+	go s.scheduleRequests(sections, dist, pend, quit, wg)
+	go s.scheduleDeliveries(pend, done, quit, wg)
+}
+
+// reset cleans up any leftovers from previous runs. This is required before a
+// restart to ensure the no previously requested but never delivered state will
+// cause a lockup.
+func (s *scheduler) reset() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	for section, res := range s.responses {
+		if res.cached == nil {
+			delete(s.responses, section)
+		}
+	}
+}
+
+// scheduleRequests reads section retrieval requests from the input channel,
+// deduplicates the stream and pushes unique retrieval tasks into the distribution
+// channel for a database or network layer to honour.
+func (s *scheduler) scheduleRequests(reqs chan uint64, dist chan *request, pend chan uint64, quit chan struct{}, wg *sync.WaitGroup) {
+	// Clean up the goroutine and pipeline when done
+	defer wg.Done()
+	defer close(pend)
+
+	// Keep reading and scheduling section requests
+	for {
+		select {
+		case <-quit:
+			return
+
+		case section, ok := <-reqs:
+			// New section retrieval requested
+			if !ok {
+				return
+			}
+			// Deduplicate retrieval requests
+			unique := false
+
+			s.lock.Lock()
+			if s.responses[section] == nil {
+				s.responses[section] = &response{
+					done: make(chan struct{}),
+				}
+				unique = true
+			}
+			s.lock.Unlock()
+
+			// Schedule the section for retrieval and notify the deliverer to expect this section
+			if unique {
+				select {
+				case <-quit:
+					return
+				case dist <- &request{bit: s.bit, section: section}:
+				}
+			}
+			select {
+			case <-quit:
+				return
+			case pend <- section:
+			}
+		}
+	}
+}
+
+// scheduleDeliveries reads section acceptance notifications and waits for them
+// to be delivered, pushing them into the output data buffer.
+func (s *scheduler) scheduleDeliveries(pend chan uint64, done chan []byte, quit chan struct{}, wg *sync.WaitGroup) {
+	// Clean up the goroutine and pipeline when done
+	defer wg.Done()
+	defer close(done)
+
+	// Keep reading notifications and scheduling deliveries
+	for {
+		select {
+		case <-quit:
+			return
+
+		case idx, ok := <-pend:
+			// New section retrieval pending
+			if !ok {
+				return
+			}
+			// Wait until the request is honoured
+			s.lock.Lock()
+			res := s.responses[idx]
+			s.lock.Unlock()
+
+			select {
+			case <-quit:
+				return
+			case <-res.done:
+			}
+			// Deliver the result
+			select {
+			case <-quit:
+				return
+			case done <- res.cached:
+			}
+		}
+	}
+}
+
+// deliver is called by the request distributor when a reply to a request arrives.
+func (s *scheduler) deliver(sections []uint64, data [][]byte) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	for i, section := range sections {
+		if res := s.responses[section]; res != nil && res.cached == nil { // Avoid non-requests and double deliveries
+			res.cached = data[i]
+			close(res.done)
+		}
+	}
+}