eth/filters: retrieve logs in async

eth/filters/filter.go

@@ -106,32 +106,32 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
  }
  return f.blockLogs(ctx, header)
  }
- // Short-cut if all we care about is pending logs
- if f.begin == rpc.PendingBlockNumber.Int64() {
- if f.end != rpc.PendingBlockNumber.Int64() {
- return nil, errors.New("invalid block range")
- }
- return f.pendingLogs()
- }
- // Figure out the limits of the filter range
- header, _ := f.sys.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
- if header == nil {
- return nil, nil
- }
+
  var (
- err error
- head = header.Number.Int64()
- pending = f.end == rpc.PendingBlockNumber.Int64()
+ beginPending = f.begin == rpc.PendingBlockNumber.Int64()
+ endPending = f.end == rpc.PendingBlockNumber.Int64()
  )
+
+ // special case for pending logs
+ if beginPending && !endPending {
+ return nil, errors.New("invalid block range")
+ }
+
+ // Short-cut if all we care about is pending logs
+ if beginPending && endPending {
+ return f.pendingLogs(), nil
+ }
+
  resolveSpecial := func(number int64) (int64, error) {
  var hdr *types.Header
  switch number {
- case rpc.LatestBlockNumber.Int64():
- return head, nil
- case rpc.PendingBlockNumber.Int64():
+ case rpc.LatestBlockNumber.Int64(), rpc.PendingBlockNumber.Int64():
  // we should return head here since we've already captured
  // that we need to get the pending logs in the pending boolean above
- return head, nil
+ hdr, _ = f.sys.backend.HeaderByNumber(ctx, rpc.LatestBlockNumber)
+ if hdr == nil {
+ return 0, errors.New("latest header not found")
+ }
  case rpc.FinalizedBlockNumber.Int64():
  hdr, _ = f.sys.backend.HeaderByNumber(ctx, rpc.FinalizedBlockNumber)
  if hdr == nil {
@@ -147,57 +147,92 @@ func (f *Filter) Logs(ctx context.Context) ([]*types.Log, error) {
  }
  return hdr.Number.Int64(), nil
  }
+
+ var err error
+ // range query need to resolve the special begin/end block number
  if f.begin, err = resolveSpecial(f.begin); err != nil {
  return nil, err
  }
  if f.end, err = resolveSpecial(f.end); err != nil {
  return nil, err
  }
- // Gather all indexed logs, and finish with non indexed ones
+
+ logChan, errChan := f.rangeLogsAsync(ctx)
+ var logs []*types.Log
+ for {
+ select {
+ case log := <-logChan:
+ logs = append(logs, log)
+ case err := <-errChan:
+ if err != nil {
+ // if an error occurs during extraction, we do return the extracted data
+ return logs, err
+ }
+ // Append the pending ones
+ if endPending {
+ pendingLogs := f.pendingLogs()
+ logs = append(logs, pendingLogs...)
+ }
+ return logs, nil
+ }
+ }
+}
+
+// rangeLogsAsync retrieves block-range logs that match the filter criteria asynchronously,
+// it creates and returns two channels: one for delivering log data, and one for reporting errors.
+func (f *Filter) rangeLogsAsync(ctx context.Context) (chan *types.Log, chan error) {
  var (
- logs []*types.Log
- end = uint64(f.end)
- size, sections = f.sys.backend.BloomStatus()
+ logChan = make(chan *types.Log)
+ errChan = make(chan error)
  )
- if indexed := sections * size; indexed > uint64(f.begin) {
- if indexed > end {
- logs, err = f.indexedLogs(ctx, end)
- } else {
- logs, err = f.indexedLogs(ctx, indexed-1)
- }
- if err != nil {
- return logs, err
+
+ go func() {
+ defer func() {
+ close(errChan)
+ close(logChan)
+ }()
+
+ // Gather all indexed logs, and finish with non indexed ones
+ var (
+ end = uint64(f.end)
+ size, sections = f.sys.backend.BloomStatus()
+ err error
+ )
+ if indexed := sections * size; indexed > uint64(f.begin) {
+ if indexed > end {
+ indexed = end + 1
+ }
+ if err = f.indexedLogs(ctx, indexed-1, logChan); err != nil {
+ errChan <- err
+ return
+ }
  }
- }
- rest, err := f.unindexedLogs(ctx, end)
- logs = append(logs, rest...)
- if pending {
- pendingLogs, err := f.pendingLogs()
- if err != nil {
- return nil, err
+
+ if err := f.unindexedLogs(ctx, end, logChan); err != nil {
+ errChan <- err
+ return
  }
- logs = append(logs, pendingLogs...)
- }
- return logs, err
+
+ errChan <- nil
+ }()
+
+ return logChan, errChan
 }
 
 // indexedLogs returns the logs matching the filter criteria based on the bloom
 // bits indexed available locally or via the network.
-func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) {
+func (f *Filter) indexedLogs(ctx context.Context, end uint64, logChan chan *types.Log) error {
  // Create a matcher session and request servicing from the backend
  matches := make(chan uint64, 64)
 
  session, err := f.matcher.Start(ctx, uint64(f.begin), end, matches)
  if err != nil {
- return nil, err
+ return err
  }
  defer session.Close()
 
  f.sys.backend.ServiceFilter(ctx, session)
 
- // Iterate over the matches until exhausted or context closed
- var logs []*types.Log
-
  for {
  select {
  case number, ok := <-matches:
@@ -207,47 +242,50 @@ func (f *Filter) indexedLogs(ctx context.Context, end uint64) ([]*types.Log, err
  if err == nil {
  f.begin = int64(end) + 1
  }
- return logs, err
+ return err
  }
  f.begin = int64(number) + 1
 
  // Retrieve the suggested block and pull any truly matching logs
  header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(number))
  if header == nil || err != nil {
- return logs, err
+ return err
  }
  found, err := f.checkMatches(ctx, header)
  if err != nil {
- return logs, err
+ return err
+ }
+ for _, log := range found {
+ logChan <- log
  }
- logs = append(logs, found...)
 
  case <-ctx.Done():
- return logs, ctx.Err()
+ return ctx.Err()
  }
  }
 }
 
 // unindexedLogs returns the logs matching the filter criteria based on raw block
 // iteration and bloom matching.
-func (f *Filter) unindexedLogs(ctx context.Context, end uint64) ([]*types.Log, error) {
- var logs []*types.Log
-
+func (f *Filter) unindexedLogs(ctx context.Context, end uint64, logChan chan *types.Log) error {
  for ; f.begin <= int64(end); f.begin++ {
- if f.begin%10 == 0 && ctx.Err() != nil {
- return logs, ctx.Err()
- }
  header, err := f.sys.backend.HeaderByNumber(ctx, rpc.BlockNumber(f.begin))
  if header == nil || err != nil {
- return logs, err
+ return err
  }
  found, err := f.blockLogs(ctx, header)
  if err != nil {
- return logs, err
+ return err
+ }
+ for _, log := range found {
+ select {
+ case logChan <- log:
+ case <-ctx.Done():
+ return ctx.Err()
+ }
  }
- logs = append(logs, found...)
  }
- return logs, nil
+ return nil
 }
 
 // blockLogs returns the logs matching the filter criteria within a single block.
@@ -294,19 +332,19 @@ func (f *Filter) checkMatches(ctx context.Context, header *types.Header) ([]*typ
 }
 
 // pendingLogs returns the logs matching the filter criteria within the pending block.
-func (f *Filter) pendingLogs() ([]*types.Log, error) {
+func (f *Filter) pendingLogs() []*types.Log {
  block, receipts := f.sys.backend.PendingBlockAndReceipts()
  if block == nil {
- return nil, errors.New("pending state not available")
+ return nil
  }
  if bloomFilter(block.Bloom(), f.addresses, f.topics) {
  var unfiltered []*types.Log
  for _, r := range receipts {
  unfiltered = append(unfiltered, r.Logs...)
  }
- return filterLogs(unfiltered, nil, nil, f.addresses, f.topics), nil
+ return filterLogs(unfiltered, nil, nil, f.addresses, f.topics)
  }
- return nil, nil
+ return nil
 }
 
 func includes(addresses []common.Address, a common.Address) bool {

eth/filters/filter_system_test.go

@@ -50,6 +50,8 @@ type testBackend struct {
  rmLogsFeed event.Feed
  pendingLogsFeed event.Feed
  chainFeed event.Feed
+ pendingBlock *types.Block
+ pendingReceipts types.Receipts
 }
 
 func (b *testBackend) ChainConfig() *params.ChainConfig {
@@ -124,7 +126,7 @@ func (b *testBackend) GetLogs(ctx context.Context, hash common.Hash, number uint
 }
 
 func (b *testBackend) PendingBlockAndReceipts() (*types.Block, types.Receipts) {
- return nil, nil
+ return b.pendingBlock, b.pendingReceipts
 }
 
 func (b *testBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {