diff --git a/core/bloombits/matcher.go b/core/bloombits/matcher.go index e33de018a5..baab3888aa 100644 --- a/core/bloombits/matcher.go +++ b/core/bloombits/matcher.go @@ -60,6 +60,7 @@ type Retrieval struct { Bit uint Sections []uint64 Bitsets [][]byte + Error error } // Matcher is a pipelined system of schedulers and logic matchers which perform @@ -502,15 +503,27 @@ func (m *Matcher) distributor(dist chan *request, session *MatcherSession) { type MatcherSession struct { matcher *Matcher - quit chan struct{} // Quit channel to request pipeline termination - kill chan struct{} // Term channel to signal non-graceful forced shutdown - pend sync.WaitGroup + quit chan struct{} // Quit channel to request pipeline termination + kill chan struct{} // Term channel to signal non-graceful forced shutdown + err error + stopping bool + lock sync.Mutex + pend sync.WaitGroup } // Close stops the matching process and waits for all subprocesses to terminate // before returning. The timeout may be used for graceful shutdown, allowing the // currently running retrievals to complete before this time. -func (s *MatcherSession) Close(timeout time.Duration) { +func (s *MatcherSession) Close() { + s.lock.Lock() + stopping := s.stopping + s.stopping = true + s.lock.Unlock() + // ensure that we only close the session once + if stopping { + return + } + // Bail out if the matcher is not running select { case <-s.quit: @@ -519,10 +532,26 @@ func (s *MatcherSession) Close(timeout time.Duration) { } // Signal termination and wait for all goroutines to tear down close(s.quit) - time.AfterFunc(timeout, func() { close(s.kill) }) + time.AfterFunc(time.Second, func() { close(s.kill) }) s.pend.Wait() } +// setError sets an error and stops the session +func (s *MatcherSession) setError(err error) { + s.lock.Lock() + s.err = err + s.lock.Unlock() + s.Close() +} + +// Error returns an error if one has happened during the session +func (s *MatcherSession) Error() error { + s.lock.Lock() + defer s.lock.Unlock() + + return s.err +} + // AllocateRetrieval assigns a bloom bit index to a client process that can either // immediately reuest and fetch the section contents assigned to this bit or wait // a little while for more sections to be requested. @@ -621,6 +650,10 @@ func (s *MatcherSession) Multiplex(batch int, wait time.Duration, mux chan chan request <- &Retrieval{Bit: bit, Sections: sections} result := <-request + if result.Error != nil { + s.setError(result.Error) + } + s.DeliverSections(result.Bit, result.Sections, result.Bitsets) } } diff --git a/core/bloombits/matcher_test.go b/core/bloombits/matcher_test.go index f95d0ea9e0..86db380a87 100644 --- a/core/bloombits/matcher_test.go +++ b/core/bloombits/matcher_test.go @@ -163,7 +163,7 @@ func testMatcher(t *testing.T, filter [][]bloomIndexes, blocks uint64, intermitt } // If we're testing intermittent mode, abort and restart the pipeline if intermittent { - session.Close(time.Second) + session.Close() close(quit) quit = make(chan struct{}) @@ -183,7 +183,7 @@ func testMatcher(t *testing.T, filter [][]bloomIndexes, blocks uint64, intermitt t.Errorf("filter = %v blocks = %v intermittent = %v: expected closed channel, got #%v", filter, blocks, intermittent, match) } // Clean up the session and ensure we match the expected retrieval count - session.Close(time.Second) + session.Close() close(quit) if retrievals != 0 && requested != retrievals { diff --git a/core/database_util.go b/core/database_util.go index 1730a048e7..fb3d1a1e8a 100644 --- a/core/database_util.go +++ b/core/database_util.go @@ -332,14 +332,13 @@ func GetReceipt(db DatabaseReader, hash common.Hash) (*types.Receipt, common.Has // GetBloomBits retrieves the compressed bloom bit vector belonging to the given // section and bit index from the. -func GetBloomBits(db DatabaseReader, bit uint, section uint64, head common.Hash) []byte { +func GetBloomBits(db DatabaseReader, bit uint, section uint64, head common.Hash) ([]byte, error) { key := append(append(bloomBitsPrefix, make([]byte, 10)...), head.Bytes()...) binary.BigEndian.PutUint16(key[1:], uint16(bit)) binary.BigEndian.PutUint64(key[3:], section) - bits, _ := db.Get(key) - return bits + return db.Get(key) } // WriteCanonicalHash stores the canonical hash for the given block number. diff --git a/eth/bloombits.go b/eth/bloombits.go index ed7b14976c..cecbb761a2 100644 --- a/eth/bloombits.go +++ b/eth/bloombits.go @@ -58,15 +58,18 @@ func (eth *Ethereum) startBloomHandlers() { case request := <-eth.bloomRequests: task := <-request - task.Bitsets = make([][]byte, len(task.Sections)) for i, section := range task.Sections { head := core.GetCanonicalHash(eth.chainDb, (section+1)*params.BloomBitsBlocks-1) - blob, err := bitutil.DecompressBytes(core.GetBloomBits(eth.chainDb, task.Bit, section, head), int(params.BloomBitsBlocks)/8) - if err != nil { - panic(err) + if compVector, err := core.GetBloomBits(eth.chainDb, task.Bit, section, head); err == nil { + if blob, err := bitutil.DecompressBytes(compVector, int(params.BloomBitsBlocks)/8); err == nil { + task.Bitsets[i] = blob + } else { + task.Error = err + } + } else { + task.Error = err } - task.Bitsets[i] = blob } request <- task } diff --git a/eth/filters/filter.go b/eth/filters/filter.go index 026cbf95ca..490fd0b906 100644 --- a/eth/filters/filter.go +++ b/eth/filters/filter.go @@ -19,7 +19,6 @@ package filters import ( "context" "math/big" - "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/core" @@ -140,7 +139,7 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err if err != nil { return nil, err } - defer session.Close(time.Second) + defer session.Close() f.backend.ServiceFilter(ctx, session) @@ -152,9 +151,13 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err case number, ok := <-matches: // Abort if all matches have been fulfilled if !ok { - f.begin = int64(end) + 1 - return logs, nil + err := session.Error() + if err == nil { + f.begin = int64(end) + 1 + } + return logs, err } + f.begin = int64(number) + 1 // Retrieve the suggested block and pull any truly matching logs header, err := f.backend.HeaderByNumber(ctx, rpc.BlockNumber(number)) if header == nil || err != nil { diff --git a/eth/filters/filter_system_test.go b/eth/filters/filter_system_test.go index bc3511f23a..7da114fda4 100644 --- a/eth/filters/filter_system_test.go +++ b/eth/filters/filter_system_test.go @@ -109,7 +109,7 @@ func (b *testBackend) ServiceFilter(ctx context.Context, session *bloombits.Matc for i, section := range task.Sections { if rand.Int()%4 != 0 { // Handle occasional missing deliveries head := core.GetCanonicalHash(b.db, (section+1)*params.BloomBitsBlocks-1) - task.Bitsets[i] = core.GetBloomBits(b.db, task.Bit, section, head) + task.Bitsets[i], _ = core.GetBloomBits(b.db, task.Bit, section, head) } } request <- task