aboutsummaryrefslogtreecommitdiff
path: root/core/bloombits/scheduler.go
diff options
context:
space:
mode:
authorDeterminant <[email protected]>2020-07-30 14:18:44 -0400
committerDeterminant <[email protected]>2020-07-30 14:18:44 -0400
commit0444e66f640999c15496066637841efcc0433934 (patch)
treec19aec2dced2e9129c880c19c52ca0f87b3d62f6 /core/bloombits/scheduler.go
parentcffa0954bbdb43821d1b71d00f99fb705cecd25b (diff)
parent1f49826de2bb8bb4f5f99f69fd2beb039b1172d9 (diff)
Merge branch 'multi-coin'
Diffstat (limited to 'core/bloombits/scheduler.go')
-rw-r--r--core/bloombits/scheduler.go181
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)
+ }
+ }
+}