diff --git a/peregrine/test/README.md b/peregrine/test/README.md index 1e0dbe3..8c74a02 100644 --- a/peregrine/test/README.md +++ b/peregrine/test/README.md @@ -8,6 +8,7 @@ | Subdirectory | Namespace | Purpose | | :------------- | :--------------------------- | :------------------------- | | `benchmark/` | `peregrine::benchmark` | Benchmarks across hosts | +| `cluster/` | `peregrine::cluster` | Multi-node cluster tests | | `integration/` | `peregrine::integration` | Tests in a single process | | `workloads/` | `peregrine::workloads` | Shared workload generators | diff --git a/peregrine/test/cluster/README.md b/peregrine/test/cluster/README.md new file mode 100644 index 0000000..a69021f --- /dev/null +++ b/peregrine/test/cluster/README.md @@ -0,0 +1,178 @@ +# Peregrine Cluster Benchmarks (`//peregrine/test/cluster`) + +This directory provides a multi-node, multi-instance cluster benchmarking +framework (`namespace peregrine::cluster`) that coordinates: + +- **$N$ Client Nodes**, each running **$N_p$ Peregrine `Transport` instances** + within a single process ($N \times N_p$ total client instances). +- **$M$ Server Nodes**, each running **$M_p$ Peregrine `Transport` instances** + within a single process ($M \times M_p$ total server instances). + +## Subdirectories + +| Subdirectory | Purpose | +| :-------------- | :---------------------------------------------------------- | +| `node/` | C++ symmetric `ClusterNode` runtime & `:node` binary | +| `orchestrator/` | Python multi-node orchestrator & `:bipartite_runner` CLI | + +## Test Workflow + +Each `bipartite_runner` session executes the following steps: + +1. **Preflight & Staging**: Inspects kernel `sysctl` / `ulimit` limits on all + hosts (warning if below recommended thresholds) and stages the `:node` binary + to remote hosts over SSH `stdin`. +2. **Server Initialization**: Spawns $M$ server `:node` processes ($M_p$ + `Transport` instances per node), registers memory buffers, and collects their + `@` targets via `PEREGRINE_NODE_READY`. +3. **Client Initialization**: Spawns $N$ client `:node` processes ($N_p$ + `Transport` instances per node) configured with the server targets and waits + for `PEREGRINE_NODE_READY`. +4. **Synchronized Workload Execution**: Broadcasts `START` across all nodes so + client instances concurrently drive transfers (`all_to_all` or `round_robin`) + and stream periodic `PEREGRINE_METRICS_SAMPLE` updates until emitting + `PEREGRINE_WORKLOAD_DONE`. +5. **Teardown & Reporting**: Broadcasts `STOP` to collect final + `PEREGRINE_NODE_DONE` snapshots (`TransportMetrics` and `CpuStats`) from all + nodes, cleans up remote binaries, and prints the cluster-wide throughput, + latency percentile, CPU, and transport metrics summary. + +## Running the Fan-In TCP Accept Test ($N=1, N_p=1,000, M=1, M_p=1$) + +### Localhost (Single Host) + +```bash +$ bazelisk run -c opt //peregrine/test/cluster/orchestrator:bipartite_runner -- \ + --client_nodes=localhost \ + --server_nodes=localhost \ + --clients_per_node=1000 \ + --servers_per_node=1 \ + --transport=tcp \ + --conn=1 \ + --workload=serial_fixed_write \ + --xfer_size=1 \ + --num_xfers=1 \ + --traffic_pattern=all_to_all +``` + +### Multi-Node Cluster (Remote Hosts over SSH) + +When remote hostnames or IPs are passed via `--client_nodes` and +`--server_nodes` (supporting both `` and `@` syntax), +`bipartite_runner` automatically streams the `:node` binary over `--ssh_cmd` +`stdin` (without requiring `scp` or `sftp-server`): + +```bash +$ bazelisk run -c opt //peregrine/test/cluster/orchestrator:bipartite_runner -- \ + --client_nodes=user@10.0.0.1,user@10.0.0.2 \ + --server_nodes=user@10.0.0.3 \ + --clients_per_node=500 \ + --servers_per_node=1 \ + --transport=tcp \ + --conn=1 \ + --workload=serial_fixed_write \ + --xfer_size=1 \ + --num_xfers=1 \ + --traffic_pattern=all_to_all \ + --output_json=/tmp/bipartite_results.json +``` + +### Running Across Virtual Machines (VMs) + +When benchmarking across cloud or local VMs where: + +- SSH uses a custom identity key, port, jump host, or `gcloud compute ssh` + (`--ssh_cmd`), or +- The data-plane NIC IPs differ from the SSH management hostnames + (`--client_ips`, `--server_ips`): + +**Standard SSH with custom key/port and separate data-plane IPs:** + +```bash +$ bazelisk run -c opt //peregrine/test/cluster/orchestrator:bipartite_runner -- \ + --client_nodes=ubuntu@vm-client-1,ubuntu@vm-client-2 \ + --server_nodes=ubuntu@vm-server-1 \ + --client_ips=192.168.100.11,192.168.100.12 \ + --server_ips=192.168.100.21 \ + --ssh_cmd="ssh -i ~/.ssh/vm_key -p 2222 -o BatchMode=yes -o StrictHostKeyChecking=no" \ + --clients_per_node=500 \ + --servers_per_node=1 \ + --transport=tcp \ + --conn=1 \ + --workload=serial_fixed_write \ + --xfer_size=1 \ + --num_xfers=1 \ + --traffic_pattern=all_to_all +``` + +**Google Cloud Compute Engine VMs (`gcloud compute ssh`):** + +```bash +$ bazelisk run -c opt //peregrine/test/cluster/orchestrator:bipartite_runner -- \ + --client_nodes=user@gce-client-1,user@gce-client-2 \ + --server_nodes=user@gce-server-1 \ + --ssh_cmd="gcloud compute ssh --project=my-project --zone=us-central1-a" \ + --clients_per_node=500 \ + --servers_per_node=1 \ + --transport=tcp \ + --conn=1 \ + --workload=serial_fixed_write \ + --xfer_size=1 \ + --num_xfers=1 \ + --traffic_pattern=all_to_all +``` + +### Running in Containers (Docker / Podman / Kubernetes) + +Because `NodeExecutor` stages the `:node` binary and exchanges control messages +purely over standard input/output (`[*ssh_cmd, , ]`), +you can target containers directly—either via SSH inside the container or via +`docker exec -i` / `kubectl exec -i` without running `sshd` inside the +containers. + +*(Tip: Start containers with `--ulimit nofile=1048576:1048576` and sufficient +`--sysctl net.core.somaxconn=20000` for high-fan-in tests.)* + +**Docker / Podman (`docker exec -i`):** + +```bash +# Start two containers on a shared bridge network with raised limits: +$ docker network create peregrine-net +$ for name in pg-server-0 pg-client-0; do + docker run -d --rm --name "${name}" --network peregrine-net \ + --ulimit nofile=1048576:1048576 \ + --sysctl net.core.somaxconn=20000 \ + ubuntu:24.04 sleep infinity + done + +# Run the cluster benchmark across the containers: +$ bazelisk run -c opt //peregrine/test/cluster/orchestrator:bipartite_runner -- \ + --client_nodes=pg-client-0 \ + --server_nodes=pg-server-0 \ + --ssh_cmd='sh -c '\''docker exec -i "$0" sh -c "$1"'\''' \ + --clients_per_node=1000 \ + --servers_per_node=1 \ + --transport=tcp \ + --conn=1 \ + --workload=serial_fixed_write \ + --xfer_size=1 \ + --num_xfers=1 \ + --traffic_pattern=all_to_all +``` + +**Kubernetes Pods (`kubectl exec -i`):** + +```bash +$ bazelisk run -c opt //peregrine/test/cluster/orchestrator:bipartite_runner -- \ + --client_nodes=peregrine-client-0,peregrine-client-1 \ + --server_nodes=peregrine-server-0 \ + --ssh_cmd='sh -c '\''kubectl exec -i "$0" -- sh -c "$1"'\''' \ + --clients_per_node=500 \ + --servers_per_node=1 \ + --transport=tcp \ + --conn=1 \ + --workload=serial_fixed_write \ + --xfer_size=1 \ + --num_xfers=1 \ + --traffic_pattern=all_to_all +``` diff --git a/peregrine/test/cluster/node/BUILD b/peregrine/test/cluster/node/BUILD new file mode 100644 index 0000000..501ee78 --- /dev/null +++ b/peregrine/test/cluster/node/BUILD @@ -0,0 +1,215 @@ +# Multi-instance C++ node agent and libraries for Peregrine cluster benchmarks. + +load("@rules_cc//cc:cc_binary.bzl", "cc_binary") +load("@rules_cc//cc:cc_library.bzl", "cc_library") +load("@rules_cc//cc:cc_test.bzl", "cc_test") + +package( + default_testonly = True, + default_visibility = ["//peregrine/test/cluster:__subpackages__"], +) + +cc_library( + name = "config", + hdrs = ["config.h"], + deps = [ + "//peregrine/src/api:transport_types", + "//peregrine/test/workloads:workload_generator", + "@abseil-cpp//absl/time", + ], +) + +cc_library( + name = "topology", + srcs = ["topology.cc"], + hdrs = ["topology.h"], + deps = [ + ":config", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/strings", + "@abseil-cpp//absl/strings:str_format", + "@abseil-cpp//absl/types:span", + ], +) + +cc_test( + name = "topology_test", + srcs = ["topology_test.cc"], + deps = [ + ":config", + ":topology", + "@googletest//:gtest_main", + ], +) + +cc_library( + name = "metrics", + srcs = ["metrics.cc"], + hdrs = ["metrics.h"], + deps = [ + "//peregrine/src/api:transport_metrics", + "@abseil-cpp//absl/time", + ], +) + +cc_test( + name = "metrics_test", + srcs = ["metrics_test.cc"], + deps = [ + ":metrics", + "//peregrine/src/api:transport_metrics", + "@abseil-cpp//absl/time", + "@googletest//:gtest_main", + ], +) + +cc_library( + name = "control_channel", + srcs = ["control_channel.cc"], + hdrs = ["control_channel.h"], + deps = [ + ":config", + ":metrics", + ":topology", + "//peregrine/src/api:transport_metrics", + "@abseil-cpp//absl/base:core_headers", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/strings", + "@abseil-cpp//absl/strings:str_format", + "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/types:span", + ], +) + +cc_test( + name = "control_channel_test", + srcs = ["control_channel_test.cc"], + deps = [ + ":config", + ":control_channel", + ":metrics", + "@googletest//:gtest_main", + ], +) + +cc_library( + name = "instance", + srcs = ["instance.cc"], + hdrs = ["instance.h"], + deps = [ + ":config", + "//peregrine/src/api:transport", + "//peregrine/src/api:transport_metrics", + "//peregrine/src/api:transport_types", + "//peregrine/src/api:transport_util", + "//peregrine/src/util", + "//peregrine/src/util:thread", + "@abseil-cpp//absl/base:core_headers", + "@abseil-cpp//absl/functional:any_invocable", + "@abseil-cpp//absl/functional:function_ref", + "@abseil-cpp//absl/log:check", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/strings", + "@abseil-cpp//absl/strings:str_format", + "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/types:span", + ], +) + +cc_test( + name = "instance_test", + srcs = ["instance_test.cc"], + deps = [ + ":config", + ":instance", + "//peregrine/src/util", + "@googletest//:gtest_main", + ], +) + +cc_library( + name = "cluster_node", + srcs = ["cluster_node.cc"], + hdrs = ["cluster_node.h"], + deps = [ + ":config", + ":control_channel", + ":instance", + ":metrics", + ":topology", + "//peregrine/src/api:transport_metrics", + "//peregrine/src/api:transport_types", + "//peregrine/src/util:thread", + "//peregrine/test/workloads:workload_generator", + "//peregrine/test/workloads:workload_util", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", + "@abseil-cpp//absl/types:span", + ], +) + +cc_test( + name = "cluster_node_test", + srcs = ["cluster_node_test.cc"], + deps = [ + ":cluster_node", + ":config", + ":control_channel", + "//peregrine/src/util", + "//peregrine/src/util:thread", + "//peregrine/test/workloads:serial_fixed_write", + "@abseil-cpp//absl/synchronization", + "@abseil-cpp//absl/time", + "@googletest//:gtest_main", + ], +) + +cc_library( + name = "flags", + srcs = ["flags.cc"], + hdrs = ["flags.h"], + deps = [ + ":config", + ":topology", + "//peregrine/src/api:transport_types", + "//peregrine/src/util:ipaddr", + "//peregrine/src/util:nic", + "//peregrine/test/workloads:workload_generator", + "@abseil-cpp//absl/flags:flag", + "@abseil-cpp//absl/status", + "@abseil-cpp//absl/status:statusor", + "@abseil-cpp//absl/strings", + "@abseil-cpp//absl/strings:str_format", + "@abseil-cpp//absl/time", + ], +) + +cc_test( + name = "flags_test", + srcs = ["flags_test.cc"], + deps = [ + ":config", + ":flags", + "@abseil-cpp//absl/flags:flag", + "@googletest//:gtest_main", + ], +) + +cc_binary( + name = "node", + srcs = ["node_main.cc"], + deps = [ + ":cluster_node", + ":control_channel", + ":flags", + ":instance", + "@abseil-cpp//absl/flags:parse", + "@abseil-cpp//absl/log:initialize", + "@abseil-cpp//absl/log", + "@abseil-cpp//absl/log:check", + ], +) diff --git a/peregrine/test/cluster/node/cluster_node.cc b/peregrine/test/cluster/node/cluster_node.cc new file mode 100644 index 0000000..273e1e4 --- /dev/null +++ b/peregrine/test/cluster/node/cluster_node.cc @@ -0,0 +1,247 @@ +#include "peregrine/test/cluster/node/cluster_node.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "absl/status/status.h" +#include "absl/synchronization/mutex.h" +#include "absl/synchronization/notification.h" +#include "absl/time/time.h" +#include "absl/types/span.h" +#include "peregrine/src/api/transport_types.h" +#include "peregrine/src/util/thread.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/control_channel.h" +#include "peregrine/test/cluster/node/instance.h" +#include "peregrine/test/cluster/node/metrics.h" +#include "peregrine/test/cluster/node/topology.h" +#include "peregrine/test/workloads/workload_generator.h" +#include "peregrine/test/workloads/workload_util.h" + +namespace peregrine::cluster { +namespace { + +struct OutboundStream { + NodeInstance* instance = nullptr; + std::string peer_endpoint; + std::vector requests; + uint32_t total_xfers = 1; + uint32_t completed_xfers = 0; +}; + +std::shared_ptr ResolveWorkload( + const NodeConfig& config) { + if (config.workload_generator != nullptr) { + return config.workload_generator; + } + return workloads::CreateWorkload(config.workload); +} + +std::vector CollectLocalEndpoints( + absl::Span> instances) { + std::vector endpoints; + endpoints.reserve(instances.size()); + for (const auto& inst : instances) { + endpoints.push_back(inst->ControlEndpoint()); + } + return endpoints; +} + +std::vector CollectLocalTargets( + absl::Span> instances) { + std::vector targets; + targets.reserve(instances.size()); + for (const auto& inst : instances) { + targets.push_back(PeerTarget{ + .endpoint = inst->ControlEndpoint(), + .raddr = inst->BufferAddr(), + }); + } + return targets; +} + +std::vector> BuildOutboundStreams( + const NodeConfig& config, const workloads::WorkloadGenerator& workload_gen, + absl::Span> instances, + absl::Span local_endpoints) { + const std::vector> target_assignments = + BuildTargetAssignments(config.node_index, config.num_instances, + config.targets, config.traffic_pattern, + config.exclude_self, local_endpoints); + const uint32_t num_xfers = std::max(1, config.num_xfers); + + std::vector> streams; + for (size_t i = 0; i < instances.size(); ++i) { + for (const PeerTarget& target : target_assignments[i]) { + auto stream = std::make_unique(); + stream->instance = instances[i].get(); + stream->peer_endpoint = target.endpoint; + stream->requests = workload_gen.GenerateRequests( + instances[i]->DataPtr(), reinterpret_cast(target.raddr)); + stream->total_xfers = num_xfers; + streams.push_back(std::move(stream)); + } + } + return streams; +} + +void ExecuteOutboundStreams( + absl::Span> streams) { + if (streams.empty()) { + return; + } + absl::Mutex mu; + std::deque ready_queue; + size_t active_streams = streams.size(); + + for (const auto& stream : streams) { + ready_queue.push_back(stream.get()); + } + + while (true) { + OutboundStream* stream = nullptr; + { + absl::MutexLock lock(mu); + auto has_work_or_done = [&]() { + return !ready_queue.empty() || active_streams == 0; + }; + mu.Await(absl::Condition(&has_work_or_done)); + if (ready_queue.empty()) { + break; + } + stream = ready_queue.front(); + ready_queue.pop_front(); + } + + auto handle_or = stream->instance->PostAsync( + stream->peer_endpoint, stream->requests, + [stream, &mu, &ready_queue, &active_streams](Status s) { + absl::MutexLock lock(mu); + if (s == Status::kSuccess && + ++stream->completed_xfers < stream->total_xfers) { + ready_queue.push_back(stream); + } else { + --active_streams; + } + }); + if (!handle_or.ok()) { + absl::MutexLock lock(mu); + --active_streams; + } + } +} + +NodeMetricsSnapshot CaptureSnapshot( + absl::Span> instances, + uint64_t xfer_size_bytes, const CpuTimer& cpu_timer) { + NodeMetricsSnapshot snap; + snap.num_instances = static_cast(instances.size()); + snap.xfer_size_bytes = xfer_size_bytes; + snap.cpu = cpu_timer.Snapshot(); + for (const auto& inst : instances) { + MergeTransportMetrics(snap.transport, inst->GetTransportMetrics()); + } + return snap; +} + +// RAII background thread that periodically emits `NodeMetricsSnapshot` samples. +class ScopedMetricsSampler { + public: + ScopedMetricsSampler(absl::Duration interval, ControlChannel& channel, + std::function capture_fn) + : interval_(interval), + channel_(channel), + capture_fn_(std::move(capture_fn)) { + if (interval_ > absl::ZeroDuration()) { + thread_ = util::Thread([this]() { + while (!stop_.WaitForNotificationWithTimeout(interval_)) { + channel_.EmitMetricsSample(capture_fn_()); + } + }); + } + } + + ~ScopedMetricsSampler() { Stop(); } + + void Stop() { + if (!stop_.HasBeenNotified()) { + stop_.Notify(); + } + if (thread_.joinable()) { + thread_.join(); + } + } + + private: + absl::Duration interval_; + ControlChannel& channel_; + std::function capture_fn_; + absl::Notification stop_; + util::Thread thread_; +}; + +} // namespace + +absl::Status ClusterNode::Run(const NodeConfig& config, + ControlChannel& channel) { + // Phase 1: Initialize workload generator, memory buffer(s), and Transports. + const std::shared_ptr workload_gen = + ResolveWorkload(config); + const uint64_t xfer_size = + std::max(1, workload_gen->TotalSizeBytes()); + + std::unique_ptr shared_buf; + if (config.share_buffer) { + shared_buf = std::make_unique(xfer_size); + } + + auto instances_or = CreateInstances(config, shared_buf.get(), xfer_size); + if (!instances_or.ok()) { + return instances_or.status(); + } + std::vector> instances = + std::move(*instances_or); + + const std::vector local_endpoints = + CollectLocalEndpoints(instances); + const std::vector local_targets = CollectLocalTargets(instances); + std::vector> streams = + BuildOutboundStreams(config, *workload_gen, instances, local_endpoints); + + channel.EmitNodeReady(local_targets); + + // Phase 2: Wait for synchronized START signal. + if (!channel.WaitForCommand(kStartCommand)) { + DestroyInstances(instances); + return absl::CancelledError("Received EOF while waiting for START command"); + } + + // Phase 3: Execute outbound workload streams (if any) + periodic sampling. + CpuTimer cpu_timer; + auto snapshot_fn = [&]() { + return CaptureSnapshot(instances, xfer_size, cpu_timer); + }; + ScopedMetricsSampler sampler(config.metrics_interval, channel, snapshot_fn); + + if (!streams.empty()) { + ExecuteOutboundStreams(streams); + channel.EmitWorkloadDone(snapshot_fn()); + } + + // Phase 4: Wait for STOP signal before tearing down Transport listeners. + channel.WaitForCommand(kStopCommand); + sampler.Stop(); + channel.EmitNodeDone(snapshot_fn()); + + DestroyInstances(instances); + return absl::OkStatus(); +} + +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/cluster_node.h b/peregrine/test/cluster/node/cluster_node.h new file mode 100644 index 0000000..96245e0 --- /dev/null +++ b/peregrine/test/cluster/node/cluster_node.h @@ -0,0 +1,28 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_CLUSTER_NODE_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_CLUSTER_NODE_H_ + +#include "absl/status/status.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/control_channel.h" + +namespace peregrine::cluster { + +// Symmetric multi-instance cluster node engine executing the 4-phase lifecycle: +// Phase 1 (Init): Create 1..P Transport instances, register memory, +// pre-generate outbound Request batches, and emit +// PEREGRINE_NODE_READY. +// Phase 2 (WaitStart): Wait for "START" on `channel`. +// Phase 3 (RunWorkload): Start CpuTimer & periodic metrics sampler. If +// outbound streams exist, dispatch PostAsync() across +// all streams, wait for completion, and emit +// PEREGRINE_WORKLOAD_DONE. +// Phase 4 (WaitStop): Wait for "STOP" on `channel`, stop sampler, emit +// PEREGRINE_NODE_DONE, and tear down cleanly. +class ClusterNode { + public: + static absl::Status Run(const NodeConfig& config, ControlChannel& channel); +}; + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_CLUSTER_NODE_H_ diff --git a/peregrine/test/cluster/node/cluster_node_test.cc b/peregrine/test/cluster/node/cluster_node_test.cc new file mode 100644 index 0000000..60362a7 --- /dev/null +++ b/peregrine/test/cluster/node/cluster_node_test.cc @@ -0,0 +1,187 @@ +#include "peregrine/test/cluster/node/cluster_node.h" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "gmock/gmock.h" +#include "gtest/gtest.h" +#include "absl/synchronization/mutex.h" +#include "absl/time/time.h" +#include "peregrine/src/util/thread.h" +#include "peregrine/src/util/util.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/control_channel.h" +#include "peregrine/test/workloads/serial_fixed_write.h" + +namespace peregrine::cluster { +namespace { + +using ::testing::HasSubstr; + +// Simple thread-safe line pipe for driving ControlChannel in-process. +class InMemoryPipe : public std::streambuf { + public: + void WriteLine(std::string_view line) { + absl::MutexLock lock(mu_); + lines_.emplace_back(line); + lines_.back().push_back('\n'); + } + + void Close() { + absl::MutexLock lock(mu_); + closed_ = true; + } + + protected: + int_type underflow() override { + absl::MutexLock lock(mu_); + auto has_data_or_closed = [this]() { return !lines_.empty() || closed_; }; + mu_.Await(absl::Condition(&has_data_or_closed)); + if (lines_.empty()) { + return traits_type::eof(); + } + current_ = lines_.front(); + lines_.erase(lines_.begin()); + setg(current_.data(), current_.data(), current_.data() + current_.size()); + return traits_type::to_int_type(*gptr()); + } + + private: + absl::Mutex mu_; + std::vector lines_; + std::string current_; + bool closed_ = false; +}; + +class PipeOStream : public std::streambuf { + public: + explicit PipeOStream(InMemoryPipe* pipe) : pipe_(pipe) {} + + protected: + int_type overflow(int_type ch) override { + if (ch != traits_type::eof()) { + char c = traits_type::to_char_type(ch); + if (c == '\n') { + pipe_->WriteLine(buf_); + buf_.clear(); + } else { + buf_.push_back(c); + } + } + return ch; + } + + private: + InMemoryPipe* pipe_; + std::string buf_; +}; + +void RunFanInLoopbackTest(bool share_buffer) { + NodeConfig server_cfg; + server_cfg.ip = "127.0.0.1"; + server_cfg.num_instances = 2; + server_cfg.base_control_port = util::FindFreePort(AF_INET, /*tcp=*/true); + server_cfg.num_conns = 1; + server_cfg.workload_generator = + std::make_shared(1); + server_cfg.share_buffer = share_buffer; + server_cfg.metrics_interval = absl::ZeroDuration(); + + InMemoryPipe server_in_buf; + std::istream server_in(&server_in_buf); + InMemoryPipe server_out_buf; + std::istream server_out_reader(&server_out_buf); + PipeOStream server_out_writer(&server_out_buf); + std::ostream server_out(&server_out_writer); + ControlChannel server_channel(server_in, server_out); + + util::Thread server_thread([&]() { + EXPECT_TRUE(ClusterNode::Run(server_cfg, server_channel).ok()); + server_out_buf.Close(); + }); + + std::string server_ready_line; + ASSERT_TRUE(std::getline(server_out_reader, server_ready_line)); + auto server_targets_or = ParseNodeReadyLine(server_ready_line); + ASSERT_TRUE(server_targets_or.ok()) << server_targets_or.status(); + ASSERT_EQ(server_targets_or->size(), 2); + if (share_buffer) { + EXPECT_EQ((*server_targets_or)[0].raddr, (*server_targets_or)[1].raddr); + } else { + EXPECT_NE((*server_targets_or)[0].raddr, (*server_targets_or)[1].raddr); + } + + NodeConfig client_cfg; + client_cfg.ip = "127.0.0.1"; + client_cfg.num_instances = 4; + client_cfg.base_control_port = + static_cast(server_cfg.base_control_port + 50); + client_cfg.num_conns = 1; + client_cfg.workload_generator = + std::make_shared(1); + client_cfg.targets = *server_targets_or; + client_cfg.traffic_pattern = TrafficPattern::kAllToAll; + client_cfg.num_xfers = 1; + client_cfg.share_buffer = share_buffer; + client_cfg.metrics_interval = absl::ZeroDuration(); + + InMemoryPipe client_in_buf; + std::istream client_in(&client_in_buf); + InMemoryPipe client_out_buf; + std::istream client_out_reader(&client_out_buf); + PipeOStream client_out_writer(&client_out_buf); + std::ostream client_out(&client_out_writer); + ControlChannel client_channel(client_in, client_out); + + util::Thread client_thread([&]() { + EXPECT_TRUE(ClusterNode::Run(client_cfg, client_channel).ok()); + client_out_buf.Close(); + }); + + std::string client_ready_line; + ASSERT_TRUE(std::getline(client_out_reader, client_ready_line)); + ASSERT_THAT(client_ready_line, HasSubstr(kNodeReadyPrefix)); + + server_in_buf.WriteLine(kStartCommand); + client_in_buf.WriteLine(kStartCommand); + + std::string workload_done_line; + ASSERT_TRUE(std::getline(client_out_reader, workload_done_line)); + EXPECT_THAT(workload_done_line, HasSubstr(kWorkloadDonePrefix)); + EXPECT_THAT(workload_done_line, HasSubstr(R"("bytes":8,"errors":0)")); + EXPECT_THAT(workload_done_line, HasSubstr(R"("count":8)")); + + client_in_buf.WriteLine(kStopCommand); + server_in_buf.WriteLine(kStopCommand); + + std::string client_done_line; + ASSERT_TRUE(std::getline(client_out_reader, client_done_line)); + EXPECT_THAT(client_done_line, HasSubstr("PEREGRINE_NODE_DONE ")); + + std::string server_done_line; + ASSERT_TRUE(std::getline(server_out_reader, server_done_line)); + EXPECT_THAT(server_done_line, HasSubstr("PEREGRINE_NODE_DONE ")); + EXPECT_THAT(server_done_line, HasSubstr(R"("rpc_requests_received":8)")); + + client_thread.join(); + server_thread.join(); +} + +TEST(ClusterNodeTest, MultiInstanceFanInOnLoopbackSharedBuffer) { + RunFanInLoopbackTest(/*share_buffer=*/true); +} + +TEST(ClusterNodeTest, MultiInstanceFanInOnLoopbackPerInstanceBuffers) { + RunFanInLoopbackTest(/*share_buffer=*/false); +} + +} // namespace +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/config.h b/peregrine/test/cluster/node/config.h new file mode 100644 index 0000000..c99b0b0 --- /dev/null +++ b/peregrine/test/cluster/node/config.h @@ -0,0 +1,54 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_CONFIG_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_CONFIG_H_ + +#include +#include +#include +#include + +#include "absl/time/time.h" +#include "peregrine/src/api/transport_types.h" +#include "peregrine/test/workloads/workload_generator.h" + +namespace peregrine::cluster { + +// Supported cluster traffic routing patterns. +enum class TrafficPattern { + kAllToAll, + kRoundRobin, +}; + +// Remote Peregrine instance target endpoint and registered buffer address. +struct PeerTarget { + std::string endpoint; + uint64_t raddr = 0; + + friend bool operator==(const PeerTarget& a, const PeerTarget& b) { + return a.endpoint == b.endpoint && a.raddr == b.raddr; + } +}; + +// Configuration for a single cluster node process (running 1..P instances). +struct NodeConfig { + std::string ip = "127.0.0.1"; + TransportType transport_type = TransportType::kTcp; + int num_instances = 1; + int node_index = 0; + uint16_t base_control_port = 10000; + int num_conns = 1; + workloads::WorkloadType workload = workloads::WorkloadType::kSerialFixedWrite; + // Optional pre-constructed workload generator (used in flag-free unit tests; + // when null, ClusterNode instantiates from `workload` via CreateWorkload()). + std::shared_ptr workload_generator = + nullptr; + std::vector targets; + TrafficPattern traffic_pattern = TrafficPattern::kAllToAll; + bool exclude_self = false; + uint32_t num_xfers = 1; + bool share_buffer = true; + absl::Duration metrics_interval = absl::Seconds(1); +}; + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_CONFIG_H_ diff --git a/peregrine/test/cluster/node/control_channel.cc b/peregrine/test/cluster/node/control_channel.cc new file mode 100644 index 0000000..3dc24ee --- /dev/null +++ b/peregrine/test/cluster/node/control_channel.cc @@ -0,0 +1,135 @@ +#include "peregrine/test/cluster/node/control_channel.h" + +#include +#include +#include +#include + +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/ascii.h" +#include "absl/strings/match.h" +#include "absl/strings/str_cat.h" +#include "absl/strings/str_format.h" +#include "absl/strings/str_join.h" +#include "absl/strings/str_split.h" +#include "absl/synchronization/mutex.h" +#include "absl/types/span.h" +#include "peregrine/src/api/transport_metrics.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/metrics.h" +#include "peregrine/test/cluster/node/topology.h" + +namespace peregrine::cluster { +namespace { + +std::string FormatHistogramJson(const Log2Histogram<32>& h) { + return absl::StrFormat(R"({"sum":%u,"count":%u,"buckets":[%s]})", h.sum, + h.Count(), absl::StrJoin(h.buckets, ",")); +} + +std::string FormatOpMetricsJson(const OpMetrics& op) { + return absl::StrFormat( + R"({"bytes":%u,"errors":%u,"e2e_latency_us":%s,"request_size_bytes":%s})", + op.bytes, op.errors, FormatHistogramJson(op.e2e_latency_us), + FormatHistogramJson(op.request_size_bytes)); +} + +std::string FormatCpuStatsJson(const CpuStats& cpu) { + return absl::StrFormat( + R"({"wall_ms":%.4f,"user_cpu_ms":%.4f,"sys_cpu_ms":%.4f,"avg_cores":%.4f})", + cpu.wall_ms, cpu.user_cpu_ms, cpu.sys_cpu_ms, cpu.avg_cores); +} + +std::string FormatTransportMetricsJson(const TransportMetrics& t) { + return absl::StrFormat( + R"({"write":%s,"read":%s,"tcp_connect_failures":%u,"rpc_requests_received":%u})", + FormatOpMetricsJson(t.write), FormatOpMetricsJson(t.read), + t.tcp_connect_failures, t.rpc_requests_received); +} + +} // namespace + +std::string FormatNodeMetricsJson(const NodeMetricsSnapshot& s) { + return absl::StrFormat( + R"({"num_instances":%d,"xfer_size_bytes":%u,"cpu":%s,"transport":%s})", + s.num_instances, s.xfer_size_bytes, FormatCpuStatsJson(s.cpu), + FormatTransportMetricsJson(s.transport)); +} + +absl::StatusOr> ParseNodeReadyLine( + std::string_view line) { + std::string_view trimmed = absl::StripAsciiWhitespace(line); + if (!absl::StartsWith(trimmed, kNodeReadyPrefix)) { + return absl::InvalidArgumentError( + absl::StrCat("Missing ", kNodeReadyPrefix, " prefix in: ", line)); + } + + const auto tg_start = trimmed.find("\"targets\":["); + if (tg_start == std::string_view::npos) { + return absl::InvalidArgumentError( + "Missing \"targets\" array in ready line"); + } + std::string_view tg_part = trimmed.substr(tg_start + 11); + const auto tg_end = tg_part.find(']'); + if (tg_end == std::string_view::npos) { + return absl::InvalidArgumentError( + "Unterminated \"targets\" array in ready line"); + } + tg_part = tg_part.substr(0, tg_end); + + std::vector entries; + for (std::string_view tok : absl::StrSplit(tg_part, ',', absl::SkipEmpty())) { + tok = absl::StripAsciiWhitespace(tok); + if (tok.size() >= 2 && tok.front() == '"' && tok.back() == '"') { + tok = tok.substr(1, tok.size() - 2); + } + if (!tok.empty()) { + entries.push_back(tok); + } + } + if (entries.empty()) { + return absl::InvalidArgumentError("Empty \"targets\" array in ready line"); + } + return ParseTargets(absl::StrJoin(entries, ",")); +} + +void ControlChannel::EmitNodeReady(absl::Span targets) { + std::string joined = + absl::StrJoin(targets, ",", [](std::string* out, const PeerTarget& t) { + absl::StrAppendFormat(out, "\"%s@%u\"", t.endpoint, t.raddr); + }); + EmitLine(kNodeReadyPrefix, absl::StrFormat(R"({"targets":[%s]})", joined)); +} + +bool ControlChannel::WaitForCommand(std::string_view expected_cmd) { + std::string line; + while (std::getline(in_, line)) { + std::string_view trimmed = absl::StripAsciiWhitespace(line); + if (trimmed == expected_cmd) { + return true; + } + } + return false; +} + +void ControlChannel::EmitMetricsSample(const NodeMetricsSnapshot& snapshot) { + EmitLine(kMetricsSamplePrefix, FormatNodeMetricsJson(snapshot)); +} + +void ControlChannel::EmitWorkloadDone(const NodeMetricsSnapshot& snapshot) { + EmitLine(kWorkloadDonePrefix, FormatNodeMetricsJson(snapshot)); +} + +void ControlChannel::EmitNodeDone(const NodeMetricsSnapshot& snapshot) { + EmitLine(kNodeDonePrefix, FormatNodeMetricsJson(snapshot)); +} + +void ControlChannel::EmitLine(std::string_view prefix, + std::string_view payload) { + absl::MutexLock lock(out_mu_); + out_ << prefix << payload << "\n"; + out_.flush(); +} + +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/control_channel.h b/peregrine/test/cluster/node/control_channel.h new file mode 100644 index 0000000..a48cc81 --- /dev/null +++ b/peregrine/test/cluster/node/control_channel.h @@ -0,0 +1,71 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_CONTROL_CHANNEL_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_CONTROL_CHANNEL_H_ + +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/status/statusor.h" +#include "absl/synchronization/mutex.h" +#include "absl/types/span.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/metrics.h" + +namespace peregrine::cluster { + +inline constexpr std::string_view kNodeReadyPrefix = "PEREGRINE_NODE_READY "; +inline constexpr std::string_view kMetricsSamplePrefix = + "PEREGRINE_METRICS_SAMPLE "; +inline constexpr std::string_view kWorkloadDonePrefix = + "PEREGRINE_WORKLOAD_DONE "; +inline constexpr std::string_view kNodeDonePrefix = "PEREGRINE_NODE_DONE "; + +inline constexpr std::string_view kStartCommand = "START"; +inline constexpr std::string_view kStopCommand = "STOP"; + +// Serializes a `NodeMetricsSnapshot` into a single-line JSON string. +std::string FormatNodeMetricsJson(const NodeMetricsSnapshot& snapshot); + +// Parses a `PEREGRINE_NODE_READY {"targets":["@",...]}` line into a +// list of `PeerTarget` structs. +absl::StatusOr> ParseNodeReadyLine( + std::string_view line); + +// Thread-safe stdio control channel used by `ClusterNode` to coordinate the +// 4-phase barrier lifecycle (`NODE_READY` -> `START` -> `WORKLOAD_DONE` -> +// `STOP` -> `NODE_DONE`) with the Python orchestrator. +class ControlChannel { + public: + explicit ControlChannel(std::istream& in = std::cin, + std::ostream& out = std::cout) + : in_(in), out_(out) {} + + // Phase 1: Emits `PEREGRINE_NODE_READY {"targets":["@",...]}\n`. + void EmitNodeReady(absl::Span targets); + + // Phase 2 & 4: Blocks reading lines from `in_` until `expected_cmd` (e.g. + // "START" or "STOP") is received, or returns false on EOF. + bool WaitForCommand(std::string_view expected_cmd); + + // Phase 3 (periodic): Emits `PEREGRINE_METRICS_SAMPLE \n`. + void EmitMetricsSample(const NodeMetricsSnapshot& snapshot); + + // Phase 3 (completion): Emits `PEREGRINE_WORKLOAD_DONE \n`. + void EmitWorkloadDone(const NodeMetricsSnapshot& snapshot); + + // Phase 4 (teardown): Emits `PEREGRINE_NODE_DONE \n`. + void EmitNodeDone(const NodeMetricsSnapshot& snapshot); + + private: + void EmitLine(std::string_view prefix, std::string_view payload); + + std::istream& in_; + absl::Mutex out_mu_; + std::ostream& out_ ABSL_GUARDED_BY(out_mu_); +}; + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_CONTROL_CHANNEL_H_ diff --git a/peregrine/test/cluster/node/control_channel_test.cc b/peregrine/test/cluster/node/control_channel_test.cc new file mode 100644 index 0000000..8542892 --- /dev/null +++ b/peregrine/test/cluster/node/control_channel_test.cc @@ -0,0 +1,71 @@ +#include "peregrine/test/cluster/node/control_channel.h" + +#include +#include +#include + +#include "gmock/gmock.h" +#include "gtest/gtest.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/metrics.h" + +namespace peregrine::cluster { +namespace { + +using ::testing::HasSubstr; +using ::testing::StartsWith; + +TEST(ControlChannelTest, EmitAndParseNodeReadyAndWaitForCommands) { + std::stringstream in("noise\n START \nSTOP\n"); + std::stringstream out; + ControlChannel channel(in, out); + + std::vector targets = { + PeerTarget{.endpoint = "127.0.0.1:10000", .raddr = 987654321ULL}, + PeerTarget{.endpoint = "127.0.0.1:10001", .raddr = 123456789ULL}, + }; + channel.EmitNodeReady(targets); + + EXPECT_THAT( + out.str(), + StartsWith( + "PEREGRINE_NODE_READY " + R"({"targets":["127.0.0.1:10000@987654321","127.0.0.1:10001@123456789"]})")); + + auto parsed_or = ParseNodeReadyLine(out.str()); + ASSERT_TRUE(parsed_or.ok()) << parsed_or.status(); + ASSERT_EQ(parsed_or->size(), 2); + EXPECT_EQ((*parsed_or)[0].endpoint, "127.0.0.1:10000"); + EXPECT_EQ((*parsed_or)[0].raddr, 987654321ULL); + EXPECT_EQ((*parsed_or)[1].endpoint, "127.0.0.1:10001"); + EXPECT_EQ((*parsed_or)[1].raddr, 123456789ULL); + + EXPECT_TRUE(channel.WaitForCommand(kStartCommand)); + EXPECT_TRUE(channel.WaitForCommand(kStopCommand)); + EXPECT_FALSE(channel.WaitForCommand("EXTRA")); +} + +TEST(ControlChannelTest, EmitMetricsAndDoneLines) { + std::stringstream in; + std::stringstream out; + ControlChannel channel(in, out); + + NodeMetricsSnapshot snap; + snap.num_instances = 4; + snap.xfer_size_bytes = 1; + snap.transport.write.bytes = 4; + snap.transport.rpc_requests_received = 4; + + channel.EmitMetricsSample(snap); + channel.EmitWorkloadDone(snap); + channel.EmitNodeDone(snap); + + const std::string output = out.str(); + EXPECT_THAT(output, HasSubstr("PEREGRINE_METRICS_SAMPLE {")); + EXPECT_THAT(output, HasSubstr("PEREGRINE_WORKLOAD_DONE {")); + EXPECT_THAT(output, HasSubstr("PEREGRINE_NODE_DONE {")); + EXPECT_THAT(output, HasSubstr(R"("rpc_requests_received":4)")); +} + +} // namespace +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/flags.cc b/peregrine/test/cluster/node/flags.cc new file mode 100644 index 0000000..2957118 --- /dev/null +++ b/peregrine/test/cluster/node/flags.cc @@ -0,0 +1,174 @@ +#include "peregrine/test/cluster/node/flags.h" + +#include +#include +#include +#include +#include +#include + +#include "absl/flags/flag.h" +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/ascii.h" +#include "absl/strings/match.h" +#include "absl/strings/str_cat.h" +#include "absl/strings/str_format.h" +#include "absl/strings/str_join.h" +#include "absl/time/time.h" +#include "peregrine/src/api/transport_types.h" +#include "peregrine/src/util/ipaddr.h" +#include "peregrine/src/util/nic.h" +#include "peregrine/test/cluster/node/config.h" +#include "peregrine/test/cluster/node/topology.h" +#include "peregrine/test/workloads/workload_generator.h" + +ABSL_FLAG(std::string, ip, "127.0.0.1", + "Local IP address to bind Peregrine instances to (supports loopback " + "127.0.0.1 / ::1 and physical NIC IPs)."); + +ABSL_FLAG(int, num_instances, 1, + "Number of Peregrine Transport instances to run in this process."); + +ABSL_FLAG(int, node_index, 0, + "0-based index of this node within its sender group."); + +ABSL_FLAG(uint16_t, base_control_port, 10000, + "Starting port for deterministic Peregrine gRPC control plane port " + "assignment."); + +ABSL_FLAG(std::string, transport, "tcp", + "Transport type to use ('tcp' or 'rdma')."); + +ABSL_FLAG(int, conn, 1, + "Number of data channels per peer (num_conns_per_peer)."); + +ABSL_FLAG(std::string, workload, "serial_fixed_write", + "Workload generator ('serial_fixed_write' or 'kv_cache')."); + +ABSL_FLAG(std::string, targets, "", + "Comma-separated list of target instances in the form " + "':@' (empty for passive server " + "nodes)."); + +ABSL_FLAG(std::string, traffic_pattern, "all_to_all", + "Traffic routing pattern ('all_to_all' or 'round_robin')."); + +ABSL_FLAG(bool, exclude_self, false, + "Exclude local instance's own endpoint when assigning targets."); + +ABSL_FLAG(uint32_t, num_xfers, 1, + "Number of transfer iterations per (local_instance, target) stream."); + +ABSL_FLAG(bool, share_buffer, true, + "Register one shared memory buffer across all instances in this " + "process."); + +ABSL_FLAG(absl::Duration, metrics_interval, absl::Seconds(1), + "Periodic sampling interval for TransportMetrics and CPU usage."); + +namespace peregrine::cluster { +namespace { + +absl::StatusOr ParseTransportFlag(std::string_view raw) { + std::string_view s = absl::StripAsciiWhitespace(raw); + if (absl::EqualsIgnoreCase(s, "tcp")) { + return TransportType::kTcp; + } + if (absl::EqualsIgnoreCase(s, "rdma")) { + return TransportType::kRdma; + } + return absl::InvalidArgumentError(absl::StrCat( + "Invalid --transport '", raw, "': expected 'tcp' or 'rdma'")); +} + +absl::StatusOr ParseWorkloadFlag( + std::string_view raw) { + std::string_view s = absl::StripAsciiWhitespace(raw); + if (absl::EqualsIgnoreCase(s, "serial_fixed_write")) { + return workloads::WorkloadType::kSerialFixedWrite; + } + if (absl::EqualsIgnoreCase(s, "kv_cache")) { + return workloads::WorkloadType::kKvCache; + } + return absl::InvalidArgumentError( + absl::StrCat("Invalid --workload '", raw, + "': expected 'serial_fixed_write' or 'kv_cache'")); +} + +} // namespace + +absl::StatusOr ValidateIp(std::string_view ip) { + ip = absl::StripAsciiWhitespace(ip); + if (ip.empty()) { + return absl::InvalidArgumentError("--ip must not be empty"); + } + const auto parsed = util::IpAddr::Create(ip); + if (!parsed.has_value()) { + return absl::InvalidArgumentError( + absl::StrCat("--ip '", ip, "' is not a valid IPv4 or IPv6 address")); + } + if (parsed->IsZero()) { + return absl::InvalidArgumentError( + absl::StrCat("--ip cannot be a wildcard address ('", ip, "')")); + } + if (parsed->IsLoopback()) { + return std::string(ip); + } + + const auto nics = util::EnumerateNics(); + std::vector available; + for (const auto& [ifname, ips] : nics) { + for (const auto& if_ip : ips) { + available.push_back(absl::StrFormat("%s: %s", ifname, if_ip)); + if (if_ip == ip || util::IpAddr::Create(if_ip) == *parsed) { + return std::string(ip); + } + } + } + return absl::InvalidArgumentError(absl::StrCat( + "Specified --ip '", ip, + "' does not match any local interface on this machine. Available:\n ", + absl::StrJoin(available, "\n "))); +} + +absl::StatusOr ReadNodeConfig() { + NodeConfig cfg; + + auto ip_or = ValidateIp(absl::GetFlag(FLAGS_ip)); + if (!ip_or.ok()) return ip_or.status(); + cfg.ip = std::move(*ip_or); + + auto transport_or = ParseTransportFlag(absl::GetFlag(FLAGS_transport)); + if (!transport_or.ok()) return transport_or.status(); + cfg.transport_type = *transport_or; + + cfg.num_instances = absl::GetFlag(FLAGS_num_instances); + if (cfg.num_instances <= 0) { + return absl::InvalidArgumentError("--num_instances must be > 0"); + } + + cfg.node_index = std::max(0, absl::GetFlag(FLAGS_node_index)); + cfg.base_control_port = absl::GetFlag(FLAGS_base_control_port); + cfg.num_conns = std::clamp(absl::GetFlag(FLAGS_conn), 1, 100); + + auto workload_or = ParseWorkloadFlag(absl::GetFlag(FLAGS_workload)); + if (!workload_or.ok()) return workload_or.status(); + cfg.workload = *workload_or; + + auto targets_or = ParseTargets(absl::GetFlag(FLAGS_targets)); + if (!targets_or.ok()) return targets_or.status(); + cfg.targets = std::move(*targets_or); + + auto pattern_or = ParseTrafficPattern(absl::GetFlag(FLAGS_traffic_pattern)); + if (!pattern_or.ok()) return pattern_or.status(); + cfg.traffic_pattern = *pattern_or; + + cfg.exclude_self = absl::GetFlag(FLAGS_exclude_self); + cfg.num_xfers = std::max(1, absl::GetFlag(FLAGS_num_xfers)); + cfg.share_buffer = absl::GetFlag(FLAGS_share_buffer); + cfg.metrics_interval = absl::GetFlag(FLAGS_metrics_interval); + return cfg; +} + +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/flags.h b/peregrine/test/cluster/node/flags.h new file mode 100644 index 0000000..256dbb9 --- /dev/null +++ b/peregrine/test/cluster/node/flags.h @@ -0,0 +1,38 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_FLAGS_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_FLAGS_H_ + +#include +#include +#include + +#include "absl/flags/declare.h" +#include "absl/status/statusor.h" +#include "absl/time/time.h" +#include "peregrine/test/cluster/node/config.h" + +ABSL_DECLARE_FLAG(std::string, ip); +ABSL_DECLARE_FLAG(int, num_instances); +ABSL_DECLARE_FLAG(int, node_index); +ABSL_DECLARE_FLAG(uint16_t, base_control_port); +ABSL_DECLARE_FLAG(std::string, transport); +ABSL_DECLARE_FLAG(int, conn); +ABSL_DECLARE_FLAG(std::string, workload); +ABSL_DECLARE_FLAG(std::string, targets); +ABSL_DECLARE_FLAG(std::string, traffic_pattern); +ABSL_DECLARE_FLAG(bool, exclude_self); +ABSL_DECLARE_FLAG(uint32_t, num_xfers); +ABSL_DECLARE_FLAG(bool, share_buffer); +ABSL_DECLARE_FLAG(absl::Duration, metrics_interval); + +namespace peregrine::cluster { + +// Validates that `ip` is a non-empty, non-wildcard address belonging to a +// local network interface (including loopback `127.0.0.1` and `::1`). +absl::StatusOr ValidateIp(std::string_view ip); + +// Reads and validates CLI flags into a `NodeConfig`. +absl::StatusOr ReadNodeConfig(); + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_FLAGS_H_ diff --git a/peregrine/test/cluster/node/flags_test.cc b/peregrine/test/cluster/node/flags_test.cc new file mode 100644 index 0000000..83c90d1 --- /dev/null +++ b/peregrine/test/cluster/node/flags_test.cc @@ -0,0 +1,51 @@ +#include "peregrine/test/cluster/node/flags.h" + +#include "gtest/gtest.h" +#include "absl/flags/flag.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { +namespace { + +TEST(FlagsTest, ValidateIpAllowsLoopbackAndRejectsWildcardOrUnknown) { + auto v4_loop = ValidateIp("127.0.0.1"); + ASSERT_TRUE(v4_loop.ok()); + EXPECT_EQ(*v4_loop, "127.0.0.1"); + + auto v6_loop = ValidateIp("::1"); + ASSERT_TRUE(v6_loop.ok()); + EXPECT_EQ(*v6_loop, "::1"); + + EXPECT_FALSE(ValidateIp("").ok()); + EXPECT_FALSE(ValidateIp("0.0.0.0").ok()); + EXPECT_FALSE(ValidateIp("::").ok()); + EXPECT_FALSE(ValidateIp("203.0.113.254").ok()); +} + +TEST(FlagsTest, ReadNodeConfigParsesFlags) { + absl::SetFlag(&FLAGS_ip, "127.0.0.1"); + absl::SetFlag(&FLAGS_num_instances, 16); + absl::SetFlag(&FLAGS_node_index, 2); + absl::SetFlag(&FLAGS_base_control_port, 20000); + absl::SetFlag(&FLAGS_transport, "tcp"); + absl::SetFlag(&FLAGS_conn, 1); + absl::SetFlag(&FLAGS_workload, "serial_fixed_write"); + absl::SetFlag(&FLAGS_targets, "127.0.0.1:10000@123456"); + absl::SetFlag(&FLAGS_traffic_pattern, "round_robin"); + absl::SetFlag(&FLAGS_num_xfers, 5); + + auto cfg = ReadNodeConfig(); + ASSERT_TRUE(cfg.ok()) << cfg.status(); + EXPECT_EQ(cfg->ip, "127.0.0.1"); + EXPECT_EQ(cfg->num_instances, 16); + EXPECT_EQ(cfg->node_index, 2); + EXPECT_EQ(cfg->base_control_port, 20000); + EXPECT_EQ(cfg->traffic_pattern, TrafficPattern::kRoundRobin); + EXPECT_EQ(cfg->num_xfers, 5); + ASSERT_EQ(cfg->targets.size(), 1); + EXPECT_EQ(cfg->targets[0].endpoint, "127.0.0.1:10000"); + EXPECT_EQ(cfg->targets[0].raddr, 123456); +} + +} // namespace +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/instance.cc b/peregrine/test/cluster/node/instance.cc new file mode 100644 index 0000000..9402b43 --- /dev/null +++ b/peregrine/test/cluster/node/instance.cc @@ -0,0 +1,266 @@ +#include "peregrine/test/cluster/node/instance.h" + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "absl/functional/any_invocable.h" +#include "absl/functional/function_ref.h" +#include "absl/log/check.h" +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/match.h" +#include "absl/strings/str_cat.h" +#include "absl/strings/str_format.h" +#include "absl/synchronization/mutex.h" +#include "absl/types/span.h" +#include "peregrine/src/api/transport.h" +#include "peregrine/src/api/transport_types.h" +#include "peregrine/src/api/transport_util.h" +#include "peregrine/src/util/thread.h" +#include "peregrine/src/util/util.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { +namespace { + +constexpr int kMaxCreateThreads = 32; +constexpr int kMaxDestroyThreads = 64; +constexpr int kMaxPortBindAttempts = 128; + +std::string FormatEndpoint(std::string_view ip, uint16_t port) { + if (absl::StrContains(ip, ':')) { + return absl::StrFormat("[%s]:%u", ip, port); + } + return absl::StrFormat("%s:%u", ip, port); +} + +// Executes `body(idx)` for `idx` in `[0, total_items)` across up to +// `max_threads` worker threads, stopping early on the first non-OK status. +absl::Status ParallelFor(int total_items, int max_threads, + absl::FunctionRef body) { + if (total_items <= 0) { + return absl::OkStatus(); + } + const int num_threads = std::min(total_items, std::max(1, max_threads)); + std::atomic next_index{0}; + absl::Mutex err_mu; + absl::Status first_error = absl::OkStatus(); + + std::vector workers; + workers.reserve(num_threads); + for (int t = 0; t < num_threads; ++t) { + workers.emplace_back([&]() { + while (true) { + const int idx = next_index.fetch_add(1, std::memory_order_relaxed); + if (idx >= total_items) { + break; + } + { + absl::MutexLock lock(err_mu); + if (!first_error.ok()) break; + } + absl::Status s = body(idx); + if (!s.ok()) { + absl::MutexLock lock(err_mu); + if (first_error.ok()) { + first_error = std::move(s); + } + break; + } + } + }); + } + for (util::Thread& w : workers) { + w.join(); + } + return first_error; +} + +absl::StatusOr>> +BindTransportWithRetry(int idx, const NodeConfig& config, + std::atomic& next_port) { + for (int attempt = 0; attempt < kMaxPortBindAttempts; ++attempt) { + const uint32_t candidate = + next_port.fetch_add(1, std::memory_order_relaxed); + if (candidate > 65535U) { + break; + } + std::string endpoint = + FormatEndpoint(config.ip, static_cast(candidate)); + std::unique_ptr transport = + CreateTransport(endpoint, config.transport_type, config.num_conns); + if (transport != nullptr) { + return std::make_pair(std::move(endpoint), std::move(transport)); + } + } + return absl::InternalError(absl::StrCat( + "Failed to bind Transport instance ", idx, " on IP ", config.ip, + " starting from port ", config.base_control_port)); +} + +absl::StatusOr> CreateSingleInstance( + int idx, const NodeConfig& config, RegisteredBuffer* shared_buf, + size_t buf_size, std::atomic& next_port) { + auto bound_or = BindTransportWithRetry(idx, config, next_port); + if (!bound_or.ok()) { + return bound_or.status(); + } + auto [bound_endpoint, transport] = std::move(*bound_or); + + std::unique_ptr owned_buf; + RegisteredBuffer* target_buf = shared_buf; + if (target_buf == nullptr) { + owned_buf = std::make_unique(buf_size); + target_buf = owned_buf.get(); + } + + absl::Status reg_status = target_buf->RegisterWith(transport.get()); + if (!reg_status.ok()) { + return reg_status; + } + + return std::make_unique(std::move(bound_endpoint), + std::move(transport), shared_buf, + std::move(owned_buf)); +} + +} // namespace + +absl::Status RaiseNoFileLimit(uint64_t target_nofile) { + struct rlimit rl{}; + if (getrlimit(RLIMIT_NOFILE, &rl) != 0) { + return absl::InternalError(absl::StrCat("getrlimit(RLIMIT_NOFILE) failed: ", + std::strerror(errno))); + } + const rlim_t desired = + (rl.rlim_max == RLIM_INFINITY) + ? static_cast(target_nofile) + : std::min(static_cast(target_nofile), rl.rlim_max); + if (rl.rlim_cur < desired) { + rl.rlim_cur = desired; + if (setrlimit(RLIMIT_NOFILE, &rl) != 0) { + return absl::InternalError(absl::StrCat( + "setrlimit(RLIMIT_NOFILE) failed: ", std::strerror(errno))); + } + } + return absl::OkStatus(); +} + +RegisteredBuffer::RegisteredBuffer(size_t size_bytes, bool zero_fill) + : data_(std::max(1, size_bytes), 0) { + if (!zero_fill) { + util::RandomNonZero(absl::MakeSpan(data_)); + } +} + +RegisteredBuffer::~RegisteredBuffer() { UnregisterAll(); } + +absl::Status RegisteredBuffer::RegisterWith(Transport* transport) { + CHECK(transport != nullptr); + absl::Status s = transport->RegisterMemory(data_.data(), data_.size()); + if (!s.ok()) { + return s; + } + absl::MutexLock lock(mu_); + registered_transports_.push_back(transport); + return absl::OkStatus(); +} + +void RegisteredBuffer::UnregisterFrom(Transport* transport) { + absl::MutexLock lock(mu_); + auto it = std::find(registered_transports_.begin(), + registered_transports_.end(), transport); + if (it != registered_transports_.end()) { + (*it)->UnregisterMemory(data_.data()).IgnoreError(); + registered_transports_.erase(it); + } +} + +void RegisteredBuffer::UnregisterAll() { + std::vector to_unregister; + { + absl::MutexLock lock(mu_); + to_unregister.swap(registered_transports_); + } + for (Transport* t : to_unregister) { + t->UnregisterMemory(data_.data()).IgnoreError(); + } +} + +NodeInstance::NodeInstance(std::string control_endpoint, + std::unique_ptr transport, + RegisteredBuffer* shared_buffer, + std::unique_ptr owned_buffer) + : control_endpoint_(std::move(control_endpoint)), + active_buffer_(shared_buffer != nullptr ? shared_buffer + : owned_buffer.get()), + owned_buffer_(std::move(owned_buffer)), + transport_(std::move(transport)) { + CHECK(transport_ != nullptr); + CHECK(active_buffer_ != nullptr); +} + +NodeInstance::~NodeInstance() { + if (active_buffer_ != nullptr && transport_ != nullptr) { + active_buffer_->UnregisterFrom(transport_.get()); + } + transport_.reset(); + owned_buffer_.reset(); +} + +absl::StatusOr NodeInstance::PostAsync( + std::string_view peer, absl::Span requests, + absl::AnyInvocable on_complete) { + return transport_->Post(peer, requests, std::move(on_complete)); +} + +absl::StatusOr>> CreateInstances( + const NodeConfig& config, RegisteredBuffer* shared_buf, size_t buf_size) { + if (config.num_instances <= 0) { + return absl::InvalidArgumentError("num_instances must be > 0"); + } + + std::vector> instances(config.num_instances); + std::atomic next_port{ + config.base_control_port > 0 ? config.base_control_port : 10000U}; + + absl::Status status = + ParallelFor(config.num_instances, kMaxCreateThreads, [&](int idx) { + auto inst_or = + CreateSingleInstance(idx, config, shared_buf, buf_size, next_port); + if (!inst_or.ok()) { + return inst_or.status(); + } + instances[idx] = std::move(*inst_or); + return absl::OkStatus(); + }); + + if (!status.ok()) { + DestroyInstances(instances); + return status; + } + return instances; +} + +void DestroyInstances(std::vector>& instances) { + ParallelFor(static_cast(instances.size()), kMaxDestroyThreads, + [&](int idx) { + instances[idx].reset(); + return absl::OkStatus(); + }) + .IgnoreError(); + instances.clear(); +} + +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/instance.h b/peregrine/test/cluster/node/instance.h new file mode 100644 index 0000000..efaa554 --- /dev/null +++ b/peregrine/test/cluster/node/instance.h @@ -0,0 +1,101 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_INSTANCE_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_INSTANCE_H_ + +#include +#include +#include +#include +#include +#include + +#include "absl/base/thread_annotations.h" +#include "absl/functional/any_invocable.h" +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/synchronization/mutex.h" +#include "absl/types/span.h" +#include "peregrine/src/api/transport.h" +#include "peregrine/src/api/transport_metrics.h" +#include "peregrine/src/api/transport_types.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { + +// Raises the soft file descriptor limit (RLIMIT_NOFILE) up to +// min(target_nofile, rlim_max) to support high instance/connection counts. +absl::Status RaiseNoFileLimit(uint64_t target_nofile = 1048576); + +// RAII memory buffer that tracks Transport registrations and unregisters memory +// cleanly before underlying Transport instances are destroyed. +class RegisteredBuffer { + public: + explicit RegisteredBuffer(size_t size_bytes, bool zero_fill = false); + ~RegisteredBuffer(); + + RegisteredBuffer(const RegisteredBuffer&) = delete; + RegisteredBuffer& operator=(const RegisteredBuffer&) = delete; + + Byte* DataPtr() { return data_.data(); } + const Byte* DataPtr() const { return data_.data(); } + size_t Size() const { return data_.size(); } + uint64_t AddrUint64() const { + return reinterpret_cast(data_.data()); + } + + absl::Status RegisterWith(Transport* transport); + void UnregisterFrom(Transport* transport); + void UnregisterAll(); + + private: + std::vector data_; + absl::Mutex mu_; + std::vector registered_transports_ ABSL_GUARDED_BY(mu_); +}; + +// Wraps a single Peregrine `Transport` instance, its bound control endpoint, +// and its associated registered memory buffer. +class NodeInstance { + public: + NodeInstance(std::string control_endpoint, + std::unique_ptr transport, + RegisteredBuffer* shared_buffer, + std::unique_ptr owned_buffer = nullptr); + ~NodeInstance(); + + NodeInstance(const NodeInstance&) = delete; + NodeInstance& operator=(const NodeInstance&) = delete; + + const std::string& ControlEndpoint() const { return control_endpoint_; } + Transport& GetTransport() const { return *transport_; } + Byte* DataPtr() const { return active_buffer_->DataPtr(); } + size_t DataSize() const { return active_buffer_->Size(); } + uint64_t BufferAddr() const { return active_buffer_->AddrUint64(); } + + absl::StatusOr PostAsync( + std::string_view peer, absl::Span requests, + absl::AnyInvocable on_complete); + + TransportMetrics GetTransportMetrics() const { + return transport_->GetTransportMetrics(); + } + + private: + std::string control_endpoint_; + RegisteredBuffer* active_buffer_ = nullptr; + std::unique_ptr owned_buffer_; + std::unique_ptr transport_; +}; + +// Creates `config.num_instances` Peregrine `NodeInstance` objects in parallel +// using deterministic atomic port allocation starting at +// `config.base_control_port`. +absl::StatusOr>> CreateInstances( + const NodeConfig& config, RegisteredBuffer* shared_buf, size_t buf_size); + +// Destroys all `NodeInstance` objects in parallel over a bounded thread pool +// so per-instance Engine thread join waits overlap rather than serializing. +void DestroyInstances(std::vector>& instances); + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_INSTANCE_H_ diff --git a/peregrine/test/cluster/node/instance_test.cc b/peregrine/test/cluster/node/instance_test.cc new file mode 100644 index 0000000..fdea80c --- /dev/null +++ b/peregrine/test/cluster/node/instance_test.cc @@ -0,0 +1,54 @@ +#include "peregrine/test/cluster/node/instance.h" + +#include + +#include "gtest/gtest.h" +#include "peregrine/src/util/util.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { +namespace { + +TEST(InstanceTest, RaiseNoFileLimitSucceeds) { + EXPECT_TRUE(RaiseNoFileLimit(65536).ok()); +} + +TEST(InstanceTest, CreateInstancesWithSharedBuffer) { + NodeConfig cfg; + cfg.ip = "127.0.0.1"; + cfg.num_instances = 4; + cfg.base_control_port = util::FindFreePort(AF_INET, /*tcp=*/true); + cfg.num_conns = 1; + + RegisteredBuffer shared_buf(/*size_bytes=*/64); + auto instances = CreateInstances(cfg, &shared_buf, /*buf_size=*/64); + ASSERT_TRUE(instances.ok()) << instances.status(); + ASSERT_EQ(instances->size(), 4); + + for (int i = 0; i < 4; ++i) { + ASSERT_NE((*instances)[i], nullptr); + EXPECT_EQ((*instances)[i]->BufferAddr(), shared_buf.AddrUint64()); + EXPECT_EQ((*instances)[i]->DataSize(), 64); + } + DestroyInstances(*instances); +} + +TEST(InstanceTest, CreateInstancesWithPerInstanceBuffers) { + NodeConfig cfg; + cfg.ip = "127.0.0.1"; + cfg.num_instances = 3; + cfg.base_control_port = util::FindFreePort(AF_INET, /*tcp=*/true); + cfg.num_conns = 1; + + auto instances = + CreateInstances(cfg, /*shared_buf=*/nullptr, /*buf_size=*/32); + ASSERT_TRUE(instances.ok()) << instances.status(); + ASSERT_EQ(instances->size(), 3); + + EXPECT_NE((*instances)[0]->BufferAddr(), (*instances)[1]->BufferAddr()); + EXPECT_NE((*instances)[1]->BufferAddr(), (*instances)[2]->BufferAddr()); + DestroyInstances(*instances); +} + +} // namespace +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/metrics.cc b/peregrine/test/cluster/node/metrics.cc new file mode 100644 index 0000000..a614b4b --- /dev/null +++ b/peregrine/test/cluster/node/metrics.cc @@ -0,0 +1,68 @@ +#include "peregrine/test/cluster/node/metrics.h" + +#include + +#include + +#include "absl/time/clock.h" +#include "absl/time/time.h" +#include "peregrine/src/api/transport_metrics.h" + +namespace peregrine::cluster { +namespace { + +void MergeHistogram(Log2Histogram<32>& dst, const Log2Histogram<32>& src) { + dst.sum += src.sum; + for (int i = 0; i < dst.NumBuckets(); ++i) { + dst.buckets[i] += src.buckets[i]; + } +} + +void MergeOpMetrics(OpMetrics& dst, const OpMetrics& src) { + MergeHistogram(dst.e2e_latency_us, src.e2e_latency_us); + MergeHistogram(dst.request_size_bytes, src.request_size_bytes); + dst.bytes += src.bytes; + dst.errors += src.errors; +} + +} // namespace + +CpuTimer::CpuTimer() { Reset(); } + +void CpuTimer::Reset() { + start_wall_ = absl::Now(); + struct rusage start_rusage{}; // NOLINT(misc-include-cleaner) + getrusage(RUSAGE_SELF, &start_rusage); + start_user_cpu_ = absl::DurationFromTimeval(start_rusage.ru_utime); + start_sys_cpu_ = absl::DurationFromTimeval(start_rusage.ru_stime); +} + +CpuStats CpuTimer::Snapshot() const { + const absl::Time now_wall = absl::Now(); + struct rusage now_rusage{}; // NOLINT(misc-include-cleaner) + getrusage(RUSAGE_SELF, &now_rusage); + + const absl::Duration user_delta = + absl::DurationFromTimeval(now_rusage.ru_utime) - start_user_cpu_; + const absl::Duration sys_delta = + absl::DurationFromTimeval(now_rusage.ru_stime) - start_sys_cpu_; + + CpuStats stats; + stats.wall_ms = absl::ToDoubleMilliseconds( + std::max(absl::ZeroDuration(), now_wall - start_wall_)); + stats.user_cpu_ms = std::max(0.0, absl::ToDoubleMilliseconds(user_delta)); + stats.sys_cpu_ms = std::max(0.0, absl::ToDoubleMilliseconds(sys_delta)); + if (stats.wall_ms > 0.0) { + stats.avg_cores = (stats.user_cpu_ms + stats.sys_cpu_ms) / stats.wall_ms; + } + return stats; +} + +void MergeTransportMetrics(TransportMetrics& dst, const TransportMetrics& src) { + MergeOpMetrics(dst.write, src.write); + MergeOpMetrics(dst.read, src.read); + dst.tcp_connect_failures += src.tcp_connect_failures; + dst.rpc_requests_received += src.rpc_requests_received; +} + +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/metrics.h b/peregrine/test/cluster/node/metrics.h new file mode 100644 index 0000000..f772f51 --- /dev/null +++ b/peregrine/test/cluster/node/metrics.h @@ -0,0 +1,49 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_METRICS_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_METRICS_H_ + +#include + +#include "absl/time/time.h" +#include "peregrine/src/api/transport_metrics.h" + +namespace peregrine::cluster { + +// Snapshot of process wall time and CPU usage. +struct CpuStats { + double wall_ms = 0.0; + double user_cpu_ms = 0.0; + double sys_cpu_ms = 0.0; + double avg_cores = 0.0; +}; + +// Tracks wall clock time and process CPU consumption via +// getrusage(RUSAGE_SELF). +class CpuTimer { + public: + CpuTimer(); + + void Reset(); + CpuStats Snapshot() const; + + private: + absl::Time start_wall_; + absl::Duration start_user_cpu_; + absl::Duration start_sys_cpu_; +}; + +// Merges `src` TransportMetrics into `dst` (summing counters and histogram +// buckets across local Transport instances). +void MergeTransportMetrics(TransportMetrics& dst, const TransportMetrics& src); + +// Complete node-level metrics snapshot (pure data struct, decoupled from wire +// formatting). +struct NodeMetricsSnapshot { + int num_instances = 0; + uint64_t xfer_size_bytes = 0; + CpuStats cpu{}; + TransportMetrics transport{}; +}; + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_METRICS_H_ diff --git a/peregrine/test/cluster/node/metrics_test.cc b/peregrine/test/cluster/node/metrics_test.cc new file mode 100644 index 0000000..462dde1 --- /dev/null +++ b/peregrine/test/cluster/node/metrics_test.cc @@ -0,0 +1,49 @@ +#include "peregrine/test/cluster/node/metrics.h" + +#include "gtest/gtest.h" +#include "absl/time/clock.h" +#include "absl/time/time.h" +#include "peregrine/src/api/transport_metrics.h" + +namespace peregrine::cluster { +namespace { + +TEST(MetricsTest, MergeTransportMetricsAccumulatesAllFields) { + TransportMetrics a; + a.write.bytes = 100; + a.write.errors = 1; + a.write.e2e_latency_us.sum = 500; + a.write.e2e_latency_us.buckets[3] = 2; + a.tcp_connect_failures = 4; + a.rpc_requests_received = 10; + + TransportMetrics b; + b.write.bytes = 250; + b.write.errors = 2; + b.write.e2e_latency_us.sum = 700; + b.write.e2e_latency_us.buckets[3] = 5; + b.read.bytes = 64; + b.tcp_connect_failures = 1; + b.rpc_requests_received = 20; + + MergeTransportMetrics(a, b); + EXPECT_EQ(a.write.bytes, 350); + EXPECT_EQ(a.write.errors, 3); + EXPECT_EQ(a.write.e2e_latency_us.sum, 1200); + EXPECT_EQ(a.write.e2e_latency_us.buckets[3], 7); + EXPECT_EQ(a.read.bytes, 64); + EXPECT_EQ(a.tcp_connect_failures, 5); + EXPECT_EQ(a.rpc_requests_received, 30); +} + +TEST(MetricsTest, CpuTimerMeasuresElapsedTime) { + CpuTimer timer; + absl::SleepFor(absl::Milliseconds(10)); + CpuStats snap = timer.Snapshot(); + EXPECT_GE(snap.wall_ms, 5.0); + EXPECT_GE(snap.user_cpu_ms, 0.0); + EXPECT_GE(snap.sys_cpu_ms, 0.0); +} + +} // namespace +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/node_main.cc b/peregrine/test/cluster/node/node_main.cc new file mode 100644 index 0000000..8496e90 --- /dev/null +++ b/peregrine/test/cluster/node/node_main.cc @@ -0,0 +1,28 @@ +// --------------------------------------------------------- +// $ bazelisk run //peregrine/test/cluster/node:node +// --------------------------------------------------------- + +#include "absl/flags/parse.h" +#include "absl/log/initialize.h" +#include "absl/log/check.h" +#include "absl/log/log.h" +#include "peregrine/test/cluster/node/cluster_node.h" +#include "peregrine/test/cluster/node/control_channel.h" +#include "peregrine/test/cluster/node/flags.h" +#include "peregrine/test/cluster/node/instance.h" + +int main(int argc, char* argv[]) { + absl::ParseCommandLine(argc, argv); + absl::InitializeLog(); + + if (auto s = peregrine::cluster::RaiseNoFileLimit(); !s.ok()) { + LOG(WARNING) << "Failed to raise RLIMIT_NOFILE: " << s; + } + + auto config_or = peregrine::cluster::ReadNodeConfig(); + QCHECK_OK(config_or.status()); + + peregrine::cluster::ControlChannel channel; + QCHECK_OK(peregrine::cluster::ClusterNode::Run(*config_or, channel)); + return 0; +} diff --git a/peregrine/test/cluster/node/topology.cc b/peregrine/test/cluster/node/topology.cc new file mode 100644 index 0000000..9cca4a2 --- /dev/null +++ b/peregrine/test/cluster/node/topology.cc @@ -0,0 +1,138 @@ +#include "peregrine/test/cluster/node/topology.h" + +#include +#include +#include +#include +#include + +#include "absl/status/status.h" +#include "absl/status/statusor.h" +#include "absl/strings/ascii.h" +#include "absl/strings/match.h" +#include "absl/strings/numbers.h" +#include "absl/strings/str_cat.h" +#include "absl/strings/str_format.h" +#include "absl/strings/str_join.h" +#include "absl/strings/str_split.h" +#include "absl/types/span.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { + +absl::StatusOr> ParseTargets(std::string_view raw) { + std::string_view trimmed = absl::StripAsciiWhitespace(raw); + if (trimmed.empty()) { + return std::vector{}; + } + + std::vector targets; + for (std::string_view token : + absl::StrSplit(trimmed, ',', absl::SkipEmpty())) { + token = absl::StripAsciiWhitespace(token); + if (token.empty()) continue; + + const size_t at_pos = token.rfind('@'); + if (at_pos == std::string_view::npos || at_pos == 0 || + at_pos + 1 >= token.size()) { + return absl::InvalidArgumentError(absl::StrCat( + "Invalid target '", token, "': expected ':@'")); + } + + std::string_view endpoint = token.substr(0, at_pos); + std::string_view raddr_str = token.substr(at_pos + 1); + if (!absl::StrContains(endpoint, ':')) { + return absl::InvalidArgumentError(absl::StrCat( + "Invalid target endpoint '", endpoint, "': missing ':'")); + } + + uint64_t raddr = 0; + if (!absl::SimpleAtoi(raddr_str, &raddr) || raddr == 0) { + return absl::InvalidArgumentError( + absl::StrCat("Invalid target raddr '", raddr_str, "' in '", token, + "': expected positive uint64")); + } + + targets.push_back(PeerTarget{ + .endpoint = std::string(endpoint), + .raddr = raddr, + }); + } + + return targets; +} + +std::string FormatTargets(absl::Span targets) { + return absl::StrJoin(targets, ",", [](std::string* out, const PeerTarget& t) { + absl::StrAppendFormat(out, "%s@%u", t.endpoint, t.raddr); + }); +} + +absl::StatusOr ParseTrafficPattern(std::string_view raw) { + std::string_view s = absl::StripAsciiWhitespace(raw); + if (absl::EqualsIgnoreCase(s, "all_to_all")) { + return TrafficPattern::kAllToAll; + } + if (absl::EqualsIgnoreCase(s, "round_robin")) { + return TrafficPattern::kRoundRobin; + } + return absl::InvalidArgumentError( + absl::StrCat("Invalid traffic_pattern '", raw, + "': expected 'all_to_all' or 'round_robin'")); +} + +std::string_view TrafficPatternToString(TrafficPattern pattern) { + switch (pattern) { + case TrafficPattern::kAllToAll: + return "all_to_all"; + case TrafficPattern::kRoundRobin: + return "round_robin"; + } + return "unknown"; +} + +std::vector> BuildTargetAssignments( + int node_index, int num_local_instances, + absl::Span all_targets, TrafficPattern pattern, + bool exclude_self, absl::Span local_endpoints) { + if (num_local_instances <= 0) { + return {}; + } + std::vector> assignments(num_local_instances); + if (all_targets.empty()) { + return assignments; + } + + for (int i = 0; i < num_local_instances; ++i) { + const std::string_view self_ep = + (exclude_self && static_cast(i) < local_endpoints.size()) + ? std::string_view(local_endpoints[i]) + : std::string_view(); + + switch (pattern) { + case TrafficPattern::kAllToAll: { + assignments[i].reserve(all_targets.size()); + for (const PeerTarget& target : all_targets) { + if (exclude_self && !self_ep.empty() && target.endpoint == self_ep) { + continue; + } + assignments[i].push_back(target); + } + break; + } + case TrafficPattern::kRoundRobin: { + const size_t global_idx = static_cast(node_index) * + static_cast(num_local_instances) + + static_cast(i); + const PeerTarget& chosen = all_targets[global_idx % all_targets.size()]; + if (!(exclude_self && !self_ep.empty() && chosen.endpoint == self_ep)) { + assignments[i].push_back(chosen); + } + break; + } + } + } + return assignments; +} + +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/node/topology.h b/peregrine/test/cluster/node/topology.h new file mode 100644 index 0000000..9f23b67 --- /dev/null +++ b/peregrine/test/cluster/node/topology.h @@ -0,0 +1,44 @@ +#ifndef PEREGRINE_TEST_CLUSTER_NODE_TOPOLOGY_H_ +#define PEREGRINE_TEST_CLUSTER_NODE_TOPOLOGY_H_ + +#include +#include +#include + +#include "absl/status/statusor.h" +#include "absl/types/span.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { + +// Parses a comma-separated list of ":@" target strings. +// Returns an empty vector if `raw` is empty or whitespace-only. +absl::StatusOr> ParseTargets(std::string_view raw); + +// Formats a list of `PeerTarget` into comma-separated "@". +std::string FormatTargets(absl::Span targets); + +// Parses traffic pattern string ("all_to_all" or "round_robin"). +absl::StatusOr ParseTrafficPattern(std::string_view raw); + +// Returns string representation of `TrafficPattern`. +std::string_view TrafficPatternToString(TrafficPattern pattern); + +// Computes the outbound target list for each local instance i in +// [0, num_local_instances). +// +// - If `all_targets` is empty (passive server mode), returns +// `num_local_instances` empty vectors. +// - `kAllToAll`: assigns every target in `all_targets` to each local instance +// (excluding `local_endpoints[i]` when `exclude_self` is true). +// - `kRoundRobin`: assigns `all_targets[(node_index * num_local_instances + i) +// % all_targets.size()]` to local instance `i`. +std::vector> BuildTargetAssignments( + int node_index, int num_local_instances, + absl::Span all_targets, TrafficPattern pattern, + bool exclude_self = false, + absl::Span local_endpoints = {}); + +} // namespace peregrine::cluster + +#endif // PEREGRINE_TEST_CLUSTER_NODE_TOPOLOGY_H_ diff --git a/peregrine/test/cluster/node/topology_test.cc b/peregrine/test/cluster/node/topology_test.cc new file mode 100644 index 0000000..b733ff7 --- /dev/null +++ b/peregrine/test/cluster/node/topology_test.cc @@ -0,0 +1,106 @@ +#include "peregrine/test/cluster/node/topology.h" + +#include +#include + +#include "gmock/gmock.h" +#include "gtest/gtest.h" +#include "peregrine/test/cluster/node/config.h" + +namespace peregrine::cluster { +namespace { + +using ::testing::ElementsAre; +using ::testing::IsEmpty; + +TEST(TopologyTest, ParseAndFormatTargets) { + auto empty = ParseTargets(" "); + ASSERT_TRUE(empty.ok()); + EXPECT_THAT(*empty, IsEmpty()); + + auto parsed = ParseTargets("127.0.0.1:10000@140001, 10.0.0.2:10001@140002 "); + ASSERT_TRUE(parsed.ok()); + EXPECT_THAT( + *parsed, + ElementsAre(PeerTarget{.endpoint = "127.0.0.1:10000", .raddr = 140001}, + PeerTarget{.endpoint = "10.0.0.2:10001", .raddr = 140002})); + + EXPECT_EQ(FormatTargets(*parsed), + "127.0.0.1:10000@140001,10.0.0.2:10001@140002"); +} + +TEST(TopologyTest, ParseTargetsRejectsInvalidInput) { + EXPECT_FALSE(ParseTargets("127.0.0.1:10000").ok()); + EXPECT_FALSE(ParseTargets("127.0.0.1@12345").ok()); + EXPECT_FALSE(ParseTargets("127.0.0.1:10000@0").ok()); + EXPECT_FALSE(ParseTargets("127.0.0.1:10000@notanumber").ok()); +} + +TEST(TopologyTest, ParseTrafficPattern) { + auto all_to_all = ParseTrafficPattern("all_to_all"); + ASSERT_TRUE(all_to_all.ok()); + EXPECT_EQ(*all_to_all, TrafficPattern::kAllToAll); + EXPECT_EQ(TrafficPatternToString(*all_to_all), "all_to_all"); + + auto rr = ParseTrafficPattern("ROUND_ROBIN"); + ASSERT_TRUE(rr.ok()); + EXPECT_EQ(*rr, TrafficPattern::kRoundRobin); + EXPECT_EQ(TrafficPatternToString(*rr), "round_robin"); + + EXPECT_FALSE(ParseTrafficPattern("one_to_one").ok()); +} + +TEST(TopologyTest, BuildTargetAssignmentsPassiveEmpty) { + auto assignments = BuildTargetAssignments( + /*node_index=*/0, /*num_local_instances=*/4, /*all_targets=*/{}, + TrafficPattern::kAllToAll); + ASSERT_EQ(assignments.size(), 4); + for (const auto& list : assignments) { + EXPECT_THAT(list, IsEmpty()); + } +} + +TEST(TopologyTest, BuildTargetAssignmentsFanInAndAllToAll) { + std::vector targets = { + {.endpoint = "127.0.0.1:10000", .raddr = 111}, + {.endpoint = "127.0.0.1:10001", .raddr = 222}, + }; + auto assignments = BuildTargetAssignments( + /*node_index=*/0, /*num_local_instances=*/3, targets, + TrafficPattern::kAllToAll); + ASSERT_EQ(assignments.size(), 3); + for (const auto& list : assignments) { + EXPECT_THAT(list, ElementsAre(targets[0], targets[1])); + } +} + +TEST(TopologyTest, BuildTargetAssignmentsExcludeSelf) { + std::vector targets = { + {.endpoint = "127.0.0.1:10000", .raddr = 111}, + {.endpoint = "127.0.0.1:10001", .raddr = 222}, + }; + std::vector local_eps = {"127.0.0.1:10000", "127.0.0.1:10001"}; + auto assignments = BuildTargetAssignments( + /*node_index=*/0, /*num_local_instances=*/2, targets, + TrafficPattern::kAllToAll, /*exclude_self=*/true, local_eps); + ASSERT_EQ(assignments.size(), 2); + EXPECT_THAT(assignments[0], ElementsAre(targets[1])); + EXPECT_THAT(assignments[1], ElementsAre(targets[0])); +} + +TEST(TopologyTest, BuildTargetAssignmentsRoundRobinAcrossNodes) { + std::vector targets = { + {.endpoint = "10.0.0.1:10000", .raddr = 1}, + {.endpoint = "10.0.0.1:10001", .raddr = 2}, + {.endpoint = "10.0.0.1:10002", .raddr = 3}, + }; + auto assignments = BuildTargetAssignments( + /*node_index=*/1, /*num_local_instances=*/2, targets, + TrafficPattern::kRoundRobin); + ASSERT_EQ(assignments.size(), 2); + EXPECT_THAT(assignments[0], ElementsAre(targets[2])); + EXPECT_THAT(assignments[1], ElementsAre(targets[0])); +} + +} // namespace +} // namespace peregrine::cluster diff --git a/peregrine/test/cluster/orchestrator/BUILD b/peregrine/test/cluster/orchestrator/BUILD new file mode 100644 index 0000000..47d94a9 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/BUILD @@ -0,0 +1,99 @@ +# Python orchestrator and CLI for Peregrine multi-node cluster benchmarks. + +load("@rules_python//python:py_binary.bzl", "py_binary") +load("@rules_python//python:py_library.bzl", "py_library") +load("@rules_python//python:py_test.bzl", "py_test") + +package( + default_testonly = True, + default_visibility = ["//peregrine/test:__subpackages__"], +) + +py_library( + name = "preflight_py", + srcs = ["preflight.py"], + deps = ["@abseil-py//absl/logging"], +) + +py_test( + name = "preflight_test", + srcs = ["preflight_test.py"], + main = "preflight_test.py", + deps = [ + ":preflight_py", + "@abseil-py//absl/testing:absltest", + ], +) + +py_library( + name = "node_executor_py", + srcs = ["node_executor.py"], + deps = [ + ":preflight_py", + "@abseil-py//absl/logging", + ], +) + +py_test( + name = "node_executor_test", + srcs = ["node_executor_test.py"], + main = "node_executor_test.py", + deps = [ + ":node_executor_py", + "@abseil-py//absl/testing:absltest", + ], +) + +py_library( + name = "report_py", + srcs = ["report.py"], +) + +py_test( + name = "report_test", + srcs = ["report_test.py"], + main = "report_test.py", + deps = [ + ":report_py", + "@abseil-py//absl/testing:absltest", + ], +) + +py_library( + name = "cluster_session_py", + srcs = ["cluster_session.py"], + data = ["//peregrine/test/cluster/node"], + deps = [ + ":node_executor_py", + ":preflight_py", + ":report_py", + "@abseil-py//absl/logging", + ], +) + +py_binary( + name = "bipartite_runner", + srcs = ["bipartite_runner.py"], + data = ["//peregrine/test/cluster/node"], + main = "bipartite_runner.py", + deps = [ + ":cluster_session_py", + ":node_executor_py", + ":report_py", + "@abseil-py//absl:app", + "@abseil-py//absl/flags", + "@abseil-py//absl/logging", + ], +) + +py_test( + name = "bipartite_runner_test", + srcs = ["bipartite_runner_test.py"], + data = ["//peregrine/test/cluster/node"], + main = "bipartite_runner_test.py", + deps = [ + ":cluster_session_py", + "//peregrine/src/util:util_py", + "@abseil-py//absl/testing:absltest", + ], +) diff --git a/peregrine/test/cluster/orchestrator/bipartite_runner.py b/peregrine/test/cluster/orchestrator/bipartite_runner.py new file mode 100644 index 0000000..8abec32 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/bipartite_runner.py @@ -0,0 +1,172 @@ +"""CLI orchestrator for the Peregrine Cluster Bipartite Benchmark.""" + +from collections.abc import Sequence +import json + +from absl import app +from absl import flags +from absl import logging + +from peregrine.test.cluster.orchestrator import cluster_session +from peregrine.test.cluster.orchestrator import node_executor +from peregrine.test.cluster.orchestrator import report + +_CLIENT_NODES = flags.DEFINE_list( + "client_nodes", + ["localhost"], + "Comma-separated list of N client hostnames or IPs.", +) +_SERVER_NODES = flags.DEFINE_list( + "server_nodes", + ["localhost"], + "Comma-separated list of M server hostnames or IPs.", +) +_CLIENTS_PER_NODE = flags.DEFINE_integer( + "clients_per_node", + 1, + "Number of Peregrine Transport instances per client node (N_p).", +) +_SERVERS_PER_NODE = flags.DEFINE_integer( + "servers_per_node", + 1, + "Number of Peregrine Transport instances per server node (M_p).", +) +_CLIENT_IPS = flags.DEFINE_list( + "client_ips", + [], + "Optional explicit Peregrine bind IPs per client node.", +) +_SERVER_IPS = flags.DEFINE_list( + "server_ips", + [], + "Optional explicit Peregrine bind IPs per server node.", +) +_SERVER_BASE_PORT = flags.DEFINE_integer( + "server_base_port", + 10000, + "Starting control plane port for server instances.", +) +_CLIENT_BASE_PORT = flags.DEFINE_integer( + "client_base_port", + 20000, + "Starting control plane port for client instances.", +) +_TRANSPORT = flags.DEFINE_string( + "transport", "tcp", "Transport type ('tcp' or 'rdma')." +) +_CONN = flags.DEFINE_integer( + "conn", 1, "Number of data channels per peer (num_conns_per_peer)." +) +_WORKLOAD = flags.DEFINE_string( + "workload", + "serial_fixed_write", + "Workload generator ('serial_fixed_write' or 'kv_cache').", +) +_XFER_SIZE = flags.DEFINE_integer( + "xfer_size", 1, "Transfer size in bytes (for serial_fixed_write)." +) +_NUM_LAYERS = flags.DEFINE_integer( + "num_layers", 1, "Number of layers (for kv_cache)." +) +_NUM_BLOCKS = flags.DEFINE_integer( + "num_blocks", 1, "Number of blocks (for kv_cache)." +) +_BLOCK_SIZE = flags.DEFINE_integer( + "block_size", 4096, "Block size in bytes (for kv_cache)." +) +_NUM_XFERS = flags.DEFINE_integer( + "num_xfers", 1, "Number of transfer iterations per stream." +) +_TRAFFIC_PATTERN = flags.DEFINE_string( + "traffic_pattern", + "all_to_all", + "Traffic routing pattern ('all_to_all' or 'round_robin').", +) +_SHARE_BUFFER = flags.DEFINE_bool( + "share_buffer", + True, + "Share one registered buffer across all instances in a node process.", +) +_METRICS_INTERVAL = flags.DEFINE_string( + "metrics_interval", + "1s", + "Periodic sampling interval for node metrics.", +) +_SSH_CMD = flags.DEFINE_string( + "ssh_cmd", + node_executor.DEFAULT_SSH_CMD, + "SSH command prefix for remote node staging and execution.", +) +_REMOTE_DIR = flags.DEFINE_string( + "remote_dir", "/tmp", "Directory on remote nodes for staging binary." +) +_NODE_BINARY = flags.DEFINE_string( + "node_binary", "", "Optional explicit path to pre-built :node binary." +) +_MIN_SOMAXCONN = flags.DEFINE_integer( + "min_somaxconn", + 0, + "Minimum required net.core.somaxconn and tcp_max_syn_backlog (0 = auto).", +) +_READY_TIMEOUT_SEC = flags.DEFINE_float( + "ready_timeout_sec", + 120.0, + "Timeout in seconds waiting for nodes to emit PEREGRINE_NODE_READY.", +) +_WORKLOAD_TIMEOUT_SEC = flags.DEFINE_float( + "workload_timeout_sec", + 600.0, + "Timeout in seconds waiting for active nodes to emit" + " PEREGRINE_WORKLOAD_DONE.", +) +_OUTPUT_JSON = flags.DEFINE_string( + "output_json", "", "Optional file path to write aggregated JSON results." +) + + +def main(argv: Sequence[str]) -> None: + """Runs the Peregrine bipartite cluster benchmark CLI.""" + if len(argv) > 1: + raise app.UsageError(f"Unrecognized positional arguments: {argv[1:]}") + + logging.set_stderrthreshold(logging.INFO) + + cfg = cluster_session.BipartiteSessionConfig( + client_nodes=_CLIENT_NODES.value, + server_nodes=_SERVER_NODES.value, + clients_per_node=_CLIENTS_PER_NODE.value, + servers_per_node=_SERVERS_PER_NODE.value, + client_ips=_CLIENT_IPS.value, + server_ips=_SERVER_IPS.value, + server_base_port=_SERVER_BASE_PORT.value, + client_base_port=_CLIENT_BASE_PORT.value, + transport=_TRANSPORT.value, + conn=_CONN.value, + workload=_WORKLOAD.value, + xfer_size=_XFER_SIZE.value, + num_layers=_NUM_LAYERS.value, + num_blocks=_NUM_BLOCKS.value, + block_size=_BLOCK_SIZE.value, + num_xfers=_NUM_XFERS.value, + traffic_pattern=_TRAFFIC_PATTERN.value, + share_buffer=_SHARE_BUFFER.value, + metrics_interval=_METRICS_INTERVAL.value, + ssh_cmd=_SSH_CMD.value, + remote_dir=_REMOTE_DIR.value, + node_binary=_NODE_BINARY.value, + min_somaxconn=_MIN_SOMAXCONN.value, + ready_timeout_sec=_READY_TIMEOUT_SEC.value, + workload_timeout_sec=_WORKLOAD_TIMEOUT_SEC.value, + ) + + results = cluster_session.run_bipartite_session(cfg) + + print(report.format_summary_table(results)) + if _OUTPUT_JSON.value: + with open(_OUTPUT_JSON.value, "w", encoding="utf-8") as f: + json.dump(results, f, indent=2) + logging.info("Wrote aggregated JSON report to %s", _OUTPUT_JSON.value) + + +if __name__ == "__main__": + app.run(main) diff --git a/peregrine/test/cluster/orchestrator/bipartite_runner_test.py b/peregrine/test/cluster/orchestrator/bipartite_runner_test.py new file mode 100644 index 0000000..1d29e17 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/bipartite_runner_test.py @@ -0,0 +1,40 @@ +"""End-to-end localhost integration test for bipartite_runner / cluster_session.""" + +import socket + +from absl.testing import absltest + +from peregrine.src.util import util +from peregrine.test.cluster.orchestrator import cluster_session + + +class BipartiteRunnerTest(absltest.TestCase): + + def test_localhost_fan_in_and_round_robin(self): + server_port = util.find_free_port(socket.AF_INET, tcp=True) + client_port = util.find_free_port(socket.AF_INET, tcp=True) + if abs(client_port - server_port) < 50: + client_port = (server_port + 200) if server_port < 60000 else 15000 + + cfg = cluster_session.BipartiteSessionConfig( + client_nodes=["localhost"], + server_nodes=["localhost"], + clients_per_node=8, + servers_per_node=1, + server_base_port=server_port, + client_base_port=client_port, + transport="tcp", + conn=1, + workload="serial_fixed_write", + xfer_size=1, + num_xfers=1, + traffic_pattern="all_to_all", + ) + results = cluster_session.run_bipartite_session(cfg) + self.assertEqual(results["latency"]["count"], 8) + self.assertEqual(results["latency"]["errors"], 0) + self.assertEqual(results["receiver_transport"]["rpc_requests_received"], 8) + + +if __name__ == "__main__": + absltest.main() diff --git a/peregrine/test/cluster/orchestrator/cluster_session.py b/peregrine/test/cluster/orchestrator/cluster_session.py new file mode 100644 index 0000000..7524f16 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/cluster_session.py @@ -0,0 +1,480 @@ +"""Reusable multi-node 4-phase barrier session coordinator.""" + +from collections.abc import Mapping, Sequence +from concurrent import futures +import dataclasses +import subprocess +import threading +from typing import Any + +from absl import logging + +from peregrine.test.cluster.orchestrator import node_executor +from peregrine.test.cluster.orchestrator import preflight +from peregrine.test.cluster.orchestrator import report + + +@dataclasses.dataclass(frozen=True) +class BipartiteSessionConfig: + """Configuration for a bipartite (N x N_p -> M x M_p) cluster run.""" + + client_nodes: Sequence[str] + server_nodes: Sequence[str] + clients_per_node: int = 1 + servers_per_node: int = 1 + client_ips: Sequence[str] = () + server_ips: Sequence[str] = () + server_base_port: int = 10000 + client_base_port: int = 20000 + transport: str = "tcp" + conn: int = 1 + workload: str = "serial_fixed_write" + xfer_size: int = 1 + num_layers: int = 1 + num_blocks: int = 1 + block_size: int = 4096 + num_xfers: int = 1 + traffic_pattern: str = "all_to_all" + share_buffer: bool = True + metrics_interval: str = "1s" + ssh_cmd: str = node_executor.DEFAULT_SSH_CMD + remote_dir: str = "/tmp" + node_binary: str = "" + min_somaxconn: int = 0 + ready_timeout_sec: float = 120.0 + workload_timeout_sec: float = 600.0 + + +class _ManagedNodeProcess: + """Wraps a running `:node` subprocess and background stdout/stderr reader threads.""" + + def __init__(self, label: str, host: str, proc: subprocess.Popen[str]): + self.label = label + self.host = host + self.proc = proc + self.ready_event = threading.Event() + self.workload_done_event = threading.Event() + self.node_done_event = threading.Event() + self.ready_payload: dict[str, Any] | None = None + self.workload_done_payload: dict[str, Any] | None = None + self.node_done_payload: dict[str, Any] | None = None + self.samples: list[dict[str, Any]] = [] + self.stderr_lines: list[str] = [] + self.error: str | None = None + self._stderr_thread = threading.Thread( + target=self._read_stderr_loop, daemon=True + ) + self._stderr_thread.start() + self._reader_thread = threading.Thread( + target=self._read_stdout_loop, daemon=True + ) + self._reader_thread.start() + + def _read_stderr_loop(self) -> None: + """Reads stderr lines from the node subprocess in the background.""" + if self.proc.stderr is None: + return + try: + for raw_line in self.proc.stderr: + line = raw_line.rstrip() + if line and len(self.stderr_lines) < 200: + self.stderr_lines.append(line) + except OSError: + pass + + def _read_stdout_loop(self) -> None: + """Parses protocol events and metrics samples from node stdout.""" + assert self.proc.stdout is not None + for raw_line in self.proc.stdout: + line = raw_line.strip() + if not line: + continue + ready = report.parse_prefixed_json(line, report.NODE_READY_PREFIX) + if ready is not None: + self.ready_payload = ready + self.ready_event.set() + continue + sample = report.parse_prefixed_json(line, report.METRICS_SAMPLE_PREFIX) + if sample is not None: + self.samples.append(sample) + lat = report.summarize_node_snapshot(sample) + if int(lat.get("count", 0)) > 0: + logging.info( + " [Sample] %s (%s): completed=%d, errors=%d, throughput=%.4f" + " Gbps", + self.label, + self.host, + int(lat.get("count", 0)), + int(lat.get("errors", 0)), + float(lat.get("throughput_gbps", 0.0)), + ) + continue + w_done = report.parse_prefixed_json(line, report.WORKLOAD_DONE_PREFIX) + if w_done is not None: + self.workload_done_payload = w_done + self.workload_done_event.set() + continue + n_done = report.parse_prefixed_json(line, report.NODE_DONE_PREFIX) + if n_done is not None: + self.node_done_payload = n_done + self.node_done_event.set() + continue + + rc = self.proc.poll() + if not self.ready_event.is_set() or not self.node_done_event.is_set(): + stderr_tail = "\n".join(self.stderr_lines[-20:]) + self.error = ( + f"Node {self.label} ({self.host}) exited prematurely (rc={rc}):" + f" {stderr_tail}" + ) + self.ready_event.set() + self.workload_done_event.set() + self.node_done_event.set() + + def debug_summary(self) -> str: + """Returns a diagnostic summary of recent samples and stderr lines.""" + last_sample = self.samples[-1] if self.samples else None + stderr_tail = "\n ".join(self.stderr_lines[-10:]) + return ( + f"{self.label} ({self.host}): samples={len(self.samples)}," + f" last_sample={last_sample}, stderr_tail=[\n {stderr_tail}\n]" + ) + + def send_command(self, cmd: str) -> None: + """Writes a control command line (`START` or `STOP`) to node stdin.""" + if self.proc.stdin is not None and self.proc.poll() is None: + try: + self.proc.stdin.write(f"{cmd}\n") + self.proc.stdin.flush() + except OSError as e: + logging.warning("Failed to send %s to %s: %s", cmd, self.label, e) + + def terminate(self) -> None: + """Terminates or kills the node subprocess if still running.""" + if self.proc.poll() is None: + self.proc.terminate() + try: + self.proc.wait(timeout=5) + except subprocess.TimeoutExpired: + self.proc.kill() + + +def _common_workload_flags(cfg: BipartiteSessionConfig) -> list[str]: + """Builds common CLI flags shared by server and client `:node` processes.""" + flags = [ + f"--transport={cfg.transport}", + f"--conn={cfg.conn}", + f"--workload={cfg.workload}", + f"--num_xfers={cfg.num_xfers}", + f"--traffic_pattern={cfg.traffic_pattern}", + f"--share_buffer={'true' if cfg.share_buffer else 'false'}", + f"--metrics_interval={cfg.metrics_interval}", + ] + if cfg.workload.lower() == "serial_fixed_write": + flags.append(f"--xfer_size={cfg.xfer_size}") + elif cfg.workload.lower() == "kv_cache": + flags.extend([ + f"--num_layers={cfg.num_layers}", + f"--num_blocks={cfg.num_blocks}", + f"--block_size={cfg.block_size}", + ]) + return flags + + +def _spawn_node_group( + role_prefix: str, + hosts: Sequence[str], + explicit_ips: Sequence[str], + instances_per_node: int, + base_port: int, + common_flags: Sequence[str], + extra_flags: Sequence[str], + executor: node_executor.NodeExecutor, + host_to_bin: Mapping[str, str], + target_nofile: int, +) -> list[_ManagedNodeProcess]: + """Spawns a group of server or client `:node` processes across `hosts`.""" + port_stride = max(100, instances_per_node * 2) + procs: list[_ManagedNodeProcess] = [] + for idx, host in enumerate(hosts): + explicit_ip = explicit_ips[idx] if idx < len(explicit_ips) else "" + bind_ip = executor.resolve_bind_ip(host, explicit_ip) + node_base_port = base_port + idx * port_stride + args = [ + f"--ip={bind_ip}", + f"--num_instances={instances_per_node}", + f"--node_index={idx}", + f"--base_control_port={node_base_port}", + *extra_flags, + *common_flags, + ] + logging.info( + " Spawning %s-%d on %s (bind_ip=%s, base_port=%d, instances=%d)", + role_prefix, + idx, + host, + bind_ip, + node_base_port, + instances_per_node, + ) + proc = executor.spawn_node( + host, host_to_bin[host], args, target_nofile=target_nofile + ) + procs.append(_ManagedNodeProcess(f"{role_prefix}-{idx}", host, proc)) + return procs + + +def _wait_for_nodes_ready( + procs: Sequence[_ManagedNodeProcess], timeout_sec: float +) -> list[str]: + """Blocks until all `procs` emit `NODE_READY` and returns `ep@raddr` targets.""" + target_entries: list[str] = [] + for p in procs: + if not p.ready_event.wait(timeout=timeout_sec): + raise RuntimeError(f"Timed out waiting for {p.label} NODE_READY") + if p.error or p.ready_payload is None: + raise RuntimeError(p.error or f"{p.label} failed during init") + targets = p.ready_payload["targets"] + logging.info( + " %s (%s) ready: %d target(s)", + p.label, + p.host, + len(targets), + ) + target_entries.extend(targets) + return target_entries + + +def _broadcast_command(procs: Sequence[_ManagedNodeProcess], cmd: str) -> None: + """Broadcasts `cmd` concurrently across all `procs`.""" + if not procs: + return + with futures.ThreadPoolExecutor(max_workers=len(procs)) as pool: + list(pool.map(lambda p: p.send_command(cmd), procs)) + + +def _wait_for_workload_done( + client_procs: Sequence[_ManagedNodeProcess], + server_procs: Sequence[_ManagedNodeProcess], + timeout_sec: float, +) -> None: + """Waits for all active client nodes to emit `WORKLOAD_DONE`.""" + for cp in client_procs: + if not cp.workload_done_event.wait(timeout=timeout_sec): + srv_diag = "\n".join(sp.debug_summary() for sp in server_procs) + raise RuntimeError( + f"Timed out waiting for {cp.label} WORKLOAD_DONE.\n" + f"Client state: {cp.debug_summary()}\n" + f"Server state: {srv_diag}" + ) + if cp.error or cp.workload_done_payload is None: + raise RuntimeError(cp.error or f"{cp.label} failed during workload") + lat = report.summarize_node_snapshot(cp.workload_done_payload) + logging.info( + " %s (%s) finished workload: count=%d, errors=%d, elapsed=%.3f ms," + " throughput=%.4f Gbps", + cp.label, + cp.host, + int(lat.get("count", 0)), + int(lat.get("errors", 0)), + float(lat.get("elapsed_ms", 0.0)), + float(lat.get("throughput_gbps", 0.0)), + ) + + +def _stop_and_collect_nodes( + all_procs: Sequence[_ManagedNodeProcess], +) -> None: + """Broadcasts `STOP` and waits for all nodes to emit `NODE_DONE` and exit.""" + _broadcast_command(all_procs, "STOP") + for p in all_procs: + if not p.node_done_event.wait(timeout=30.0): + raise RuntimeError(f"Timed out waiting for {p.label} NODE_DONE") + if p.error or p.node_done_payload is None: + raise RuntimeError(p.error or f"{p.label} failed during teardown") + p.proc.wait(timeout=15.0) + + +def run_bipartite_session(cfg: BipartiteSessionConfig) -> dict[str, Any]: + """Runs the full 4-phase bipartite benchmark session across all nodes.""" + all_hosts = [*cfg.server_nodes, *cfg.client_nodes] + unique_hosts = list(dict.fromkeys(all_hosts)) + server_set = { + "localhost" if node_executor.is_local_node(h) else h + for h in cfg.server_nodes + } + client_set = { + "localhost" if node_executor.is_local_node(h) else h + for h in cfg.client_nodes + } + colocated = bool(server_set & client_set) + + requirements = preflight.compute_requirements( + num_client_nodes=len(cfg.client_nodes), + clients_per_node=cfg.clients_per_node, + num_server_nodes=len(cfg.server_nodes), + servers_per_node=cfg.servers_per_node, + conn=cfg.conn, + traffic_pattern=cfg.traffic_pattern, + colocated=colocated, + min_somaxconn=cfg.min_somaxconn, + ) + + executor = node_executor.NodeExecutor( + ssh_cmd=cfg.ssh_cmd, remote_dir=cfg.remote_dir + ) + server_procs: list[_ManagedNodeProcess] = [] + client_procs: list[_ManagedNodeProcess] = [] + + try: + # Step 1: Preflight check on all nodes (logs warnings on low limits). + logging.info( + "[Step 1/7] Running preflight checks on %d unique host(s): %s " + "(recommended somaxconn>=%d, ulimit_nofile>=%d," + " ephemeral_ports>=%d)...", + len(unique_hosts), + ", ".join(unique_hosts), + requirements.somaxconn, + requirements.ulimit_nofile, + requirements.ephemeral_ports, + ) + preflight.verify_nodes(all_hosts, requirements, executor.inspect_node) + logging.info( + "[Step 1/7] Preflight check complete on %d host(s).", + len(unique_hosts), + ) + + # Step 2: Locate `:node` binary and stage to all nodes. + logging.info( + "[Step 2/7] Locating :node binary and staging to %d host(s)...", + len(unique_hosts), + ) + local_bin = node_executor.locate_node_binary(explicit_path=cfg.node_binary) + host_to_bin = executor.stage_binary(all_hosts, local_bin) + logging.info( + "[Step 2/7] Binary ready on all %d host(s).", len(unique_hosts) + ) + + target_nofile = max(65536, requirements.ulimit_nofile) + common_flags = _common_workload_flags(cfg) + + # Step 3: Launch passive server nodes and await NODE_READY. + logging.info( + "[Step 3/7] Launching %d passive server node(s) (%d" + " instance(s)/node)...", + len(cfg.server_nodes), + cfg.servers_per_node, + ) + server_procs = _spawn_node_group( + role_prefix="server", + hosts=cfg.server_nodes, + explicit_ips=cfg.server_ips, + instances_per_node=cfg.servers_per_node, + base_port=cfg.server_base_port, + common_flags=common_flags, + extra_flags=(), + executor=executor, + host_to_bin=host_to_bin, + target_nofile=target_nofile, + ) + target_entries = _wait_for_nodes_ready(server_procs, cfg.ready_timeout_sec) + logging.info( + "[Step 3/7] All %d server node(s) ready (%d total server endpoint(s)).", + len(server_procs), + len(target_entries), + ) + + # Step 4: Launch active client nodes with --targets and await NODE_READY. + logging.info( + "[Step 4/7] Launching %d active client node(s) (%d instance(s)/node," + " pattern=%s, targeting %d server endpoint(s))...", + len(cfg.client_nodes), + cfg.clients_per_node, + cfg.traffic_pattern, + len(target_entries), + ) + client_procs = _spawn_node_group( + role_prefix="client", + hosts=cfg.client_nodes, + explicit_ips=cfg.client_ips, + instances_per_node=cfg.clients_per_node, + base_port=cfg.client_base_port, + common_flags=common_flags, + extra_flags=(f"--targets={','.join(target_entries)}",), + executor=executor, + host_to_bin=host_to_bin, + target_nofile=target_nofile, + ) + _wait_for_nodes_ready(client_procs, cfg.ready_timeout_sec) + logging.info("[Step 4/7] All %d client node(s) ready.", len(client_procs)) + + # Step 5: Global START barrier across all server and client nodes. + logging.info( + "[Step 5/7] Releasing global START barrier across %d server(s) and %d" + " client(s)...", + len(server_procs), + len(client_procs), + ) + _broadcast_command(server_procs, "START") + _broadcast_command(client_procs, "START") + + # Step 6: Wait for all active client nodes to emit WORKLOAD_DONE. + logging.info( + "[Step 6/7] Running workload (%s, num_xfers=%d) and waiting for" + " WORKLOAD_DONE from %d client node(s) (timeout=%.1fs)...", + cfg.workload, + cfg.num_xfers, + len(client_procs), + cfg.workload_timeout_sec, + ) + _wait_for_workload_done( + client_procs, server_procs, cfg.workload_timeout_sec + ) + logging.info( + "[Step 6/7] All %d client node(s) completed workload.", + len(client_procs), + ) + + # Step 7: Broadcast STOP to all nodes and collect NODE_DONE. + logging.info( + "[Step 7/7] Broadcasting STOP command and collecting final metrics from" + " all %d node(s)...", + len(client_procs) + len(server_procs), + ) + _stop_and_collect_nodes([*client_procs, *server_procs]) + logging.info( + "[Step 7/7] All nodes stopped cleanly. Aggregating cluster report." + ) + + config_summary: Mapping[str, Any] = { + "num_client_nodes": len(cfg.client_nodes), + "clients_per_node": cfg.clients_per_node, + "num_server_nodes": len(cfg.server_nodes), + "servers_per_node": cfg.servers_per_node, + "traffic_pattern": cfg.traffic_pattern, + "workload": cfg.workload, + "conn": cfg.conn, + "num_xfers": cfg.num_xfers, + } + return report.aggregate_cluster_results( + config_summary=config_summary, + sender_workload_snapshots=[ + cp.workload_done_payload + for cp in client_procs + if cp.workload_done_payload is not None + ], + sender_final_snapshots=[ + cp.node_done_payload + for cp in client_procs + if cp.node_done_payload is not None + ], + receiver_final_snapshots=[ + sp.node_done_payload + for sp in server_procs + if sp.node_done_payload is not None + ], + ) + finally: + for p in [*client_procs, *server_procs]: + p.terminate() + executor.cleanup() diff --git a/peregrine/test/cluster/orchestrator/node_executor.py b/peregrine/test/cluster/orchestrator/node_executor.py new file mode 100644 index 0000000..636270d --- /dev/null +++ b/peregrine/test/cluster/orchestrator/node_executor.py @@ -0,0 +1,290 @@ +"""Local subprocess and SSH-pipe node execution for Peregrine cluster tests.""" + +from collections.abc import Sequence +from concurrent import futures +import ipaddress +import os +import resource +import shlex +import shutil +import socket +import subprocess +import uuid + +from absl import logging + +from peregrine.test.cluster.orchestrator import preflight + +DEFAULT_SSH_CMD = ( + "ssh -o BatchMode=yes -o StrictHostKeyChecking=no " + "-o ControlMaster=auto -o ControlPath=/tmp/peregrine_ssh_%r@%h:%p " + "-o ControlPersist=60s" +) + + +def is_local_node(host: str) -> bool: + """Returns True if `host` refers to the local machine.""" + normalized = host.strip().lower() + if normalized in {"localhost", "127.0.0.1", "::1", "[::1]"}: + return True + local_names = {socket.gethostname().lower(), socket.getfqdn().lower()} + return normalized in local_names + + +def _find_prebuilt_binary() -> str | None: + """Searches runfiles and blaze-bin/bazel-bin for an existing `:node` binary.""" + candidates: list[str] = [] + # 1. Sibling `../node/node` directory (works inside Blaze/Bazel runfiles tree) + here = os.path.dirname(os.path.abspath(__file__)) + candidates.append(os.path.normpath(os.path.join(here, "..", "node", "node"))) + + # 2. Runfiles environment variables. + runfiles_dir = os.environ.get("RUNFILES_DIR") or os.environ.get("TEST_SRCDIR") + if runfiles_dir: + for rel in ( + "google3/third_party/peregrine/test/cluster/node/node", + "_main/peregrine/test/cluster/node/node", + "peregrine/test/cluster/node/node", + ): + candidates.append(os.path.join(runfiles_dir, rel)) + + # 3. Workspace blaze-bin / bazel-bin outputs. + cwd = os.environ.get("BUILD_WORKING_DIRECTORY", os.getcwd()) + candidates.extend([ + os.path.join( + cwd, "blaze-bin/third_party/peregrine/test/cluster/node/node" + ), + os.path.join(cwd, "bazel-bin/peregrine/test/cluster/node/node"), + ]) + + for path in candidates: + if os.path.isfile(path) and os.access(path, os.X_OK): + resolved = os.path.abspath(path) + logging.info(" Found pre-built :node binary: %s", resolved) + return resolved + return None + + +def locate_node_binary(explicit_path: str = "") -> str: + """Locates the `:node` binary via `--node_binary`, runfiles, or bin dir.""" + if explicit_path: + abs_path = os.path.abspath(explicit_path) + if not os.path.isfile(abs_path) or not os.access(abs_path, os.X_OK): + raise FileNotFoundError( + "Specified --node_binary does not exist or is not executable:" + f" {abs_path}" + ) + logging.info(" Using explicit --node_binary: %s", abs_path) + return abs_path + + found = _find_prebuilt_binary() + if found is not None: + return found + + raise FileNotFoundError( + "Could not locate :node binary in runfiles or bin directory; " + "build it first or pass --node_binary." + ) + + +def raise_local_nofile_limit(target_nofile: int = 1048576) -> None: + """Raises local soft RLIMIT_NOFILE up to min(target_nofile, hard_limit).""" + try: + soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + desired = ( + target_nofile + if hard == resource.RLIM_INFINITY + else min(target_nofile, hard) + ) + if soft < desired: + resource.setrlimit(resource.RLIMIT_NOFILE, (desired, hard)) + except (OSError, ValueError): + pass + + +class NodeExecutor: + """Executes commands and stages/runs `:node` locally or via `--ssh_cmd` pipe.""" + + def __init__( + self, + ssh_cmd: str = DEFAULT_SSH_CMD, + remote_dir: str = "/tmp", + session_id: str | None = None, + ): + self._ssh_argv = shlex.split(ssh_cmd) + self._remote_dir = remote_dir.rstrip("/") or "/tmp" + self._session_id = session_id or uuid.uuid4().hex[:8] + self._staged_remotes: dict[str, str] = {} + + def _build_ssh_cmd(self, host: str, remote_shell_cmd: str) -> list[str]: + if self._ssh_argv and os.path.basename(self._ssh_argv[0]) == "gcloud": + return [*self._ssh_argv, host, f"--command={remote_shell_cmd}"] + return [*self._ssh_argv, host, remote_shell_cmd] + + def run_shell(self, host: str, shell_cmd: str, timeout: float = 30.0) -> str: + """Runs a shell command on `host` and returns stdout.""" + if is_local_node(host): + res = subprocess.run( + ["bash", "-c", shell_cmd], + capture_output=True, + text=True, + timeout=timeout, + check=True, + ) + return res.stdout + res = subprocess.run( + self._build_ssh_cmd(host, shell_cmd), + capture_output=True, + text=True, + timeout=timeout, + check=True, + ) + return res.stdout + + def inspect_node(self, host: str) -> preflight.NodeSysInfo: + """Inspects kernel sysctl and ulimit values on `host`.""" + if is_local_node(host): + return preflight.inspect_local_node(host) + out = self.run_shell(host, preflight.REMOTE_PROBE_SCRIPT) + return preflight.parse_probe_output(host, out) + + def resolve_bind_ip(self, host: str, explicit_ip: str = "") -> str: + """Resolves the Peregrine bind IP for `host` (supports `user@host` or `user@ip`).""" + if explicit_ip: + return explicit_ip + if is_local_node(host): + return "127.0.0.1" + host_only = host.split("@", 1)[-1].strip() + try: + ipaddress.ip_address(host_only) + return host_only + except ValueError: + pass + out = self.run_shell(host, "hostname -I | awk '{print $1}'").strip() + if not out: + raise RuntimeError(f"Failed to resolve bind IP on remote host '{host}'") + return out + + def _stage_to_single_remote(self, host: str, local_binary_path: str) -> str: + """Stages `local_binary_path` to `host` via SSH stdin pipe.""" + remote_path = f"{self._remote_dir}/peregrine_node_{self._session_id}" + remote_cmd = ( + f"cat > {shlex.quote(remote_path)} && " + f"chmod +x {shlex.quote(remote_path)}" + ) + ssh_cmd = self._build_ssh_cmd(host, remote_cmd) + logging.info(" Staging command for %s: %s", host, shlex.join(ssh_cmd)) + with open(local_binary_path, "rb") as src_file: + proc = subprocess.Popen( + ssh_cmd, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + assert proc.stdin is not None + copy_err: OSError | None = None + try: + shutil.copyfileobj(src_file, proc.stdin, length=1024 * 1024) + except OSError as e: + copy_err = e + _, stderr_bytes = proc.communicate(timeout=120) + if proc.returncode != 0 or copy_err is not None: + err_msg = stderr_bytes.decode("utf-8", errors="replace").strip() + if copy_err is not None: + err_msg = ( + f"{err_msg} (copyfileobj error: {copy_err})" + if err_msg + else f"copyfileobj error: {copy_err}" + ) + raise RuntimeError( + f"Failed to stage binary to '{host}' via SSH pipe: {err_msg}" + ) + return remote_path + + def stage_binary( + self, hosts: Sequence[str], local_binary_path: str + ) -> dict[str, str]: + """Stages `local_binary_path` to all unique hosts (zero-copy on localhost).""" + unique_hosts = list(dict.fromkeys(hosts)) + host_to_bin: dict[str, str] = {} + remote_hosts: list[str] = [] + + for h in unique_hosts: + if is_local_node(h): + host_to_bin[h] = local_binary_path + else: + remote_hosts.append(h) + + if remote_hosts: + logging.info( + " Staging :node binary via SSH pipe to %d remote host(s): %s...", + len(remote_hosts), + ", ".join(remote_hosts), + ) + with futures.ThreadPoolExecutor( + max_workers=min(16, len(remote_hosts)) + ) as pool: + future_map = { + pool.submit(self._stage_to_single_remote, h, local_binary_path): h + for h in remote_hosts + } + for fut in futures.as_completed(future_map): + h = future_map[fut] + remote_path = fut.result() + host_to_bin[h] = remote_path + self._staged_remotes[h] = remote_path + logging.info(" Staged :node to %s:%s", h, remote_path) + + return host_to_bin + + def spawn_node( + self, + host: str, + binary_path: str, + args: Sequence[str], + target_nofile: int = 1048576, + ) -> subprocess.Popen[str]: + """Spawns a `:node` process on `host` with line-buffered stdin/stdout/stderr.""" + if is_local_node(host): + raise_local_nofile_limit(target_nofile) + return subprocess.Popen( + [binary_path, *args], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + + quoted_cmd = " ".join(shlex.quote(x) for x in [binary_path, *args]) + remote_cmd = ( + "echo $$ 2>/dev/null >/dev/cgroup/memory/cgroup.procs || true; " + f"ulimit -n {int(target_nofile)} 2>/dev/null || true; exec {quoted_cmd}" + ) + return subprocess.Popen( + self._build_ssh_cmd(host, remote_cmd), + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + bufsize=1, + ) + + def cleanup(self) -> None: + """Removes any remote staged binaries.""" + if self._staged_remotes: + logging.info( + "Cleaning up staged :node binary on %d remote host(s)...", + len(self._staged_remotes), + ) + for host, remote_path in list(self._staged_remotes.items()): + try: + subprocess.run( + self._build_ssh_cmd(host, f"rm -f {shlex.quote(remote_path)}"), + capture_output=True, + timeout=15, + check=False, + ) + except OSError: + pass + self._staged_remotes.clear() diff --git a/peregrine/test/cluster/orchestrator/node_executor_test.py b/peregrine/test/cluster/orchestrator/node_executor_test.py new file mode 100644 index 0000000..d0dc023 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/node_executor_test.py @@ -0,0 +1,87 @@ +"""Unit tests for node_executor.py.""" + +import socket +import tempfile +from unittest import mock + +from absl.testing import absltest + +from peregrine.test.cluster.orchestrator import node_executor + + +class NodeExecutorTest(absltest.TestCase): + + def test_is_local_node(self): + self.assertTrue(node_executor.is_local_node("localhost")) + self.assertTrue(node_executor.is_local_node("127.0.0.1")) + self.assertTrue(node_executor.is_local_node("::1")) + self.assertTrue(node_executor.is_local_node(socket.gethostname())) + self.assertFalse(node_executor.is_local_node("10.128.0.99")) + + def test_stage_binary_skips_ssh_on_localhost(self): + executor = node_executor.NodeExecutor() + mapping = executor.stage_binary(["localhost", "127.0.0.1"], "/bin/true") + self.assertEqual(mapping["localhost"], "/bin/true") + self.assertEqual(mapping["127.0.0.1"], "/bin/true") + + def test_stage_binary_streams_via_ssh_stdin_on_remote(self): + with tempfile.NamedTemporaryFile(mode="wb", delete=False) as tmp: + tmp.write(b"fake_elf_payload") + tmp_path = tmp.name + + executor = node_executor.NodeExecutor( + ssh_cmd="ssh -p 2222", remote_dir="/tmp", session_id="test1234" + ) + fake_proc = mock.MagicMock() + fake_proc.stdin = mock.MagicMock() + fake_proc.communicate.return_value = (b"", b"") + fake_proc.returncode = 0 + + with mock.patch.object( + node_executor.subprocess, "Popen", return_value=fake_proc + ) as mock_popen: + mapping = executor.stage_binary(["remote-host-1"], tmp_path) + self.assertEqual(mapping["remote-host-1"], "/tmp/peregrine_node_test1234") + called_argv = mock_popen.call_args[0][0] + self.assertEqual( + called_argv[:4], + ["ssh", "-p", "2222", "remote-host-1"], + ) + self.assertIn("cat > /tmp/peregrine_node_test1234", called_argv[4]) + + def test_stage_binary_handles_copyfileobj_broken_pipe(self): + with tempfile.NamedTemporaryFile(mode="wb", delete=False) as tmp: + tmp.write(b"x" * (2 * 1024 * 1024)) + tmp_path = tmp.name + + executor = node_executor.NodeExecutor( + ssh_cmd="bash -c 'echo Permission denied >&2; exit 255' --", + remote_dir="/tmp", + session_id="test1234", + ) + with self.assertRaisesRegex( + RuntimeError, + r"Permission denied.*copyfileobj error", + ): + executor.stage_binary(["remote-host-1"], tmp_path) + + def test_build_ssh_cmd_supports_gcloud_compute_ssh(self): + executor = node_executor.NodeExecutor( + ssh_cmd="gcloud compute ssh --project=my-project --zone=us-central1-a" + ) + self.assertEqual( + executor._build_ssh_cmd("user@gce-vm-1", "echo ready"), + [ + "gcloud", + "compute", + "ssh", + "--project=my-project", + "--zone=us-central1-a", + "user@gce-vm-1", + "--command=echo ready", + ], + ) + + +if __name__ == "__main__": + absltest.main() diff --git a/peregrine/test/cluster/orchestrator/preflight.py b/peregrine/test/cluster/orchestrator/preflight.py new file mode 100644 index 0000000..8ad078c --- /dev/null +++ b/peregrine/test/cluster/orchestrator/preflight.py @@ -0,0 +1,325 @@ +"""Preflight kernel sysctl and ulimit validator for Peregrine cluster tests.""" + +from collections.abc import Callable, Sequence +from concurrent import futures +import dataclasses +import math +import os +import resource + +from absl import logging + + +@dataclasses.dataclass(frozen=True) +class NodeRequirements: + """Minimum OS kernel and resource limits required on each cluster node.""" + + somaxconn: int + tcp_max_syn_backlog: int + netdev_max_backlog: int + ephemeral_ports: int + vm_max_map_count: int + kernel_threads_max: int + ulimit_nofile: int + + +@dataclasses.dataclass(frozen=True) +class NodeSysInfo: + """Observed OS kernel and resource limits on a single node.""" + + host: str + somaxconn: int + tcp_max_syn_backlog: int + netdev_max_backlog: int | None + ephemeral_port_low: int + ephemeral_port_high: int + vm_max_map_count: int + kernel_threads_max: int + ulimit_nofile_hard: int + + @property + def ephemeral_ports(self) -> int: + """Returns the total number of available ephemeral ports.""" + return max(0, self.ephemeral_port_high - self.ephemeral_port_low + 1) + + +class Error(RuntimeError): + """Raised when preflight probing or parsing fails on a cluster node.""" + + +def compute_requirements( + num_client_nodes: int, + clients_per_node: int, + num_server_nodes: int, + servers_per_node: int, + conn: int, + traffic_pattern: str = "all_to_all", + colocated: bool = False, + min_somaxconn: int = 0, +) -> NodeRequirements: + """Computes minimum required kernel sysctl and ulimit thresholds.""" + total_clients = max(1, num_client_nodes * clients_per_node) + total_servers = max(1, num_server_nodes * servers_per_node) + conn = max(1, conn) + + if traffic_pattern.lower() == "round_robin": + max_clients_per_server = math.ceil(total_clients / total_servers) + targets_per_client = 1 + else: + max_clients_per_server = total_clients + targets_per_client = total_servers + + c_in = max_clients_per_server * conn + c_out = clients_per_node * targets_per_client * (conn + 1) + p_node = ( + (clients_per_node + servers_per_node) + if colocated + else max(clients_per_node, servers_per_node) + ) + + somaxconn = max(1024, c_in, min_somaxconn) + tcp_max_syn_backlog = max(1024, c_in, min_somaxconn) + netdev_max_backlog = max(1000, c_in) + ephemeral_ports = 2 * p_node + c_out + 1024 + vm_max_map_count = 20 * p_node + 4 * c_in + kernel_threads_max = 20 * p_node + 4 * c_in + ulimit_nofile = 16 * p_node + 4 * (c_in + c_out) + 1024 + + return NodeRequirements( + somaxconn=somaxconn, + tcp_max_syn_backlog=tcp_max_syn_backlog, + netdev_max_backlog=netdev_max_backlog, + ephemeral_ports=ephemeral_ports, + vm_max_map_count=vm_max_map_count, + kernel_threads_max=kernel_threads_max, + ulimit_nofile=ulimit_nofile, + ) + + +def _read_local_proc_int(path: str) -> int | None: + """Reads an integer sysctl value from `/proc`, or None if unavailable.""" + try: + with open(path, "r", encoding="utf-8") as f: + return int(f.read().strip()) + except OSError: + return None + + +def inspect_local_node(host: str = "localhost") -> NodeSysInfo: + """Reads sysctl and RLIMIT_NOFILE values directly from the local machine.""" + somaxconn = _read_local_proc_int("/proc/sys/net/core/somaxconn") or 0 + syn_backlog = ( + _read_local_proc_int("/proc/sys/net/ipv4/tcp_max_syn_backlog") or 0 + ) + netdev_backlog = _read_local_proc_int("/proc/sys/net/core/netdev_max_backlog") + vm_map_count = _read_local_proc_int("/proc/sys/vm/max_map_count") or 0 + threads_max = _read_local_proc_int("/proc/sys/kernel/threads-max") or 0 + + port_low, port_high = 32768, 60999 + if os.path.exists("/proc/sys/net/ipv4/ip_local_port_range"): + with open( + "/proc/sys/net/ipv4/ip_local_port_range", "r", encoding="utf-8" + ) as f: + parts = f.read().split() + if len(parts) >= 2: + port_low, port_high = int(parts[0]), int(parts[1]) + + _, hard_nofile = resource.getrlimit(resource.RLIMIT_NOFILE) + if hard_nofile == resource.RLIM_INFINITY or hard_nofile < 0: + hard_nofile = 1048576 + + return NodeSysInfo( + host=host, + somaxconn=somaxconn, + tcp_max_syn_backlog=syn_backlog, + netdev_max_backlog=netdev_backlog, + ephemeral_port_low=port_low, + ephemeral_port_high=port_high, + vm_max_map_count=vm_map_count, + kernel_threads_max=threads_max, + ulimit_nofile_hard=hard_nofile, + ) + + +REMOTE_PROBE_SCRIPT = ( + "cat /proc/sys/net/core/somaxconn; " + "cat /proc/sys/net/ipv4/tcp_max_syn_backlog; " + "if [ -f /proc/sys/net/core/netdev_max_backlog ]; then " + "cat /proc/sys/net/core/netdev_max_backlog; else echo NA; fi; " + "cat /proc/sys/net/ipv4/ip_local_port_range; " + "cat /proc/sys/vm/max_map_count; " + "cat /proc/sys/kernel/threads-max; " + "ulimit -Hn" +) + + +def parse_probe_output(host: str, stdout: str) -> NodeSysInfo: + """Parses the 7-line output of REMOTE_PROBE_SCRIPT into NodeSysInfo.""" + lines = [line.strip() for line in stdout.strip().splitlines() if line.strip()] + if len(lines) < 7: + raise Error( + f"Incomplete preflight probe output from node '{host}': {stdout!r}" + ) + somaxconn = int(lines[0]) + syn_backlog = int(lines[1]) + netdev_backlog = None if lines[2] == "NA" else int(lines[2]) + port_parts = lines[3].split() + port_low, port_high = int(port_parts[0]), int(port_parts[1]) + vm_map_count = int(lines[4]) + threads_max = int(lines[5]) + ulimit_raw = lines[6] + ulimit_hard = ( + 1048576 if ulimit_raw.lower() == "unlimited" else int(ulimit_raw) + ) + return NodeSysInfo( + host=host, + somaxconn=somaxconn, + tcp_max_syn_backlog=syn_backlog, + netdev_max_backlog=netdev_backlog, + ephemeral_port_low=port_low, + ephemeral_port_high=port_high, + vm_max_map_count=vm_map_count, + kernel_threads_max=threads_max, + ulimit_nofile_hard=ulimit_hard, + ) + + +@dataclasses.dataclass(frozen=True) +class _CheckSpec: + param_name: str + get_actual: Callable[[NodeSysInfo], int | None] + get_required: Callable[[NodeRequirements], int] + format_remediation: Callable[[int], str] + + +_PREFLIGHT_CHECKS: tuple[_CheckSpec, ...] = ( + _CheckSpec( + param_name="net.core.somaxconn", + get_actual=lambda i: i.somaxconn, + get_required=lambda r: r.somaxconn, + format_remediation=lambda req: ( + f"sudo sysctl -w net.core.somaxconn={max(8192, req)}" + ), + ), + _CheckSpec( + param_name="net.ipv4.tcp_max_syn_backlog", + get_actual=lambda i: i.tcp_max_syn_backlog, + get_required=lambda r: r.tcp_max_syn_backlog, + format_remediation=lambda req: ( + f"sudo sysctl -w net.ipv4.tcp_max_syn_backlog={max(8192, req)}" + ), + ), + _CheckSpec( + param_name="net.core.netdev_max_backlog", + get_actual=lambda i: i.netdev_max_backlog, + get_required=lambda r: r.netdev_max_backlog, + format_remediation=lambda req: ( + f"sudo sysctl -w net.core.netdev_max_backlog={max(16384, req)}" + ), + ), + _CheckSpec( + param_name="net.ipv4.ip_local_port_range (width)", + get_actual=lambda i: i.ephemeral_ports, + get_required=lambda r: r.ephemeral_ports, + format_remediation=lambda _: ( + 'sudo sysctl -w net.ipv4.ip_local_port_range="10000 65535"' + ), + ), + _CheckSpec( + param_name="vm.max_map_count", + get_actual=lambda i: i.vm_max_map_count, + get_required=lambda r: r.vm_max_map_count, + format_remediation=lambda req: ( + f"sudo sysctl -w vm.max_map_count={max(262144, req)}" + ), + ), + _CheckSpec( + param_name="kernel.threads-max", + get_actual=lambda i: i.kernel_threads_max, + get_required=lambda r: r.kernel_threads_max, + format_remediation=lambda req: ( + f"sudo sysctl -w kernel.threads-max={max(131072, req)}" + ), + ), + _CheckSpec( + param_name="ulimit -Hn (RLIMIT_NOFILE)", + get_actual=lambda i: i.ulimit_nofile_hard, + get_required=lambda r: r.ulimit_nofile, + format_remediation=lambda req: ( + f"Raise hard nofile limit >= {req} in /etc/security/limits.conf" + ), + ), +) + + +def evaluate_node( + info: NodeSysInfo, req: NodeRequirements +) -> list[tuple[str, str, int, int, str]]: + """Returns a list of (host, param, actual, required, remediation) failures.""" + failures: list[tuple[str, str, int, int, str]] = [] + for spec in _PREFLIGHT_CHECKS: + actual = spec.get_actual(info) + required = spec.get_required(req) + if actual is not None and actual < required: + failures.append(( + info.host, + spec.param_name, + actual, + required, + spec.format_remediation(required), + )) + return failures + + +def format_failures_warning( + failures: Sequence[tuple[str, str, int, int, str]], +) -> str: + """Formats preflight threshold mismatches into a human-readable warning table.""" + if not failures: + return "" + lines = [ + ( + "Preflight kernel/resource check found sub-optimal limits on one or" + " more nodes (continuing anyway):" + ), + ( + f" {'NODE':<20} {'PARAMETER':<36} {'ACTUAL':>10}" + f" {'RECOMMENDED':>11} REMEDIATION" + ), + " " + "-" * 106, + ] + for host, param, actual, required, remediation in failures: + lines.append( + f" {host:<20} {param:<36} {actual:>10} {required:>11} {remediation}" + ) + return "\n".join(lines) + + +def verify_nodes( + hosts: Sequence[str], + requirements: NodeRequirements, + inspect_fn: Callable[[str], NodeSysInfo], +) -> list[NodeSysInfo]: + """Inspects all unique hosts in parallel and logs warnings on sub-optimal limits.""" + unique_hosts = list(dict.fromkeys(hosts)) + infos: list[NodeSysInfo] = [] + with futures.ThreadPoolExecutor( + max_workers=max(1, min(16, len(unique_hosts))) + ) as pool: + future_map = {h: pool.submit(inspect_fn, h) for h in unique_hosts} + for h in unique_hosts: + try: + infos.append(future_map[h].result()) + except Exception as e: # pylint: disable=broad-exception-caught + logging.warning( + "Preflight probe failed on node '%s' (continuing anyway): %s", h, e + ) + + all_failures: list[tuple[str, str, int, int, str]] = [] + for info in infos: + all_failures.extend(evaluate_node(info, requirements)) + + if all_failures: + logging.warning("%s", format_failures_warning(all_failures)) + + return infos diff --git a/peregrine/test/cluster/orchestrator/preflight_test.py b/peregrine/test/cluster/orchestrator/preflight_test.py new file mode 100644 index 0000000..66b7636 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/preflight_test.py @@ -0,0 +1,77 @@ +"""Unit tests for preflight.py.""" + +from absl.testing import absltest +from peregrine.test.cluster.orchestrator import preflight + + +class PreflightTest(absltest.TestCase): + + def test_compute_requirements_fan_in_1000(self): + req = preflight.compute_requirements( + num_client_nodes=1, + clients_per_node=1000, + num_server_nodes=1, + servers_per_node=1, + conn=1, + traffic_pattern="all_to_all", + colocated=True, + ) + self.assertGreaterEqual(req.somaxconn, 1000) + self.assertGreaterEqual(req.tcp_max_syn_backlog, 1000) + self.assertGreaterEqual(req.ulimit_nofile, 20000) + + def test_parse_and_evaluate_node_passes_when_limits_sufficient(self): + req = preflight.compute_requirements( + num_client_nodes=1, + clients_per_node=1000, + num_server_nodes=1, + servers_per_node=1, + conn=1, + traffic_pattern="all_to_all", + colocated=True, + ) + probe_out = "\n".join([ + "4096", + "4096", + "NA", + "32768 60999", + "65530", + "2054695", + "1048576", + ]) + info = preflight.parse_probe_output("node-a", probe_out) + self.assertIsNone(info.netdev_max_backlog) + self.assertEqual(preflight.evaluate_node(info, req), []) + + def test_verify_nodes_warns_without_raising_on_insufficient_somaxconn(self): + req = preflight.compute_requirements( + num_client_nodes=1, + clients_per_node=2000, + num_server_nodes=1, + servers_per_node=1, + conn=1, + ) + bad_info = preflight.NodeSysInfo( + host="node-low", + somaxconn=128, + tcp_max_syn_backlog=4096, + netdev_max_backlog=2000, + ephemeral_port_low=10000, + ephemeral_port_high=65000, + vm_max_map_count=262144, + kernel_threads_max=262144, + ulimit_nofile_hard=1048576, + ) + with self.assertLogs(level="WARNING") as cm: + infos = preflight.verify_nodes(["node-low"], req, lambda _: bad_info) + self.assertLen(infos, 1) + self.assertIn("net.core.somaxconn", "\n".join(cm.output)) + + def test_inspect_local_node_succeeds(self): + info = preflight.inspect_local_node("localhost") + self.assertGreater(info.somaxconn, 0) + self.assertGreater(info.ephemeral_ports, 0) + + +if __name__ == "__main__": + absltest.main() diff --git a/peregrine/test/cluster/orchestrator/report.py b/peregrine/test/cluster/orchestrator/report.py new file mode 100644 index 0000000..3982a00 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/report.py @@ -0,0 +1,302 @@ +"""Multi-node JSON metrics parser and summary report generator.""" + +from collections.abc import Mapping, Sequence +import json +from typing import Any + +NODE_READY_PREFIX = "PEREGRINE_NODE_READY " +METRICS_SAMPLE_PREFIX = "PEREGRINE_METRICS_SAMPLE " +WORKLOAD_DONE_PREFIX = "PEREGRINE_WORKLOAD_DONE " +NODE_DONE_PREFIX = "PEREGRINE_NODE_DONE " + + +def parse_prefixed_json(line: str, prefix: str) -> dict[str, Any] | None: + """Extracts and parses the JSON payload if `line` starts with `prefix`.""" + stripped = line.strip() + if not stripped.startswith(prefix): + return None + payload = stripped[len(prefix) :].strip() + return json.loads(payload) + + +def _merge_histogram(dst: dict[str, Any], src: Mapping[str, Any]) -> None: + """Accumulates `src` Log2Histogram counts and sum into `dst`.""" + dst["sum"] = dst.get("sum", 0) + int(src.get("sum", 0)) + dst["count"] = dst.get("count", 0) + int(src.get("count", 0)) + src_buckets = list(src.get("buckets", [0] * 32)) + dst_buckets = dst.setdefault("buckets", [0] * len(src_buckets)) + for i, val in enumerate(src_buckets): + if i < len(dst_buckets): + dst_buckets[i] += int(val) + + +def _merge_op_metrics(dst: dict[str, Any], src: Mapping[str, Any]) -> None: + """Accumulates `src` operation bytes, errors, and histograms into `dst`.""" + dst["bytes"] = dst.get("bytes", 0) + int(src.get("bytes", 0)) + dst["errors"] = dst.get("errors", 0) + int(src.get("errors", 0)) + _merge_histogram( + dst.setdefault("e2e_latency_us", {}), src.get("e2e_latency_us", {}) + ) + _merge_histogram( + dst.setdefault("request_size_bytes", {}), + src.get("request_size_bytes", {}), + ) + + +def merge_transport_dicts( + snapshots: Sequence[Mapping[str, Any]], +) -> dict[str, Any]: + """Merges the `transport` section across multiple node snapshots.""" + merged: dict[str, Any] = { + "write": { + "bytes": 0, + "errors": 0, + "e2e_latency_us": {"sum": 0, "count": 0, "buckets": [0] * 32}, + "request_size_bytes": {"sum": 0, "count": 0, "buckets": [0] * 32}, + }, + "read": { + "bytes": 0, + "errors": 0, + "e2e_latency_us": {"sum": 0, "count": 0, "buckets": [0] * 32}, + "request_size_bytes": {"sum": 0, "count": 0, "buckets": [0] * 32}, + }, + "tcp_connect_failures": 0, + "rpc_requests_received": 0, + } + for snap in snapshots: + t = snap.get("transport", {}) + _merge_op_metrics(merged["write"], t.get("write", {})) + _merge_op_metrics(merged["read"], t.get("read", {})) + merged["tcp_connect_failures"] += int(t.get("tcp_connect_failures", 0)) + merged["rpc_requests_received"] += int(t.get("rpc_requests_received", 0)) + return merged + + +def _percentile_from_log2_histogram_ms( + hist: Mapping[str, Any], q: float +) -> float: + """Estimates the q-th quantile (in ms) from a Log2Histogram<32> (in us).""" + buckets = [int(x) for x in hist.get("buckets", ())] + total = sum(buckets) + if total <= 0: + return 0.0 + target = q * total + cum = 0 + for i, count in enumerate(buckets): + if count <= 0: + continue + prev_cum = cum + cum += count + if cum >= target: + low_us = 0.0 if i == 0 else float(1 << (i - 1)) + high_us = 1.0 if i == 0 else float(1 << i) + frac = max(0.0, min(1.0, (target - prev_cum) / count)) + return (low_us + frac * (high_us - low_us)) / 1000.0 + return 0.0 + + +def _summarize_op_latency(op: Mapping[str, Any]) -> dict[str, Any]: + """Computes count, mean_ms, and percentiles from an OpMetrics dict.""" + hist = op.get("e2e_latency_us", {}) + bucket_sum = sum(int(x) for x in hist.get("buckets", ())) + count = int(hist.get("count", 0)) or bucket_sum + if count > 0: + mean_ms = (float(hist.get("sum", 0)) / count) / 1000.0 + p50_ms = _percentile_from_log2_histogram_ms(hist, 0.50) + p90_ms = _percentile_from_log2_histogram_ms(hist, 0.90) + p99_ms = _percentile_from_log2_histogram_ms(hist, 0.99) + p999_ms = _percentile_from_log2_histogram_ms(hist, 0.999) + else: + mean_ms = p50_ms = p90_ms = p99_ms = p999_ms = 0.0 + return { + "count": count, + "mean_ms": mean_ms, + "p50_ms": p50_ms, + "p90_ms": p90_ms, + "p99_ms": p99_ms, + "p999_ms": p999_ms, + } + + +def _aggregate_latency( + active_snaps: Sequence[Mapping[str, Any]], + sender_transport: Mapping[str, Any], +) -> dict[str, Any]: + """Computes cluster-wide latency percentiles, elapsed time, and throughput.""" + write_op = sender_transport.get("write", {}) + read_op = sender_transport.get("read", {}) + write_lat = _summarize_op_latency(write_op) + read_lat = _summarize_op_latency(read_op) + + combined_hist: dict[str, Any] = {"sum": 0, "count": 0, "buckets": [0] * 32} + _merge_histogram(combined_hist, write_op.get("e2e_latency_us", {})) + _merge_histogram(combined_hist, read_op.get("e2e_latency_us", {})) + combined_lat = _summarize_op_latency({"e2e_latency_us": combined_hist}) + + total_count = combined_lat["count"] + total_errors = int(write_op.get("errors", 0)) + int(read_op.get("errors", 0)) + xfer_size_bytes = ( + max(int(s.get("xfer_size_bytes", 0)) for s in active_snaps) + if active_snaps + else 0 + ) + elapsed_ms = ( + max(float(s.get("cpu", {}).get("wall_ms", 0.0)) for s in active_snaps) + if active_snaps + else 0.0 + ) + + transfers_per_sec = ( + (total_count / (elapsed_ms / 1000.0)) if elapsed_ms > 0.0 else 0.0 + ) + total_bytes = int(write_op.get("bytes", 0)) + int(read_op.get("bytes", 0)) + if total_bytes <= 0: + total_bytes = total_count * xfer_size_bytes + throughput_gbps = ( + (total_bytes * 8.0) / ((elapsed_ms / 1000.0) * 1e9) + if elapsed_ms > 0.0 + else 0.0 + ) + + return { + "count": total_count, + "errors": total_errors, + "xfer_size_bytes": xfer_size_bytes, + "p50_ms": combined_lat["p50_ms"], + "p90_ms": combined_lat["p90_ms"], + "p99_ms": combined_lat["p99_ms"], + "p999_ms": combined_lat["p999_ms"], + "mean_ms": combined_lat["mean_ms"], + "write": write_lat, + "read": read_lat, + "elapsed_ms": elapsed_ms, + "transfers_per_sec": transfers_per_sec, + "throughput_gbps": throughput_gbps, + } + + +def summarize_node_snapshot(snap: Mapping[str, Any]) -> dict[str, Any]: + """Computes latency and throughput summary for a single node snapshot.""" + return _aggregate_latency([snap], merge_transport_dicts([snap])) + + +def aggregate_cluster_results( + config_summary: Mapping[str, Any], + sender_workload_snapshots: Sequence[Mapping[str, Any]], + sender_final_snapshots: Sequence[Mapping[str, Any]], + receiver_final_snapshots: Sequence[Mapping[str, Any]], +) -> dict[str, Any]: + """Aggregates final metrics across all sender (client) and receiver (server) nodes.""" + active_snaps = ( + sender_workload_snapshots + if sender_workload_snapshots + else sender_final_snapshots + ) + sender_transport = merge_transport_dicts( + sender_final_snapshots or active_snaps + ) + + sender_cpu_cores = sum( + float(s.get("cpu", {}).get("avg_cores", 0.0)) for s in active_snaps + ) + receiver_cpu_cores = sum( + float(s.get("cpu", {}).get("avg_cores", 0.0)) + for s in receiver_final_snapshots + ) + + return { + "config": dict(config_summary), + "latency": _aggregate_latency(active_snaps, sender_transport), + "cpu": { + "sender_total_avg_cores": sender_cpu_cores, + "receiver_total_avg_cores": receiver_cpu_cores, + }, + "sender_transport": sender_transport, + "receiver_transport": merge_transport_dicts(receiver_final_snapshots), + } + + +def format_summary_table(aggregated: Mapping[str, Any]) -> str: + """Formats the aggregated cluster benchmark results into a readable table.""" + cfg = aggregated.get("config", {}) + lat = aggregated.get("latency", {}) + cpu = aggregated.get("cpu", {}) + st = aggregated.get("sender_transport", {}) + rt = aggregated.get("receiver_transport", {}) + write_lat = lat.get("write", _summarize_op_latency(st.get("write", {}))) + read_lat = lat.get("read", _summarize_op_latency(st.get("read", {}))) + + lines = [ + "=" * 76, + "Peregrine Cluster Bipartite Benchmark Results", + "=" * 76, + ( + f"Topology : N={cfg.get('num_client_nodes', 1)} client" + f" node(s) x N_p={cfg.get('clients_per_node', 1)} instances ->" + f" M={cfg.get('num_server_nodes', 1)} server node(s) x" + f" M_p={cfg.get('servers_per_node', 1)} instances" + ), + ( + "Traffic / Workload:" + f" pattern={cfg.get('traffic_pattern', 'all_to_all')}," + f" workload={cfg.get('workload', 'serial_fixed_write')}," + f" conn={cfg.get('conn', 1)}, num_xfers={cfg.get('num_xfers', 1)}" + ), + f"Transfer Size : {lat.get('xfer_size_bytes', 0)} B", + "-" * 76, + ( + f"Completed Xfers : {lat.get('count', 0)}" + f" (errors: {lat.get('errors', 0)})" + ), + f"Elapsed Time : {lat.get('elapsed_ms', 0.0):.3f} ms", + f"Transfer Rate : {lat.get('transfers_per_sec', 0.0):.2f} xfers/s", + f"Throughput : {lat.get('throughput_gbps', 0.0):.4f} Gbps", + "-" * 76, + "End-to-End Post() Completion Latency (ms):", + ( + f" Write: Count {write_lat.get('count', 0)}," + f" Mean {write_lat.get('mean_ms', 0.0):.3f} ms" + ), + ( + " p50/90/99/99.9 :" + f" {write_lat.get('p50_ms', 0.0):.3f}/{write_lat.get('p90_ms', 0.0):.3f}/" + f"{write_lat.get('p99_ms', 0.0):.3f}/{write_lat.get('p999_ms', 0.0):.3f}" + " ms" + ), + ( + f" Read: Count {read_lat.get('count', 0)}," + f" Mean {read_lat.get('mean_ms', 0.0):.3f} ms" + ), + ( + " p50/90/99/99.9 :" + f" {read_lat.get('p50_ms', 0.0):.3f}/{read_lat.get('p90_ms', 0.0):.3f}/" + f"{read_lat.get('p99_ms', 0.0):.3f}/{read_lat.get('p999_ms', 0.0):.3f}" + " ms" + ), + "-" * 76, + "Transport & CPU Telemetry:", + ( + " Client Write :" + f" {st.get('write', {}).get('bytes', 0)} B," + f" errors={st.get('write', {}).get('errors', 0)}," + f" tcp_connect_failures={st.get('tcp_connect_failures', 0)}" + ), + ] + read_bytes = int(st.get("read", {}).get("bytes", 0)) + read_errors = int(st.get("read", {}).get("errors", 0)) + if read_bytes > 0 or read_errors > 0: + lines.append(f" Client Read : {read_bytes} B, errors={read_errors}") + lines.extend([ + ( + " Server RPCs :" + f" rpc_requests_received={rt.get('rpc_requests_received', 0)}," + f" tcp_connect_failures={rt.get('tcp_connect_failures', 0)}" + ), + ( + " Avg CPU Cores :" + f" clients={cpu.get('sender_total_avg_cores', 0.0):.2f} cores," + f" servers={cpu.get('receiver_total_avg_cores', 0.0):.2f} cores" + ), + "=" * 76, + ]) + return "\n".join(lines) diff --git a/peregrine/test/cluster/orchestrator/report_test.py b/peregrine/test/cluster/orchestrator/report_test.py new file mode 100644 index 0000000..9139df5 --- /dev/null +++ b/peregrine/test/cluster/orchestrator/report_test.py @@ -0,0 +1,105 @@ +"""Unit tests for report.py.""" + +from absl.testing import absltest +from peregrine.test.cluster.orchestrator import report + + +class ReportTest(absltest.TestCase): + + def test_parse_prefixed_json(self): + line = 'PEREGRINE_NODE_READY {"targets":["127.0.0.1:10000@12345"]}\n' + parsed = report.parse_prefixed_json(line, report.NODE_READY_PREFIX) + self.assertIsNotNone(parsed) + self.assertEqual(parsed["targets"], ["127.0.0.1:10000@12345"]) + self.assertIsNone( + report.parse_prefixed_json(line, report.WORKLOAD_DONE_PREFIX) + ) + + def test_aggregate_and_format_summary_table(self): + buckets = [0] * 32 + buckets[14] = 1000 # [8192, 16384) us + sender_snap = { + "num_instances": 1000, + "xfer_size_bytes": 1, + "cpu": { + "wall_ms": 40.0, + "user_cpu_ms": 80.0, + "sys_cpu_ms": 40.0, + "avg_cores": 3.0, + }, + "transport": { + "write": { + "bytes": 1000, + "errors": 0, + "e2e_latency_us": { + "sum": 16500000, + "count": 1000, + "buckets": buckets, + }, + "request_size_bytes": { + "sum": 1000, + "count": 1000, + "buckets": [0] * 32, + }, + }, + "read": { + "bytes": 0, + "errors": 0, + "e2e_latency_us": {"sum": 0, "count": 0, "buckets": [0] * 32}, + "request_size_bytes": { + "sum": 0, + "count": 0, + "buckets": [0] * 32, + }, + }, + "tcp_connect_failures": 0, + "rpc_requests_received": 0, + }, + } + receiver_snap = { + "num_instances": 1, + "xfer_size_bytes": 1, + "cpu": { + "wall_ms": 45.0, + "user_cpu_ms": 10.0, + "sys_cpu_ms": 20.0, + "avg_cores": 0.67, + }, + "transport": { + "write": {"bytes": 0, "errors": 0}, + "read": {"bytes": 0, "errors": 0}, + "tcp_connect_failures": 0, + "rpc_requests_received": 1000, + }, + } + agg = report.aggregate_cluster_results( + config_summary={ + "num_client_nodes": 1, + "clients_per_node": 1000, + "num_server_nodes": 1, + "servers_per_node": 1, + "traffic_pattern": "all_to_all", + "workload": "serial_fixed_write", + "conn": 1, + "num_xfers": 1, + }, + sender_workload_snapshots=[sender_snap], + sender_final_snapshots=[sender_snap], + receiver_final_snapshots=[receiver_snap], + ) + self.assertEqual(agg["latency"]["count"], 1000) + self.assertAlmostEqual(agg["latency"]["mean_ms"], 16.5) + self.assertAlmostEqual(agg["latency"]["p50_ms"], 12.288) + self.assertEqual(agg["receiver_transport"]["rpc_requests_received"], 1000) + table = report.format_summary_table(agg) + self.assertIn("Completed Xfers : 1000 (errors: 0)", table) + self.assertIn("Elapsed Time : 40.000 ms", table) + self.assertIn(" Write: Count 1000, Mean 16.500 ms", table) + self.assertIn(" p50/90/99/99.9 : 12.288/15.565/16.302/16.376 ms", table) + self.assertIn(" Read: Count 0, Mean 0.000 ms", table) + self.assertIn(" p50/90/99/99.9 : 0.000/0.000/0.000/0.000 ms", table) + self.assertIn("rpc_requests_received=1000", table) + + +if __name__ == "__main__": + absltest.main()