diff --git a/.gitignore b/.gitignore index 53ef0c3d..0a3c299c 100644 --- a/.gitignore +++ b/.gitignore @@ -28,6 +28,7 @@ z-scratch/ .scratch/ # Binary files +benchkit/cmd/benchmark/benchmark examples/storage/storage # Gorums generated backup files diff --git a/AGENTS.md b/AGENTS.md index 85ffd0db..da37b40e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -29,6 +29,8 @@ gorums/ │ └── gengorums/ # Compiler logic + templates ├── benchkit/ # Separate module: measurement and benchmarking │ ├── proto/ # .proto sources for the benchkit module +│ ├── benchmark/ # Gorums workloads built on benchkit +│ ├── cmd/benchmark/ # Benchmark node binary ├── examples/ # Separate module: example implementations ├── internal/ # Internal packages ├── doc/ # Documentation @@ -256,6 +258,7 @@ Before making significant changes, consult: ## Performance Considerations - Gorums is used in performance-critical distributed systems +- Benchmarking tools are available in `benchkit/benchmark/` and `benchkit/cmd/benchmark/` - Profile before optimizing - use Go's pprof tools ## Communication with Project Maintainer diff --git a/Makefile b/Makefile index a27e8af3..217083b5 100644 --- a/Makefile +++ b/Makefile @@ -17,10 +17,11 @@ 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 +benchmark_deps := $(benchkit_deps) benchkit/benchmark/benchmark.pb.go benchkit/benchmark/benchmark_gorums.pb.go -.PHONY: all dev tools bootstrapgorums installgorums benchkit test compiletests genproto benchtest bench lint deadcode modernize goplscheck +.PHONY: all dev tools bootstrapgorums installgorums benchmark benchkit test compiletests genproto benchtest bench lint deadcode modernize goplscheck -all: dev compiletests +all: dev benchmark compiletests dev: installgorums $(runtime_deps) @rm -f $(dev_path)/zorums*.pb.go @@ -30,6 +31,9 @@ dev: installgorums $(runtime_deps) --go_opt=default_api_level=API_OPAQUE \ $(zorums_proto) +benchmark: installgorums $(benchmark_deps) + @go build -C benchkit -o cmd/benchmark/benchmark ./cmd/benchmark + benchkit: installgorums $(benchkit_deps) # The benchkit module's generated code is written back into the module root @@ -48,6 +52,15 @@ benchkit/control_gorums.pb.go: $(bk_path)/benchkit/control.proto @protoc -I=$(bk_proto_path) \ --gorums_out=benchkit --gorums_opt=module=$(bk_module) $< +benchkit/benchmark/benchmark.pb.go: $(bk_path)/benchmark/benchmark.proto + @protoc -I=$(bk_proto_path) \ + --go_out=benchkit --go_opt=module=$(bk_module) \ + --go_opt=default_api_level=API_OPAQUE $< + +benchkit/benchmark/benchmark_gorums.pb.go: $(bk_path)/benchmark/benchmark.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) @@ -158,12 +171,12 @@ goplscheck: exit 1; \ fi -# Regenerate all Gorums and protobuf generated files across the repo (dev, benchkit, internal/tests, examples). +# Regenerate all Gorums and protobuf generated files across the repo (dev, benchkit, benchmark, internal/tests, examples). # This will force regeneration even though the proto files have not changed. genproto: installgorums dev - @echo "Regenerating all proto files (dev, benchkit, internal/tests, examples)" + @echo "Regenerating all proto files (dev, benchkit, benchmark, internal/tests, examples)" @$(MAKE) -B -s dev - @$(MAKE) -B -s $(benchkit_deps) + @$(MAKE) -B -s $(benchmark_deps) @$(MAKE) -B -s --no-print-directory -C ./internal/tests all @$(MAKE) -B -s --no-print-directory -C ./examples all diff --git a/benchkit/benchmark/benchmark.go b/benchkit/benchmark/benchmark.go new file mode 100644 index 00000000..82e66b62 --- /dev/null +++ b/benchkit/benchmark/benchmark.go @@ -0,0 +1,389 @@ +// Package benchmark drives Gorums quorum calls against a configuration of +// server nodes to measure their performance. +// +// It builds on [github.com/relab/gorums/benchkit] to run configurable load +// against local or remote targets, including the symmetric peer-to-peer setup +// where every server also calls the others, and reports latency and throughput. +package benchmark + +import ( + "context" + "errors" + "fmt" + "os" + "regexp" + "sync/atomic" + "time" + + "github.com/relab/gorums/benchkit" + "golang.org/x/sync/errgroup" +) + +// BenchTarget carries the targets for benchmark execution. Set Config for +// traditional (coordinator→servers) benchmarks and Symmetric for peer-to-peer +// benchmarks; either or both may be set. Benchmarks for unset targets are +// omitted from GetBenchmarks. +type BenchTarget struct { + Config Config + Symmetric *SymmetricTarget +} + +// asyncQCFunc issues one async quorum call and returns its future. +type asyncQCFunc func(*ConfigContext, *Echo, int) AsyncEcho + +// ErrAsyncQCRampUnsupported is returned by runAsyncQCBenchmark when rate +// ramping (-rate-step/-rate-step-max) is requested; AsyncQuorumCall does not +// support it. +var ErrAsyncQCRampUnsupported = errors.New("rate ramping is not supported by AsyncQuorumCall") + +// asyncQCComplete processes the outcome of one async quorum call and reports +// whether the run should continue. [benchkit.ErrRunOver] ends it, recording the +// op neither as a success nor as a failure; any other error is counted via +// [benchkit.Measurement.RecordError] and the run continues, so a saturating +// workload degrades gracefully instead of aborting on the first failure; a nil +// error records elapsed as the round-trip latency. +func asyncQCComplete(err error, elapsed time.Duration, m *benchkit.Measurement) (continueRun bool) { + switch { + case errors.Is(err, benchkit.ErrRunOver): + return false + case err != nil: + m.RecordError() + default: + m.Stats.AddLatency(elapsed) + } + return true +} + +// runAsyncQCBenchmark drives the async quorum-call lifecycle: opts.Workers +// dispatchers keep up to MaxAsync calls in flight, so it is a custom +// [benchkit.Bench].Run closure built on benchkit's primitives instead of +// [benchkit.ClientMeasured]. A failed call is counted without aborting the run, +// [benchkit.ErrRunOver] ends it cleanly, and a paced run that falls behind its +// offered-load schedule warns on stderr. +func runAsyncQCBenchmark(opts benchkit.Options, config Config, f asyncQCFunc) (*benchkit.Result, error) { + if opts.RateStep > 0 || opts.RateStepMax > 0 { + return nil, fmt.Errorf("benchmark %q: %w", opts.BenchName, ErrAsyncQCRampUnsupported) + } + ctx, cancel := benchkit.BenchContext(opts) + defer cancel() + cfgCtx := config.Context(ctx) + msg := Echo_builder{Payload: make([]byte, opts.Payload)}.Build() + var g errgroup.Group + var sent atomic.Uint64 + offered := opts.OfferedOps() + + if err := benchkit.StartRemote(cfgCtx, opts); err != nil { + return nil, err + } + + m := benchkit.StartMeasurement(opts) + measureStart := time.Now() + endTime := measureStart.Add(opts.Duration) + // gate paces new sends to opts.Rate across all dispatching goroutines; nil + // when unlimited, so a shared gate replaces a per-worker pacer. + gate := benchkit.NewRatedGate(opts.Rate, measureStart) + // inFlight holds one token per dispatched call, so acquiring a token is + // what bounds concurrency at MaxAsync. A counter compared before dispatch + // would not: the comparison and the increment are separate steps, and every + // dispatcher can pass the same comparison before any of them increments. + inFlight := make(chan struct{}, max(opts.MaxAsync, 1)) + var runOver atomic.Bool + dispatch := func() error { + for !time.Now().After(endTime) && ctx.Err() == nil && !runOver.Load() { + select { + case inFlight <- struct{}{}: + case <-ctx.Done(): + return nil + } + if !gate.Wait(ctx) { + <-inFlight + return nil + } + if offered > 0 { + sent.Add(1) + } + start := time.Now() + fut := f(cfgCtx, msg, opts.QuorumSize) + // The completion goroutine times the call and does nothing else. + // Dispatching from here instead would put this goroutine's own work + // between the call completing and the clock being read, so the + // harness's scheduling would land inside the measurement. + g.Go(func() error { + _, err := fut.Get() + elapsed := time.Since(start) + <-inFlight + if !asyncQCComplete(err, elapsed, m) { + runOver.Store(true) + } + return nil + }) + } + return nil + } + + for range opts.Workers { + g.Go(dispatch) + } + if err := g.Wait(); err != nil { + return nil, err + } + if warning := benchkit.PaceWarning(sent.Load(), offered); warning != "" { + fmt.Fprintln(os.Stderr, warning) + } + + result := m.Finish() + if _, err := benchkit.StopRemote(cfgCtx, opts, result); err != nil { + return nil, err + } + + return result, nil +} + +// asyncSends bounds how many one-way sends a benchmark keeps outstanding, so a +// one-way benchmark can pipeline the way an Async caller does. It is safe for +// concurrent use by the workers sharing one benchmark closure. +type asyncSend interface { + Wait() error +} + +type asyncSends struct { + handles chan asyncSend + slots chan struct{} +} + +// newAsyncSends bounds the outstanding sends at depth, the benchmark's +// -max-async. A depth below one is raised to one. +func newAsyncSends(depth int) *asyncSends { + depth = max(depth, 1) + return &asyncSends{ + handles: make(chan asyncSend, depth), + slots: make(chan struct{}, depth), + } +} + +// dispatch reserves outstanding capacity before invoking send. Once depth +// sends are outstanding, it reaps the oldest to make room. An error belongs to +// that earlier send, so the pending send is not dispatched after the run has +// already failed. +func (a *asyncSends) dispatch(send func() asyncSend) error { + for { + select { + case a.slots <- struct{}{}: + a.handles <- send() + return nil + default: + } + + oldest := <-a.handles + err := oldest.Wait() + <-a.slots + if err != nil { + return err + } + } +} + +// drain collects every send still outstanding and reports the first failure. +// It runs after the send window so the servers observe the whole pipeline. +func (a *asyncSends) drain() error { + var firstErr error + for { + select { + case h := <-a.handles: + if err := h.Wait(); err != nil && firstErr == nil { + firstErr = err + } + <-a.slots + default: + return firstErr + } + } +} + +// benchTargetNeeds identifies which BenchTarget field a benchDesc's build +// function requires, so GetBenchmarks can filter benchDescs by what the +// caller's target actually provides. +type benchTargetNeeds int + +const ( + needsConfig benchTargetNeeds = iota // requires a non-nil Config (traditional coordinator→servers) + needsSymmetric // requires a non-nil *SymmetricTarget (peer-to-peer) +) + +// benchDesc is the single source of truth for one benchmark: its name and +// description (used by -list via [BenchmarkDescriptions]) and how to build +// its runnable closure (used by [GetBenchmarks]). +type benchDesc struct { + Name string + Description string + Needs benchTargetNeeds + build func(cfg Config, sym *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) +} + +// benchDescs lists every known benchmark. Traditional benchmarks (needsConfig) +// receive cfg, which [GetBenchmarks] derives from t.Config or, when unset, +// from the symmetric target's server 0 outbound config; peer-to-peer +// benchmarks (needsSymmetric) receive sym. +var benchDescs = []benchDesc{ + { + Name: "QuorumCall", + Description: "NodeStream based quorum call implementation with FIFO ordering", + Needs: needsConfig, + build: func(cfg Config, _ *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + return benchkit.ClientMeasured(cfg, func(opts benchkit.Options, cc *ConfigContext) func() error { + msg := Echo_builder{Payload: make([]byte, opts.Payload)}.Build() + if opts.CallTimeout <= 0 { + return func() error { + _, err := QuorumCall(cc, msg).Threshold(opts.QuorumSize) + return err + } + } + // With -call-timeout, each call carries its own deadline so a + // call stalled behind an unresponsive peer fails with + // DeadlineExceeded instead of hanging until run end. + callCfg := cc.Config() + return func() error { + callCtx, cancel := context.WithTimeout(cc, opts.CallTimeout) + defer cancel() + _, err := QuorumCall(callCfg.Context(callCtx), msg).Threshold(opts.QuorumSize) + return err + } + }) + }, + }, + { + Name: "AsyncQuorumCall", + Description: "NodeStream based async quorum call implementation with FIFO ordering", + Needs: needsConfig, + build: func(cfg Config, _ *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + return func(opts benchkit.Options) (*benchkit.Result, error) { + return runAsyncQCBenchmark(opts, cfg, func(ctx *ConfigContext, in *Echo, quorumSize int) AsyncEcho { + return QuorumCall(ctx, in).AsyncThreshold(quorumSize) + }) + } + }, + }, + { + Name: "SlowServer", + Description: "Quorum Call with a 10ms processing time on the server", + Needs: needsConfig, + build: func(cfg Config, _ *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + return benchkit.ClientMeasured(cfg, func(opts benchkit.Options, cc *ConfigContext) func() error { + msg := Echo_builder{Payload: make([]byte, opts.Payload)}.Build() + return func() error { + _, err := SlowServer(cc, msg).Threshold(opts.QuorumSize) + return err + } + }) + }, + }, + { + Name: "Multicast", + Description: "NodeStream based multicast implementation (servers measure latency and throughput)", + Needs: needsConfig, + build: func(cfg Config, _ *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + return benchkit.ServerMeasured(cfg, func(opts benchkit.Options, cc *ConfigContext) func() error { + payload := make([]byte, opts.Payload) + return func() error { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano(), Payload: payload}.Build() + return Multicast(cc, msg).Send() + } + }) + }, + }, + { + Name: "AsyncMulticast", + Description: "Multicast pipelined with Async, up to -max-async sends outstanding (servers measure latency and throughput)", + Needs: needsConfig, + build: func(cfg Config, _ *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + // outstanding is rebound by setup on every run and drained by the + // quiesce hook after that run's send window; both run on the run + // goroutine, in that order. + var outstanding *asyncSends + return benchkit.ServerMeasured(cfg, + func(opts benchkit.Options, cc *ConfigContext) func() error { + payload := make([]byte, opts.Payload) + outstanding = newAsyncSends(opts.MaxAsync) + return func() error { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano(), Payload: payload}.Build() + return outstanding.dispatch(func() asyncSend { return Multicast(cc, msg).Async() }) + } + }, + benchkit.WithQuiesce(func(context.Context) error { + return outstanding.drain() + })) + }, + }, + { + Name: "SymmetricQuorumCall", + Description: "Peer-to-peer quorum call benchmark; each node is both client and server", + Needs: needsSymmetric, + build: func(_ Config, sym *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + return func(opts benchkit.Options) (*benchkit.Result, error) { + return runSymmetricQuorumCall(sym, opts) + } + }, + }, + { + Name: "SymmetricMulticast", + Description: "Peer-to-peer multicast benchmark; servers measure one-way latency", + Needs: needsSymmetric, + build: func(_ Config, sym *SymmetricTarget) func(benchkit.Options) (*benchkit.Result, error) { + return func(opts benchkit.Options) (*benchkit.Result, error) { + return runSymmetricMulticast(sym, opts) + } + }, + }, +} + +// BenchmarkDescriptions returns name and description for every known +// benchmark, regardless of which targets are available. Used by -list. +func BenchmarkDescriptions() []benchkit.Bench { + descs := make([]benchkit.Bench, len(benchDescs)) + for i, d := range benchDescs { + descs[i] = benchkit.Bench{Name: d.Name, Description: d.Description} + } + return descs +} + +// GetBenchmarks returns runnable benchmarks for the given targets. Traditional +// (needsConfig) benchmarks are included when t.Config is set, or when +// t.Symmetric is a single-process local target, in which case server 0's +// outbound config serves as the Config. They are excluded for a distributed +// (multi-process) symmetric target: every node there runs the same binary, so +// if more than one selected a needsConfig benchmark, each would issue its own +// Control.Start/Stop against the same peer group concurrently, resetting and +// reading every other node's Stats window mid-run. Symmetric benchmarks are +// included whenever t.Symmetric is set, local or distributed. +func GetBenchmarks(t BenchTarget) []benchkit.Bench { + cfg := t.Config + if cfg == nil && t.Symmetric != nil && t.Symmetric.selfAddr == "" && len(t.Symmetric.servers) > 0 { + cfg = t.Symmetric.servers[0].PeerConfig() + } + var m []benchkit.Bench + for _, d := range benchDescs { + switch d.Needs { + case needsConfig: + if cfg == nil { + continue + } + case needsSymmetric: + if t.Symmetric == nil { + continue + } + } + m = append(m, benchkit.Bench{ + Name: d.Name, + Description: d.Description, + Run: d.build(cfg, t.Symmetric), + }) + } + return m +} + +// RunBenchmarks runs all the benchmarks that match the given regex with the +// given options against the target, delegating selection, the per-benchmark +// metadata, and ordering to the benchkit harness. +func RunBenchmarks(benchRegex *regexp.Regexp, options benchkit.Options, t BenchTarget) ([]*benchkit.Result, error) { + return benchkit.Run(benchRegex, options, GetBenchmarks(t)) +} diff --git a/benchkit/benchmark/benchmark.pb.go b/benchkit/benchmark/benchmark.pb.go new file mode 100644 index 00000000..709ef9a6 --- /dev/null +++ b/benchkit/benchmark/benchmark.pb.go @@ -0,0 +1,235 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc v7.35.1 +// source: benchmark/benchmark.proto + +package benchmark + +import ( + _ "github.com/relab/gorums" + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + emptypb "google.golang.org/protobuf/types/known/emptypb" + reflect "reflect" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +// Echo is a simple message used for echo benchmarks. +type Echo struct { + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_Payload []byte `protobuf:"bytes,1,opt,name=payload"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Echo) Reset() { + *x = Echo{} + mi := &file_benchmark_benchmark_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Echo) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Echo) ProtoMessage() {} + +func (x *Echo) ProtoReflect() protoreflect.Message { + mi := &file_benchmark_benchmark_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +func (x *Echo) GetPayload() []byte { + if x != nil { + return x.xxx_hidden_Payload + } + return nil +} + +func (x *Echo) SetPayload(v []byte) { + if v == nil { + v = []byte{} + } + x.xxx_hidden_Payload = v +} + +type Echo_builder struct { + _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. + + Payload []byte +} + +func (b0 Echo_builder) Build() *Echo { + m0 := &Echo{} + b, x := &b0, m0 + _, _ = b, x + x.xxx_hidden_Payload = b.Payload + return m0 +} + +// TimedMsg is a message with a send time and a payload used for multicast benchmarks. +type TimedMsg struct { + state protoimpl.MessageState `protogen:"opaque.v1"` + xxx_hidden_SendTime int64 `protobuf:"varint,1,opt,name=send_time,json=sendTime"` + xxx_hidden_Payload []byte `protobuf:"bytes,2,opt,name=payload"` + xxx_hidden_SenderId uint32 `protobuf:"varint,3,opt,name=sender_id,json=senderId"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *TimedMsg) Reset() { + *x = TimedMsg{} + mi := &file_benchmark_benchmark_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *TimedMsg) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*TimedMsg) ProtoMessage() {} + +func (x *TimedMsg) ProtoReflect() protoreflect.Message { + mi := &file_benchmark_benchmark_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +func (x *TimedMsg) GetSendTime() int64 { + if x != nil { + return x.xxx_hidden_SendTime + } + return 0 +} + +func (x *TimedMsg) GetPayload() []byte { + if x != nil { + return x.xxx_hidden_Payload + } + return nil +} + +func (x *TimedMsg) GetSenderId() uint32 { + if x != nil { + return x.xxx_hidden_SenderId + } + return 0 +} + +func (x *TimedMsg) SetSendTime(v int64) { + x.xxx_hidden_SendTime = v +} + +func (x *TimedMsg) SetPayload(v []byte) { + if v == nil { + v = []byte{} + } + x.xxx_hidden_Payload = v +} + +func (x *TimedMsg) SetSenderId(v uint32) { + x.xxx_hidden_SenderId = v +} + +type TimedMsg_builder struct { + _ [0]func() // Prevents comparability and use of unkeyed literals for the builder. + + SendTime int64 + Payload []byte + SenderId uint32 +} + +func (b0 TimedMsg_builder) Build() *TimedMsg { + m0 := &TimedMsg{} + b, x := &b0, m0 + _, _ = b, x + x.xxx_hidden_SendTime = b.SendTime + x.xxx_hidden_Payload = b.Payload + x.xxx_hidden_SenderId = b.SenderId + return m0 +} + +var File_benchmark_benchmark_proto protoreflect.FileDescriptor + +const file_benchmark_benchmark_proto_rawDesc = "" + + "\n" + + "\x19benchmark/benchmark.proto\x12\tbenchmark\x1a\x1bgoogle/protobuf/empty.proto\x1a\fgorums.proto\" \n" + + "\x04Echo\x12\x18\n" + + "\apayload\x18\x01 \x01(\fR\apayload\"^\n" + + "\bTimedMsg\x12\x1b\n" + + "\tsend_time\x18\x01 \x01(\x03R\bsendTime\x12\x18\n" + + "\apayload\x18\x02 \x01(\fR\apayload\x12\x1b\n" + + "\tsender_id\x18\x03 \x01(\rR\bsenderId2\xb7\x01\n" + + "\tBenchmark\x124\n" + + "\n" + + "QuorumCall\x12\x0f.benchmark.Echo\x1a\x0f.benchmark.Echo\"\x04\xa0\xb5\x18\x01\x124\n" + + "\n" + + "SlowServer\x12\x0f.benchmark.Echo\x1a\x0f.benchmark.Echo\"\x04\xa0\xb5\x18\x01\x12>\n" + + "\tMulticast\x12\x13.benchmark.TimedMsg\x1a\x16.google.protobuf.Empty\"\x04\x98\xb5\x18\x01B1Z*github.com/relab/gorums/benchkit/benchmark\x92\x03\x02\b\x02b\beditionsp\xe9\a" + +var file_benchmark_benchmark_proto_msgTypes = make([]protoimpl.MessageInfo, 2) +var file_benchmark_benchmark_proto_goTypes = []any{ + (*Echo)(nil), // 0: benchmark.Echo + (*TimedMsg)(nil), // 1: benchmark.TimedMsg + (*emptypb.Empty)(nil), // 2: google.protobuf.Empty +} +var file_benchmark_benchmark_proto_depIdxs = []int32{ + 0, // 0: benchmark.Benchmark.QuorumCall:input_type -> benchmark.Echo + 0, // 1: benchmark.Benchmark.SlowServer:input_type -> benchmark.Echo + 1, // 2: benchmark.Benchmark.Multicast:input_type -> benchmark.TimedMsg + 0, // 3: benchmark.Benchmark.QuorumCall:output_type -> benchmark.Echo + 0, // 4: benchmark.Benchmark.SlowServer:output_type -> benchmark.Echo + 2, // 5: benchmark.Benchmark.Multicast:output_type -> google.protobuf.Empty + 3, // [3:6] is the sub-list for method output_type + 0, // [0:3] is the sub-list for method input_type + 0, // [0:0] is the sub-list for extension type_name + 0, // [0:0] is the sub-list for extension extendee + 0, // [0:0] is the sub-list for field type_name +} + +func init() { file_benchmark_benchmark_proto_init() } +func file_benchmark_benchmark_proto_init() { + if File_benchmark_benchmark_proto != nil { + return + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_benchmark_benchmark_proto_rawDesc), len(file_benchmark_benchmark_proto_rawDesc)), + NumEnums: 0, + NumMessages: 2, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_benchmark_benchmark_proto_goTypes, + DependencyIndexes: file_benchmark_benchmark_proto_depIdxs, + MessageInfos: file_benchmark_benchmark_proto_msgTypes, + }.Build() + File_benchmark_benchmark_proto = out.File + file_benchmark_benchmark_proto_goTypes = nil + file_benchmark_benchmark_proto_depIdxs = nil +} diff --git a/benchkit/benchmark/benchmark_gorums.pb.go b/benchkit/benchmark/benchmark_gorums.pb.go new file mode 100644 index 00000000..6120f998 --- /dev/null +++ b/benchkit/benchmark/benchmark_gorums.pb.go @@ -0,0 +1,104 @@ +// Code generated by protoc-gen-gorums. DO NOT EDIT. +// versions: +// protoc-gen-gorums v0.11.0-devel +// protoc v7.35.1 +// source: benchmark/benchmark.proto + +package benchmark + +import ( + gorums "github.com/relab/gorums" + gorumsimpl "github.com/relab/gorums/runtime/gorumsimpl" + emptypb "google.golang.org/protobuf/types/known/emptypb" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = gorumsimpl.EnforceVersion(11 - gorumsimpl.MinVersion) + // Verify that the gorums runtime is sufficiently up-to-date. + _ = gorumsimpl.EnforceVersion(gorumsimpl.MaxVersion - 11) +) + +// The type aliases below are useful Gorums types that we make accessible +// from generated code. These names therefore become reserved identifiers, +// meaning that proto message types with these names would collide with the +// generated aliases and cause a compile error. +// +// The bundler (gorums_bundle.go) is responsible for discovering these +// aliases and any other identifiers defined herein, and adding them to +// the reserved identifiers list. +// +// If necessary, additional aliases and other identifiers should be added in +// the generator's cmd/protoc-gen-gorums/dev directory, and the bundler will +// automatically discover them and add them to the reserved identifiers list. + +type ( + Config = gorums.Config + Node = gorums.Node + NodeContext = gorums.NodeContext + ConfigContext = gorums.ConfigContext +) + +// AsyncEcho is a future for async quorum calls returning *Echo. +type AsyncEcho = *gorums.Async[*Echo] + +// CorrectableEcho is a correctable object for quorum calls returning *Echo. +type CorrectableEcho = *gorums.Correctable[*Echo] + +// Reference imports to suppress errors if they are not otherwise used. +var _ emptypb.Empty + +// QuorumCall performs an echo quorum call on all servers. +func QuorumCall(ctx *ConfigContext, in *Echo) *gorums.Call[*Echo, *Echo] { + return gorumsimpl.QuorumCall[*Echo, *Echo]( + ctx, in, "benchmark.Benchmark.QuorumCall", + ) +} + +// SlowServer performs an echo quorum call on slow servers. +func SlowServer(ctx *ConfigContext, in *Echo) *gorums.Call[*Echo, *Echo] { + return gorumsimpl.QuorumCall[*Echo, *Echo]( + ctx, in, "benchmark.Benchmark.SlowServer", + ) +} + +// Multicast performs a multicast call to all servers. +// +// Example: +// +// err := Multicast(ctx, in).Send() +// h := Multicast(ctx, in).Async(); err := h.Wait() +func Multicast(ctx *ConfigContext, in *TimedMsg) *gorums.OnewayCall[*TimedMsg] { + return gorumsimpl.Multicast(ctx, in, "benchmark.Benchmark.Multicast") +} + +// Benchmark is the server-side API for the Benchmark Service +type BenchmarkServer interface { + QuorumCall(gorums.ServerContext, *Echo) (*Echo, error) + SlowServer(gorums.ServerContext, *Echo) (*Echo, error) + Multicast(gorums.ServerContext, *TimedMsg) +} + +func RegisterBenchmarkServer(srv *gorums.Server, impl BenchmarkServer) { + srv.RegisterHandler("benchmark.Benchmark.QuorumCall", func(ctx gorums.ServerContext, in *gorums.Message) (*gorums.Message, error) { + req := gorums.AsProto[*Echo](in) + resp, err := impl.QuorumCall(ctx, req) + if err != nil { + return nil, err + } + return gorums.NewResponseMessage(in, resp), nil + }) + srv.RegisterHandler("benchmark.Benchmark.SlowServer", func(ctx gorums.ServerContext, in *gorums.Message) (*gorums.Message, error) { + req := gorums.AsProto[*Echo](in) + resp, err := impl.SlowServer(ctx, req) + if err != nil { + return nil, err + } + return gorums.NewResponseMessage(in, resp), nil + }) + srv.RegisterHandler("benchmark.Benchmark.Multicast", func(ctx gorums.ServerContext, in *gorums.Message) (*gorums.Message, error) { + req := gorums.AsProto[*TimedMsg](in) + impl.Multicast(ctx, req) + return nil, nil + }) +} diff --git a/benchkit/benchmark/benchmark_test.go b/benchkit/benchmark/benchmark_test.go new file mode 100644 index 00000000..5ddfa735 --- /dev/null +++ b/benchkit/benchmark/benchmark_test.go @@ -0,0 +1,899 @@ +package benchmark + +import ( + "context" + "errors" + "fmt" + "slices" + "sync/atomic" + "testing" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/benchkit" + "github.com/relab/gorums/gorumstest" +) + +func TestBenchmarkDescriptions(t *testing.T) { + descs := BenchmarkDescriptions() + wantNames := []string{ + "QuorumCall", + "AsyncQuorumCall", + "SlowServer", + "Multicast", + "AsyncMulticast", + "SymmetricQuorumCall", + "SymmetricMulticast", + } + if len(descs) != len(wantNames) { + t.Fatalf("got %d descriptions, want %d", len(descs), len(wantNames)) + } + seen := make(map[string]bool, len(wantNames)) + for _, d := range descs { + if d.Description == "" { + t.Errorf("%q: empty description", d.Name) + } + seen[d.Name] = true + } + for _, want := range wantNames { + if !seen[want] { + t.Errorf("missing benchmark %q", want) + } + } +} + +func TestRunComplete(t *testing.T) { + tests := []struct { + name string + outSize, done, need int + want bool + }{ + {"NoneDone", 5, 0, 3, false}, + {"OneDoneQuorumStillPossible", 5, 1, 3, false}, // alive 4 >= 3 + {"EnoughDoneQuorumImpossible", 5, 3, 3, true}, // alive 2 < 3 + {"AllPeersNeededOneDone", 5, 1, 5, true}, // alive 4 < 5 + {"AllPeersNeededNoneDone", 5, 0, 5, false}, // never masks a startup fault + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := runComplete(tt.outSize, tt.done, tt.need); got != tt.want { + t.Errorf("runComplete(%d, %d, %d) = %v, want %v", + tt.outSize, tt.done, tt.need, got, tt.want) + } + }) + } +} + +// TestSymmetricRunOver checks the live classifiers flip from false to true only +// after peers signal Done, over a real (in-process) symmetric target. +func TestSymmetricRunOver(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + size := target.servers[0].PeerConfig().Size() + quorum := size/2 + 1 + + // No peer has signaled Done yet: nothing is over, so a failure now would + // still be reported as a fault. + if anyPeerFinished(target) { + t.Error("anyPeerFinished = true before any Done, want false") + } + if quorumRunOver(target, quorum) { + t.Error("quorumRunOver = true before any Done, want false") + } + + // Signal every peer done on the first server; the run is now winding down. + target.controls[0].ArmDone(size) + for id := 1; id <= size; id++ { + target.controls[0].Done(gorums.ServerContext{}, benchkit.DoneRequest_builder{SenderId: uint32(id)}.Build()) + } + if !anyPeerFinished(target) { + t.Error("anyPeerFinished = false after all peers Done, want true") + } + if !quorumRunOver(target, quorum) { + t.Error("quorumRunOver = false after all peers Done, want true") + } +} + +// TestRunSymmetricQuorumCallDefaultsToMajority verifies that +// runSymmetricQuorumCall completes a short run without an explicit +// QuorumSize, falling back to a majority of the outbound peer count. +func TestRunSymmetricQuorumCallDefaultsToMajority(t *testing.T) { + targets := localSymmetricTargets(t, 3, nil) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + for _, target := range targets { + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + } + + opts := benchkit.Options{Duration: 100 * time.Millisecond, Rate: 100, Workers: 1} + result, err := runSymmetricQuorumCall(targets[0], opts) + if err != nil { + t.Fatalf("runSymmetricQuorumCall: %v", err) + } + if result.GetTotalOps() == 0 { + t.Error("TotalOps = 0, want at least one completed quorum call") + } +} + +// TestRunSymmetricQuorumCallStragglerEndsCleanly verifies the straggler path: +// when a quorum call fails after enough peers have signaled Done that the +// quorum can no longer form, runSymmetricQuorumCall treats it as the expected +// end of the run (benchkit.ErrRunOver) rather than propagating the failure, +// so a node outliving its peers still returns a usable partial Result instead +// of failing the whole benchmark. +func TestRunSymmetricQuorumCallStragglerEndsCleanly(t *testing.T) { + servers := localServers(t, 2, nil) + target := benchTarget(servers[0], 2) // numPeers=2 arms Done tracking for IDs 1 and 2 + target2 := benchTarget(servers[1], 2) + go func() { _ = servers[0].ListenAndServe() }() + go func() { _ = servers[1].ListenAndServe() }() + + // Both sides probe: a dual-mode stream is only fully up once each side has + // dialed out, so waiting on only one direction can stall the other peer's + // half of the handshake. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + errCh := make(chan error, 2) + go func() { errCh <- AwaitReady(ctx, target) }() + go func() { errCh <- AwaitReady(ctx, target2) }() + var errs error + for range 2 { + if err := <-errCh; err != nil { + errs = errors.Join(errs, err) + } + } + if errs != nil { + t.Fatalf("AwaitReady: %v", errs) + } + + // Node 2 (the only outbound peer) finishes and exits; its calls now fail, + // and it has signaled Done, so quorumRunOver must recognize this as the + // expected end of the run rather than a fault. + servers[1].Stop() + target.controls[0].Done(gorums.ServerContext{}, benchkit.DoneRequest_builder{SenderId: 2}.Build()) + + // Duration is deliberately long relative to the expected stop: a regular + // (non-run-over) failure is only counted via Stats.RecordError and the + // run keeps retrying until Duration elapses, so a prompt return is what + // distinguishes the run-over path (which calls cancel() on the first + // failed call) from ordinary failure counting — both paths return a nil + // top-level error either way. + const duration = 2 * time.Second + opts := benchkit.Options{Duration: duration, Rate: 50, Workers: 1} + start := time.Now() + result, err := runSymmetricQuorumCall(target, opts) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("runSymmetricQuorumCall = %v, want nil error (straggler run-over should be a clean stop)", err) + } + if elapsed >= duration/2 { + t.Errorf("runSymmetricQuorumCall took %v, want well under %v (run-over should cancel on the first failed call)", elapsed, duration) + } + if got := result.GetFailedOps(); got != 0 { + t.Errorf("FailedOps = %d, want 0 (the failing call must be classified as run-over, not counted as a failure)", got) + } +} + +// TestRunSymmetricQuorumCallHonorsCallTimeout verifies that the symmetric +// runner's -call-timeout branch bounds each call to opts.CallTimeout instead +// of hanging behind an unresponsive peer until the run's own deadline +// elapses, mirroring TestQuorumCallHonorsCallTimeout for the coordinator +// path. +func TestRunSymmetricQuorumCallHonorsCallTimeout(t *testing.T) { + servers := localServers(t, 2, nil) + target := benchTarget(servers[0], 2) + // Node 2 drops every reply from the start, mirroring + // TestQuorumCallHonorsCallTimeout: a call without CallTimeout would hang + // until BenchContext's own (30s+) deadline instead of the short one + // below. AwaitReady is not used here since its own probe is a QuorumCall + // against the same handler and would never succeed against a + // permanently unresponsive peer. + registerReplyDroppingPeer(servers[1], 1<<30) + go func() { _ = servers[0].ListenAndServe() }() + go func() { _ = servers[1].ListenAndServe() }() + + // QuorumSize=2 requires both replies: server0's own PeerConfig includes + // itself alongside server1, so a threshold of 1 would be satisfied by the + // self entry alone without ever needing server1's (dropped) reply. + opts := benchkit.Options{ + Workers: 1, Duration: 100 * time.Millisecond, QuorumSize: 2, + CallTimeout: 20 * time.Millisecond, + } + start := time.Now() + result, err := runSymmetricQuorumCall(target, opts) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("runSymmetricQuorumCall: %v", err) + } + const wantBound = 5 * time.Second // generous vs. what an ignored CallTimeout would look like + if elapsed > wantBound { + t.Errorf("took %v, want well under %v (CallTimeout should bound each call against the unresponsive peer)", elapsed, wantBound) + } + if result.GetFailedOps() == 0 { + t.Error("FailedOps = 0, want > 0 (every call should time out against the unresponsive peer)") + } +} + +func TestGetBenchmarksTargetRouting(t *testing.T) { + symTarget := &SymmetricTarget{} + + tests := []struct { + name string + t BenchTarget + count int + }{ + {"empty", BenchTarget{}, 0}, + {"symmetric only", BenchTarget{Symmetric: symTarget}, 2}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := GetBenchmarks(tt.t) + if len(got) != tt.count { + t.Errorf("got %d benchmarks, want %d", len(got), tt.count) + } + }) + } +} + +// TestGetBenchmarksExcludesConfigBenchmarksForDistributedTarget verifies that +// a distributed (multi-process) symmetric target excludes needsConfig +// benchmarks (QuorumCall, Multicast, ...), unlike a local symmetric target. +// Every distributed node runs the same binary; if a needsConfig benchmark +// were exposed, more than one node selecting it would each issue their own +// Control.Start/Stop against the same peer group concurrently, corrupting +// every other node's Stats window mid-run. Only the needsSymmetric +// benchmarks (SymmetricQuorumCall, SymmetricMulticast), designed for +// concurrent per-node execution, are safe here. +func TestGetBenchmarksExcludesConfigBenchmarksForDistributedTarget(t *testing.T) { + peers := []string{"127.0.0.1:0", "127.0.0.2:0"} + target, stop, err := SetupRemoteServer(peers[0], peers, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupRemoteServer: %v", err) + } + t.Cleanup(stop) + + got := GetBenchmarks(BenchTarget{Symmetric: target}) + for _, b := range got { + if b.Name == "QuorumCall" || b.Name == "Multicast" || b.Name == "AsyncMulticast" || b.Name == "AsyncQuorumCall" || b.Name == "SlowServer" { + t.Errorf("GetBenchmarks(distributed target) included needsConfig benchmark %q, want excluded", b.Name) + } + } + const wantCount = 2 // SymmetricQuorumCall, SymmetricMulticast + if len(got) != wantCount { + t.Errorf("GetBenchmarks(distributed target) returned %d benchmarks, want %d", len(got), wantCount) + } +} + +// TestGetBenchmarksMatchesDescriptionsForFullTarget verifies that a target +// exposing both a Config and a SymmetricTarget produces exactly the +// runnable benchmarks BenchmarkDescriptions lists, by name and count: both +// views are derived from the one benchDescs table (see benchmark.go), so +// they cannot drift the way two hand-written lists could. +func TestGetBenchmarksMatchesDescriptionsForFullTarget(t *testing.T) { + target, stop, err := SetupSymmetricServers(2, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + // Symmetric alone is enough: GetBenchmarks derives cfg from server 0's + // outbound config when t.Config is unset, so needsConfig benchmarks are + // also included. + got := GetBenchmarks(BenchTarget{Symmetric: target}) + gotNames := make(map[string]bool, len(got)) + for _, b := range got { + gotNames[b.Name] = true + } + + wantDescs := BenchmarkDescriptions() + if len(got) != len(wantDescs) { + t.Fatalf("GetBenchmarks returned %d benchmarks, want %d (BenchmarkDescriptions)", len(got), len(wantDescs)) + } + for _, d := range wantDescs { + if !gotNames[d.Name] { + t.Errorf("BenchmarkDescriptions lists %q but GetBenchmarks did not return it", d.Name) + } + } +} + +// TestAsyncQCBoundsInFlight verifies that -max-async bounds the calls actually +// in flight, and that the recorded latency describes the calls rather than the +// harness. Both are checked against the same run: Little's law ties throughput +// and mean latency to the concurrency the bound permits, so a latency inflated +// by the harness's own scheduling shows up as an impossible concurrency. +func TestAsyncQCBoundsInFlight(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + const maxAsync = 64 + var inFlight, peak atomic.Int64 + opts := benchkit.Options{Workers: 2, MaxAsync: maxAsync, Duration: 2 * time.Second, QuorumSize: 2} + res, err := runAsyncQCBenchmark(opts, target.servers[0].PeerConfig(), + func(cc *ConfigContext, in *Echo, quorumSize int) AsyncEcho { + fut := QuorumCall(cc, in).AsyncThreshold(quorumSize) + cur := inFlight.Add(1) + for { + old := peak.Load() + if cur <= old || peak.CompareAndSwap(old, cur) { + break + } + } + // Async.Get reads an already-closed channel, so observing the same + // future from a second goroutine is safe. + go func() { _, _ = fut.Get(); inFlight.Add(-1) }() + return fut + }) + if err != nil { + t.Fatalf("run: %v", err) + } + + // The observer decrements after the benchmark's own completion path, so a + // small overshoot is measurement skew rather than a broken bound. + if got := peak.Load(); got > maxAsync*2 { + t.Errorf("peak in flight = %d, want <= %d (2x -max-async=%d)", got, maxAsync*2, maxAsync) + } + + lat := res.GetLatencies() + if len(lat) == 0 { + t.Fatal("no latency samples recorded") + } + var sum int64 + for _, l := range lat { + sum += l + } + mean := time.Duration(sum / int64(len(lat))) + elapsed := time.Duration(res.GetTotalTime()) + throughput := float64(res.GetTotalOps()) / elapsed.Seconds() + concurrency := throughput * mean.Seconds() + t.Logf("throughput=%.0f/s mean=%v peak_in_flight=%d littles_law_concurrency=%.1f", + throughput, mean, peak.Load(), concurrency) + if concurrency > maxAsync*4 { + t.Errorf("throughput %.0f/s at mean latency %v implies %.0f concurrent calls, but -max-async=%d; "+ + "the recorded latency is measuring the harness, not the calls", + throughput, mean, concurrency, maxAsync) + } +} + +// TestAsyncQCComplete verifies the per-call completion handling that keeps +// AsyncQuorumCall aligned with [benchkit.MeasureLatency]'s contract: +// [benchkit.ErrRunOver] stops the send chain without refiring and records +// neither a success nor a failure; any other error is counted via +// [benchkit.Measurement.RecordError] and the chain still refires, so a +// saturating workload degrades gracefully instead of aborting on the first +// failure; a nil error records the latency. +func TestAsyncQCComplete(t *testing.T) { + tests := []struct { + name string + err error + wantRefire bool + wantTotalOps uint64 + wantFailedOps uint64 + }{ + {"Success", nil, true, 1, 0}, + {"FailureRefiresAndCounts", errors.New("quorum call failed"), true, 0, 1}, + {"RunOverStopsWithoutCounting", benchkit.ErrRunOver, false, 0, 0}, + {"WrappedRunOverStopsWithoutCounting", fmt.Errorf("node 3: %w", benchkit.ErrRunOver), false, 0, 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := benchkit.StartMeasurement(benchkit.Options{}) + refire := asyncQCComplete(tt.err, time.Millisecond, m) + if refire != tt.wantRefire { + t.Errorf("refire = %v, want %v", refire, tt.wantRefire) + } + result := m.Finish() + if got := result.GetTotalOps(); got != tt.wantTotalOps { + t.Errorf("TotalOps = %d, want %d", got, tt.wantTotalOps) + } + if got := result.GetFailedOps(); got != tt.wantFailedOps { + t.Errorf("FailedOps = %d, want %d", got, tt.wantFailedOps) + } + }) + } +} + +// TestRunAsyncQCBenchmarkRejectsRateRamp verifies that runAsyncQCBenchmark +// rejects rate-ramp options instead of silently ignoring them: sends fire +// from completion callbacks gated by a shared RatedGate, not the +// runMeasure/runSchedule path that implements ramping for ClientMeasured and +// ServerMeasured. config is nil because the rejection happens before it is +// touched. +func TestRunAsyncQCBenchmarkRejectsRateRamp(t *testing.T) { + opts := benchkit.Options{Workers: 1, Duration: time.Second, RateStep: 10, RateStepMax: 100} + _, err := runAsyncQCBenchmark(opts, nil, func(*ConfigContext, *Echo, int) AsyncEcho { + t.Fatal("asyncQCFunc invoked despite rejected rate-ramp options") + return nil + }) + if !errors.Is(err, ErrAsyncQCRampUnsupported) { + t.Fatalf("err = %v, want %v", err, ErrAsyncQCRampUnsupported) + } +} + +// registerFailingQuorumCallPeer registers a QuorumCall handler on srv that +// always fails, mirroring a peer whose quorum calls are erroring (as opposed +// to registerReplyDroppingPeer's silent drop, which instead times out). Call +// before serving srv, since it registers a service. +func registerFailingQuorumCallPeer(srv *gorums.Server, errMsg string) { + srv.RegisterHandler("benchmark.Benchmark.QuorumCall", func(_ gorums.ServerContext, _ *gorums.Message) (*gorums.Message, error) { + return nil, errors.New(errMsg) + }) +} + +// TestQuorumCallHonorsCallTimeout verifies that the coordinator QuorumCall +// benchmark bounds each call to opts.CallTimeout instead of hanging behind an +// unresponsive peer until the run's own deadline elapses. +func TestQuorumCallHonorsCallTimeout(t *testing.T) { + servers := localServers(t, 1, nil) + registerReplyDroppingPeer(servers[0], 1<<30) // drops every reply for the test's duration + go func() { _ = servers[0].ListenAndServe() }() + + cfg, err := gorums.NewConfig(gorums.WithNodeList([]string{servers[0].Addr()}), gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("NewConfig: %v", err) + } + t.Cleanup(gorumstest.Closer(t, cfg)) + + run := benchDescs[0].build(cfg, nil) // "QuorumCall" + opts := benchkit.Options{ + Workers: 1, Duration: 100 * time.Millisecond, QuorumSize: 1, + CallTimeout: 20 * time.Millisecond, + } + start := time.Now() + result, err := run(opts) + elapsed := time.Since(start) + if err != nil { + t.Fatalf("QuorumCall benchmark: %v", err) + } + const wantBound = 5 * time.Second // generous vs. what an ignored CallTimeout would look like + if elapsed > wantBound { + t.Errorf("took %v, want well under %v (CallTimeout should bound each call against the unresponsive peer)", elapsed, wantBound) + } + if result.GetFailedOps() == 0 { + t.Error("FailedOps = 0, want > 0 (every call should time out against the unresponsive peer)") + } +} + +// TestRunAsyncQCBenchmarkCountsErrorsWithoutAborting verifies that a failing +// quorum call does not abort runAsyncQCBenchmark: a failed call is counted +// and the run continues, matching [benchkit.ClientMeasured]'s +// [benchkit.MeasureLatency] contract. +func TestRunAsyncQCBenchmarkCountsErrorsWithoutAborting(t *testing.T) { + servers := localServers(t, 1, nil) + registerFailingQuorumCallPeer(servers[0], "quorum call failed") + go func() { _ = servers[0].ListenAndServe() }() + + cfg, err := gorums.NewConfig(gorums.WithNodeList([]string{servers[0].Addr()}), gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("NewConfig: %v", err) + } + t.Cleanup(gorumstest.Closer(t, cfg)) + + opts := benchkit.Options{Workers: 1, Duration: 30 * time.Millisecond, MaxAsync: 10, QuorumSize: 1} + result, err := runAsyncQCBenchmark(opts, cfg, + func(ctx *ConfigContext, in *Echo, quorumSize int) AsyncEcho { + return QuorumCall(ctx, in).AsyncThreshold(quorumSize) + }) + if err != nil { + t.Fatalf("runAsyncQCBenchmark aborted on call error: %v", err) + } + if result.GetFailedOps() == 0 { + t.Error("FailedOps = 0, want > 0 (failed calls must be counted, not abort the run)") + } +} + +func TestRunSymmetricMulticastDrainsServerMessages(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + result, err := runSymmetricMulticast(target, benchkit.Options{ + Workers: 1, + Duration: 20 * time.Millisecond, + }) + if err != nil { + t.Fatalf("runSymmetricMulticast: %v", err) + } + + wantSamples := int(result.GetTotalOps()) * target.numPeers + if got := len(result.GetLatencies()); got != wantSamples { + t.Fatalf("len(Latencies) = %d, want %d", got, wantSamples) + } +} + +// TestRunSymmetricMulticastHDR verifies that a symmetric server-measured run +// honors StatsMode_HDR end to end: the per-sender stores, offset correction, +// and cross-server aggregation carry a bounded histogram (Result.Histogram) +// instead of raw samples (Result.Latencies nil). +func TestRunSymmetricMulticastHDR(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + result, err := runSymmetricMulticast(target, benchkit.Options{ + Workers: 1, + Duration: 20 * time.Millisecond, + StatsMode: benchkit.StatsMode_HDR, + }) + if err != nil { + t.Fatalf("runSymmetricMulticast: %v", err) + } + if got := result.GetLatencies(); got != nil { + t.Errorf("Latencies in HDR mode = %v, want nil", got) + } + h := result.GetHistogram() + if h == nil { + t.Fatal("Histogram in HDR mode = nil, want non-nil") + } + var total uint64 + for _, c := range h.GetCount() { + total += c + } + if wantSamples := result.GetTotalOps() * uint64(target.numPeers); total != wantSamples { + t.Errorf("histogram counts sum = %d, want %d", total, wantSamples) + } +} + +// TestAsyncSendsPipelinesAndDrains verifies the two properties AsyncMulticast +// relies on: dispatch keeps at most depth sends outstanding without waiting for +// them, and drain collects the ones the send window left behind so they are not +// lost from the run. +func TestAsyncSendsPipelinesAndDrains(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + cc := target.servers[0].PeerConfig().Context(ctx) + + const depth = 4 + outstanding := newAsyncSends(depth) + // The first depth dispatches must not block on completion; the ones after + // reap an earlier send to make room. + for range depth * 3 { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano()}.Build() + if err := outstanding.dispatch(func() asyncSend { return Multicast(cc, msg).Async() }); err != nil { + t.Fatalf("dispatch: %v", err) + } + } + if got := len(outstanding.handles); got != depth { + t.Errorf("outstanding sends = %d, want %d", got, depth) + } + if err := outstanding.drain(); err != nil { + t.Errorf("drain: %v", err) + } + if got := len(outstanding.handles); got != 0 { + t.Errorf("outstanding sends after drain = %d, want 0", got) + } +} + +type blockingAsyncSend struct { + done <-chan struct{} + active *atomic.Int32 +} + +func (s *blockingAsyncSend) Wait() error { + <-s.done + s.active.Add(-1) + return nil +} + +// TestAsyncSendsReservesCapacityBeforeDispatch verifies that -max-async is a +// bound on sends actually dispatched, not merely on handles retained after +// dispatch. The third closure must not run until one of the first two handles +// has completed. +func TestAsyncSendsReservesCapacityBeforeDispatch(t *testing.T) { + const depth = 2 + outstanding := newAsyncSends(depth) + done := make(chan struct{}, 3) + var active atomic.Int32 + newSend := func() asyncSend { + active.Add(1) + return &blockingAsyncSend{done: done, active: &active} + } + + for range depth { + if err := outstanding.dispatch(newSend); err != nil { + t.Fatalf("dispatch: %v", err) + } + } + if got := active.Load(); got != depth { + t.Fatalf("active sends = %d, want %d", got, depth) + } + + dispatchStarted := make(chan struct{}) + thirdDispatched := make(chan struct{}) + errCh := make(chan error, 1) + go func() { + close(dispatchStarted) + errCh <- outstanding.dispatch(func() asyncSend { + h := newSend() + close(thirdDispatched) + return h + }) + }() + <-dispatchStarted + + select { + case <-thirdDispatched: + t.Fatal("third send dispatched before outstanding capacity was released") + case <-time.After(20 * time.Millisecond): + } + + done <- struct{}{} + select { + case <-thirdDispatched: + case <-time.After(time.Second): + t.Fatal("third send did not dispatch after outstanding capacity was released") + } + if err := <-errCh; err != nil { + t.Fatalf("third dispatch: %v", err) + } + if got := active.Load(); got != depth { + t.Errorf("active sends after third dispatch = %d, want %d", got, depth) + } + + done <- struct{}{} + done <- struct{}{} + if err := outstanding.drain(); err != nil { + t.Fatalf("drain: %v", err) + } +} + +// TestAsyncMulticastBenchmarkRuns verifies that the AsyncMulticast benchmark +// completes a server-measured run and records operations, exercising the +// dispatch and quiesce-drain wiring together. +func TestAsyncMulticastBenchmarkRuns(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + benches := GetBenchmarks(BenchTarget{Config: target.servers[0].PeerConfig()}) + idx := slices.IndexFunc(benches, func(b benchkit.Bench) bool { return b.Name == "AsyncMulticast" }) + if idx < 0 { + t.Fatal("AsyncMulticast benchmark not registered") + } + result, err := benches[idx].Run(benchkit.Options{ + Workers: 2, MaxAsync: 8, Duration: 50 * time.Millisecond, + }) + if err != nil { + t.Fatalf("run: %v", err) + } + if result.GetTotalOps() == 0 { + t.Error("TotalOps = 0, want > 0") + } +} + +// TestServerMeasuredMulticastHDR verifies that the coordinator server-measured +// Multicast lifecycle honors StatsMode_HDR end to end: the mode reaches the +// server over the Start RPC, so Stop returns a histogram, clock-offset +// correction shifts the histogram, and the aggregate carries it (Latencies nil). +func TestServerMeasuredMulticastHDR(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + run := benchkit.ServerMeasured(target.servers[0].PeerConfig(), + func(_ benchkit.Options, cc *ConfigContext) func() error { + return func() error { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano()}.Build() + return Multicast(cc, msg).Send() + } + }) + + result, err := run(benchkit.Options{Workers: 1, Duration: 20 * time.Millisecond, StatsMode: benchkit.StatsMode_HDR}) + if err != nil { + t.Fatalf("run: %v", err) + } + if result.GetTotalOps() == 0 { + t.Fatal("TotalOps = 0, want > 0") + } + if got := result.GetLatencies(); got != nil { + t.Errorf("Latencies in HDR mode = %v, want nil", got) + } + if result.GetHistogram() == nil { + t.Error("Histogram in HDR mode = nil, want non-nil") + } +} + +// TestServerMeasuredQuiesce verifies that benchkit.ServerMeasured invokes the +// WithQuiesce drain hook after the send window and before Control.Stop +// collects the server-side statistics. +func TestServerMeasuredQuiesce(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + quiesceCalls := 0 + run := benchkit.ServerMeasured(target.servers[0].PeerConfig(), + func(_ benchkit.Options, cc *ConfigContext) func() error { + return func() error { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano()}.Build() + return Multicast(cc, msg).Send() + } + }, + benchkit.WithQuiesce(func(context.Context) error { + quiesceCalls++ + return nil + })) + + result, err := run(benchkit.Options{Workers: 1, Duration: 20 * time.Millisecond}) + if err != nil { + t.Fatalf("run: %v", err) + } + if quiesceCalls != 1 { + t.Errorf("quiesce calls = %d, want 1", quiesceCalls) + } + if result.GetTotalOps() == 0 { + t.Error("TotalOps = 0, want > 0") + } +} + +// TestServerMeasuredVerify verifies that benchkit.WithVerify receives the +// per-server Stop replies of a server-measured run and that a verify error +// fails the run before aggregation. +func TestServerMeasuredVerify(t *testing.T) { + target, stop, err := SetupSymmetricServers(3, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + setup := func(_ benchkit.Options, cc *ConfigContext) func() error { + return func() error { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano()}.Build() + return Multicast(cc, msg).Send() + } + } + + var gotNodes int + run := benchkit.ServerMeasured(target.servers[0].PeerConfig(), setup, + benchkit.WithVerify(func(replies map[uint32]*benchkit.Result) error { + gotNodes = len(replies) + return nil + })) + if _, err := run(benchkit.Options{Workers: 1, Duration: 20 * time.Millisecond}); err != nil { + t.Fatalf("run: %v", err) + } + if gotNodes != 3 { + t.Errorf("verify saw %d replies, want 3", gotNodes) + } + + errVerify := errors.New("per-server ops diverged") + failing := benchkit.ServerMeasured(target.servers[0].PeerConfig(), setup, + benchkit.WithVerify(func(map[uint32]*benchkit.Result) error { return errVerify })) + if _, err := failing(benchkit.Options{Workers: 1, Duration: 20 * time.Millisecond}); !errors.Is(err, errVerify) { + t.Errorf("run with failing verify = %v, want %v", err, errVerify) + } +} + +// TestServerMeasuredWindowExcludesClockSync verifies that the server-measured +// throughput window (Control.Start to Control.Stop) brackets only the send +// window, not the two clock-offset estimation phases that run around it. Each +// phase is 50 sequential ClockSync round trips, so on a real network the +// window would otherwise report a TotalTime and Throughput skewed by +// clock-sync time; a bufconn round trip is normally too fast for that skew to +// show up in a wall-clock assertion, so a per-round delay on ClockSync alone +// (via a server interceptor) stands in for that network cost and makes the +// leak deterministically detectable. +func TestServerMeasuredWindowExcludesClockSync(t *testing.T) { + const clockSyncDelay = 2 * time.Millisecond + delayClockSync := func(ctx gorums.ServerContext, in *gorums.Message, next gorums.Handler) (*gorums.Message, error) { + if in.GetMethod() == "benchkit.Control.ClockSync" { + time.Sleep(clockSyncDelay) + } + return next(ctx, in) + } + serverOpts := []gorums.ServerOption{gorums.WithServerInterceptors(delayClockSync)} + target, stop, err := SetupSymmetricServers(3, serverOpts, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + + setup := func(_ benchkit.Options, cc *ConfigContext) func() error { + return func() error { + msg := TimedMsg_builder{SendTime: time.Now().UnixNano()}.Build() + return Multicast(cc, msg).Send() + } + } + + run := benchkit.ServerMeasured(target.servers[0].PeerConfig(), setup) + const duration = 20 * time.Millisecond + result, err := run(benchkit.Options{Workers: 1, Duration: duration, Interval: 5 * time.Millisecond}) + if err != nil { + t.Fatalf("run: %v", err) + } + // Each of the two clock-sync phases pays clockSyncRounds (50) sequential + // ClockSync round trips, so if either phase leaked into the window, + // TotalTime would grow by tens of clockSyncDelay on top of duration. A + // window correctly bounded to the send phase stays within a few + // durations' worth of slack for scheduling and the Stop RPC. + if got, want := time.Duration(result.GetTotalTime()), duration+20*clockSyncDelay; got > want { + t.Errorf("TotalTime = %v, want <= %v (clock-sync phases leaking into the measurement window?)", got, want) + } + var eventDuration time.Duration + for _, event := range result.GetEvents() { + if throughput := event.GetThroughput(); throughput != nil { + eventDuration += time.Duration(throughput.GetDuration()) + } + } + if eventDuration == 0 { + t.Fatal("throughput event duration = 0, want a measured interval") + } + if want := duration + 20*clockSyncDelay; eventDuration > want { + t.Errorf("throughput event duration = %v, want <= %v (clock-sync phases leaking into the event stream?)", eventDuration, want) + } +} diff --git a/benchkit/benchmark/server.go b/benchkit/benchmark/server.go new file mode 100644 index 00000000..c9bc5f7f --- /dev/null +++ b/benchkit/benchmark/server.go @@ -0,0 +1,66 @@ +package benchmark + +import ( + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/benchkit" +) + +// workloadServer implements the gorums Benchmark workload RPCs. Each handled +// operation and each server-measured Multicast latency is recorded into the +// shared benchkit.Control, so the control plane's Stop reply observes this +// server's work. The measurement control plane (Start/Stop/ClockSync) lives in +// benchkit.Control, registered alongside this server on one listener. +type workloadServer struct { + ctrl *benchkit.Control +} + +func (srv *workloadServer) QuorumCall(_ gorums.ServerContext, in *Echo) (*Echo, error) { + srv.ctrl.RecordOp() + return in, nil +} + +func (srv *workloadServer) SlowServer(ctx gorums.ServerContext, in *Echo) (*Echo, error) { + ctx.Release() + srv.ctrl.RecordOp() + time.Sleep(10 * time.Millisecond) + return in, nil +} + +func (srv *workloadServer) Multicast(_ gorums.ServerContext, msg *TimedMsg) { + srv.ctrl.RecordOp() + latency := time.Duration(time.Now().UnixNano() - msg.GetSendTime()) + // A symmetric sender tags the message with its node ID (>= 1) so its + // samples can be corrected by that sender's clock offset; an untagged + // message (sender ID 0) records into the flat samples instead. + if id := msg.GetSenderId(); id != 0 { + srv.ctrl.Stats().AddLatencyBySender(id, latency) + } else { + srv.ctrl.Stats().AddLatency(latency) + } +} + +// attachBenchServer registers benchkit's Control server and the gorums workload +// server on srv, sharing one Control instance, and returns the Control handle. +func attachBenchServer(srv *gorums.Server) *benchkit.Control { + ctrl := benchkit.NewControl() + w := &workloadServer{ctrl: ctrl} + ctrl.SetID(srv.NodeID()) + benchkit.RegisterControlServer(srv, ctrl) + RegisterBenchmarkServer(srv, w) + return ctrl +} + +// Server is a unified server registering both benchkit's Control plane and the +// gorums workload service on one listener. +type Server struct { + *gorums.Server + ctrl *benchkit.Control +} + +// NewBenchServer returns a new benchmark server. +func NewBenchServer(opts ...gorums.ServerOption) *Server { + srv := gorums.NewServer(opts...) + return &Server{Server: srv, ctrl: attachBenchServer(srv)} +} diff --git a/benchkit/benchmark/symmetric.go b/benchkit/benchmark/symmetric.go new file mode 100644 index 00000000..a32a52c0 --- /dev/null +++ b/benchkit/benchmark/symmetric.go @@ -0,0 +1,636 @@ +package benchmark + +import ( + "cmp" + "context" + "fmt" + "io" + "net" + "os" + "runtime/pprof" + "slices" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/benchkit" +) + +// SymmetricTarget bundles the gorums servers and their registered control +// planes for symmetric (peer-to-peer) benchmarks. +type SymmetricTarget struct { + servers []*gorums.Server + controls []*benchkit.Control + numPeers int // cluster size (including self); sizes the exit grace period + labels []string + selfAddr string // distributed mode only: this node's address in the peer list; enables the probe-stall self-diagnosis +} + +// SetupSymmetricServers creates n local Gorums servers with the benchkit Control +// plane and the gorums workload server registered and serving. Returns a +// SymmetricTarget, a stop function, and any error. +func SetupSymmetricServers(n int, serverOpts []gorums.ServerOption, dialOpts ...gorums.DialOption) (*SymmetricTarget, func(), error) { + servers, stop, err := gorums.NewLocalServers( + n, + gorums.WithLocalServerOptions(serverOpts...), + gorums.WithLocalDialOptions(dialOpts...), + ) + if err != nil { + return nil, nil, err + } + controls := make([]*benchkit.Control, n) + labels := make([]string, n) + for i, srv := range servers { + controls[i] = attachBenchServer(srv) + labels[i] = fmt.Sprintf("server %d (%s)", i+1, srv.Addr()) + } + for _, srv := range servers { + go func() { _ = srv.ListenAndServe() }() + } + return &SymmetricTarget{servers: servers, controls: controls, numPeers: n, labels: labels}, stop, nil +} + +// SetupRemoteServer creates a single Gorums server for distributed +// benchmarking. selfAddr must appear in peerAddrs; the slice is sorted to +// assign stable node IDs (1..N) across all machines. +// +// Exit synchronization is not handled here: after a node finishes its own +// benchmark it must keep its listener open for a short grace period (see +// ExitGrace) before exiting, so that slower peers can complete their final +// cross-node RPCs without hitting a closed listener. The caller is responsible +// for that linger; see cmd/benchmark. +func SetupRemoteServer(selfAddr string, peerAddrs []string, serverOpts []gorums.ServerOption, dialOpts ...gorums.DialOption) (*SymmetricTarget, func(), error) { + sorted := slices.Clone(peerAddrs) + slices.Sort(sorted) + idx := slices.Index(sorted, selfAddr) + if idx < 0 { + return nil, nil, fmt.Errorf("self address %q not found in peer list", selfAddr) + } + myID := uint32(idx + 1) + peerList := gorums.WithNodeList(sorted) + + // Listen on the wildcard address rather than selfAddr: listening on a + // hostname binds the single IP the local resolver returns for it, and + // hosts following the Debian convention resolve their own name to the + // loopback address 127.0.1.1, leaving the listener unreachable for all + // peers (see doc/benchkit-troubleshooting.html). selfAddr is still used + // for node identity and ID assignment above. + _, port, err := net.SplitHostPort(selfAddr) + if err != nil { + return nil, nil, fmt.Errorf("invalid self address %q: %w", selfAddr, err) + } + srv := gorums.NewServer(append([]gorums.ServerOption{ + gorums.WithAddr(net.JoinHostPort("", port)), + gorums.WithPeers(myID, peerList, dialOpts...), + }, serverOpts...)...) + + ctrl := attachBenchServer(srv) + ctrl.ArmDone(len(peerAddrs)) // same count used for numPeers below + label := fmt.Sprintf("node %d (%s)", myID, selfAddr) + + go func() { _ = srv.ListenAndServe() }() + benchkit.Logf("[%s] listener bound to %s\n", label, srv.Addr()) + t := &SymmetricTarget{ + servers: []*gorums.Server{srv}, + controls: []*benchkit.Control{ctrl}, + numPeers: len(peerAddrs), + labels: []string{label}, + selfAddr: selfAddr, + } + return t, func() { + srv.Stop() + benchkit.Logf("[%s %s] server stopped\n", time.Now().Format(time.TimeOnly), label) + }, nil +} + +// ExitGrace returns the fallback ceiling a distributed-mode replica waits, +// after finishing its own benchmark, for its peers to also finish before it +// gives up and exits anyway. +// +// The symmetric topology has no exit barrier: a node multicasts an advisory +// Done signal (see [SignalDone]) and races it against this ceiling (see +// [AwaitPeersDoneOrGrace]), exiting as soon as every peer has signaled or this +// timeout elapses, whichever comes first. Because most runs finish within +// milliseconds of each other and exit via the Done signal long before the +// ceiling is reached, this is a worst-case bound, not the expected wait — it +// only matters when Done itself doesn't arrive (e.g. a peer's inbound +// channel is broken, the same class of issue that made the previous Done +// barrier unreliable) or the whole cluster is unusually slow. The bound is +// the inter-node completion skew, which is dominated by mesh-formation/ +// release skew in AwaitReady and therefore grows with cluster size; the +// trailing round-trips are sub-second and absorbed by the base term. The +// result is clamped so a very large cluster does not linger excessively. +func ExitGrace(numNodes int) time.Duration { + const ( + base = 3 * time.Second + perNode = 300 * time.Millisecond + maxGrace = 20 * time.Second + ) + return min(base+time.Duration(numNodes)*perNode, maxGrace) +} + +// SignalDone notifies every outbound peer that this node has finished its +// own benchmark work and trailing flush and will issue no further calls. The +// send is best-effort: per-peer errors are discarded, since a dropped or failed +// notification only costs the receiving peer its fast exit via +// AwaitPeersDoneOrGrace, never correctness, because that wait always falls +// back to its own grace deadline. +// +// The whole notification is bounded by a deadline. Done is a one-way Multicast +// whose enqueue waits for send-queue space until the request context is done +// (the one-way path in internal/stream.(*Channel).Enqueue). Callers pass a +// deadline-free context (cmd/benchmark passes context.Background), so without a +// bound a full or backpressured send queue during the teardown broadcast traps +// this call — and hence the caller — forever; this deadlocked node teardown on +// the cluster. The bound mirrors the peers' own wait (ExitGrace over the +// cluster size): never spend longer announcing Done than a peer will wait to +// hear it. Wait blocks until each send completes or the deadline fires, +// keeping the deferred cancel from aborting a send still in flight. +func SignalDone(ctx context.Context, t *SymmetricTarget) { + ctx, cancel := context.WithTimeout(ctx, ExitGrace(t.numPeers)) + defer cancel() + for i, srv := range t.servers { + out := srv.PeerConfig() + if out.Size() == 0 { + continue + } + req := benchkit.DoneRequest_builder{SenderId: t.controls[i].SelfID()}.Build() + // Report rather than return: Done is advisory, and a peer that already + // finished and exited makes a failed send the expected outcome, not a + // fault. Logging it still distinguishes that from a run where every + // send failed and no peer was ever told. + if err := benchkit.Done(out.Context(ctx), req).Send(); err != nil { + benchkit.Logf("Done signal from node %d: %v\n", t.controls[i].SelfID(), err) + } + } +} + +// AwaitPeersDoneOrGrace blocks until every peer has signaled Done or grace +// elapses, whichever is first, using one shared deadline across all of t's +// local servers. It returns true only if every server's peers all signaled +// Done before the deadline; false means at least one server fell back to +// the grace timeout. It never errors: the timeout is a safe, expected +// fallback, not a failure. +func AwaitPeersDoneOrGrace(ctx context.Context, t *SymmetricTarget, grace time.Duration) bool { + ctx, cancel := context.WithTimeout(ctx, grace) + defer cancel() + for _, ctrl := range t.controls { + ch := ctrl.DoneCh() + if ch == nil { + continue + } + select { + case <-ch: + case <-ctx.Done(): + return false + } + } + return true +} + +// MissingDoneSenders returns the peer node IDs across all of t's local +// servers that have not yet signaled Done, for diagnostics when +// AwaitPeersDoneOrGrace falls back to its timeout. +func MissingDoneSenders(t *SymmetricTarget) []uint32 { + var missing []uint32 + for _, ctrl := range t.controls { + missing = append(missing, ctrl.MissingDone()...) + } + return missing +} + +// runComplete reports whether so many of outSize outbound peers have finished +// (done) that a phase needing `need` still-live peers can no longer succeed. It +// is the pure decision behind quorumRunOver: false until at least one peer has +// signaled Done, so a failure before any peer finishes (a genuine startup or +// mid-run fault) is never mistaken for the expected end of the run. +func runComplete(outSize, done, need int) bool { + return done > 0 && outSize-done < need +} + +// quorumRunOver reports whether enough outbound peers have signaled Done that a +// quorum call needing quorumSize replies can no longer form a quorum from the +// live peers. When true, an incomplete/connection-refused quorum call is the +// expected end of the run (peers finished and closed their listeners), not a +// fault, so a straggler treats it as a clean stop rather than a failure. +func quorumRunOver(t *SymmetricTarget, quorumSize int) bool { + for i, srv := range t.servers { + if runComplete(srv.PeerConfig().Size(), t.controls[i].DoneCount(), quorumSize) { + return true + } + } + return false +} + +// anyPeerFinished reports whether any outbound peer has signaled Done. The +// all-peers phases (offset estimation and the trailing flush) require every +// peer, a requirement no live-peer set can meet once a peer exits, so a single +// Done marks their failure as the expected end of the run. It is false until a +// peer finishes, so startup failures before anyone is Done are not masked. +func anyPeerFinished(t *SymmetricTarget) bool { + for i := range t.servers { + if t.controls[i].DoneCount() > 0 { + return true + } + } + return false +} + +// readyStallTimeout bounds how long the outbound probe in AwaitReady may go +// without any peer responding. A peer that died during startup (e.g. because +// its listen port was taken) never responds, so waiting out the rest of the +// readiness deadline only delays the inevitable failure; each response resets +// the timer, so a large mesh that is still forming keeps the probe alive. It +// is a variable so tests can shorten it. +var readyStallTimeout = 20 * time.Second + +// probeAttemptTimeout bounds one per-peer echo attempt in the outbound probe +// of awaitReady. A reply that is silently lost (e.g. to stream churn during +// setup) costs one attempt round rather than the caller's whole +// readiness deadline; the next round re-sends a fresh echo over whatever +// stream is then current. It is a variable so tests can shorten it. +var probeAttemptTimeout = 2 * time.Second + +// probeLogf emits outbound-probe progress, notably the per-round straggler +// list. It is a variable so tests can capture and assert on the straggler +// log; production code logs via benchkit.Logf. +var probeLogf = benchkit.Logf + +// AwaitReady waits until every server in t is ready to run the benchmark: it +// validates round-trip connectivity to every outbound peer of every server, +// sending each peer echoes until it responds, so every outbound connection is +// established before the benchmark starts (gRPC dials lazily and may still be +// in backoff after setup). See probeOutbound for the per-peer retry/stall +// semantics. In distributed mode a probe failure additionally writes a +// network self-diagnosis (see diagnoseProbeStall). +// +// In dedup mode, call this after [gorums.Server.WaitForAll] so every shared +// stream is live before the probe exercises it. +func AwaitReady(ctx context.Context, t *SymmetricTarget) error { + for i, srv := range t.servers { + label := t.label(i) + if err := probeOutbound(ctx, srv, label); err != nil { + if t.selfAddr != "" { + diagnoseProbeStall(diagWriter, srv, t.selfAddr) + } + return fmt.Errorf("%s: outbound peers not ready: %w", label, err) + } + benchkit.Logf("[ready] %s: outbound ready\n", label) + } + return nil +} + +// probeOutbound validates round-trip connectivity to every outbound peer of +// srv before the benchmark starts. Each peer that has not yet responded is +// probed in parallel with a single-node echo call under its own +// probeAttemptTimeout deadline, so a silently lost reply costs one round, not +// the caller's whole readiness deadline (the all-or-nothing Threshold(N) +// probe it replaces blocked on one missing reply with "incomplete call +// (errors: 0)" until ctx expired). Responders leave the pending set; the +// remaining stragglers are logged by node ID each round. The probe fails +// early, naming the pending peers, when no peer has responded for +// readyStallTimeout: with nobody making progress, waiting out the rest of the +// deadline only delays the inevitable failure. +func probeOutbound(ctx context.Context, srv *gorums.Server, label string) error { + pending := srv.PeerConfig() + msg := Echo_builder{}.Build() + probeLogf("[ready] %s: probing %d outbound connections...\n", label, pending.Size()) + lastResponse := time.Now() + for pending.Size() > 0 { + attemptCtx, cancel := context.WithTimeout(ctx, probeAttemptTimeout) + var mu sync.Mutex + var responded []uint32 + var wg sync.WaitGroup + for _, node := range pending { + wg.Go(func() { + if _, err := QuorumCall(gorums.Config{node}.Context(attemptCtx), msg).Threshold(1); err == nil { + mu.Lock() + responded = append(responded, node.ID()) + mu.Unlock() + } + }) + } + wg.Wait() + cancel() + if len(responded) > 0 { + lastResponse = time.Now() + } + pending = pending.Remove(responded...) + if pending.Size() == 0 { + break + } + probeLogf("[ready] %s: outbound probe: %d peers pending: %s\n", label, pending.Size(), pendingPeerDetails(pending)) + if time.Since(lastResponse) >= readyStallTimeout { + return fmt.Errorf("no outbound peer responded for %v; pending: %s", readyStallTimeout, pendingPeerDetails(pending)) + } + select { + case <-ctx.Done(): + return fmt.Errorf("pending: %s: %w", pendingPeerDetails(pending), ctx.Err()) + case <-time.After(200 * time.Millisecond): + } + } + return nil +} + +// nodeDetail formats node as "node ID (address)", appending its last recorded +// error when one exists, so peer-naming diagnostics identify both the peer +// and its known cause without cross-referencing logs. +func nodeDetail(node *gorums.Node) string { + detail := fmt.Sprintf("node %d (%s)", node.ID(), node.Address()) + if err := node.LastErr(); err != nil { + detail += fmt.Sprintf(": %v", err) + } + return detail +} + +// pendingPeerDetails names each peer still pending in the outbound probe. +func pendingPeerDetails(pending gorums.Config) string { + details := make([]string, 0, len(pending)) + for _, node := range pending { + details = append(details, nodeDetail(node)) + } + return strings.Join(details, ", ") +} + +// diagWriter receives the probe-stall self-diagnosis. It is a variable so +// tests can capture and assert on the diagnosis instead of spamming stderr. +var diagWriter io.Writer = os.Stderr + +// diagnoseProbeStall writes a network self-diagnosis to w when the outbound +// readiness probe fails in distributed mode. At that point this process is +// still alive on the affected host, which makes it ideally placed to +// discriminate the known failure classes (see doc/benchkit-troubleshooting.html): +// +// - self-dial fails and the bound address differs from the advertised +// address: the listener is bound to the wrong interface because the +// host resolved its own name unexpectedly (e.g. to a loopback address). +// - self-dial fails with matching addresses: a local firewall is refusing +// new connections, or the listener died. +// - self-dial succeeds: the listener works from this host, so the blockage +// is between the peers and this host (network filtering, peer-side state). +// +// The goroutine dump at the end shows whether the probe's own calls stalled +// inside gorums rather than on the network, which would indicate a gorums +// bug rather than a host or network condition. +func diagnoseProbeStall(w io.Writer, srv *gorums.Server, selfAddr string) { + fmt.Fprintf(w, "[diag] probe stall: listener bound to %s; self peer-list address %s\n", srv.Addr(), selfAddr) + tcpAddr, err := net.ResolveTCPAddr("tcp", selfAddr) + if err != nil { + fmt.Fprintf(w, "[diag] cannot resolve self address %s: %v\n", selfAddr, err) + return + } + advertised := tcpAddr.String() + if tcpAddr.IP.IsLoopback() { + fmt.Fprintf(w, "[diag] note: %s resolves to a loopback address on this host; remote peers dialing the host's real IP cannot reach a loopback-bound listener\n", selfAddr) + } + conn, err := net.DialTimeout("tcp", advertised, 2*time.Second) + if err != nil { + fmt.Fprintf(w, "[diag] self-dial %s failed: %v (listener unreachable on advertised address: wrong bind or local firewall)\n", advertised, err) + } else { + _ = conn.Close() + fmt.Fprintf(w, "[diag] self-dial %s ok (listener reachable from this host; inbound blockage is between peers and this host)\n", advertised) + } + fmt.Fprintf(w, "[diag] goroutine dump:\n") + _ = pprof.Lookup("goroutine").WriteTo(w, 2) +} + +func (t *SymmetricTarget) label(i int) string { + if i >= 0 && i < len(t.labels) && t.labels[i] != "" { + return t.labels[i] + } + return fmt.Sprintf("server %d", i+1) +} + +// estimateAllOffsets runs benchkit.EstimateOffsets from every server in t to its +// outbound peers, returning one offset map per server, indexed to match +// t.servers (and t.controls). +func estimateAllOffsets(ctx context.Context, t *SymmetricTarget) ([]map[uint32]int64, error) { + offsets := make([]map[uint32]int64, len(t.servers)) + for i, srv := range t.servers { + off, err := benchkit.EstimateOffsets(ctx, srv.PeerConfig()) + if err != nil { + return nil, fmt.Errorf("%s: %w", t.label(i), err) + } + offsets[i] = off + } + return offsets, nil +} + +func flushSymmetricOutbound(ctx context.Context, t *SymmetricTarget) error { + for i, srv := range t.servers { + if err := flushOutbound(ctx, srv); err != nil { + return fmt.Errorf("%s: %w", t.label(i), err) + } + } + return nil +} + +func flushOutbound(ctx context.Context, srv *gorums.Server) error { + out := srv.PeerConfig() + if out == nil { + return fmt.Errorf("missing outbound configuration") + } + n := out.Size() + if n == 0 { + return gorums.ErrIncomplete + } + _, err := QuorumCall(out.Context(ctx), Echo_builder{}.Build()).Threshold(n) + if err != nil { + // A flush failure means a peer stopped responding mid-run (e.g. it + // exited before the grace period elapsed). Name the unresponsive peers + // so the cause is identifiable without cross-referencing logs. + if unresponsive := unresponsiveOutbound(srv); unresponsive != "" { + return fmt.Errorf("%w; unresponsive peers: %s", err, unresponsive) + } + } + return err +} + +// unresponsiveOutbound names the outbound peers with a recorded dial/call error, +// formatted as "node ID (address): error". It returns "" when every peer is +// healthy, so callers can include it only when there is something to report. +func unresponsiveOutbound(srv *gorums.Server) string { + out := srv.PeerConfig() + if out == nil { + return "" + } + details := make([]string, 0, out.Size()) + for _, node := range out { + if err := node.LastErr(); err != nil { + details = append(details, nodeDetail(node)) + } + } + return strings.Join(details, ", ") +} + +// runSymmetricQuorumCall benchmarks quorum call round-trip latency and +// throughput in a symmetric (peer-to-peer) topology. Each server issues +// QuorumCall RPCs to its outbound peers; the client side measures latency. +// quorumSize defaults to a majority of the outbound peer count. +func runSymmetricQuorumCall(t *SymmetricTarget, opts benchkit.Options) (*benchkit.Result, error) { + ctx, cancel := benchkit.BenchContext(opts) + defer cancel() + + // Yield each server's outbound configuration so MeasureLatency drives one + // concurrent send target per server in a single measurement phase. + configs := func(yield func(gorums.Config) bool) { + for _, srv := range t.servers { + if !yield(srv.PeerConfig()) { + return + } + } + } + setup := func(opts benchkit.Options, cc *gorums.ConfigContext) func() error { + msg := Echo_builder{Payload: make([]byte, opts.Payload)}.Build() + // Honor the configured quorum size, matching the coordinator QuorumCall + // benchmark; fall back to a majority of the outbound peers when unset. + quorumSize := cmp.Or(opts.QuorumSize, cc.Config().Size()/2+1) + call := func(cc *gorums.ConfigContext) error { + _, err := QuorumCall(cc, msg).Threshold(quorumSize) + if err != nil && quorumRunOver(t, quorumSize) { + // Peers finished and closed their listeners; the incomplete + // quorum is the expected end of the run, not a fault. Cancel the + // window so every worker stops promptly, and return ErrRunOver + // so this op is recorded neither as a success nor as a failure + // and MeasureLatency returns the samples gathered so far + // instead of failing. + cancel() + return benchkit.ErrRunOver + } + return err + } + if opts.CallTimeout <= 0 { + return func() error { return call(cc) } + } + // With -call-timeout, each call carries its own deadline so a call + // stalled behind an unresponsive peer fails with DeadlineExceeded + // instead of hanging until run end. The branch is taken here at setup + // so the default path stays free of per-op timers. + cfg := cc.Config() + return func() error { + callCtx, cancelCall := context.WithTimeout(cc, opts.CallTimeout) + defer cancelCall() + return call(cfg.Context(callCtx)) + } + } + return benchkit.MeasureLatency(ctx, opts, configs, setup) +} + +// runSymmetricMulticast benchmarks multicast throughput and server-side +// one-way latency in a symmetric topology. Each server sends Multicast RPCs +// to its outbound peers; the server side measures latency via TimedMsg.SendTime. +// Throughput is reported as total sends per second across all servers. +func runSymmetricMulticast(t *SymmetricTarget, opts benchkit.Options) (*benchkit.Result, error) { + ctx, cancel := benchkit.BenchContext(opts) + defer cancel() + + payload := make([]byte, opts.Payload) + // Each sender tags the message with its own node ID so the receiving server + // can bucket samples per sender and correct them by that sender's estimated + // clock offset (see [benchkit.EstimateOffsets] and + // [benchkit.Stats.GetResultCorrected]). + newMsg := func(senderID uint32) *TimedMsg { + return TimedMsg_builder{SendTime: time.Now().UnixNano(), SenderId: senderID, Payload: payload}.Build() + } + + // resetServerStats clears and restarts every local server's counters so the + // next phase measures from zero; used before the connection flush and again + // at the start of the measurement window. The stats mode carries opts.StatsMode + // so a symmetric server-measured run honors -stats-mode like the coordinator + // path does via the Start RPC. + resetServerStats := func() { + for _, ctrl := range t.controls { + ctrl.Reset(opts.StatsMode) + } + } + + resetServerStats() + if err := flushSymmetricOutbound(ctx, t); err != nil && !anyPeerFinished(t) { + return nil, err + } + + var totalSent atomic.Int64 + var elapsed time.Duration + + // One send target per server; each tags its message with its own node ID so + // the receiving server can bucket and clock-correct samples per sender. + sends := make([]func() error, len(t.servers)) + for i, srv := range t.servers { + cfgCtx := srv.PeerConfig().Context(ctx) + senderID := t.controls[i].SelfID() + sends[i] = func() error { + if err := Multicast(cfgCtx, newMsg(senderID)).Send(); err != nil { + return err + } + totalSent.Add(1) + return nil + } + } + + // Latency is measured server-side; the client-side Measurement carries only + // the op count (via Stats.AddOp) so the ticker can emit the throughput + // time-series, and honors opts.Interval / opts.StatsMode like the other + // runners. Going through MeasureOneWay also gives the symmetric multicast the + // rate-ramp and ticker rate-step support the coordinator runners already have. + m, window := benchkit.MeasureOneWay(ctx, opts, sends...) + + estimate := func() ([]map[uint32]int64, error) { + off, err := estimateAllOffsets(ctx, t) + if err != nil && anyPeerFinished(t) { + // Peers finished and exited before this straggler could sync clocks; + // the incomplete offset set is the expected end of the run, not a + // fault. estimateAllOffsets returns nil on error, so hand back one + // empty offset map per server to keep buildReplies' per-server + // indexing in bounds; AverageOffsets tolerates the empty side, so + // the samples are left uncorrected rather than failing the run. + return make([]map[uint32]int64, len(t.servers)), nil + } + return off, err + } + measure := func() error { + // Reset server stats so the measurement phase counts from zero. + resetServerStats() + startTime := time.Now() + if err := window(); err != nil && !anyPeerFinished(t) { + return err + } + elapsed = time.Since(startTime) + if err := flushSymmetricOutbound(ctx, t); err != nil && !anyPeerFinished(t) { + return err + } + return nil + } + buildReplies := func(before, after []map[uint32]int64) (map[uint32]*benchkit.Result, error) { + for _, ctrl := range t.controls { + ctrl.Stats().End() + } + // Aggregate server-side latency samples across all local servers, + // correcting each sample by its sender's averaged clock offset. + replies := make(map[uint32]*benchkit.Result, len(t.controls)) + for i, ctrl := range t.controls { + benchkit.LogOffsets(t.label(i), before[i], after[i]) + offsets := benchkit.AverageOffsets(before[i], after[i]) + replies[uint32(i+1)] = ctrl.Stats().GetResultCorrected(offsets) + } + return replies, nil + } + r, err := benchkit.RunOffsetCorrected(estimate, measure, buildReplies) + if err != nil { + m.Abandon() + return nil, err + } + m.Attach(r) + + // Override TotalOps and Throughput with client-side send counts, which + // are the authoritative measure of work done. Server receives are N times + // sends in this topology, so server TotalOps would overcount. + n := uint64(totalSent.Load()) + r.SetTotalOps(n) + r.SetTotalTime(int64(elapsed)) + if n > 0 { + r.SetThroughput(float64(n) / elapsed.Seconds()) + } + return r, nil +} diff --git a/benchkit/benchmark/target.go b/benchkit/benchmark/target.go new file mode 100644 index 00000000..abbd7946 --- /dev/null +++ b/benchkit/benchmark/target.go @@ -0,0 +1,145 @@ +package benchmark + +import ( + "context" + "fmt" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/benchkit" +) + +// SetupTarget builds the BenchTarget for one of the three run modes and fills +// in the topology-derived Options fields (Remote, NumNodes): +// +// - distributed: self is this node's listen address and remotes lists all +// peers (including self); every node runs the same binary. +// - local: neither self nor remotes is set; configSize in-process servers are +// created. +// - coordinator: remotes lists the servers and self is unset; this process +// coordinates against them, using configSize nodes when 1 <= configSize <= +// len(remotes) and all of them otherwise. +// +// It returns the target and a cleanup function the caller must invoke (e.g. +// defer) once the benchmarks finish. In distributed mode the caller must also +// linger for ExitGrace before invoking cleanup, so slower peers can finish +// their trailing cross-node RPCs before this node closes its listener. +func SetupTarget(opts *benchkit.Options, self string, remotes []string, configSize int, dialOpts ...gorums.DialOption) (BenchTarget, func(), error) { + switch { + case self != "": + return setupDistributed(opts, self, remotes, dialOpts) + case len(remotes) < 1: + return setupLocal(opts, configSize, dialOpts) + default: + return setupCoordinator(opts, remotes, configSize, dialOpts) + } +} + +// setupDistributed builds the symmetric peer-to-peer target for one node of a +// distributed run and waits until every peer is reachable. +func setupDistributed(opts *benchkit.Options, self string, remotes []string, dialOpts []gorums.DialOption) (BenchTarget, func(), error) { + var target BenchTarget + if len(remotes) < 2 { + return target, nil, fmt.Errorf("distributed mode requires at least 2 remotes (including self)") + } + opts.Remote = true + opts.NumNodes = len(remotes) + + symTarget, symStop, err := SetupRemoteServer(self, remotes, opts.ServerOptions(), dialOpts...) + if err != nil { + return target, nil, fmt.Errorf("remote server setup: %w", err) + } + + // In dedup mode, wait for every shared stream to be live first, so the + // probe below exercises the shared topology instead of failing fast with + // ErrStreamDown for lower-ID peers that have not yet connected. + if opts.StreamMode == "dedup" { + dedupCtx, dedupCancel := context.WithTimeout(context.Background(), 2*time.Minute) + err := awaitStreamDedup(dedupCtx, symTarget) + dedupCancel() + if err != nil { + symStop() + return target, nil, err + } + } + // The deadline is only an upper bound for very large clusters that are + // still making progress; a dead peer is detected within readyStallTimeout + // by the probe's stall check. + readyCtx, readyCancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer readyCancel() + if err := AwaitReady(readyCtx, symTarget); err != nil { + symStop() + return target, nil, fmt.Errorf("remote peers not ready: %w", err) + } + + target.Symmetric = symTarget + return target, symStop, nil +} + +// setupLocal builds the symmetric target backed by configSize in-process +// servers. +func setupLocal(opts *benchkit.Options, configSize int, dialOpts []gorums.DialOption) (BenchTarget, func(), error) { + var target BenchTarget + if configSize < 1 { + return target, nil, fmt.Errorf("local mode requires config-size >= 1, got %d", configSize) + } + opts.Remote = false + opts.NumNodes = configSize + + symTarget, symStop, err := SetupSymmetricServers(configSize, opts.ServerOptions(), dialOpts...) + if err != nil { + return target, nil, fmt.Errorf("symmetric servers setup: %w", err) + } + + // In dedup mode, wait for every shared stream to be live first, so the + // probe below exercises the shared topology instead of failing fast with + // ErrStreamDown for lower-ID peers that have not yet connected. + if opts.StreamMode == "dedup" { + dedupCtx, dedupCancel := context.WithTimeout(context.Background(), 10*time.Second) + err := awaitStreamDedup(dedupCtx, symTarget) + dedupCancel() + if err != nil { + symStop() + return target, nil, err + } + } + readyCtx, readyCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer readyCancel() + if err := AwaitReady(readyCtx, symTarget); err != nil { + symStop() + return target, nil, fmt.Errorf("symmetric servers not ready: %w", err) + } + + target.Symmetric = symTarget + return target, symStop, nil +} + +func awaitStreamDedup(ctx context.Context, t *SymmetricTarget) error { + for i, srv := range t.servers { + if _, err := srv.WaitForAll(ctx); err != nil { + return fmt.Errorf("%s: stream dedup setup: %w", t.label(i), err) + } + } + return nil +} + +// setupCoordinator builds the traditional coordinator-side configuration +// against the given remote servers. +func setupCoordinator(opts *benchkit.Options, remotes []string, configSize int, dialOpts []gorums.DialOption) (BenchTarget, func(), error) { + var target BenchTarget + opts.Remote = true + numNodes := len(remotes) + if configSize < 1 || configSize > numNodes { + opts.NumNodes = numNodes + } else { + opts.NumNodes = configSize + } + + cfg, err := gorums.NewConfig(gorums.WithNodeList(remotes[:opts.NumNodes]), dialOpts...) + if err != nil { + return target, nil, fmt.Errorf("configuration setup: %w", err) + } + + target.Config = cfg + return target, func() { _ = cfg.Close() }, nil +} diff --git a/benchkit/benchmark/target_test.go b/benchkit/benchmark/target_test.go new file mode 100644 index 00000000..13124d34 --- /dev/null +++ b/benchkit/benchmark/target_test.go @@ -0,0 +1,750 @@ +package benchmark + +import ( + "bytes" + "context" + "errors" + "fmt" + "math" + "net" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/benchkit" + "github.com/relab/gorums/gorumstest" + "google.golang.org/grpc" + "google.golang.org/grpc/keepalive" +) + +// captureDiag redirects the probe-stall self-diagnosis to a buffer for the +// duration of the test, so failure-path tests can assert on the diagnosis +// content without spamming the test log with goroutine dumps. +func captureDiag(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + orig := diagWriter + diagWriter = &buf + t.Cleanup(func() { diagWriter = orig }) + return &buf +} + +// localServers builds n peers with gorums.NewLocalServers and returns the raw, +// unstarted servers. It reuses the Gorums test framework's listener allocation, +// which binds every listener once and keeps it open for the lifetime of the +// server. Keeping each listener open avoids a close-to-rebind race under +// repeated test runs. +// +// The servers are returned unstarted so a test can register per-node handlers, +// serve only a subset, or stagger serving — the independent per-node control a +// real multi-process distributed run has, which single-target helpers hide. +func localServers(t *testing.T, n int, serverOpt gorums.ServerOption) []*gorums.Server { + t.Helper() + servers, stop, err := gorums.NewLocalServers( + n, + gorums.WithLocalServerOptions(serverOpt), + gorums.WithLocalDialOptions(gorumstest.InsecureDialOptions(t)), + ) + if err != nil { + t.Fatalf("NewLocalServers: %v", err) + } + t.Cleanup(stop) + return servers +} + +// benchTarget wraps one server as a single-server SymmetricTarget with the +// benchkit Control plane and workload server attached, so it can be passed to +// AwaitReady and the other per-node setup helpers. numPeers is the full cluster +// size (arms Done tracking and sizes the exit grace period). Call before +// serving srv, since attaching registers services. +func benchTarget(srv *gorums.Server, numPeers int) *SymmetricTarget { + ctrl := attachBenchServer(srv) + ctrl.ArmDone(numPeers) // match SetupRemoteServer, which arms Done tracking for the exit barrier + return &SymmetricTarget{ + servers: []*gorums.Server{srv}, + controls: []*benchkit.Control{ctrl}, + numPeers: numPeers, + selfAddr: srv.Addr(), + labels: []string{fmt.Sprintf("node %d (%s)", ctrl.SelfID(), srv.Addr())}, + } +} + +// localSymmetricTargets builds n single-server SymmetricTargets over one local +// node list, each wrapping one node (target[i] is node ID i+1) and already +// serving, so a test can drive per-node setup — dedup wait, probe — in a +// controlled order, the way separate SetupRemoteServer instances do in a +// real distributed run, but without freeTCPAddrs's port-reuse race. +func localSymmetricTargets(t *testing.T, n int, serverOpt gorums.ServerOption) []*SymmetricTarget { + t.Helper() + servers := localServers(t, n, serverOpt) + targets := make([]*SymmetricTarget, n) + for i, srv := range servers { + targets[i] = benchTarget(srv, n) + } + for _, srv := range servers { + go func() { _ = srv.ListenAndServe() }() + } + return targets +} + +// TestSetupTargetLocal verifies that local mode (no self, no remotes) builds a +// ready symmetric target and fills in the topology-derived Options fields. +func TestSetupTargetLocal(t *testing.T) { + var opts benchkit.Options + target, cleanup, err := SetupTarget(&opts, "", nil, 3, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupTarget: %v", err) + } + t.Cleanup(cleanup) + + if target.Symmetric == nil { + t.Error("Symmetric target is nil, want local symmetric servers") + } + if opts.Remote { + t.Error("opts.Remote = true, want false in local mode") + } + if opts.NumNodes != 3 { + t.Errorf("opts.NumNodes = %d, want 3", opts.NumNodes) + } +} + +// TestSetupTargetLocalRejectsNonPositiveConfigSize verifies that local mode +// (no self, no remotes) rejects a config size below 1 instead of creating a +// degenerate zero-server target that "succeeds" without doing any work. +func TestSetupTargetLocalRejectsNonPositiveConfigSize(t *testing.T) { + for _, configSize := range []int{0, -1} { + var opts benchkit.Options + _, _, err := SetupTarget(&opts, "", nil, configSize, gorumstest.InsecureDialOptions(t)) + if err == nil { + t.Errorf("SetupTarget(local, config-size=%d) = nil error, want error", configSize) + } + } +} + +// TestSetupSymmetricServersAppliesAllServerOptions verifies that +// SetupSymmetricServers forwards every option in the given slice to the +// in-process servers, not just the first. setupLocal previously called +// SetupSymmetricServers with a single gorums.ServerOption argument (only +// opts.StreamDedupOption()), so any other option opts.ServerOptions() would +// have supplied — e.g. buffer sizes — was silently dropped for local-mode +// runs; a local buffer-size sweep ran every arm with the default capacities +// while the recorded results claimed otherwise. +// +// Stream deduplication is the observable option here: it makes a peer with a +// lower ID than this node borrow that peer's channel instead of dialing its +// own, which Node.IsShared reports structurally, before any peer connects. A +// connect callback is the second, independent option, confirming that both +// elements of the slice reached the server rather than only the first. +func TestSetupSymmetricServersAppliesAllServerOptions(t *testing.T) { + var connects atomic.Int32 + opts := []gorums.ServerOption{ + gorums.WithStreamDedup(), + gorums.WithConnectCallback(func(context.Context) { connects.Add(1) }), + } + target, stop, err := SetupSymmetricServers(3, opts, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupSymmetricServers: %v", err) + } + t.Cleanup(stop) + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + for _, srv := range target.servers { + if _, err := srv.WaitForAll(ctx); err != nil { + t.Fatalf("WaitForAll: %v", err) + } + } + + srv3 := target.servers[2] + for _, node := range srv3.PeerConfig() { + if node.ID() >= 3 { + continue + } + if !node.IsShared() { + t.Errorf("node %d: IsShared() = false, want true; the stream-dedup option did not reach the server", node.ID()) + } + } + if got := connects.Load(); got == 0 { + t.Error("connect callback never fired; the connect-callback option did not reach the server") + } +} + +// TestSetupTargetDistributedRequiresPeers verifies that distributed mode with +// fewer than two remotes fails instead of running a degenerate benchmark. +func TestSetupTargetDistributedRequiresPeers(t *testing.T) { + var opts benchkit.Options + _, _, err := SetupTarget(&opts, "127.0.0.1:9000", []string{"127.0.0.1:9000"}, 0, gorumstest.InsecureDialOptions(t)) + if err == nil { + t.Fatal("SetupTarget(distributed, 1 remote) = nil error, want error") + } +} + +// TestSetupRemoteServerAppliesServerOption verifies that the ServerOption +// reaches the server built for distributed mode. The option carries the run's +// stream topology, so dropping it would leave every cluster sweep running the +// default topology while its results were labeled otherwise. +// +// Stream deduplication is the observable case: it makes a peer with a lower ID +// than this node borrow that peer's channel instead of dialing its own, which +// Node.IsShared reports structurally, before any peer connects. +func TestSetupRemoteServerAppliesServerOption(t *testing.T) { + // Self is the higher address, so the sorted peer list gives it ID 2 and the + // remaining peer ID 1; only a lower-ID peer is borrowed under dedup. + peers := []string{"127.0.0.1:0", "127.0.0.2:0"} + tests := []struct { + name string + serverOpts []gorums.ServerOption + wantShared bool + }{ + {name: "WithoutDedup"}, + {name: "WithDedup", serverOpts: []gorums.ServerOption{gorums.WithStreamDedup()}, wantShared: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + target, stop, err := SetupRemoteServer(peers[1], peers, tt.serverOpts, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupRemoteServer(%s): %v", peers[1], err) + } + t.Cleanup(stop) + + cfg := target.servers[0].PeerConfig() + var peer *gorums.Node + for _, n := range cfg.Nodes() { + if n.ID() == 1 { + peer = n + } + } + if peer == nil { + t.Fatalf("peer with ID 1 not in peer config %v", cfg.NodeIDs()) + } + if got := peer.IsShared(); got != tt.wantShared { + t.Errorf("peer 1 IsShared() = %v, want %v; the ServerOption did not reach the server", + got, tt.wantShared) + } + }) + } +} + +// TestSetupRemoteServerBindsWildcard verifies the distributed-mode listener +// binds the wildcard address rather than whatever the local host resolves its +// own name to. Hosts following the Debian convention map their own hostname +// to 127.0.1.1 in /etc/hosts, which would put the listener on loopback and +// make it unreachable for all peers (see doc/benchkit-troubleshooting.html). +func TestSetupRemoteServerBindsWildcard(t *testing.T) { + // This test must exercise SetupRemoteServer directly, because it is + // SetupRemoteServer (not the local test framework, which binds 127.0.0.1) + // that binds the wildcard host. Port 0 lets SetupRemoteServer pick its own + // free port, so no port is reserved and released beforehand — avoiding the + // bind-reuse race. The two peers differ only so the sort/self-index is + // stable; only self (the lower address) is bound. + peers := []string{"127.0.0.1:0", "127.0.0.2:0"} + target, stop, err := SetupRemoteServer(peers[0], peers, nil, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupRemoteServer(%s): %v", peers[0], err) + } + t.Cleanup(stop) + + // ListenAndServe binds the wildcard listener in a goroutine, so wait until + // Addr reports the concrete bound port rather than the configured ":0". + var addr string + if !gorumstest.WaitUntil(t, 2*time.Second, func() bool { + addr = target.servers[0].Addr() + _, p, e := net.SplitHostPort(addr) + return e == nil && p != "" && p != "0" + }) { + t.Fatalf("listener did not bind a concrete port; Addr = %q", addr) + } + host, port, err := net.SplitHostPort(addr) + if err != nil { + t.Fatalf("SplitHostPort(%s): %v", addr, err) + } + // The host must be the wildcard (empty or unspecified), never the loopback + // or resolved hostname — the Debian 127.0.1.1 trap this guards against. + if ip := net.ParseIP(host); host != "" && (ip == nil || !ip.IsUnspecified()) { + t.Errorf("listener bound to host %q, want wildcard", host) + } + // Port 0 randomizes the bound port, so exact port preservation is not + // asserted here; the binding must still resolve to a concrete port. + if port == "" || port == "0" { + t.Errorf("listener bound to port %q, want a concrete port", port) + } +} + +// TestExitGrace verifies the distributed-mode exit grace grows with cluster +// size, stays at or above the base floor, and is clamped for large clusters. +func TestExitGrace(t *testing.T) { + const ( + base = 3 * time.Second + perNode = 300 * time.Millisecond + maxGrace = 20 * time.Second + ) + tests := []struct { + numNodes int + want time.Duration + }{ + {0, base}, + {3, base + 3*perNode}, + {25, base + 25*perNode}, + {120, maxGrace}, // base + 120*perNode = 21s, clamped to maxGrace + } + for _, tt := range tests { + if got := ExitGrace(tt.numNodes); got != tt.want { + t.Errorf("ExitGrace(%d) = %v, want %v", tt.numNodes, got, tt.want) + } + } + // The grace must never decrease as the cluster grows. + prev := ExitGrace(1) + for n := 2; n <= 200; n++ { + got := ExitGrace(n) + if got < prev { + t.Fatalf("ExitGrace(%d) = %v < ExitGrace(%d) = %v; want non-decreasing", n, got, n-1, prev) + } + prev = got + } +} + +// TestAwaitReadyStaggeredRemoteStartup verifies distributed readiness tolerates +// one node starting before its peer. Both listeners are bound up front by the +// framework, so the stagger is in when each node begins serving (accepting gRPC +// streams): node 1 serves 500ms before node 2, and node 1's outbound stream to +// node 2 must retry across that gap rather than fail readiness. +func TestAwaitReadyStaggeredRemoteStartup(t *testing.T) { + servers := localServers(t, 2, nil) + target1, target2 := benchTarget(servers[0], 2), benchTarget(servers[1], 2) + + go func() { _ = servers[0].ListenAndServe() }() + time.Sleep(500 * time.Millisecond) + go func() { _ = servers[1].ListenAndServe() }() + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + errCh := make(chan error, 2) + go func() { errCh <- AwaitReady(ctx, target1) }() + go func() { errCh <- AwaitReady(ctx, target2) }() + + var errs error + for range 2 { + if err := <-errCh; err != nil { + errs = errors.Join(errs, err) + } + } + if errs != nil { + t.Fatalf("AwaitReady after staggered startup: %v", errs) + } + if got := target1.servers[0].ConnectedPeers().Size(); got != 2 { + t.Errorf("target1 connected config size = %d, want 2", got) + } + if got := target2.servers[0].ConnectedPeers().Size(); got != 2 { + t.Errorf("target2 connected config size = %d, want 2", got) + } +} + +// TestDualReconnectsDroppedIdleStream verifies that in dual mode a symmetric +// server re-establishes an outbound stream that dropped while idle — with no +// local send prompting it — so the peer stays reachable in its connected-peer +// configuration. During setup no node sends application traffic, so a stream +// that is never re-established leaves the node without its peer and stalls +// the readiness probe. +// +// The drop is forced deterministically with a short server-side +// MaxConnectionAge: gRPC sends GOAWAY and closes the connection, ending the +// outbound stream the other node dialed in. This is more reliable than a +// refused initial connect, which gRPC papers over by retrying the connection +// underneath a still-pending stream. The age limit recurs on every reconnected +// stream, so the gap is observed even if a reconnect follows immediately. +// +// The servers are built through the shared test framework +// ([gorumstest.LocalServers]), which owns listener allocation and cleanup for +// the whole test. The age limit is applied to both symmetric servers; in a +// two-node group the observed node's single outbound stream is dropped by its +// peer's age limit either way. The drop and reconnect are observed from one +// node, whose connected-peer view tracks its outbound stream state. +func TestDualReconnectsDroppedIdleStream(t *testing.T) { + const maxAge = 300 * time.Millisecond + servers := gorumstest.LocalServers(t, 2, gorums.WithGRPCServerOptions( + grpc.KeepaliveParams(keepalive.ServerParameters{ + MaxConnectionAge: maxAge, + MaxConnectionAgeGrace: 50 * time.Millisecond, + }), + )) + + observer := servers[0] + const peerID = 2 + hasPeer := func(cfg gorums.Config) bool { return cfg.Contains(peerID) } + missingPeer := func(cfg gorums.Config) bool { return !cfg.Contains(peerID) } + + ctx := gorumstest.Context(t, 10*time.Second) + + // The mesh forms from the senders' eager initial connect, with no sends. + if err := observer.WaitForPeers(ctx, hasPeer); err != nil { + t.Fatalf("outbound stream to the peer never came up: %v", err) + } + + // The peer's MaxConnectionAge closes the connection the observer dialed in, + // dropping the observer's outbound stream, so the peer leaves the observer's + // connected view. The stream-state change broadcasts a config change, and + // the age limit recurs on every reconnected stream, so the gap is observed + // even if a reconnect follows immediately. + if err := observer.WaitForPeers(ctx, missingPeer); err != nil { + t.Fatalf("MaxConnectionAge never dropped the idle outbound stream: %v", err) + } + + // The observer must re-establish its dropped stream on its own — no sends + // happen here — so the peer returns to its connected view. Without a + // self-initiated reconnect the observer has lost the peer for good. + if err := observer.WaitForPeers(ctx, hasPeer); err != nil { + t.Fatalf("did not reconnect the dropped idle stream: %v", err) + } +} + +// TestDedupSetupProbesSharedTopology verifies that the setup sequence +// setupDistributed/setupLocal use in dedup mode — the dedup wait +// (awaitStreamDedup, which calls Server.WaitForAll), then the outbound +// probe — leaves every lower-ID outbound peer backed by its live shared +// inbound stream before the probe runs, and that the probe then succeeds +// against that shared topology. Probing before the dedup wait would fail +// fast with ErrStreamDown for any lower-ID peer that has not yet connected. +func TestDedupSetupProbesSharedTopology(t *testing.T) { + targets := localSymmetricTargets(t, 3, gorums.WithStreamDedup()) + + // One ctx is shared across the sequential dedup-wait and probe steps + // below. Setup is in-process over kept-open listeners, so it completes in + // milliseconds; the timeout only bounds a genuinely stuck peer. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + // Step 1: the dedup wait, exactly as setupDistributed does before probing. + for _, target := range targets { + if err := awaitStreamDedup(ctx, target); err != nil { + t.Fatalf("awaitStreamDedup: %v", err) + } + } + + // By the time the probe runs, every peer with a lower ID than node 3 + // must be a shared node backed by a live inbound stream. + srv3 := targets[2].servers[0] + for _, node := range srv3.PeerConfig() { + if node.ID() >= 3 { + continue + } + if node.IsOutbound() { + t.Errorf("node %d: IsOutbound() = true, want false (expected a shared inbound stream before the probe)", node.ID()) + } + if !node.IsInbound() { + t.Errorf("node %d: IsInbound() = false, want true (expected a shared inbound stream before the probe)", node.ID()) + } + } + + // Step 2: the probe must succeed against that shared topology. + for _, target := range targets { + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady: %v", err) + } + } +} + +// TestAwaitPeersDoneOrGraceReturnsEarlyWhenAllSignal verifies that once every +// peer has called SignalDone, AwaitPeersDoneOrGrace returns true well before +// a deliberately long grace period elapses, instead of always sleeping it out. +func TestAwaitPeersDoneOrGraceReturnsEarlyWhenAllSignal(t *testing.T) { + targets := localSymmetricTargets(t, 2, nil) + target1, target2 := targets[0], targets[1] + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := AwaitReady(ctx, target1); err != nil { + t.Fatalf("AwaitReady(target1): %v", err) + } + if err := AwaitReady(ctx, target2); err != nil { + t.Fatalf("AwaitReady(target2): %v", err) + } + + const grace = 10 * time.Second + SignalDone(ctx, target1) + SignalDone(ctx, target2) + + type result struct { + allDone bool + elapsed time.Duration + } + results := make(chan result, 2) + for _, target := range []*SymmetricTarget{target1, target2} { + go func(target *SymmetricTarget) { + start := time.Now() + allDone := AwaitPeersDoneOrGrace(context.Background(), target, grace) + results <- result{allDone, time.Since(start)} + }(target) + } + for range 2 { + r := <-results + if !r.allDone { + t.Error("AwaitPeersDoneOrGrace = false, want true when all peers signal Done") + } + if r.elapsed > grace/2 { + t.Errorf("AwaitPeersDoneOrGrace took %v, want well under grace=%v", r.elapsed, grace) + } + } +} + +// TestAwaitPeersDoneOrGraceFallsBackWhenPeerNeverSignals verifies that a peer +// which never calls SignalDone does not hang or fail the waiter: the waiter +// falls back to the grace deadline and returns false. +func TestAwaitPeersDoneOrGraceFallsBackWhenPeerNeverSignals(t *testing.T) { + targets := localSymmetricTargets(t, 2, nil) + target1, target2 := targets[0], targets[1] + + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := AwaitReady(ctx, target1); err != nil { + t.Fatalf("AwaitReady(target1): %v", err) + } + if err := AwaitReady(ctx, target2); err != nil { + t.Fatalf("AwaitReady(target2): %v", err) + } + + // target2 never signals Done; only target1 does. + const grace = 500 * time.Millisecond + SignalDone(context.Background(), target1) + + start := time.Now() + allDone := AwaitPeersDoneOrGrace(context.Background(), target1, grace) + elapsed := time.Since(start) + + if allDone { + t.Error("AwaitPeersDoneOrGrace = true, want false when a peer never signals Done") + } + if elapsed < grace { + t.Errorf("AwaitPeersDoneOrGrace returned after %v, want at least grace=%v", elapsed, grace) + } + if elapsed > grace+2*time.Second { + t.Errorf("AwaitPeersDoneOrGrace returned after %v, want close to grace=%v", elapsed, grace) + } + if got := target1.controls[0].MissingDone(); len(got) != 1 { + t.Errorf("MissingDone() = %v, want exactly 1 missing peer", got) + } +} + +// captureProbeLog redirects the outbound-probe progress log to a buffer for +// the duration of the test, so probe tests can assert that stragglers were +// logged by node ID. The probe logs from the calling goroutine only, so the +// buffer needs no locking as long as it is read after AwaitReady returns. +func captureProbeLog(t *testing.T) *bytes.Buffer { + t.Helper() + var buf bytes.Buffer + orig := probeLogf + probeLogf = func(format string, args ...any) { fmt.Fprintf(&buf, format, args...) } + t.Cleanup(func() { probeLogf = orig }) + return &buf +} + +// registerReplyDroppingPeer registers a QuorumCall handler on srv that silently +// sends no reply for the first drop echo requests it receives — mirroring a +// reply lost to stream churn (no error reaches the caller). Later requests echo +// normally. Call before serving srv, since it registers a service. Used to make +// one peer in a localServers mesh a reply-dropping node. +func registerReplyDroppingPeer(srv *gorums.Server, drop int64) { + var remaining atomic.Int64 + remaining.Store(drop) + srv.RegisterHandler("benchmark.Benchmark.QuorumCall", func(_ gorums.ServerContext, in *gorums.Message) (*gorums.Message, error) { + if remaining.Add(-1) >= 0 { + return nil, nil // no response and no error: nothing is sent back + } + return gorums.NewResponseMessage(in, gorums.AsProto[*Echo](in)), nil + }) +} + +// TestAwaitReadyProbeRetriesDroppedReply verifies the outbound probe survives a +// peer that silently loses exactly one echo reply: the per-peer attempt times +// out after probeAttemptTimeout instead of consuming the whole readiness +// deadline, the straggler is logged by node ID, and a later round succeeds. +func TestAwaitReadyProbeRetriesDroppedReply(t *testing.T) { + defer func(d time.Duration) { probeAttemptTimeout = d }(probeAttemptTimeout) + probeAttemptTimeout = 300 * time.Millisecond + probeLog := captureProbeLog(t) + + // node 1 probes; node 2 drops one reply. + servers := localServers(t, 2, nil) + target := benchTarget(servers[0], 2) + registerReplyDroppingPeer(servers[1], 1) + for _, srv := range servers { + go func() { _ = srv.ListenAndServe() }() + } + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + start := time.Now() + if err := AwaitReady(ctx, target); err != nil { + t.Fatalf("AwaitReady with one dropped echo reply: %v", err) + } + if elapsed := time.Since(start); elapsed > 5*time.Second { + t.Errorf("AwaitReady took %v, want one lost reply to cost roughly one probe round", elapsed) + } + if got := probeLog.String(); !strings.Contains(got, "node 2") { + t.Errorf("probe log does not name straggler node 2; got:\n%s", got) + } +} + +// TestAwaitReadyProbeFailsFastOnSilentPeer verifies that a peer which never +// answers echo probes fails the probe within the stall window — naming the +// silent peer — instead of blocking until the context deadline with the +// unattributable "incomplete call (errors: 0)" of the all-or-nothing probe. +func TestAwaitReadyProbeFailsFastOnSilentPeer(t *testing.T) { + defer func(d time.Duration) { readyStallTimeout = d }(readyStallTimeout) + readyStallTimeout = 500 * time.Millisecond + defer func(d time.Duration) { probeAttemptTimeout = d }(probeAttemptTimeout) + probeAttemptTimeout = 100 * time.Millisecond + captureProbeLog(t) + + // node 1 probes; node 2 never answers echoes. + servers := localServers(t, 2, nil) + target := benchTarget(servers[0], 2) + registerReplyDroppingPeer(servers[1], math.MaxInt64) + for _, srv := range servers { + go func() { _ = srv.ListenAndServe() }() + } + node2Addr := servers[1].Addr() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + start := time.Now() + err := AwaitReady(ctx, target) + elapsed := time.Since(start) + if err == nil { + t.Fatal("AwaitReady = nil error, want silent-peer probe failure") + } + for _, want := range []string{"outbound peers not ready", "node 2", node2Addr} { + if !strings.Contains(err.Error(), want) { + t.Errorf("AwaitReady error %q does not contain %q", err, want) + } + } + if elapsed > 5*time.Second { + t.Errorf("AwaitReady took %v, want fail-fast well under the 30s deadline", elapsed) + } +} + +// TestAwaitReadyReportsMissingRemotePeers verifies distributed readiness errors +// identify peer addresses that never respond. +func TestAwaitReadyReportsMissingRemotePeers(t *testing.T) { + captureDiag(t) + captureProbeLog(t) + // node 2's address stays in node 1's node list, but node 2 is shut down + // immediately so nothing listens there: node 1's probe never gets a + // response and readiness reports the peer pending by address. Its port is + // closed and never rebound, so unlike freeTCPAddrs there is no bind-reuse + // race. + servers := localServers(t, 2, nil) + node2Addr := servers[1].Addr() + servers[1].Stop() + target := benchTarget(servers[0], 2) + go func() { _ = servers[0].ListenAndServe() }() + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + err := AwaitReady(ctx, target) + if err == nil { + t.Fatal("AwaitReady = nil error, want missing peer error") + } + for _, want := range []string{"outbound peers not ready", "node 2", node2Addr} { + if !strings.Contains(err.Error(), want) { + t.Errorf("AwaitReady error %q does not contain %q", err, want) + } + } +} + +// TestAwaitReadyFailsFastOnStalledPeer verifies the outbound probe gives up +// readyStallTimeout after the last peer responded, instead of waiting out the +// full context deadline when a peer never starts. It also verifies the +// failure emits the probe-stall self-diagnosis: the bound listener address, a +// self-dial probe of the advertised address, and a goroutine dump (see +// doc/benchkit-troubleshooting.html). +func TestAwaitReadyFailsFastOnStalledPeer(t *testing.T) { + defer func(d time.Duration) { readyStallTimeout = d }(readyStallTimeout) + readyStallTimeout = 500 * time.Millisecond + defer func(d time.Duration) { probeAttemptTimeout = d }(probeAttemptTimeout) + probeAttemptTimeout = 100 * time.Millisecond + diag := captureDiag(t) + captureProbeLog(t) + + // node 2's address stays in node 1's node list, but node 2 is shut down + // immediately so nothing listens there: node 1's probe stalls and must + // give up readyStallTimeout after the last response rather than waiting + // out the full context deadline. Its port is closed and never rebound, so + // unlike freeTCPAddrs there is no bind-reuse race. + servers := localServers(t, 2, nil) + node1Addr, node2Addr := servers[0].Addr(), servers[1].Addr() + servers[1].Stop() + target := benchTarget(servers[0], 2) + go func() { _ = servers[0].ListenAndServe() }() + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + start := time.Now() + err := AwaitReady(ctx, target) + elapsed := time.Since(start) + if err == nil { + t.Fatal("AwaitReady = nil error, want stalled readiness error") + } + for _, want := range []string{"outbound peers not ready", "no outbound peer responded", node2Addr} { + if !strings.Contains(err.Error(), want) { + t.Errorf("AwaitReady error %q does not contain %q", err, want) + } + } + if elapsed > 5*time.Second { + t.Errorf("AwaitReady took %v, want fail-fast well under the 30s deadline", elapsed) + } + + // The listener is up (wildcard-bound), so the self-dial probe of the + // advertised address must succeed and the diagnosis must point the + // blockage away from this host. + got := diag.String() + for _, want := range []string{ + "listener bound to ", + "self-dial " + node1Addr + " ok", + "goroutine dump", + } { + if !strings.Contains(got, want) { + t.Errorf("probe-stall diagnosis does not contain %q; got:\n%s", want, got) + } + } +} + +// TestSetupTargetCoordinatorNumNodes verifies the coordinator-mode node-count +// clamping: configSize selects a prefix of the remotes when within range and +// all remotes otherwise. +func TestSetupTargetCoordinatorNumNodes(t *testing.T) { + remotes := []string{"127.0.0.1:9001", "127.0.0.1:9002", "127.0.0.1:9003"} + tests := []struct { + name string + configSize int + wantNodes int + }{ + {"WithinRangeSelectsPrefix", 2, 2}, + {"ZeroUsesAll", 0, 3}, + {"TooLargeUsesAll", 5, 3}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var opts benchkit.Options + target, cleanup, err := SetupTarget(&opts, "", remotes, tt.configSize, gorumstest.InsecureDialOptions(t)) + if err != nil { + t.Fatalf("SetupTarget: %v", err) + } + t.Cleanup(cleanup) + + if target.Config == nil { + t.Error("Config target is nil, want coordinator configuration") + } + if !opts.Remote { + t.Error("opts.Remote = false, want true in coordinator mode") + } + if opts.NumNodes != tt.wantNodes { + t.Errorf("opts.NumNodes = %d, want %d", opts.NumNodes, tt.wantNodes) + } + if got := target.Config.Size(); got != tt.wantNodes { + t.Errorf("config size = %d, want %d", got, tt.wantNodes) + } + }) + } +} diff --git a/benchkit/cmd/benchmark/main.go b/benchkit/cmd/benchmark/main.go new file mode 100644 index 00000000..8f73a97d --- /dev/null +++ b/benchkit/cmd/benchmark/main.go @@ -0,0 +1,219 @@ +package main + +import ( + "context" + "flag" + "fmt" + "net" + "os" + "os/signal" + "syscall" + "time" + + "github.com/relab/gorums" + "github.com/relab/gorums/benchkit" + "github.com/relab/gorums/benchkit/benchmark" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +// flags embeds the benchkit standard flag contract (benchmarks, self, remotes, +// workers, payload, rate, time, output, verbose) and adds the +// gorums-specific extras the reference tool exposes. +type flags struct { + *benchkit.StandardFlags + maxAsync int + server string + serverStats bool + configSize int + qSize int + sendBuffer uint + recvBuffer uint + list bool + label string + compare string +} + +func parseFlags() *flags { + // The standard contract flags are registered by benchkit so this binary + // complies with what sweep launches; the extras below are gorums-specific. + f := &flags{StandardFlags: benchkit.RegisterFlags(flag.CommandLine)} + flag.IntVar(&f.maxAsync, "max-async", 1000, "Maximum number of async calls that can be in flight at once.") + flag.StringVar(&f.server, "server", "", "Run a benchmark server on given `address`.") + flag.BoolVar(&f.serverStats, "server-stats", false, "Show server statistics separately.") + flag.IntVar(&f.configSize, "config-size", 4, "Size of the configuration to use. In local mode this is the number of in-process servers and must be >= 1. In coordinator mode, values < 1 or greater than the number of remotes use all remotes. Ignored in distributed mode (-self), where every remote is used.") + flag.IntVar(&f.qSize, "quorum-size", 0, "Number of replies to wait for before completing a quorum call.") + flag.UintVar(&f.sendBuffer, "send-buffer", 0, "The size of the client's (and server's reverse channel) send buffer.") + flag.UintVar(&f.recvBuffer, "recv-buffer", 0, "The size of the server's receive buffer.") + flag.BoolVar(&f.list, "list", false, "List all available benchmarks.") + flag.StringVar(&f.label, "label", "", "Label for this run, stored in the output file.") + flag.StringVar(&f.compare, "compare", "", "Compare against results in this `file`.") + flag.Parse() + mode, err := normalizeStreamMode(f.StreamMode) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(2) + } + f.StreamMode = mode + return f +} + +// normalizeStreamMode validates the -stream-mode flag value and normalizes +// its default: "" and "dual" both resolve to "dual"; "dedup" passes through +// unchanged. Any other value is rejected. +func normalizeStreamMode(mode string) (string, error) { + switch mode { + case "", "dual": + return "dual", nil + case "dedup": + return mode, nil + default: + return "", fmt.Errorf("invalid -stream-mode %q (want: dual or dedup)", mode) + } +} + +func (f *flags) options() benchkit.Options { + opts := f.StandardFlags.Options() + opts.MaxAsync = f.maxAsync + opts.SendBuffer, opts.RecvBuffer = f.bufferSizes() + return opts +} + +// bufferSizes resolves the -send-buffer and -recv-buffer flags to the actual +// capacities this run uses. A send-buffer flag left at its zero default is +// resolved here to gorums.DefaultSendBufferSize, the same substitution +// gorums.WithSendBufferSize applies internally, so recorded results show what +// actually ran rather than a zero that does not reflect it. Zero is already +// the real receive-buffer default, so it needs no such resolution. +func (f *flags) bufferSizes() (sendBuffer, recvBuffer uint) { + sendBuffer = f.sendBuffer + if sendBuffer == 0 { + sendBuffer = gorums.DefaultSendBufferSize + } + return sendBuffer, f.recvBuffer +} + +func (f *flags) dialOpts() []gorums.DialOption { + return []gorums.DialOption{ + gorums.WithGRPCDialOptions(grpc.WithTransportCredentials(insecure.NewCredentials())), + gorums.WithSendBufferSize(f.sendBuffer), + } +} + +// target configures the benchmark target for one of three modes: +// distributed (-self set), local (no -remotes), or coordinator (-remotes set). +// It returns the target and a cleanup function that must be deferred by the +// caller. The mode selection and setup live in benchmark.SetupTarget. +func (f *flags) target(opts *benchkit.Options, dialOpts []gorums.DialOption) (benchmark.BenchTarget, func()) { + target, cleanup, err := benchmark.SetupTarget(opts, f.Self, f.Remotes, f.configSize, dialOpts...) + checkf("Failed to set up benchmark target: %v", err) + return target, cleanup +} + +// quorumSize resolves the -quorum-size flag against the configuration size, +// clamping it to numNodes. Unset, it is a majority. The threshold counts the +// local node's in-process reply, so anything below a majority can be satisfied +// without contacting a peer. +func (f *flags) quorumSize(numNodes int) int { + switch { + case f.qSize < 1: + return numNodes/2 + 1 + case f.qSize > numNodes: + return numNodes + default: + return f.qSize + } +} + +func (f *flags) report(results []*benchkit.Result, opts benchkit.Options) { + benchkit.PrintResults(os.Stdout, results, opts, f.serverStats, f.Self) + // In distributed mode -self names this node, so it labels the run when + // -label is unset. The written report and the comparison use the same one. + label := f.label + if label == "" && f.Self != "" { + label = f.Self + } + if f.Output != "" { + checkf("Failed to write results: %v", benchkit.WriteLabeledReport(results, label, f.Output)) + } + if f.compare != "" { + checkf("Failed to compare results: %v", benchkit.CompareWithBaseline(f.compare, label, results, os.Stdout)) + } +} + +func runServer(addr string, recvSize, sendSize uint) { + signals := make(chan os.Signal, 1) + signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM) + + lis, err := net.Listen("tcp", addr) + checkf("Failed to listen on '%s': %v", addr, err) + + srv := benchmark.NewBenchServer(gorums.WithBufferSizes(recvSize, sendSize)) + go func() { checkf("serve failed: %v", srv.Serve(lis)) }() + benchkit.Logf("Running benchmark server on '%s'\n", addr) + + <-signals + srv.Stop() +} + +func main() { + f := parseFlags() + benchkit.SetVerbose(f.Verbose) + + if f.list { + benchkit.ListBenches(os.Stdout, benchmark.BenchmarkDescriptions()) + return + } + + stopProfilers, err := benchkit.StartProfilers(f.CPUProfile, f.MemProfile, f.Trace) + checkf("Failed to start profiling: %v", err) + defer func() { checkf("Failed to stop profiling: %v", stopProfilers()) }() + + benchkit.ArmFaultInjection(f.FaultKillAfter) + + if f.server != "" { + runServer(f.server, f.recvBuffer, f.sendBuffer) + return + } + + opts := f.options() + target, cleanup := f.target(&opts, f.dialOpts()) + defer cleanup() + + opts.QuorumSize = f.quorumSize(opts.NumNodes) + + results, err := benchmark.RunBenchmarks(f.Benchmarks, opts, target) + checkf("Error running benchmarks: %v", err) + + f.report(results, opts) + + // In distributed mode the symmetric topology has no exit barrier: signal + // peers that this node is done, then race that signal against a grace + // deadline (deferred cleanup happens once one of the two resolves). See + // benchmark.ExitGrace and benchmark.AwaitPeersDoneOrGrace. + if f.Self != "" { + nodeLabel := f.label + if nodeLabel == "" { + nodeLabel = f.Self + } + grace := benchmark.ExitGrace(opts.NumNodes) + benchkit.Logf("[%s %s] Benchmark complete; signaling done, waiting up to %v for peers...\n", time.Now().Format(time.TimeOnly), nodeLabel, grace) + benchmark.SignalDone(context.Background(), target.Symmetric) + waitStart := time.Now() + if benchmark.AwaitPeersDoneOrGrace(context.Background(), target.Symmetric, grace) { + benchkit.Logf("[%s %s] All peers done after %v; exiting early.\n", time.Now().Format(time.TimeOnly), nodeLabel, time.Since(waitStart)) + } else { + benchkit.Logf("[%s %s] Grace exhausted after %v; missing done from node(s) %v; exiting anyway.\n", + time.Now().Format(time.TimeOnly), nodeLabel, time.Since(waitStart), benchmark.MissingDoneSenders(target.Symmetric)) + } + } +} + +func checkf(format string, args ...any) { + for _, arg := range args { + if err, _ := arg.(error); err != nil { + fmt.Fprintf(os.Stderr, format, args...) + os.Exit(1) // skipcq: RVV-A0003 + } + } +} diff --git a/benchkit/cmd/benchmark/main_test.go b/benchkit/cmd/benchmark/main_test.go new file mode 100644 index 00000000..4d0269ea --- /dev/null +++ b/benchkit/cmd/benchmark/main_test.go @@ -0,0 +1,114 @@ +package main + +import ( + "testing" + + "github.com/relab/gorums" +) + +// TestQuorumSize verifies that an unset -quorum-size resolves to a majority of +// the configuration, and that an explicit value is honored but clamped to the +// configuration size. +// +// The majority matters beyond arithmetic: the threshold counts the local node's +// in-process reply, so a sub-majority threshold lets a call complete without +// contacting any peer, which makes the benchmark measure local dispatch. +func TestQuorumSize(t *testing.T) { + tests := []struct { + name string + qSize int + numNodes int + want int + }{ + {"UnsetOneNode", 0, 1, 1}, + {"UnsetThreeNodes", 0, 3, 2}, + {"UnsetFourNodes", 0, 4, 3}, + {"UnsetFiveNodes", 0, 5, 3}, + {"UnsetSevenNodes", 0, 7, 4}, + {"NegativeTreatedAsUnset", -1, 5, 3}, + {"ExplicitBelowMajority", 2, 7, 2}, + {"ExplicitAboveMajority", 6, 7, 6}, + {"ExplicitClampedToNodes", 9, 7, 7}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := &flags{qSize: tt.qSize} + if got := f.quorumSize(tt.numNodes); got != tt.want { + t.Errorf("quorumSize(%d) with -quorum-size=%d = %d, want %d", + tt.numNodes, tt.qSize, got, tt.want) + } + }) + } +} + +// TestQuorumSizeExceedsLocalReply verifies the property the default exists for: +// a majority always needs at least one reply beyond the local node's, so the +// benchmark cannot be satisfied by in-process dispatch alone. +func TestQuorumSizeExceedsLocalReply(t *testing.T) { + f := &flags{} + for numNodes := 2; numNodes <= 33; numNodes++ { + if got := f.quorumSize(numNodes); got < 2 { + t.Errorf("quorumSize(%d) = %d, want at least 2 so a peer reply is required", numNodes, got) + } + } +} + +// TestBufferSizes verifies that an unset -send-buffer resolves to +// gorums.DefaultSendBufferSize, so recorded results show the capacity that +// actually ran instead of a zero that does not reflect it, while an explicit +// value and -recv-buffer (whose zero is already the real default) pass +// through unchanged. +func TestBufferSizes(t *testing.T) { + tests := []struct { + name string + sendBuffer uint + recvBuffer uint + wantSendBuffer uint + wantRecvBuffer uint + }{ + {"UnsetSendBufferResolvesToDefault", 0, 0, gorums.DefaultSendBufferSize, 0}, + {"ExplicitSendBufferPassesThrough", 4096, 0, 4096, 0}, + {"ExplicitRecvBufferPassesThrough", 0, 2048, gorums.DefaultSendBufferSize, 2048}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + f := &flags{sendBuffer: tt.sendBuffer, recvBuffer: tt.recvBuffer} + gotSend, gotRecv := f.bufferSizes() + if gotSend != tt.wantSendBuffer { + t.Errorf("sendBuffer = %d, want %d", gotSend, tt.wantSendBuffer) + } + if gotRecv != tt.wantRecvBuffer { + t.Errorf("recvBuffer = %d, want %d", gotRecv, tt.wantRecvBuffer) + } + }) + } +} + +// TestNormalizeStreamMode verifies -stream-mode's validation and default: +// an unset value normalizes to "dual", "dedup" passes through, and any other +// value is rejected instead of silently falling back to a mode the user did +// not choose. +func TestNormalizeStreamMode(t *testing.T) { + tests := []struct { + name string + mode string + want string + wantErr bool + }{ + {"UnsetDefaultsToDual", "", "dual", false}, + {"ExplicitDual", "dual", "dual", false}, + {"ExplicitDedup", "dedup", "dedup", false}, + {"InvalidRejected", "bogus", "", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := normalizeStreamMode(tt.mode) + if (err != nil) != tt.wantErr { + t.Fatalf("normalizeStreamMode(%q) error = %v, wantErr %v", tt.mode, err, tt.wantErr) + } + if err == nil && got != tt.want { + t.Errorf("normalizeStreamMode(%q) = %q, want %q", tt.mode, got, tt.want) + } + }) + } +}