benchkit: add the measurement harness as a second module

AGENTS.md

@@ -27,14 +27,19 @@ gorums/
 ├── cmd/protoc-gen-gorums/     # Compiler plugin for code generation
 │   ├── dev/                   # Static code + generated code examples
 │   └── gengorums/             # Compiler logic + templates
+├── benchkit/                  # Separate module: measurement and benchmarking
+│   ├── proto/                 # .proto sources for the benchkit module
 ├── examples/                  # Separate module: example implementations
 ├── internal/                  # Internal packages
 ├── doc/                       # Documentation
 └── *.go                       # Core library files
 ```
 
-The repository holds two modules: `github.com/relab/gorums` at the root and
-`github.com/relab/gorums/examples`, joined by `go.work`.
+The repository holds three modules: `github.com/relab/gorums` at the root,
+`github.com/relab/gorums/benchkit`, and `github.com/relab/gorums/examples`.
+They are joined by `go.work`.
+The dependency edge runs one way: benchkit imports gorums, never the reverse.
+Keep the root `go.mod` free of benchmarking and orchestration dependencies.
 
 ## Development Rules
 

Makefile

@@ -7,12 +7,18 @@ static_file				:= $(gen_path)/template_static.go
 static_files			:= $(shell find $(dev_path) -name "*.go" -not -name "zorums*" -not -name "*_test.go")
 proto_path 				:= $(dev_path):third_party:.
 
-workspace_packages		:= ./... ./examples/...
+# The benchkit module keeps its .proto files under benchkit/proto so that the
+# import paths protoc records stay "benchkit/*.proto" and "benchmark/*.proto".
+bk_path					:= benchkit/proto
+bk_proto_path			:= $(bk_path):third_party:.
+bk_module				:= github.com/relab/gorums/benchkit
+workspace_packages		:= ./... ./examples/... ./benchkit/...
 
 plugin_deps				:= gorums.pb.go $(static_file)
 runtime_deps			:= internal/stream/stream.pb.go internal/stream/stream_grpc.pb.go
+benchkit_deps			:= benchkit/benchkit.pb.go benchkit/control.pb.go benchkit/control_gorums.pb.go
 
-.PHONY: all dev tools bootstrapgorums installgorums test compiletests genproto benchtest bench lint deadcode modernize goplscheck
+.PHONY: all dev tools bootstrapgorums installgorums benchkit test compiletests genproto benchtest bench lint deadcode modernize goplscheck
 
 all: dev compiletests
 
@@ -24,6 +30,24 @@ dev: installgorums $(runtime_deps)
 		--go_opt=default_api_level=API_OPAQUE \
 		$(zorums_proto)
 
+benchkit: installgorums $(benchkit_deps)
+
+# The benchkit module's generated code is written back into the module root
+# rather than next to its .proto file, so these cannot use the pattern rules.
+benchkit/benchkit.pb.go: $(bk_path)/benchkit/benchkit.proto
+	@protoc -I=$(bk_proto_path) \
+		--go_out=benchkit --go_opt=module=$(bk_module) \
+		--go_opt=default_api_level=API_OPAQUE $<
+
+benchkit/control.pb.go: $(bk_path)/benchkit/control.proto
+	@protoc -I=$(bk_proto_path) \
+		--go_out=benchkit --go_opt=module=$(bk_module) \
+		--go_opt=default_api_level=API_OPAQUE $<
+
+benchkit/control_gorums.pb.go: $(bk_path)/benchkit/control.proto
+	@protoc -I=$(bk_proto_path) \
+		--gorums_out=benchkit --gorums_opt=module=$(bk_module) $<
+
 $(static_file): $(static_files)
 	@cp $(static_file) $(static_file).bak
 	@protoc-gen-gorums --bundle=$(static_file)
@@ -100,20 +124,20 @@ stressgen: tools
 	rm ./internal/testprotos/testprotos.test
 
 lint: deadcode
-	@golangci-lint-v2 run ./... ./examples/...
+	@golangci-lint-v2 run ./... ./examples/... ./benchkit/...
 
 # deadcode reports functions unreachable from any main or test across all
-# workspace modules (root and examples), so cross-module usage is
+# workspace modules (root, examples, benchkit), so cross-module usage is
 # accounted for. It is advisory: exported library API with no in-repo caller
 # (e.g. optional dial/server options) and example-only helpers are expected to
 # appear. Review new entries for genuinely dead internal code.
 deadcode:
-	@go run golang.org/x/tools/cmd/deadcode@latest -test ./... ./examples/...
+	@go run golang.org/x/tools/cmd/deadcode@latest -test ./... ./examples/... ./benchkit/...
 
 modernize:
-	@go fix ./... ./examples/...
+	@go fix ./... ./examples/... ./benchkit/...
 	@go run golang.org/x/tools/go/analysis/passes/modernize/cmd/modernize@latest \
-		-fix ./... ./examples/...
+		-fix ./... ./examples/... ./benchkit/...
 
 # Report all gopls diagnostics, including hint-level style and modernization
 # suggestions. Generated Go files are excluded because their generators own them.
@@ -134,11 +158,12 @@ goplscheck:
 			exit 1; \
 		fi
 
-# Regenerate all Gorums and protobuf generated files across the repo (dev, internal/tests, examples).
+# Regenerate all Gorums and protobuf generated files across the repo (dev, benchkit, internal/tests, examples).
 # This will force regeneration even though the proto files have not changed.
 genproto: installgorums dev
-	@echo "Regenerating all proto files (dev, internal/tests, examples)"
+	@echo "Regenerating all proto files (dev, benchkit, internal/tests, examples)"
 	@$(MAKE) -B -s dev
+	@$(MAKE) -B -s $(benchkit_deps)
 	@$(MAKE) -B -s --no-print-directory -C ./internal/tests all
 	@$(MAKE) -B -s --no-print-directory -C ./examples all
 

benchkit/aggregate.go

@@ -0,0 +1,67 @@
+package benchkit
+
+import (
+	"maps"
+	"slices"
+
+	"github.com/relab/gorums"
+)
+
+// AppendServerStats attaches per-server memory statistics from Stop RPC replies
+// to an existing result. It is used by client-measured benchmarks (QuorumCall,
+// AsyncQuorumCall) where the client's latency samples are already in result but
+// the server-side allocations must be collected separately.
+func AppendServerStats(result *Result, replies map[uint32]*Result) {
+	for _, id := range slices.Sorted(maps.Keys(replies)) {
+		r := replies[id]
+		result.SetServerStats(append(result.GetServerStats(), MemoryStat_builder{
+			Allocs: r.GetAllocsPerOp() * r.GetTotalOps(),
+			Memory: r.GetMemPerOp() * r.GetTotalOps(),
+		}.Build()))
+	}
+}
+
+// AggregateServerResults combines per-server Stop replies into a single
+// cluster-wide Result. TotalOps and Throughput are summed across servers so
+// the reported value reflects the cluster's aggregate work; TotalTime is the
+// maximum across servers so it reflects the measurement window's wall-clock
+// time rather than an N-fold sum. Latency samples from every reply are
+// concatenated so that LatencyMean, LatencyMeanAndStdDev and Percentiles
+// recompute from the full cluster-wide distribution. In StatsMode_HDR, where
+// replies carry a histogram instead of raw samples, the per-server histograms
+// are merged onto one canonical histogram (see [mergeHistograms]) instead.
+//
+// Per-server memory and alloc counters are attached as ServerStats in a stable
+// node-ID order so the output columns do not reshuffle between runs.
+func AggregateServerResults(replies map[uint32]*Result) (*Result, error) {
+	if len(replies) == 0 {
+		return nil, gorums.ErrIncomplete
+	}
+
+	resp := &Result{}
+	var allSamples []int64
+	var hists []*LatencyHistogram
+	for _, id := range slices.Sorted(maps.Keys(replies)) {
+		reply := replies[id]
+		// The benchmark name lives in RunConfig and is stamped by Run on the
+		// aggregated result; per-server replies carry no config.
+		resp.SetTotalOps(resp.GetTotalOps() + reply.GetTotalOps())
+		resp.SetTotalTime(max(resp.GetTotalTime(), reply.GetTotalTime()))
+		resp.SetThroughput(resp.GetThroughput() + reply.GetThroughput())
+		allSamples = append(allSamples, reply.GetLatencies()...)
+		if h := reply.GetHistogram(); h != nil {
+			hists = append(hists, h)
+		}
+		resp.SetServerStats(append(resp.GetServerStats(), MemoryStat_builder{
+			Allocs: reply.GetAllocsPerOp() * reply.GetTotalOps(),
+			Memory: reply.GetMemPerOp() * reply.GetTotalOps(),
+		}.Build()))
+	}
+	if len(allSamples) > 0 {
+		resp.SetLatencies(allSamples)
+	}
+	if len(hists) > 0 {
+		resp.SetHistogram(mergeHistograms(hists...))
+	}
+	return resp, nil
+}

benchkit/aggregate_test.go

@@ -0,0 +1,186 @@
+package benchkit
+
+import (
+	"errors"
+	"math"
+	"testing"
+	"time"
+
+	"github.com/relab/gorums"
+)
+
+func TestAggregateServerResults(t *testing.T) {
+	const eps = 1e-9
+
+	// reply is a compact per-node stub used to build the input map.
+	// samples stands in for per-op latency measurements in nanoseconds; the
+	// aggregator concatenates them across servers so the cluster-wide mean
+	// and stddev can be recomputed from the full distribution.
+	type reply struct {
+		tTime    int64
+		tput     float64
+		samples  []int64
+		allocsPO uint64
+		memPO    uint64
+	}
+	newReplies := func(in map[uint32]reply) map[uint32]*Result {
+		out := make(map[uint32]*Result, len(in))
+		for id, r := range in {
+			out[id] = Result_builder{
+				TotalOps:    uint64(len(r.samples)),
+				TotalTime:   r.tTime,
+				Throughput:  r.tput,
+				Latencies:   r.samples,
+				AllocsPerOp: r.allocsPO,
+				MemPerOp:    r.memPO,
+			}.Build()
+		}
+		return out
+	}
+
+	tests := []struct {
+		name    string
+		replies map[uint32]reply
+		wantErr error
+		// For non-error cases, the expected aggregate values.
+		wantTotalOps      uint64
+		wantTotalTime     int64
+		wantThroughput    float64
+		wantSamplesConcat []int64  // feeds expected mean/stddev via Result methods
+		wantServerMem     []uint64 // per-server memory in sorted node-ID order
+		wantServerAlloc   []uint64 // per-server allocs in sorted node-ID order
+	}{
+		{
+			name:    "EmptyRepliesReturnsIncomplete",
+			replies: nil,
+			wantErr: gorums.ErrIncomplete,
+		},
+		{
+			name: "SingleReplyPassesThrough",
+			replies: map[uint32]reply{
+				1: {tTime: 1_000_000, tput: 100, samples: []int64{10, 20, 30, 40}, allocsPO: 3, memPO: 64},
+			},
+			wantTotalOps:      4,
+			wantTotalTime:     1_000_000,
+			wantThroughput:    100,
+			wantSamplesConcat: []int64{10, 20, 30, 40},
+			wantServerMem:     []uint64{64 * 4},
+			wantServerAlloc:   []uint64{3 * 4},
+		},
+		{
+			name: "ThreeRepliesConcatenateSamples",
+			replies: map[uint32]reply{
+				1: {tTime: 1, tput: 10, samples: []int64{10, 12}, allocsPO: 2, memPO: 8},
+				2: {tTime: 2, tput: 20, samples: []int64{20, 22, 24}, allocsPO: 3, memPO: 16},
+				3: {tTime: 3, tput: 30, samples: []int64{30, 32, 34, 36}, allocsPO: 5, memPO: 32},
+			},
+			wantTotalOps:      9,
+			wantTotalTime:     3, // max across servers, not the sum
+			wantThroughput:    60,
+			wantSamplesConcat: []int64{10, 12, 20, 22, 24, 30, 32, 34, 36},
+			wantServerMem:     []uint64{8 * 2, 16 * 3, 32 * 4}, // sorted by node ID: 1, 2, 3
+			wantServerAlloc:   []uint64{2 * 2, 3 * 3, 5 * 4},
+		},
+		{
+			name: "UnsortedIDsProduceSortedServerStats",
+			replies: map[uint32]reply{
+				// Insertion order does not matter; sorted ID order determines ServerStats[i].
+				7: {tTime: 1, tput: 1, samples: []int64{5, 5, 5}, allocsPO: 1, memPO: 1},
+				3: {tTime: 1, tput: 1, samples: []int64{5, 5, 5}, allocsPO: 2, memPO: 2},
+				5: {tTime: 1, tput: 1, samples: []int64{5, 5, 5}, allocsPO: 3, memPO: 3},
+			},
+			wantTotalOps:      9,
+			wantTotalTime:     1, // max across servers, not the sum
+			wantThroughput:    3,
+			wantSamplesConcat: []int64{5, 5, 5, 5, 5, 5, 5, 5, 5},
+			wantServerMem:     []uint64{2 * 3, 3 * 3, 1 * 3}, // node IDs sorted: 3, 5, 7 -> mem 2,3,1
+			wantServerAlloc:   []uint64{2 * 3, 3 * 3, 1 * 3},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got, err := AggregateServerResults(newReplies(tt.replies))
+			if tt.wantErr != nil {
+				if !errors.Is(err, tt.wantErr) {
+					t.Fatalf("err = %v, want %v", err, tt.wantErr)
+				}
+				return
+			}
+			if err != nil {
+				t.Fatalf("unexpected error: %v", err)
+			}
+			if got.GetTotalOps() != tt.wantTotalOps {
+				t.Errorf("TotalOps = %d, want %d", got.GetTotalOps(), tt.wantTotalOps)
+			}
+			if got.GetTotalTime() != tt.wantTotalTime {
+				t.Errorf("TotalTime = %d, want %d", got.GetTotalTime(), tt.wantTotalTime)
+			}
+			if math.Abs(got.GetThroughput()-tt.wantThroughput) > eps {
+				t.Errorf("Throughput = %v, want %v", got.GetThroughput(), tt.wantThroughput)
+			}
+
+			// Reference values come from invoking the same methods on a
+			// Result built from the expected concatenation. time.Duration
+			// is int64 so equality is exact once the truncation boundary
+			// is shared between got and want.
+			want := Result_builder{Latencies: tt.wantSamplesConcat}.Build()
+			gotMean, gotSD := got.LatencyMeanAndStdDev()
+			wantMean, wantSD := want.LatencyMeanAndStdDev()
+			if gotMean != wantMean {
+				t.Errorf("LatencyMean = %v, want %v", gotMean, wantMean)
+			}
+			if gotSD != wantSD {
+				t.Errorf("LatencyStdDev = %v, want %v", gotSD, wantSD)
+			}
+
+			if n := len(got.GetServerStats()); n != len(tt.wantServerMem) {
+				t.Fatalf("len(ServerStats) = %d, want %d", n, len(tt.wantServerMem))
+			}
+			for i, s := range got.GetServerStats() {
+				if s.GetMemory() != tt.wantServerMem[i] {
+					t.Errorf("ServerStats[%d].Memory = %d, want %d", i, s.GetMemory(), tt.wantServerMem[i])
+				}
+				if s.GetAllocs() != tt.wantServerAlloc[i] {
+					t.Errorf("ServerStats[%d].Allocs = %d, want %d", i, s.GetAllocs(), tt.wantServerAlloc[i])
+				}
+			}
+		})
+	}
+}
+
+// TestAggregateServerResultsHistogram verifies that in StatsMode_HDR, where
+// per-server replies carry a histogram instead of raw samples, the aggregate
+// merges the per-server histograms (Latencies nil, Histogram set) with counts
+// summed across servers, and that TotalOps and Throughput still sum as usual.
+func TestAggregateServerResultsHistogram(t *testing.T) {
+	replies := map[uint32]*Result{
+		1: Result_builder{TotalOps: 3, TotalTime: 2, Throughput: 30, Histogram: hist(1_000, 1_000, 1_000)}.Build(),
+		2: Result_builder{TotalOps: 2, TotalTime: 1, Throughput: 20, Histogram: hist(1_000, 2_000)}.Build(),
+	}
+	got, err := AggregateServerResults(replies)
+	if err != nil {
+		t.Fatalf("AggregateServerResults: %v", err)
+	}
+	if got.GetLatencies() != nil {
+		t.Errorf("Latencies = %v, want nil in HDR mode", got.GetLatencies())
+	}
+	if got.GetTotalOps() != 5 {
+		t.Errorf("TotalOps = %d, want 5", got.GetTotalOps())
+	}
+	if got.GetThroughput() != 50 {
+		t.Errorf("Throughput = %v, want 50", got.GetThroughput())
+	}
+	h := got.GetHistogram()
+	if h == nil {
+		t.Fatal("Histogram = nil, want merged histogram")
+	}
+	if n := totalCount(h); n != 5 {
+		t.Errorf("merged histogram count = %d, want 5", n)
+	}
+	// Merged distribution {1,1,1,1,2}µs has median 1µs, reproduced within
+	// HDR precision.
+	if p := got.Percentiles(0.5); p == nil || math.Abs(float64(p[0]-time.Microsecond)) > 50 {
+		t.Errorf("Percentiles(0.5) = %v, want ≈1µs", p)
+	}
+}