Fetch ahead blocks from delivery service

Signed-off-by: Alexandros Filios <alexandros.filios@ibm.com>

Author: alexandrosfilios

platform/fabric/core/generic/config/ds.go

@@ -106,6 +106,7 @@ type Finality struct {
 }
 
 type Delivery struct {
+	BufferSize          int           `yaml:"BufferSize,omitempty"`
 	WaitForEventTimeout time.Duration `yaml:"WaitForEventTimeout,omitempty"`
 	SleepAfterFailure   time.Duration `yaml:"SleepAfterFailure,omitempty"`
 }
@@ -205,6 +206,13 @@ func (c *Channel) CommitterWaitForEventTimeout() time.Duration {
 	return c.Committer.WaitForEventTimeout
 }
 
+func (c *Channel) DeliveryBufferSize() int {
+	if c.Delivery.BufferSize <= 0 {
+		return 1
+	}
+	return c.Delivery.BufferSize
+}
+
 func (c *Channel) DiscoveryTimeout() time.Duration {
 	if c.Discovery.Timeout == 0 {
 		return 20 * time.Second

platform/fabric/core/generic/delivery/delivery.go

@@ -8,7 +8,9 @@ package delivery
 
 import (
 	"context"
+	"fmt"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/services/logging"
@@ -38,6 +40,11 @@ var (
 	}
 )
 
+type blockResponse struct {
+	ctx   context.Context
+	block *common.Block
+}
+
 type messageType = string
 
 const (
@@ -73,7 +80,8 @@ type Delivery struct {
 	client              services.PeerClient
 	tracer              trace.Tracer
 	lastBlockReceived   uint64
-	stop                chan bool
+	bufferSize          int
+	stop                chan struct{}
 }
 
 func New(
@@ -87,6 +95,7 @@ func New(
 	callback driver.BlockCallback,
 	vault Vault,
 	waitForEventTimeout time.Duration,
+	bufferSize int,
 	tracerProvider trace.TracerProvider,
 	_ metrics.Provider,
 ) (*Delivery, error) {
@@ -108,9 +117,10 @@ func New(
 			Namespace:  "fabricsdk",
 			LabelNames: []tracing.LabelName{messageTypeLabel},
 		})),
-		callback: callback,
-		vault:    vault,
-		stop:     make(chan bool),
+		callback:   callback,
+		vault:      vault,
+		bufferSize: max(bufferSize, 1),
+		stop:       make(chan struct{}),
 	}
 	return d, nil
 }
@@ -123,22 +133,56 @@ func (d *Delivery) Start(ctx context.Context) {
 }
 
 func (d *Delivery) Stop() {
-	d.stop <- true
+	close(d.stop)
 }
 
+var ctr = atomic.Uint32{}
+
 func (d *Delivery) Run(ctx context.Context) error {
+	logger.Infof("Running delivery service [%d]", ctr.Add(1))
 	if ctx == nil {
 		ctx = context.Background()
 	}
+	ch := make(chan blockResponse, d.bufferSize)
+	go d.readBlocks(ch)
+	return d.runReceiver(ctx, ch)
+}
+
+func (d *Delivery) readBlocks(ch <-chan blockResponse) {
+	for {
+		select {
+		case b := <-ch:
+			logger.Debugf("Invoking callback for block [%d]", b.block.Header)
+			stop, err := d.callback(b.ctx, b.block)
+			if err != nil {
+				logger.Errorf("callback errored for block %d: %v", b.block.Header.Number, err)
+			}
+			if stop {
+				logger.Infof("Stopping delivery at block [%d]", b.block.Header.Number)
+				close(d.stop)
+				return
+			}
+		case <-d.stop:
+			logger.Infof("Stopping delivery service")
+			return
+		}
+	}
+}
+
+func (d *Delivery) runReceiver(ctx context.Context, ch chan<- blockResponse) error {
+	if ctx == nil || ch == nil {
+		return errors.New("ctx and channel must be provided")
+	}
 	var df DeliverStream
 	var err error
 	waitTime := d.channelConfig.DeliverySleepAfterFailure()
 	for {
 		select {
 		case <-d.stop:
-			// Time to stop
+			logger.Infof("Stopped receiver")
 			return nil
 		case <-ctx.Done():
+			logger.Infof("Context done")
 			// Time to cancel
 			return errors.New("context done")
 		default:
@@ -192,19 +236,14 @@ func (d *Delivery) Run(ctx context.Context) error {
 				}
 				d.lastBlockReceived = r.Block.Header.Number
 
-				span.AddEvent("invoke_callback")
-				stop, err := d.callback(deliveryCtx, r.Block)
-				span.AddEvent("invoked_callback")
-				if err != nil {
-					span.RecordError(err)
-					logger.Errorf("error occurred when processing filtered block [%s], retry...", err)
-					time.Sleep(waitTime)
-					df = nil
-				}
-				if stop {
-					span.End()
-					return nil
+				span.AddEvent(fmt.Sprintf("push_%d_to_channel", r.Block.Header.Number))
+				logger.Debugf("Pushing block [%d] to channel with current length %d", r.Block.Header.Number, len(ch))
+				ch <- blockResponse{
+					ctx:   deliveryCtx,
+					block: r.Block,
 				}
+				logger.Debugf("Pushed block [%d] to channel", r.Block.Header.Number)
+				span.AddEvent("pushed_to_channel")
 			case *pb.DeliverResponse_Status:
 				span.SetAttributes(tracing.String(messageTypeLabel, responseStatus))
 				if r.Status == common.Status_NOT_FOUND {

platform/fabric/core/generic/delivery/service.go

@@ -68,6 +68,7 @@ func NewService(
 		callback,
 		txIDStore,
 		channelConfig.CommitterWaitForEventTimeout(),
+		channelConfig.DeliveryBufferSize(),
 		tracerProvider,
 		metricsProvider,
 	)
@@ -114,6 +115,7 @@ func (c *Service) scanBlock(ctx context.Context, vault Vault, callback driver.Bl
 		callback,
 		vault,
 		c.channelConfig.CommitterWaitForEventTimeout(),
+		c.channelConfig.DeliveryBufferSize(),
 		c.tracerProvider,
 		c.metricsProvider,
 	)

platform/fabric/driver/config.go

@@ -51,6 +51,7 @@ type ChannelConfig interface {
 	CommitterFinalityNumRetries() int
 	CommitterFinalityUnknownTXTimeout() time.Duration
 	CommitterWaitForEventTimeout() time.Duration
+	DeliveryBufferSize() int
 	DeliverySleepAfterFailure() time.Duration
 	CommitParallelism() int
 	ChaincodeConfigs() []ChaincodeConfig