perf(protocols): improve performance across all protocols

- Increase default queue sizes for better buffering
- Eliminate unnecessary goroutines for channel cleanup
- Apply optimizations consistently across all protocols

Signed-off-by: Chris Gianelloni <wolf31o2@blinklabs.io>

Author: wolf31o2

protocol/blockfetch/blockfetch.go

@@ -119,7 +119,7 @@ type Config struct {
 const MaxRecvQueueSize = 512
 
 // DefaultRecvQueueSize is the default receive queue size.
-const DefaultRecvQueueSize = 256
+const DefaultRecvQueueSize = 384
 
 // MaxPendingMessageBytes is the maximum allowed pending message bytes (5MB).
 const MaxPendingMessageBytes = 5242880

protocol/blockfetch/client.go

@@ -18,6 +18,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"github.com/blinklabs-io/gouroboros/cbor"
 	"github.com/blinklabs-io/gouroboros/ledger"
@@ -36,6 +37,7 @@ type Client struct {
 	blockUseCallback     bool              // Whether to use callback for blocks
 	onceStart            sync.Once         // Ensures Start is only called once
 	onceStop             sync.Once         // Ensures Stop is only called once
+	started              atomic.Bool       // Whether the protocol has been started
 }
 
 // NewClient creates a new Block Fetch protocol client with the given options and configuration.
@@ -93,13 +95,8 @@ func (c *Client) Start() {
 				"protocol", ProtocolName,
 				"connection_id", c.callbackContext.ConnectionId.String(),
 			)
+		c.started.Store(true)
 		c.Protocol.Start()
-		// Start goroutine to cleanup resources on protocol shutdown
-		go func() {
-			<-c.DoneChan()
-			close(c.blockChan)
-			close(c.startBatchResultChan)
-		}()
 	})
 }
 
@@ -115,6 +112,21 @@ func (c *Client) Stop() error {
 			)
 		msg := NewMsgClientDone()
 		err = c.SendMessage(msg)
+		if err == nil {
+			c.Protocol.Stop()
+		}
+		// Defer closing channels until protocol fully shuts down (only if started)
+		if c.started.Load() {
+			go func() {
+				<-c.DoneChan()
+				close(c.blockChan)
+				close(c.startBatchResultChan)
+			}()
+		} else {
+			// If protocol was never started, close channels immediately
+			close(c.blockChan)
+			close(c.startBatchResultChan)
+		}
 	})
 	return err
 }
@@ -222,13 +234,11 @@ func (c *Client) handleStartBatch() error {
 			"role", "client",
 			"connection_id", c.callbackContext.ConnectionId.String(),
 		)
-	// Check for shutdown
 	select {
 	case <-c.DoneChan():
 		return protocol.ErrProtocolShuttingDown
-	default:
+	case c.startBatchResultChan <- nil:
 	}
-	c.startBatchResultChan <- nil
 	return nil
 }
 
@@ -241,14 +251,12 @@ func (c *Client) handleNoBlocks() error {
 			"role", "client",
 			"connection_id", c.callbackContext.ConnectionId.String(),
 		)
-	// Check for shutdown
+	err := errors.New("block(s) not found")
 	select {
 	case <-c.DoneChan():
 		return protocol.ErrProtocolShuttingDown
-	default:
+	case c.startBatchResultChan <- err:
 	}
-	err := errors.New("block(s) not found")
-	c.startBatchResultChan <- err
 	return nil
 }
 
@@ -298,7 +306,11 @@ func (c *Client) handleBlock(msgGeneric protocol.Message) error {
 			return errors.New("received block-fetch Block message but no callback function is defined")
 		}
 	} else {
-		c.blockChan <- block
+		select {
+		case <-c.DoneChan():
+			return protocol.ErrProtocolShuttingDown
+		case c.blockChan <- block:
+		}
 	}
 	return nil
 }

protocol/blockfetch/client_test.go

@@ -207,3 +207,24 @@ func TestGetBlockNoBlocks(t *testing.T) {
 		},
 	)
 }
+
+func TestClientShutdown(t *testing.T) {
+	runTest(
+		t,
+		[]ouroboros_mock.ConversationEntry{
+			ouroboros_mock.ConversationEntryHandshakeRequestGeneric,
+			ouroboros_mock.ConversationEntryHandshakeNtNResponse,
+		},
+		func(t *testing.T, oConn *ouroboros.Connection) {
+			if oConn.BlockFetch() == nil {
+				t.Fatalf("BlockFetch client is nil")
+			}
+			// Start the client
+			oConn.BlockFetch().Client.Start()
+			// Stop the client
+			if err := oConn.BlockFetch().Client.Stop(); err != nil {
+				t.Fatalf("unexpected error when stopping client: %s", err)
+			}
+		},
+	)
+}

