[loadgen] Close broadcast streams only after the workload has completed (#136)

#### Type of change

- Bug fix
 
#### Description

- Fix bug in load-geneator: Close broadcast streams only after the
workload has completed.
This ensures that any pending messages in the stream’s internal queue
are fully processed. Even if all transactions have already been
submitted, additional time may be required for them to be transmitted.
- Use a connection cache when opening all the connections in the
connection manager to avoid re-opening the same connection multiple
times.
- Fix broken rebase from prior commit.

#### Related issues

- resolves #117

Signed-off-by: Liran Funaro <liran.funaro@gmail.com>

Author: liran-funaro

integration/runner/runtime.go

@@ -243,14 +243,14 @@ func NewRuntime(t *testing.T, conf *Config) *CommitterRuntime {
 		ConsensusType: ordererconn.Bft,
 	})
 	require.NoError(t, err)
-	t.Cleanup(c.ordererStream.Close)
+	t.Cleanup(c.ordererStream.CloseConnections)
 
 	c.sidecarClient, err = sidecarclient.New(&sidecarclient.Parameters{
 		ChannelID: s.Policy.ChannelID,
 		Client:    test.NewTLSClientConfig(s.ClientTLS, s.Endpoints.Sidecar.Server),
 	})
 	require.NoError(t, err)
-	t.Cleanup(c.sidecarClient.Close)
+	t.Cleanup(c.sidecarClient.CloseConnections)
 	return c
 }
 

loadgen/adapters/orderer.go

@@ -13,6 +13,7 @@ import (
 	"golang.org/x/sync/errgroup"
 
 	"github.com/hyperledger/fabric-x-committer/loadgen/workload"
+	"github.com/hyperledger/fabric-x-committer/utils/connection"
 	"github.com/hyperledger/fabric-x-committer/utils/deliver"
 	"github.com/hyperledger/fabric-x-committer/utils/test"
 )
@@ -39,7 +40,7 @@ func (c *OrdererAdapter) RunWorkload(ctx context.Context, txStream *workload.Str
 	if err != nil {
 		return errors.Wrap(err, "failed to create orderer clients")
 	}
-	defer client.Close()
+	defer client.CloseConnections()
 
 	dCtx, dCancel := context.WithCancel(ctx)
 	defer dCancel()
@@ -61,13 +62,19 @@ func (c *OrdererAdapter) RunWorkload(ctx context.Context, txStream *workload.Str
 		})
 	}
 
