cmd/bench: Implement sub-command to run load generator

cmd/bench/cmd/client.go

@@ -0,0 +1,95 @@
+// Copyright Contributors to the Mir project
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package cmd
+
+import (
+	"context"
+	"crypto"
+	"errors"
+	"fmt"
+	"math/rand"
+	"time"
+
+	rateLimiter "golang.org/x/time/rate"
+
+	"github.com/spf13/cobra"
+
+	"github.com/filecoin-project/mir/pkg/dummyclient"
+	"github.com/filecoin-project/mir/pkg/logging"
+	t "github.com/filecoin-project/mir/pkg/types"
+)
+
+var (
+	reqSize  int
+	rate     float64
+	burst    int
+	duration time.Duration
+
+	clientCmd = &cobra.Command{
+		Use:   "client",
+		Short: "Generate and submit requests to a Mir cluster",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return runClient()
+		},
+	}
+)
+
+func init() {
+	rootCmd.AddCommand(clientCmd)
+	clientCmd.Flags().IntVarP(&reqSize, "reqSize", "s", 256, "size of each request in bytes")
+	clientCmd.Flags().Float64VarP(&rate, "rate", "r", 1000, "average number of requests per second")
+	clientCmd.Flags().IntVarP(&burst, "burst", "b", 1, "maximum number of requests in a burst")
+	clientCmd.Flags().DurationVarP(&duration, "duration", "T", 10*time.Second, "benchmarking duration")
+}
+
+func runClient() error {
+	var logger logging.Logger
+	if verbose {
+		logger = logging.ConsoleDebugLogger
+	} else {
+		logger = logging.ConsoleWarnLogger
+	}
+
+	nodeIDs := make([]t.NodeID, nrNodes)
+	for i := range nodeIDs {
+		nodeIDs[i] = t.NewNodeIDFromInt(i)
+	}
+
+	reqReceiverAddrs := make(map[t.NodeID]string)
+	for i := range nodeIDs {
+		reqReceiverAddrs[t.NewNodeIDFromInt(i)] = fmt.Sprintf("127.0.0.1:%d", ReqReceiverBasePort+i)
+	}
+
+	ctx, stop := context.WithCancel(context.Background())
+
+	client := dummyclient.NewDummyClient(
+		t.ClientID(id),
+		crypto.SHA256,
+		logger,
+	)
+	client.Connect(ctx, reqReceiverAddrs)
+	defer client.Disconnect()
+
+	go func() {
+		time.Sleep(duration)
+		stop()
+	}()
+
+	limiter := rateLimiter.NewLimiter(rateLimiter.Limit(rate), 1)
+	reqBytes := make([]byte, reqSize)
+	for i := 0; ; i++ {
+		if err := limiter.Wait(ctx); err != nil {
+			if errors.Is(err, context.Canceled) {
+				err = nil
+			}
+			return err
+		}
+		rand.Read(reqBytes) //nolint:gosec
+		logger.Log(logging.LevelDebug, fmt.Sprintf("Submitting request #%d", i))
+		if err := client.SubmitRequest(reqBytes); err != nil {
+			return err
+		}
+	}
+}

go.mod

@@ -30,6 +30,7 @@ require (
 	github.com/multiformats/go-multiaddr v0.5.0
 	github.com/spf13/cobra v1.5.0
 	github.com/whyrusleeping/cbor-gen v0.0.0-20220514204315-f29c37e9c44c
+	golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9
 	golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f
 )
 

go.sum

@@ -1141,6 +1141,8 @@ golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxb
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9 h1:ftMN5LMiBFjbzleLqtoBZk7KdJwhuybIU+FckUHgoyQ=
+golang.org/x/time v0.0.0-20220722155302-e5dcc9cfc0b9/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180828015842-6cd1fcedba52/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=