protocol/chainsync/chainsync.go

@@ -223,8 +223,8 @@ type Config struct {
 const (
 	MaxPipelineLimit       = 100    // Max pipelined requests
 	MaxRecvQueueSize       = 100    // Max receive queue size (messages)
-	DefaultPipelineLimit   = 50     // Default pipeline limit
-	DefaultRecvQueueSize   = 50     // Default queue size
+	DefaultPipelineLimit   = 75     // Default pipeline limit
+	DefaultRecvQueueSize   = 75     // Default queue size
 	MaxPendingMessageBytes = 102400 // Max pending message bytes (100KB)
 )
 

protocol/chainsync/client.go

@@ -34,10 +34,8 @@ type Client struct {
 	readyForNextBlockChan    chan bool
 	onceStart                sync.Once
 	onceStop                 sync.Once
+	started                  atomic.Bool
 	syncPipelinedRequestNext int
-
-	// waitingForCurrentTipChan will process all the requests for the current tip until the channel
-	// is empty.
 	//
 	// want* only processes one request per message reply received from the server. If the message
 	// request fails, it is the responsibility of the caller to clear the channel.
@@ -126,12 +124,8 @@ func (c *Client) Start() {
 				"protocol", ProtocolName,
 				"connection_id", c.callbackContext.ConnectionId.String(),
 			)
+		c.started.Store(true)
 		c.Protocol.Start()
-		// Start goroutine to cleanup resources on protocol shutdown
-		go func() {
-			<-c.DoneChan()
-			close(c.readyForNextBlockChan)
-		}()
 	})
 }
 
@@ -151,6 +145,17 @@ func (c *Client) Stop() error {
 		if err = c.SendMessage(msg); err != nil {
 			return
 		}
+		c.Protocol.Stop()
+		// Defer closing channel until protocol fully shuts down (only if started)
+		if c.started.Load() {
+			go func() {
+				<-c.DoneChan()
+				close(c.readyForNextBlockChan)
+			}()
+		} else {
+			// If protocol was never started, close channel immediately
+			close(c.readyForNextBlockChan)
+		}
 	})
 	return err
 }
@@ -334,7 +339,15 @@ func (c *Client) GetAvailableBlockRange(
 				)
 			}
 			start = firstBlock.point
-		case <-c.readyForNextBlockChan:
+		case ready, ok := <-c.readyForNextBlockChan:
+			if !ok {
+				// Channel closed, protocol shutting down
+				return start, end, protocol.ErrProtocolShuttingDown
+			}
+			// Only proceed if ready is true
+			if !ready {
+				return start, end, ErrSyncCancelled
+			}
 			// Request the next block
 			msg := NewMsgRequestNext()
 			if err := c.SendMessage(msg); err != nil {
@@ -721,14 +734,22 @@ func (c *Client) handleRollForward(msgGeneric protocol.Message) error {
 	if callbackErr != nil {
 		if errors.Is(callbackErr, ErrStopSyncProcess) {
 			// Signal that we're cancelling the sync
-			c.readyForNextBlockChan <- false
+			select {
+			case <-c.DoneChan():
+				return protocol.ErrProtocolShuttingDown
+			case c.readyForNextBlockChan <- false:
+			}
 			return nil
 		} else {
 			return callbackErr
 		}
 	}
 	// Signal that we're ready for the next block
-	c.readyForNextBlockChan <- true
+	select {
+	case <-c.DoneChan():
+		return protocol.ErrProtocolShuttingDown
+	case c.readyForNextBlockChan <- true:
+	}
 	return nil
 }
 
@@ -752,15 +773,23 @@ func (c *Client) handleRollBackward(msgGeneric protocol.Message) error {
 		if callbackErr := c.config.RollBackwardFunc(c.callbackContext, msgRollBackward.Point, msgRollBackward.Tip); callbackErr != nil {
 			if errors.Is(callbackErr, ErrStopSyncProcess) {
 				// Signal that we're cancelling the sync
-				c.readyForNextBlockChan <- false
+				select {
+				case <-c.DoneChan():
+					return protocol.ErrProtocolShuttingDown
+				case c.readyForNextBlockChan <- false:
+				}
 				return nil
 			} else {
 				return callbackErr
 			}
 		}
 	}
 	// Signal that we're ready for the next block
-	c.readyForNextBlockChan <- true
+	select {
+	case <-c.DoneChan():
+		return protocol.ErrProtocolShuttingDown
+	case c.readyForNextBlockChan <- true:
+	}
 	return nil
 }