-	for range c.config.BroadcastParallelism {
+	streams := make([]*test.BroadcastStream, c.config.BroadcastParallelism)
+	for i := range streams {
+		streams[i], err = test.NewBroadcastStream(gCtx, &c.config.Orderer)
+		if err != nil {
+			connection.CloseConnectionsLog(streams[:i]...)
+			return errors.Wrap(err, "failed to create a broadcast stream")
+		}
+	}
+	defer connection.CloseConnectionsLog(streams...)
+
+	for _, stream := range streams {
+		stream := stream
 		g.Go(func() error {
-			stream, streamErr := test.NewBroadcastStream(gCtx, &c.config.Orderer)
-			if streamErr != nil {
-				return errors.Wrap(streamErr, "failed to create a broadcast stream")
-			}
-			defer stream.Close()
 			return sendBlocks(gCtx, &c.commonAdapter, txStream, workload.MapToEnvelopeBatch, stream.SendBatch)
 		})
 	}

loadgen/adapters/sigverifier.go

@@ -69,6 +69,7 @@ func (c *SvAdapter) RunWorkload(ctx context.Context, txStream *workload.StreamWi
 	}
 
 	for _, stream := range streams {
+		stream := stream
 		g.Go(func() error {
 			return sendBlocks(gCtx, &c.commonAdapter, txStream, workload.MapToVerifierBatch, stream.Send)
 		})

loadgen/adapters/vcservice.go

@@ -78,6 +78,7 @@ func (c *VcAdapter) RunWorkload(ctx context.Context, txStream *workload.StreamWi
 	defer dCancel()
 	g, gCtx := errgroup.WithContext(dCtx)
 	for _, stream := range streams {
+		stream := stream
 		g.Go(func() error {
 			return sendBlocks(ctx, &c.commonAdapter, txStream, workload.MapToVcBatch, stream.Send)
 		})

loadgen/client_test.go

@@ -302,6 +302,7 @@ func TestLoadGenForOnlyOrderer(t *testing.T) {
 						Endpoints: endpoints,
 					},
 					ChannelID:     clientConf.LoadProfile.Transaction.Policy.ChannelID,
+					Identity:      clientConf.LoadProfile.Transaction.Policy.Identity,
 					ConsensusType: ordererconn.Bft,
 				},
 				BroadcastParallelism: 5,

service/sidecar/sidecarclient/client.go

@@ -73,9 +73,9 @@ func New(config *Parameters) (*Client, error) {
 	}, nil
 }
 
-// Close closes all the connections.
-func (c *Client) Close() {
-	c.ConnectionManager.Close()
+// CloseConnections closes all the connections.
+func (c *Client) CloseConnections() {
+	c.ConnectionManager.CloseConnections()
 }
 
 // Deliver start receiving blocks starting from config.StartBlkNum to config.OutputBlock.

utils/connection/client_util.go

@@ -201,7 +201,7 @@ func Connect(config *DialConfig) (*grpc.ClientConn, error) {
 
 // OpenConnections opens connections with multiple remotes.
 func OpenConnections(config MultiClientConfig) ([]*grpc.ClientConn, error) {
-	logger.Infof("Opening connections to %d endpoints: %v.\n", len(config.Endpoints), config.Endpoints)
+	logger.Infof("Opening connections to %d endpoints: %v.", len(config.Endpoints), config.Endpoints)
 	dialConfigs, err := NewDialConfigPerEndpoint(&config)
 	if err != nil {
 		return nil, errors.Wrapf(err, "error while creating dial configs")

utils/deliver/client.go

@@ -52,9 +52,9 @@ func New(config *ordererconn.Config) (*Client, error) {
 	}, nil
 }
 
-// Close closes all the connections for the client.
-func (s *Client) Close() {
-	s.connectionManager.Close()
+// CloseConnections closes all the connections for the client.
+func (s *Client) CloseConnections() {
+	s.connectionManager.CloseConnections()
 }
 
 // UpdateConnections updates the connection config.

utils/deliver/client_test.go

@@ -48,7 +48,7 @@ func TestBroadcastDeliver(t *testing.T) {
 	conf.Connection.Endpoints = allEndpoints[:6]
 	client, err := New(&conf)
 	require.NoError(t, err)
-	t.Cleanup(client.Close)
+	t.Cleanup(client.CloseConnections)
 
 	ctx, cancel := context.WithTimeout(t.Context(), time.Minute)
 	t.Cleanup(cancel)
@@ -152,7 +152,7 @@ func submit(
 	defer cancel()
 	stream, err := test.NewBroadcastStream(ctx, conf)
 	require.NoError(t, err)
-	defer stream.Close()
+	defer stream.CloseConnections()
 
 	err = stream.SendBatch(workload.MapToEnvelopeBatch(0, []*protoloadgen.TX{tx}))
 	if err != nil {
@@ -233,7 +233,7 @@ func waitUntilGrpcServerIsReady(ctx context.Context, t *testing.T, endpoint *con
 
 func waitUntilGrpcServerIsDown(ctx context.Context, t *testing.T, endpoint *connection.Endpoint) {
 	t.Helper()
-	newConn, err := connection.Connect(connection.NewInsecureDialConfig(endpoint))
+	newConn, err := connection.Connect(test.NewInsecureDialConfig(endpoint))
 	require.NoError(t, err)
 	defer connection.CloseConnectionsLog(newConn)
 	test.WaitUntilGrpcServerIsDown(ctx, t, newConn)

utils/ordererconn/conn_manager.go

@@ -12,6 +12,7 @@ import (
 	"math"
 	"math/rand"
 	"slices"
+	"strings"
 	"sync"
 	"sync/atomic"
 
@@ -110,17 +111,26 @@ func (c *ConnectionManager) Update(config *ConnectionConfig) error {
 
 	// We pre create all the connections to ensure correct form.
 	connections := make(map[string]*grpc.ClientConn)
-	allAPis := []string{anyAPI, Broadcast, Deliver}
+	// We use a connection cache to avoid opening the same connection multiple times.
+	connCache := make(map[string]*grpc.ClientConn)
+	allAPIs := []string{anyAPI, Broadcast, Deliver}
 	for _, id := range append(getAllIDs(config.Endpoints), anyID) {
-		for _, api := range allAPis {
+		for _, api := range allAPIs {
 			filter := aggregateFilter(WithAPI(api), WithID(id))
-			conn, err := openConnection(config, filter)
-			if errors.Is(err, ErrNoConnections) {
+			endpoints := filterOrdererEndpoints(config.Endpoints, filter)
+			if len(endpoints) == 0 {
 				continue
 			}
-			if err != nil {
-				closeConnection(connections)
-				return err
+			endpointsKey := makeEndpointsKey(endpoints)
+			conn, connInCache := connCache[endpointsKey]
+			if !connInCache {
+				var err error
+				conn, err = openConnection(config, endpoints)
+				if err != nil {
+					closeConnection(connections)
+					return err
+				}
+				connCache[endpointsKey] = conn
 			}
 			connections[filterKey(filter)] = conn
 		}
@@ -179,14 +189,8 @@ func getAllIDs(endpoints []*Endpoint) []uint32 {
 
 func openConnection(
 	conf *ConnectionConfig,
-	filter ...ConnFilter,
+	endpoints []*connection.Endpoint,
 ) (*grpc.ClientConn, error) {
-	key := aggregateFilter(filter...)
-
-	endpoints := filterOrdererEndpoints(conf.Endpoints, key)
-	if len(endpoints) == 0 {
-		return nil, ErrNoConnections
-	}
 	// We shuffle the endpoints for load balancing.
 	shuffle(endpoints)
 	logger.Infof("Opening connections to %d endpoints: %v.", len(endpoints), endpoints)
@@ -201,8 +205,17 @@ func openConnection(
 	return connection.Connect(dialConfig)
 }
 
-// Close closes all the connections.
-func (c *ConnectionManager) Close() {
+func makeEndpointsKey(endpoint []*connection.Endpoint) string {
+	addresses := make([]string, len(endpoint))
+	for i, e := range endpoint {
+		addresses[i] = e.Address()
+	}
+	slices.Sort(addresses)
+	return strings.Join(addresses, ";")
+}
+
+// CloseConnections closes all the connections.
+func (c *ConnectionManager) CloseConnections() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
 	closeConnection(c.connections)