GCC Code Coverage Report

src/
Coverage:
low: ≥ 0%
medium: ≥ 75.0%
high: ≥ 90.0%
Lines:
564 / 615
91.7%
Branches:
664 / 1376
48.3%
Functions:
47 / 47
100.0%
ps/rdma/control_plane.cc
Line Branch Exec Source
1 #include "ps/rdma/control_plane.h"
2
3 #include <grpcpp/grpcpp.h>
4
5 #include <chrono>
6 #include <cstdint>
7 #include <cstring>
8 #include <memory>
9 #include <mutex>
10 #include <stdexcept>
11 #include <string>
12 #include <utility>
13
14 #include "base/log.h"
15 #include "rdma_control_plane.grpc.pb.h"
16
17 namespace petps {
18
19 namespace {
20
21 using recstoreps::rdma::GetMetaRequest;
22 using recstoreps::rdma::GetMetaResponse;
23 using recstoreps::rdma::GetServerHealthRequest;
24 using recstoreps::rdma::GetServerHealthResponse;
25 using recstoreps::rdma::ProbeRequest;
26 using recstoreps::rdma::ProbeResponse;
27 using recstoreps::rdma::PublishMetaRequest;
28 using recstoreps::rdma::PublishMetaResponse;
29 using recstoreps::rdma::PublishServerDrainingRequest;
30 using recstoreps::rdma::PublishServerDrainingResponse;
31 using recstoreps::rdma::PublishServerReadyRequest;
32 using recstoreps::rdma::PublishServerReadyResponse;
33 using recstoreps::rdma::RawVerbsMetadata;
34 using recstoreps::rdma::RdmaControlPlane;
35 using recstoreps::rdma::WaitServerReadyRequest;
36 using recstoreps::rdma::WaitServerReadyResponse;
37 using recstoreps::rdma::WaitServerRequest;
38 using recstoreps::rdma::WaitServerResponse;
39
40 60 std::string EndpointString(const RdmaControlPlaneEndpoint& endpoint) {
41
2/4
✓ Branch 2 taken 60 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 60 times.
✗ Branch 6 not taken.
 120 return endpoint.host + ":" + std::to_string(endpoint.port);
42 }
43
44 92 std::chrono::system_clock::time_point DeadlineFromNow(int timeout_ms) {
45 92 return std::chrono::system_clock::now() +
46
1/2
✓ Branch 2 taken 92 times.
✗ Branch 3 not taken.
 276 std::chrono::milliseconds(timeout_ms);
47 }
48
49 26 std::string GrpcStatusText(const grpc::Status& status) {
50
1/2
✗ Branch 3 not taken.
✓ Branch 4 taken 26 times.
 26 if (status.error_message().empty()) {
51 ✗ return status.error_code() == grpc::StatusCode::OK
52 ? std::string("OK")
53
0/10
✗ Branch 0 not taken.
✗ Branch 1 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✗ Branch 11 not taken.
✗ Branch 13 not taken.
✗ Branch 14 not taken.
 ✗ : std::to_string(status.error_code());
54 }
55 26 return status.error_message();
56 }
57
58 78 void ThrowIfNotOk(const grpc::Status& status, const std::string& operation) {
59
2/2
✓ Branch 1 taken 52 times.
✓ Branch 2 taken 26 times.
 78 if (status.ok()) {
60 52 return;
61 }
62
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 throw std::runtime_error(
63
4/8
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 26 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 26 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 26 times.
✗ Branch 12 not taken.
 52 "control-plane " + operation + " failed: " + GrpcStatusText(status));
64 }
65
66 6 std::string EncodeMetaBytes(const RawVerbsNodeMeta& meta) {
67 6 return EncodeRawVerbsNodeMeta(meta);
68 }
69
70 4 RawVerbsNodeMeta DecodeMetaBytes(const std::string& payload) {
71 4 return DecodeRawVerbsNodeMeta(payload);
72 }
73
74 2 grpc::Status MakeDeadlineExceeded(const std::string& message) {
75 2 return grpc::Status(grpc::StatusCode::DEADLINE_EXCEEDED, message);
76 }
77
78 6 grpc::Status MakeUnavailable(const std::string& message) {
79 6 return grpc::Status(grpc::StatusCode::UNAVAILABLE, message);
80 }
81
82 } // namespace
83
84 // One channel/stub per client. A shard handshakes and heartbeats, so opening a
85 // connection per RPC would be wasteful; gRPC channels are thread-safe and serve
86 // concurrent RPCs, so every RPC of one client shares this cache. Creating a
87 // channel does not connect, so an unused client costs nothing.
88 struct RdmaControlPlaneChannelCache {
89 34 explicit RdmaControlPlaneChannelCache(
90 const RdmaControlPlaneEndpoint& endpoint)
91
2/4
✓ Branch 2 taken 34 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 34 times.
✗ Branch 7 not taken.
 68 : stub(recstoreps::rdma::RdmaControlPlane::NewStub(grpc::CreateChannel(
92
2/4
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 34 times.
✗ Branch 5 not taken.
 68 EndpointString(endpoint), grpc::InsecureChannelCredentials()))) {}
93
94 // The generated stub keeps the channel alive.
95 std::unique_ptr<recstoreps::rdma::RdmaControlPlane::Stub> stub;
96 };
97
98 10 std::string EncodeRawVerbsNodeMeta(const RawVerbsNodeMeta& meta) {
99
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 RawVerbsMetadata wire;
100
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_magic(kRawVerbsMetadataMagic);
101
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_protocol_version(meta.protocol_version);
102
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_deployment_id(meta.deployment_id);
103
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_deployment_epoch(meta.deployment_epoch);
104
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_configuration_digest(meta.configuration_digest);
105
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_fabric_digest(meta.fabric_digest);
106
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_node_id(meta.node_id);
107
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_logical_id(meta.logical_id);
108
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_node_role(meta.node_role);
109
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_lid(meta.lid);
110
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_qpn(meta.qpn);
111
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_psn(meta.psn);
112
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_rkey(meta.rkey);
113
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_base_addr(meta.base_addr);
114
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_gid(meta.gid, sizeof(meta.gid));
115
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_port_num(meta.port_num);
116
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_link_layer(meta.link_layer);
117
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_gid_index(meta.gid_index);
118
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_active_mtu(meta.active_mtu);
119
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_fabric_mode(meta.fabric_mode);
120
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 wire.set_lane_generation(meta.lane_generation);
121 10 std::string payload;
122
2/4
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 10 times.
 10 if (!wire.SerializeToString(&payload)) {
123
0/2
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 ✗ throw std::runtime_error("failed to serialize RawVerbsNodeMeta");
124 }
125 20 return payload;
126 10 }
127
128 12 RawVerbsNodeMeta DecodeRawVerbsNodeMeta(const std::string& payload) {
129
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 RawVerbsMetadata wire;
130
2/2
✓ Branch 2 taken 10 times.
✓ Branch 3 taken 2 times.
 24 if (payload.empty() || payload.size() > kMaxRawVerbsMetadataBytes ||
131
3/4
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 8 times.
✓ Branch 4 taken 2 times.
 10 !wire.ParseFromString(payload) ||
132
2/4
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 8 times.
✗ Branch 4 not taken.
 8 wire.magic() != kRawVerbsMetadataMagic ||
133
4/6
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✓ Branch 4 taken 2 times.
✓ Branch 5 taken 6 times.
✗ Branch 6 not taken.
 14 wire.protocol_version() != recstore::kRdmaDeploymentProtocolVersion ||
134
4/8
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 6 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 6 times.
✗ Branch 8 not taken.
✓ Branch 9 taken 6 times.
✗ Branch 10 not taken.
 18 wire.deployment_id().empty() || wire.deployment_epoch() == 0 ||
135
3/6
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 6 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 6 times.
✗ Branch 9 not taken.
 12 wire.configuration_digest().empty() || wire.fabric_digest().empty() ||
136
4/8
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 6 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 6 times.
✗ Branch 9 not taken.
 6 wire.node_id() < 0 || wire.node_id() > UINT16_MAX ||
137
2/4
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
 6 wire.node_role() >
138 6 static_cast<std::uint32_t>(recstore::RdmaNodeRole::kClient) ||
139
4/8
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 6 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 6 times.
✗ Branch 10 not taken.
 6 wire.lid() > UINT16_MAX || wire.gid().size() != 16 ||
140
4/8
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 6 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 6 times.
✗ Branch 9 not taken.
 6 wire.port_num() == 0 || wire.port_num() > UINT8_MAX ||
141
4/8
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 6 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 6 times.
✗ Branch 9 not taken.
 6 wire.gid_index() < 0 || wire.active_mtu() < IBV_MTU_256 ||
142
2/4
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
 6 wire.active_mtu() > IBV_MTU_4096 ||
143
3/4
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2 times.
✓ Branch 4 taken 4 times.
 6 (wire.link_layer() != IBV_LINK_LAYER_INFINIBAND &&
144
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
 2 wire.link_layer() != IBV_LINK_LAYER_ETHERNET) ||
145
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 4 times.
✗ Branch 4 not taken.
 4 (wire.fabric_mode() ==
146 4 static_cast<std::uint32_t>(recstore::RdmaFabricMode::kIb) &&
147
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 4 times.
✗ Branch 4 not taken.
 4 wire.link_layer() != IBV_LINK_LAYER_INFINIBAND) ||
148
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 4 times.
 4 (wire.fabric_mode() !=
149 ✗ static_cast<std::uint32_t>(recstore::RdmaFabricMode::kIb) &&
150
3/8
✓ Branch 0 taken 12 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 8 times.
✓ Branch 8 taken 4 times.
 24 wire.link_layer() != IBV_LINK_LAYER_ETHERNET) ||
151
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 4 times.
 4 wire.fabric_mode() >
152 static_cast<std::uint32_t>(recstore::RdmaFabricMode::kRocEv2)) {
153
1/2
✓ Branch 2 taken 8 times.
✗ Branch 3 not taken.
 8 throw std::runtime_error("invalid versioned RawVerbsNodeMeta payload");
154 }
155 4 RawVerbsNodeMeta meta{};
156
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.protocol_version = wire.protocol_version();
157
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 4 meta.deployment_id = wire.deployment_id();
158
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.deployment_epoch = wire.deployment_epoch();
159
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 4 meta.configuration_digest = wire.configuration_digest();
160
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 4 meta.fabric_digest = wire.fabric_digest();
161
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.node_id = static_cast<std::uint16_t>(wire.node_id());
162
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.logical_id = wire.logical_id();
163
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.node_role = static_cast<std::uint8_t>(wire.node_role());
164
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.lid = static_cast<std::uint16_t>(wire.lid());
165
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.qpn = wire.qpn();
166
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.psn = wire.psn();
167
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.rkey = wire.rkey();
168
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.base_addr = wire.base_addr();
169
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 std::memcpy(meta.gid, wire.gid().data(), sizeof(meta.gid));
170
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.port_num = static_cast<std::uint8_t>(wire.port_num());
171
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.link_layer = static_cast<std::uint8_t>(wire.link_layer());
172
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.gid_index = wire.gid_index();
173
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.active_mtu = static_cast<std::uint8_t>(wire.active_mtu());
174
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.fabric_mode = static_cast<std::uint8_t>(wire.fabric_mode());
175
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 meta.lane_generation = wire.lane_generation();
176 8 return meta;
177 12 }
178
179 8 void ValidateRawVerbsPeerMeta(const RawVerbsConfig& local,
180 int expected_node_id,
181 const RawVerbsNodeMeta& remote) {
182 8 const bool expected_server = expected_node_id < local.num_servers;
183 8 const int expected_logical_id =
184
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 expected_server ? expected_node_id : expected_node_id - local.num_servers;
185
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 const auto expected_role = static_cast<std::uint8_t>(
186 expected_server ? recstore::RdmaNodeRole::kServer
187 : recstore::RdmaNodeRole::kClient);
188 8 const auto expected_link_layer = static_cast<std::uint8_t>(
189
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 local.fabric_mode == recstore::RdmaFabricMode::kIb
190 ? IBV_LINK_LAYER_INFINIBAND
191 : IBV_LINK_LAYER_ETHERNET);
192 12 const auto incompatible = [expected_node_id](const char* field) {
193
1/2
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
 6 throw std::runtime_error(
194
3/6
✓ Branch 2 taken 6 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 6 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 6 times.
✗ Branch 9 not taken.
 12 "incompatible RDMA peer metadata field=" + std::string(field) +
195
2/4
✓ Branch 2 taken 6 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 6 times.
✗ Branch 6 not taken.
 24 " node_id=" + std::to_string(expected_node_id));
196 8 };
197
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (remote.node_id != expected_node_id)
198
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("node_id");
199
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (remote.logical_id != expected_logical_id)
200
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("logical_id");
201
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (remote.node_role != expected_role)
202
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("node_role");
203
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8 times.
 8 if (remote.protocol_version != local.protocol_version)
204
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("protocol_version");
205
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
 8 if (remote.deployment_id != local.deployment_id)
206
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("deployment_id");
207
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 6 times.
 8 if (remote.deployment_epoch != local.deployment_epoch)
208
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
 2 incompatible("deployment_epoch");
209
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 6 times.
 6 if (remote.configuration_digest != local.configuration_digest)
210
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("configuration_digest");
211
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 6 times.
 6 if (remote.fabric_digest != local.fabric_digest)
212
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("fabric_digest");
213
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (remote.link_layer != expected_link_layer)
214
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("link_layer");
215
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 6 times.
 6 if (remote.fabric_mode != static_cast<std::uint8_t>(local.fabric_mode))
216
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("fabric_mode");
217
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 4 times.
 6 if (remote.port_num == 0)
218
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
 2 incompatible("port_num");
219
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 2 times.
 4 if (remote.gid_index < 0)
220
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
 2 incompatible("gid_index");
221
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 if (remote.active_mtu == 0)
222
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ incompatible("active_mtu");
223 2 }
224
225 class RdmaControlPlaneService final : public RdmaControlPlane::Service {
226 public:
227 26 explicit RdmaControlPlaneService(RdmaControlPlaneServer* owner)
228 26 : owner_(owner) {}
229
230 90 grpc::Status ValidateDeploymentIdentity(
231 const std::string& deployment_id, std::uint64_t deployment_epoch) const {
232
3/6
✓ Branch 1 taken 90 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 90 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 90 times.
 90 if (deployment_id.empty() || deployment_epoch == 0) {
233 return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
234
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "control-plane deployment identity is required");
235 }
236
4/4
✓ Branch 1 taken 80 times.
✓ Branch 2 taken 10 times.
✓ Branch 3 taken 12 times.
✓ Branch 4 taken 78 times.
 170 if (deployment_id != owner_->endpoint_.deployment_id ||
237
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 78 times.
 80 deployment_epoch != owner_->endpoint_.deployment_epoch) {
238 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
239
2/4
✓ Branch 2 taken 12 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 12 times.
✗ Branch 6 not taken.
 12 "control-plane deployment identity mismatch");
240 }
241 78 return grpc::Status::OK;
242 }
243
244 4 grpc::Status PublishMeta(grpc::ServerContext*,
245 const PublishMetaRequest* request,
246 PublishMetaResponse*) override {
247 const grpc::Status identity = ValidateDeploymentIdentity(
248
3/6
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 4 times.
✗ Branch 8 not taken.
 4 request->deployment_id(), request->deployment_epoch());
249
2/2
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 2 times.
 4 if (!identity.ok()) {
250
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return identity;
251 }
252
3/6
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
 4 if (request->publisher_node_id() < 0 || request->publisher_lane() < 0 ||
253
6/12
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 2 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2 times.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✓ Branch 11 taken 2 times.
✗ Branch 12 not taken.
✓ Branch 13 taken 2 times.
 4 request->receiver_node_id() < 0 || request->receiver_lane() < 0) {
254 return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
255
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "metadata node and lane ids must be non-negative");
256 }
257 const RdmaControlPlaneServer::MetaKey key{
258
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 request->deployment_id(),
259
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 request->deployment_epoch(),
260 4 request->publisher_node_id(),
261 4 request->publisher_lane(),
262 4 request->receiver_node_id(),
263 4 request->receiver_lane(),
264
5/10
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 2 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 2 times.
✗ Branch 14 not taken.
 4 };
265 2 RawVerbsNodeMeta meta;
266 try {
267
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 2 meta = DecodeMetaBytes(request->meta());
268
0/2
✗ Branch 0 not taken.
✗ Branch 1 not taken.
 ✗ } catch (const std::exception& error) {
269
0/4
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
 ✗ return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, error.what());
270 ✗ }
271
1/2
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
 2 if (meta.protocol_version != owner_->endpoint_.protocol_version ||
272
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 4 meta.configuration_digest != owner_->endpoint_.configuration_digest ||
273
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 4 meta.fabric_digest != owner_->endpoint_.fabric_digest ||
274
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 meta.deployment_id != request->deployment_id() ||
275
4/8
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 2 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
✓ Branch 8 taken 2 times.
 6 meta.deployment_epoch != request->deployment_epoch() ||
276
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
 2 meta.node_id != request->publisher_node_id()) {
277 return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
278
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "metadata deployment identity mismatch");
279 }
280 {
281
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 std::lock_guard<std::mutex> guard(owner_->mu_);
282
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 2 owner_->metadata_[key] = meta;
283 2 }
284 2 owner_->cv_.notify_all();
285
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return grpc::Status::OK;
286 4 }
287
288 4 grpc::Status GetMeta(grpc::ServerContext*,
289 const GetMetaRequest* request,
290 GetMetaResponse* response) override {
291 const grpc::Status identity = ValidateDeploymentIdentity(
292
3/6
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 4 times.
✗ Branch 8 not taken.
 4 request->deployment_id(), request->deployment_epoch());
293
2/2
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 2 times.
 4 if (!identity.ok()) {
294
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return identity;
295 }
296
3/6
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 2 times.
✗ Branch 7 not taken.
 4 if (request->publisher_node_id() < 0 || request->publisher_lane() < 0 ||
297
6/12
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 2 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2 times.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✓ Branch 11 taken 2 times.
✗ Branch 12 not taken.
✓ Branch 13 taken 2 times.
 4 request->receiver_node_id() < 0 || request->receiver_lane() < 0) {
298 return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
299
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "metadata node and lane ids must be non-negative");
300 }
301 const RdmaControlPlaneServer::MetaKey key{
302
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 request->deployment_id(),
303
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 request->deployment_epoch(),
304 4 request->publisher_node_id(),
305 4 request->publisher_lane(),
306 4 request->receiver_node_id(),
307 4 request->receiver_lane(),
308
5/10
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 2 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 2 times.
✗ Branch 14 not taken.
 4 };
309 const int timeout_ms =
310
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 request->timeout_ms() > 0
311
2/4
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 2 times.
✗ Branch 4 not taken.
 2 ? request->timeout_ms()
312 ✗ : owner_->endpoint_.timeout_ms;
313
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 std::unique_lock<std::mutex> lock(owner_->mu_);
314 const bool ready =
315
1/2
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 2 owner_->cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms), [&] {
316
3/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2 times.
✓ Branch 4 taken 2 times.
 8 return owner_->stop_requested_.load(std::memory_order_relaxed) ||
317
1/2
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
 8 owner_->metadata_.find(key) != owner_->metadata_.end();
318 });
319
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2 times.
 2 if (!ready) {
320 return MakeDeadlineExceeded(
321
0/6
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
 ✗ "get_meta timeout key=" + std::to_string(key.publisher_node_id) +
322
0/8
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
✗ Branch 10 not taken.
✗ Branch 11 not taken.
 ✗ ":" + std::to_string(key.publisher_lane) + "->" +
323
0/6
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
 ✗ std::to_string(key.receiver_node_id) + ":" +
324
0/4
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 ✗ std::to_string(key.receiver_lane));
325 }
326
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
 2 if (owner_->stop_requested_.load(std::memory_order_relaxed)) {
327
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ return MakeUnavailable("control-plane stopping");
328 }
329
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 4 response->set_meta(EncodeMetaBytes(owner_->metadata_.at(key)));
330
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return grpc::Status::OK;
331 4 }
332
333 24 grpc::Status PublishServerReady(grpc::ServerContext*,
334 const PublishServerReadyRequest* request,
335 PublishServerReadyResponse*) override {
336 const grpc::Status identity = ValidateDeploymentIdentity(
337
3/6
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 24 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 24 times.
✗ Branch 8 not taken.
 24 request->deployment_id(), request->deployment_epoch());
338
2/2
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 22 times.
 24 if (!identity.ok()) {
339
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return identity;
340 }
341
2/4
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 22 times.
✗ Branch 4 not taken.
 44 if (request->protocol_version() != owner_->endpoint_.protocol_version ||
342
2/4
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22 times.
✗ Branch 5 not taken.
 44 request->configuration_digest().empty() ||
343
2/4
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22 times.
✗ Branch 5 not taken.
 44 request->fabric_digest().empty() ||
344 22 request->configuration_digest() !=
345
2/4
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 22 times.
✗ Branch 4 not taken.
 44 owner_->endpoint_.configuration_digest ||
346
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 request->fabric_digest() != owner_->endpoint_.fabric_digest ||
347
4/8
✓ Branch 0 taken 22 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 22 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 22 times.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
✓ Branch 8 taken 22 times.
 66 request->server_id() < 0 ||
348
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 14 times.
 22 (owner_->endpoint_.num_servers > 0 &&
349
2/4
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 8 times.
 8 request->server_id() >= owner_->endpoint_.num_servers)) {
350 return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
351
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "invalid versioned server READY record");
352 }
353 {
354
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 std::lock_guard<std::mutex> guard(owner_->mu_);
355 44 auto& record = owner_->ready_servers_[{request->deployment_id(),
356
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 request->deployment_epoch(),
357
4/8
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 22 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 22 times.
✗ Branch 11 not taken.
 44 request->server_id()}];
358
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 20 times.
 22 if (record.state ==
359 RdmaControlPlaneServer::ReadyRecord::State::kDraining) {
360 return grpc::Status(
361 grpc::StatusCode::FAILED_PRECONDITION,
362
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
 2 "draining server cannot become ready in same epoch");
363 }
364
1/2
✓ Branch 1 taken 20 times.
✗ Branch 2 not taken.
 20 record.protocol_version = request->protocol_version();
365
2/4
✓ Branch 1 taken 20 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 20 times.
✗ Branch 5 not taken.
 20 record.configuration_digest = request->configuration_digest();
366
2/4
✓ Branch 1 taken 20 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 20 times.
✗ Branch 5 not taken.
 20 record.fabric_digest = request->fabric_digest();
367 20 record.state = RdmaControlPlaneServer::ReadyRecord::State::kServing;
368 20 record.last_seen = std::chrono::steady_clock::now();
369
2/2
✓ Branch 1 taken 20 times.
✓ Branch 2 taken 2 times.
 22 }
370 20 owner_->cv_.notify_all();
371
1/2
✓ Branch 1 taken 20 times.
✗ Branch 2 not taken.
 20 return grpc::Status::OK;
372 24 }
373
374 4 grpc::Status PublishServerDraining(
375 grpc::ServerContext*,
376 const PublishServerDrainingRequest* request,
377 PublishServerDrainingResponse*) override {
378 const grpc::Status identity = ValidateDeploymentIdentity(
379
3/6
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 4 times.
✗ Branch 8 not taken.
 4 request->deployment_id(), request->deployment_epoch());
380
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 4 times.
 4 if (!identity.ok()) {
381
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ return identity;
382 }
383
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 4 times.
✗ Branch 4 not taken.
 8 if (request->protocol_version() != owner_->endpoint_.protocol_version ||
384 4 request->configuration_digest() !=
385
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 4 times.
✗ Branch 4 not taken.
 8 owner_->endpoint_.configuration_digest ||
386
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request->fabric_digest() != owner_->endpoint_.fabric_digest ||
387
4/8
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 4 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 4 times.
✗ Branch 6 not taken.
✗ Branch 7 not taken.
✓ Branch 8 taken 4 times.
 12 request->server_id() < 0 ||
388
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 (owner_->endpoint_.num_servers > 0 &&
389
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 4 times.
 4 request->server_id() >= owner_->endpoint_.num_servers)) {
390 return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT,
391
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "invalid versioned server DRAINING record");
392 }
393 {
394
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 std::lock_guard<std::mutex> guard(owner_->mu_);
395 const RdmaControlPlaneServer::ReadyKey key{
396
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request->deployment_id(),
397
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request->deployment_epoch(),
398
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 8 request->server_id()};
399
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 const auto it = owner_->ready_servers_.find(key);
400
2/2
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 2 times.
 4 if (it == owner_->ready_servers_.end()) {
401 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
402
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
 2 "server must be ready before draining");
403 }
404
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 4 if (it->second.protocol_version != request->protocol_version() ||
405
4/8
✓ Branch 0 taken 2 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 2 times.
✗ Branch 4 not taken.
✗ Branch 7 not taken.
✓ Branch 8 taken 2 times.
✗ Branch 9 not taken.
✓ Branch 10 taken 2 times.
 4 it->second.configuration_digest != request->configuration_digest() ||
406
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 it->second.fabric_digest != request->fabric_digest()) {
407 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
408
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "server DRAINING deployment contract mismatch");
409 }
410 2 it->second.state = RdmaControlPlaneServer::ReadyRecord::State::kDraining;
411
4/4
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 2 times.
✓ Branch 4 taken 2 times.
✓ Branch 5 taken 2 times.
 6 }
412 2 owner_->cv_.notify_all();
413
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return grpc::Status::OK;
414 4 }
415
416 12 grpc::Status WaitServer(grpc::ServerContext*,
417 const WaitServerRequest* request,
418 WaitServerResponse*) override {
419 const grpc::Status identity = ValidateDeploymentIdentity(
420
3/6
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 12 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 12 times.
✗ Branch 8 not taken.
 12 request->deployment_id(), request->deployment_epoch());
421
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 12 times.
 12 if (!identity.ok()) {
422
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ return identity;
423 }
424
3/4
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 10 times.
✓ Branch 4 taken 2 times.
 24 if (request->protocol_version() != owner_->endpoint_.protocol_version ||
425 12 request->configuration_digest() !=
426
2/4
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 10 times.
✗ Branch 4 not taken.
 22 owner_->endpoint_.configuration_digest ||
427
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 request->fabric_digest() != owner_->endpoint_.fabric_digest ||
428
5/8
✓ Branch 0 taken 12 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 10 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 10 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 2 times.
✓ Branch 8 taken 10 times.
 34 request->server_id() < 0 ||
429
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 6 times.
 10 (owner_->endpoint_.num_servers > 0 &&
430
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 4 times.
 4 request->server_id() >= owner_->endpoint_.num_servers)) {
431 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
432
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
 2 "server READY deployment contract mismatch");
433 }
434 const int timeout_ms =
435
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 request->timeout_ms() > 0
436
2/4
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 10 times.
✗ Branch 4 not taken.
 10 ? request->timeout_ms()
437 ✗ : owner_->endpoint_.timeout_ms;
438 const RdmaControlPlaneServer::ReadyKey key{
439
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 request->deployment_id(),
440
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 request->deployment_epoch(),
441
2/4
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 10 times.
✗ Branch 5 not taken.
 20 request->server_id()};
442
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 std::unique_lock<std::mutex> lock(owner_->mu_);
443 const bool ready =
444
1/2
✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
 10 owner_->cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms), [&] {
445
4/4
✓ Branch 1 taken 10 times.
✓ Branch 2 taken 2 times.
✓ Branch 3 taken 8 times.
✓ Branch 4 taken 2 times.
 22 return owner_->stop_requested_.load(std::memory_order_relaxed) ||
446
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 owner_->ready_servers_.find(key) !=
447 22 owner_->ready_servers_.end();
448 });
449
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 10 times.
 10 if (!ready) {
450
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ return MakeDeadlineExceeded("wait_server timeout server_id=" +
451
0/6
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
 ✗ std::to_string(request->server_id()));
452 }
453
2/2
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 8 times.
 10 if (owner_->stop_requested_.load(std::memory_order_relaxed)) {
454
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
 2 return MakeUnavailable("control-plane stopping");
455 }
456
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 const auto& record = owner_->ready_servers_.at(key);
457
2/4
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 8 times.
✗ Branch 4 not taken.
 16 if (record.protocol_version != request->protocol_version() ||
458
4/8
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 8 times.
✗ Branch 4 not taken.
✗ Branch 6 not taken.
✓ Branch 7 taken 8 times.
✗ Branch 8 not taken.
✓ Branch 9 taken 8 times.
 16 record.configuration_digest != request->configuration_digest() ||
459
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 record.fabric_digest != request->fabric_digest()) {
460 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
461
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "server READY deployment contract mismatch");
462 }
463
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 6 times.
 8 if (record.state == RdmaControlPlaneServer::ReadyRecord::State::kDraining) {
464 return MakeUnavailable(
465
4/8
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 2 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
 4 "server draining server_id=" + std::to_string(request->server_id()));
466 }
467
1/2
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
 6 return grpc::Status::OK;
468 12 }
469
470 8 grpc::Status WaitServerReady(grpc::ServerContext*,
471 const WaitServerReadyRequest* request,
472 WaitServerReadyResponse*) override {
473 const grpc::Status identity = ValidateDeploymentIdentity(
474
3/6
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 8 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 8 times.
✗ Branch 8 not taken.
 8 request->deployment_id(), request->deployment_epoch());
475
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 8 times.
 8 if (!identity.ok()) {
476
0/2
✗ Branch 1 not taken.
✗ Branch 2 not taken.
 ✗ return identity;
477 }
478
3/4
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✓ Branch 4 taken 2 times.
 16 if (request->protocol_version() != owner_->endpoint_.protocol_version ||
479 8 request->configuration_digest() !=
480
2/4
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
 14 owner_->endpoint_.configuration_digest ||
481
1/2
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
 6 request->fabric_digest() != owner_->endpoint_.fabric_digest ||
482
5/8
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 6 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 2 times.
✓ Branch 8 taken 6 times.
 22 request->num_servers() <= 0 ||
483
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 4 times.
 6 (owner_->endpoint_.num_servers > 0 &&
484
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 2 times.
 2 request->num_servers() > owner_->endpoint_.num_servers)) {
485 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
486
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
 2 "server READY deployment contract mismatch");
487 }
488 const int timeout_ms =
489
1/2
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
 6 request->timeout_ms() > 0
490
2/4
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
 6 ? request->timeout_ms()
491 ✗ : owner_->endpoint_.timeout_ms;
492
1/2
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
 6 std::unique_lock<std::mutex> lock(owner_->mu_);
493 const bool ready =
494
1/2
✓ Branch 2 taken 6 times.
✗ Branch 3 not taken.
 6 owner_->cv_.wait_for(lock, std::chrono::milliseconds(timeout_ms), [&] {
495 10 return owner_->stop_requested_.load(std::memory_order_relaxed) ||
496 10 owner_->ReadyCount(
497 10 request->deployment_id(),
498 request->deployment_epoch(),
499 request->protocol_version(),
500 request->configuration_digest(),
501
2/2
✓ Branch 1 taken 8 times.
✓ Branch 2 taken 2 times.
 10 request->fabric_digest()) >= request->num_servers() ||
502
1/2
✓ Branch 6 taken 8 times.
✗ Branch 7 not taken.
 8 owner_->HasReadyContractMismatch(
503 request->deployment_id(),
504 request->deployment_epoch(),
505 request->protocol_version(),
506 request->configuration_digest(),
507
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 20 request->fabric_digest()) ||
508
2/2
✓ Branch 4 taken 2 times.
✓ Branch 5 taken 6 times.
 8 owner_->HasDrainingServer(request->deployment_id(),
509 request->deployment_epoch(),
510 10 request->num_servers());
511 });
512
7/14
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 6 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 6 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 6 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 6 times.
✗ Branch 14 not taken.
✓ Branch 16 taken 6 times.
✗ Branch 17 not taken.
✗ Branch 18 not taken.
✓ Branch 19 taken 6 times.
 6 if (owner_->HasReadyContractMismatch(
513 request->deployment_id(),
514 request->deployment_epoch(),
515 request->protocol_version(),
516 request->configuration_digest(),
517 request->fabric_digest())) {
518 return grpc::Status(grpc::StatusCode::FAILED_PRECONDITION,
519
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ "server READY deployment contract mismatch");
520 }
521
6/10
✓ Branch 1 taken 6 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 6 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 6 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 6 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 2 times.
✓ Branch 13 taken 4 times.
 6 if (owner_->HasDrainingServer(request->deployment_id(),
522 request->deployment_epoch(),
523 request->num_servers())) {
524
2/4
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 2 times.
✗ Branch 6 not taken.
 2 return MakeUnavailable("one or more RDMA servers are draining");
525 }
526
2/2
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 2 times.
 4 if (!ready) {
527 return MakeDeadlineExceeded(
528
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 "wait_server_ready timeout ready=" +
529
7/14
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 2 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 2 times.
✗ Branch 14 not taken.
✓ Branch 16 taken 2 times.
✗ Branch 17 not taken.
✓ Branch 19 taken 2 times.
✗ Branch 20 not taken.
 4 std::to_string(owner_->ReadyCount(
530 request->deployment_id(),
531 request->deployment_epoch(),
532 request->protocol_version(),
533 request->configuration_digest(),
534
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 4 request->fabric_digest())) +
535
4/8
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 2 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 2 times.
✗ Branch 11 not taken.
 8 "/" + std::to_string(request->num_servers()));
536 }
537
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 2 times.
 2 if (owner_->stop_requested_.load(std::memory_order_relaxed)) {
538
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ return MakeUnavailable("control-plane stopping");
539 }
540
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return grpc::Status::OK;
541 8 }
542
543 22 grpc::Status Probe(grpc::ServerContext*,
544 const ProbeRequest* request,
545 ProbeResponse*) override {
546 const grpc::Status identity = ValidateDeploymentIdentity(
547
3/6
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 22 times.
✗ Branch 8 not taken.
 22 request->deployment_id(), request->deployment_epoch());
548
2/2
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 18 times.
 22 if (!identity.ok()) {
549
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 return identity;
550 }
551
2/4
✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 18 times.
 18 if (request->server_id() < 0) {
552 return grpc::Status(
553
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ grpc::StatusCode::INVALID_ARGUMENT, "server id must be non-negative");
554 }
555
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 18 times.
 18 if (owner_->stop_requested_.load(std::memory_order_relaxed)) {
556
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ return MakeUnavailable("control-plane stopping");
557 }
558 // Only a shard that already published READY has a liveness record; a
559 // heartbeat never creates one, and READY remains the only way to serve.
560
2/4
✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 18 times.
✗ Branch 5 not taken.
 18 owner_->RecordHeartbeat(request->server_id());
561
1/2
✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
 18 return grpc::Status::OK;
562 22 }
563
564 12 grpc::Status GetServerHealth(grpc::ServerContext*,
565 const GetServerHealthRequest* request,
566 GetServerHealthResponse* response) override {
567 const grpc::Status identity = ValidateDeploymentIdentity(
568
3/6
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 12 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 12 times.
✗ Branch 8 not taken.
 12 request->deployment_id(), request->deployment_epoch());
569
2/2
✓ Branch 1 taken 2 times.
✓ Branch 2 taken 10 times.
 12 if (!identity.ok()) {
570
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 return identity;
571 }
572
2/4
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 10 times.
 10 if (request->server_id() < 0) {
573 return grpc::Status(
574
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ grpc::StatusCode::INVALID_ARGUMENT, "server id must be non-negative");
575 }
576
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 10 times.
 10 if (owner_->stop_requested_.load(std::memory_order_relaxed)) {
577
0/4
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
 ✗ return MakeUnavailable("control-plane stopping");
578 }
579 const RdmaServerHealthSnapshot snapshot =
580
2/4
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 10 times.
✗ Branch 5 not taken.
 10 owner_->ServerHealthSnapshot(request->server_id());
581
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 response->set_known(snapshot.known);
582
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 response->set_state(
583
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 2 times.
 10 snapshot.known ? static_cast<std::int32_t>(snapshot.health) : -1);
584
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 response->set_last_seen_age_ms(
585
2/2
✓ Branch 0 taken 8 times.
✓ Branch 1 taken 2 times.
 10 snapshot.known ? snapshot.last_seen_age_ms : -1);
586
1/2
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
 10 return grpc::Status::OK;
587 12 }
588
589 private:
590 RdmaControlPlaneServer* owner_;
591 };
592
593 34 RdmaControlPlaneClient::RdmaControlPlaneClient(
594 34 RdmaControlPlaneEndpoint endpoint)
595 34 : endpoint_(std::move(endpoint)),
596
1/2
✓ Branch 1 taken 34 times.
✗ Branch 2 not taken.
 34 channels_(std::make_unique<RdmaControlPlaneChannelCache>(endpoint_)) {}
597
598 // Out of line because the cache type is incomplete in the header.
599 34 RdmaControlPlaneClient::~RdmaControlPlaneClient() = default;
600
601 4 void RdmaControlPlaneClient::PublishMeta(
602 int publisher_node_id,
603 int publisher_lane,
604 int receiver_node_id,
605 int receiver_lane,
606 const RawVerbsNodeMeta& meta) const {
607 4 RdmaControlPlane::Stub* stub = channels_->stub.get();
608
609
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 PublishMetaRequest request;
610
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_publisher_node_id(publisher_node_id);
611
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_publisher_lane(publisher_lane);
612
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_receiver_node_id(receiver_node_id);
613
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_receiver_lane(receiver_lane);
614
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 8 request.set_meta(EncodeMetaBytes(meta));
615
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_deployment_id(endpoint_.deployment_id);
616
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_deployment_epoch(endpoint_.deployment_epoch);
617
618
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 PublishMetaResponse response;
619
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 grpc::ClientContext context;
620
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 4 context.set_deadline(DeadlineFromNow(endpoint_.timeout_ms));
621
4/6
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 4 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2 times.
✓ Branch 9 taken 2 times.
 10 ThrowIfNotOk(stub->PublishMeta(&context, request, &response), "publish_meta");
622 8 }
623
624 4 RawVerbsNodeMeta RdmaControlPlaneClient::GetMeta(
625 int publisher_node_id,
626 int publisher_lane,
627 int receiver_node_id,
628 int receiver_lane,
629 int timeout_ms) const {
630 4 const int effective_timeout_ms =
631
1/2
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
 4 timeout_ms > 0 ? timeout_ms : endpoint_.timeout_ms;
632 4 RdmaControlPlane::Stub* stub = channels_->stub.get();
633
634
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 GetMetaRequest request;
635
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_publisher_node_id(publisher_node_id);
636
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_publisher_lane(publisher_lane);
637
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_receiver_node_id(receiver_node_id);
638
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_receiver_lane(receiver_lane);
639
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_timeout_ms(effective_timeout_ms);
640
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_deployment_id(endpoint_.deployment_id);
641
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_deployment_epoch(endpoint_.deployment_epoch);
642
643
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 GetMetaResponse response;
644
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 grpc::ClientContext context;
645
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 4 context.set_deadline(DeadlineFromNow(effective_timeout_ms));
646
4/6
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 4 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2 times.
✓ Branch 9 taken 2 times.
 10 ThrowIfNotOk(stub->GetMeta(&context, request, &response), "get_meta");
647
2/4
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2 times.
✗ Branch 5 not taken.
 4 return DecodeMetaBytes(response.meta());
648 8 }
649
650 24 void RdmaControlPlaneClient::PublishServerReady(int server_id) const {
651 24 RdmaControlPlane::Stub* stub = channels_->stub.get();
652
653
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 PublishServerReadyRequest request;
654
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 request.set_server_id(server_id);
655
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 request.set_protocol_version(endpoint_.protocol_version);
656
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 request.set_deployment_id(endpoint_.deployment_id);
657
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 request.set_deployment_epoch(endpoint_.deployment_epoch);
658
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 request.set_configuration_digest(endpoint_.configuration_digest);
659
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 request.set_fabric_digest(endpoint_.fabric_digest);
660
661
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 PublishServerReadyResponse response;
662
1/2
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
 24 grpc::ClientContext context;
663
2/4
✓ Branch 1 taken 24 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 24 times.
✗ Branch 5 not taken.
 24 context.set_deadline(DeadlineFromNow(endpoint_.timeout_ms));
664
4/6
✓ Branch 2 taken 24 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 24 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 20 times.
✓ Branch 9 taken 4 times.
 36 ThrowIfNotOk(stub->PublishServerReady(&context, request, &response),
665 "server_ready");
666 32 }
667
668 4 void RdmaControlPlaneClient::PublishServerDraining(int server_id) const {
669 4 RdmaControlPlane::Stub* stub = channels_->stub.get();
670
671
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 PublishServerDrainingRequest request;
672
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_server_id(server_id);
673
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_protocol_version(endpoint_.protocol_version);
674
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_deployment_id(endpoint_.deployment_id);
675
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_deployment_epoch(endpoint_.deployment_epoch);
676
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_configuration_digest(endpoint_.configuration_digest);
677
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 request.set_fabric_digest(endpoint_.fabric_digest);
678
679
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 PublishServerDrainingResponse response;
680
1/2
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
 4 grpc::ClientContext context;
681
2/4
✓ Branch 1 taken 4 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4 times.
✗ Branch 5 not taken.
 4 context.set_deadline(DeadlineFromNow(endpoint_.timeout_ms));
682
4/6
✓ Branch 2 taken 4 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 4 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2 times.
✓ Branch 9 taken 2 times.
 10 ThrowIfNotOk(stub->PublishServerDraining(&context, request, &response),
683 "server_draining");
684 8 }
685
686 22 void RdmaControlPlaneClient::SendHeartbeat(int server_id) const {
687 22 RdmaControlPlane::Stub* stub = channels_->stub.get();
688
689
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 ProbeRequest request;
690
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 request.set_server_id(server_id);
691
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 request.set_deployment_id(endpoint_.deployment_id);
692
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 request.set_deployment_epoch(endpoint_.deployment_epoch);
693
694
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 ProbeResponse response;
695
1/2
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
 22 grpc::ClientContext context;
696
2/4
✓ Branch 1 taken 22 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22 times.
✗ Branch 5 not taken.
 22 context.set_deadline(DeadlineFromNow(endpoint_.timeout_ms));
697
4/6
✓ Branch 2 taken 22 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 22 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 18 times.
✓ Branch 9 taken 4 times.
 34 ThrowIfNotOk(stub->Probe(&context, request, &response), "probe");
698 30 }
699
700 RdmaServerHealthSnapshot
701 14 RdmaControlPlaneClient::GetServerHealth(int server_id, int timeout_ms) const {
702 14 const int effective_timeout_ms =
703
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 14 times.
 14 timeout_ms > 0 ? timeout_ms : endpoint_.timeout_ms;
704 14 RdmaServerHealthSnapshot snapshot;
705
706 14 RdmaControlPlane::Stub* stub = channels_->stub.get();
707
708
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 GetServerHealthRequest request;
709
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 request.set_server_id(server_id);
710
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 request.set_deployment_id(endpoint_.deployment_id);
711
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 request.set_deployment_epoch(endpoint_.deployment_epoch);
712
713
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 GetServerHealthResponse response;
714
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 grpc::ClientContext context;
715
2/4
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 14 times.
✗ Branch 5 not taken.
 14 context.set_deadline(DeadlineFromNow(effective_timeout_ms));
716 const grpc::Status status =
717
1/2
✓ Branch 1 taken 14 times.
✗ Branch 2 not taken.
 14 stub->GetServerHealth(&context, request, &response);
718
2/2
✓ Branch 1 taken 4 times.
✓ Branch 2 taken 10 times.
 14 if (!status.ok()) {
719
3/12
✓ Branch 0 taken 4 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 4 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 4 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✗ Branch 11 not taken.
✗ Branch 12 not taken.
✗ Branch 15 not taken.
✗ Branch 16 not taken.
 4 VLOG(1) << "component=rdma_control_plane event=get_server_health_failed"
720
0/4
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 ✗ << " server_id=" << server_id
721
2/14
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✗ Branch 7 not taken.
✗ Branch 8 not taken.
✗ Branch 10 not taken.
✗ Branch 11 not taken.
✗ Branch 13 not taken.
✓ Branch 14 taken 4 times.
✗ Branch 16 not taken.
✓ Branch 17 taken 4 times.
✗ Branch 18 not taken.
✗ Branch 19 not taken.
 4 << " status=" << GrpcStatusText(status);
722 4 return snapshot;
723 }
724
3/4
✓ Branch 1 taken 10 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2 times.
✓ Branch 4 taken 8 times.
 10 if (!response.known()) {
725 2 return snapshot;
726 }
727
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 const auto state = response.state();
728
2/4
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✓ Branch 3 taken 8 times.
 8 if (state < static_cast<std::int32_t>(RdmaServerHealth::kServing) ||
729 state > static_cast<std::int32_t>(RdmaServerHealth::kDead)) {
730 ✗ return snapshot;
731 }
732 8 snapshot.known = true;
733 8 snapshot.health = static_cast<RdmaServerHealth>(state);
734
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 snapshot.last_seen_age_ms = response.last_seen_age_ms();
735 8 return snapshot;
736 14 }
737
738 12 void RdmaControlPlaneClient::WaitServer(int server_id, int timeout_ms) const {
739 12 const int effective_timeout_ms =
740
1/2
✓ Branch 0 taken 12 times.
✗ Branch 1 not taken.
 12 timeout_ms > 0 ? timeout_ms : endpoint_.timeout_ms;
741 12 RdmaControlPlane::Stub* stub = channels_->stub.get();
742
743
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 WaitServerRequest request;
744
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_server_id(server_id);
745
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_timeout_ms(effective_timeout_ms);
746
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_protocol_version(endpoint_.protocol_version);
747
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_deployment_id(endpoint_.deployment_id);
748
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_deployment_epoch(endpoint_.deployment_epoch);
749
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_configuration_digest(endpoint_.configuration_digest);
750
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 request.set_fabric_digest(endpoint_.fabric_digest);
751
752
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 WaitServerResponse response;
753
1/2
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
 12 grpc::ClientContext context;
754
2/4
✓ Branch 1 taken 12 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 12 times.
✗ Branch 5 not taken.
 12 context.set_deadline(DeadlineFromNow(effective_timeout_ms));
755
4/6
✓ Branch 2 taken 12 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 12 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 6 times.
✓ Branch 9 taken 6 times.
 30 ThrowIfNotOk(stub->WaitServer(&context, request, &response), "wait_server");
756 24 }
757
758 8 void RdmaControlPlaneClient::WaitServerReady(int num_servers,
759 int timeout_ms) const {
760 8 const int effective_timeout_ms =
761
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 timeout_ms > 0 ? timeout_ms : endpoint_.timeout_ms;
762 8 RdmaControlPlane::Stub* stub = channels_->stub.get();
763
764
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 WaitServerReadyRequest request;
765
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_num_servers(num_servers);
766
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_timeout_ms(effective_timeout_ms);
767
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_protocol_version(endpoint_.protocol_version);
768
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_deployment_id(endpoint_.deployment_id);
769
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_deployment_epoch(endpoint_.deployment_epoch);
770
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_configuration_digest(endpoint_.configuration_digest);
771
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 request.set_fabric_digest(endpoint_.fabric_digest);
772
773
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 WaitServerReadyResponse response;
774
1/2
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
 8 grpc::ClientContext context;
775
2/4
✓ Branch 1 taken 8 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 8 times.
✗ Branch 5 not taken.
 8 context.set_deadline(DeadlineFromNow(effective_timeout_ms));
776
4/6
✓ Branch 2 taken 8 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 8 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 2 times.
✓ Branch 9 taken 6 times.
 26 ThrowIfNotOk(stub->WaitServerReady(&context, request, &response),
777 "wait_server_ready");
778 20 }
779
780 26 RdmaControlPlaneServer::RdmaControlPlaneServer(
781 26 RdmaControlPlaneEndpoint endpoint)
782 26 : endpoint_(std::move(endpoint)) {}
783
784 26 RdmaControlPlaneServer::~RdmaControlPlaneServer() { Stop(); }
785
786 std::size_t
787 8 RdmaControlPlaneServer::MetaKeyHash::operator()(const MetaKey& key) const {
788 8 std::size_t hash = std::hash<std::string>{}(key.deployment_id);
789 8 hash = hash * 1315423911u + static_cast<std::size_t>(key.deployment_epoch);
790 8 hash = hash * 1315423911u + static_cast<std::size_t>(key.publisher_node_id);
791 8 hash = hash * 1315423911u + static_cast<std::size_t>(key.publisher_lane);
792 8 hash = hash * 1315423911u + static_cast<std::size_t>(key.receiver_node_id);
793 8 hash = hash * 1315423911u + static_cast<std::size_t>(key.receiver_lane);
794 8 return hash;
795 }
796
797 std::size_t
798 98 RdmaControlPlaneServer::ReadyKeyHash::operator()(const ReadyKey& key) const {
799 98 std::size_t hash = std::hash<std::string>{}(key.deployment_id);
800 98 hash = hash * 1315423911u + static_cast<std::size_t>(key.deployment_epoch);
801 98 hash = hash * 1315423911u + static_cast<std::size_t>(key.server_id);
802 98 return hash;
803 }
804
805 12 int RdmaControlPlaneServer::ReadyCount(
806 const std::string& deployment_id,
807 std::uint64_t deployment_epoch,
808 std::uint32_t protocol_version,
809 const std::string& configuration_digest,
810 const std::string& fabric_digest) const {
811 12 int count = 0;
812
2/2
✓ Branch 5 taken 8 times.
✓ Branch 6 taken 12 times.
 20 for (const auto& entry : ready_servers_) {
813 8 if (entry.first.deployment_id == deployment_id &&
814
1/2
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
 8 entry.first.deployment_epoch == deployment_epoch &&
815
2/2
✓ Branch 0 taken 6 times.
✓ Branch 1 taken 2 times.
 8 entry.second.state == ReadyRecord::State::kServing &&
816
2/4
✓ Branch 0 taken 6 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 6 times.
✗ Branch 3 not taken.
 12 entry.second.protocol_version == protocol_version &&
817
4/6
✓ Branch 0 taken 8 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 6 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 6 times.
✓ Branch 6 taken 2 times.
 22 entry.second.configuration_digest == configuration_digest &&
818 6 entry.second.fabric_digest == fabric_digest) {
819 6 ++count;
820 }
821 }
822 12 return count;
823 }
824
825 14 bool RdmaControlPlaneServer::HasReadyContractMismatch(
826 const std::string& deployment_id,
827 std::uint64_t deployment_epoch,
828 std::uint32_t protocol_version,
829 const std::string& configuration_digest,
830 const std::string& fabric_digest) const {
831
2/2
✓ Branch 5 taken 10 times.
✓ Branch 6 taken 14 times.
 24 for (const auto& entry : ready_servers_) {
832 10 if (entry.first.deployment_id == deployment_id &&
833
3/6
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 10 times.
 20 entry.first.deployment_epoch == deployment_epoch &&
834
2/4
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
 20 (entry.second.protocol_version != protocol_version ||
835
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 10 times.
 20 entry.second.configuration_digest != configuration_digest ||
836 10 entry.second.fabric_digest != fabric_digest)) {
837 ✗ return true;
838 }
839 }
840 14 return false;
841 }
842
843 14 bool RdmaControlPlaneServer::HasDrainingServer(
844 const std::string& deployment_id,
845 std::uint64_t deployment_epoch,
846 int num_servers) const {
847
2/2
✓ Branch 5 taken 10 times.
✓ Branch 6 taken 10 times.
 20 for (const auto& entry : ready_servers_) {
848 10 if (entry.first.deployment_id == deployment_id &&
849
1/2
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
 10 entry.first.deployment_epoch == deployment_epoch &&
850
5/8
✓ Branch 0 taken 10 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 10 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 10 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 4 times.
✓ Branch 7 taken 6 times.
 30 entry.first.server_id >= 0 && entry.first.server_id < num_servers &&
851
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 6 times.
 10 entry.second.state == ReadyRecord::State::kDraining) {
852 4 return true;
853 }
854 }
855 10 return false;
856 }
857
858 RdmaServerHealthSnapshot
859 36 RdmaControlPlaneServer::ServerHealthSnapshot(int server_id) const {
860
1/2
✓ Branch 1 taken 36 times.
✗ Branch 2 not taken.
 36 std::lock_guard<std::mutex> guard(mu_);
861 72 const auto it = ready_servers_.find(
862
2/4
✓ Branch 1 taken 36 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 36 times.
✗ Branch 5 not taken.
 36 ReadyKey{endpoint_.deployment_id, endpoint_.deployment_epoch, server_id});
863
2/2
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 32 times.
 36 if (it == ready_servers_.end()) {
864 4 return RdmaServerHealthSnapshot{};
865 }
866 32 const auto now = std::chrono::steady_clock::now();
867 32 RdmaServerHealthSnapshot snapshot;
868 32 snapshot.known = true;
869
1/2
✓ Branch 1 taken 32 times.
✗ Branch 2 not taken.
 32 snapshot.health = EvaluateServerHealth(
870 32 it->second.last_seen,
871 now,
872 32 endpoint_.health_stale_ms,
873 32 endpoint_.health_dead_ms);
874
2/4
✓ Branch 3 taken 32 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 32 times.
✗ Branch 6 not taken.
 32 if (it->second.last_seen != std::chrono::steady_clock::time_point{}) {
875 32 snapshot.last_seen_age_ms =
876
1/2
✓ Branch 1 taken 32 times.
✗ Branch 2 not taken.
 32 std::chrono::duration_cast<std::chrono::milliseconds>(
877
1/2
✓ Branch 2 taken 32 times.
✗ Branch 3 not taken.
 64 now - it->second.last_seen)
878 32 .count();
879 }
880 32 return snapshot;
881 36 }
882
883 18 void RdmaControlPlaneServer::RecordHeartbeat(int server_id) {
884
1/2
✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
 18 std::lock_guard<std::mutex> guard(mu_);
885 36 const auto it = ready_servers_.find(
886
2/4
✓ Branch 1 taken 18 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 18 times.
✗ Branch 5 not taken.
 18 ReadyKey{endpoint_.deployment_id, endpoint_.deployment_epoch, server_id});
887
2/2
✓ Branch 2 taken 4 times.
✓ Branch 3 taken 14 times.
 18 if (it == ready_servers_.end()) {
888 // Unknown shard: not ready yet, or never ready. A heartbeat never creates a
889 // record; READY remains the only way to become serving.
890 4 return;
891 }
892 14 it->second.last_seen = std::chrono::steady_clock::now();
893
2/2
✓ Branch 1 taken 14 times.
✓ Branch 2 taken 4 times.
 18 }
894
895 26 void RdmaControlPlaneServer::Start() {
896
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
 26 if (server_ != nullptr) {
897 ✗ return;
898 }
899
2/4
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 26 times.
✗ Branch 4 not taken.
 52 if (endpoint_.host.empty() || endpoint_.port < 1 || endpoint_.port > 65535 ||
900
2/4
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 26 times.
✗ Branch 4 not taken.
 26 endpoint_.timeout_ms <= 0 || endpoint_.deployment_id.empty() ||
901
1/2
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
 26 endpoint_.deployment_epoch == 0 ||
902
2/4
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
 52 endpoint_.protocol_version != recstore::kRdmaDeploymentProtocolVersion ||
903
3/6
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 26 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 26 times.
 78 endpoint_.configuration_digest.empty() ||
904 26 endpoint_.fabric_digest.empty()) {
905
0/2
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 ✗ throw std::invalid_argument("invalid RDMA control-plane endpoint contract");
906 }
907 26 stop_requested_.store(false, std::memory_order_relaxed);
908
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 service_ = std::make_unique<RdmaControlPlaneService>(this);
909
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 grpc::ServerBuilder builder;
910
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 builder.AddListeningPort(
911
2/4
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 26 times.
✗ Branch 5 not taken.
 52 EndpointString(endpoint_), grpc::InsecureServerCredentials());
912
1/2
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
 26 builder.RegisterService(service_.get());
913
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 server_ = builder.BuildAndStart();
914
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
 26 if (server_ == nullptr) {
915
0/4
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 ✗ throw std::runtime_error("control-plane gRPC server failed to listen on " +
916
0/2
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 ✗ EndpointString(endpoint_));
917 }
918 26 }
919
920 52 void RdmaControlPlaneServer::Stop() {
921
2/2
✓ Branch 1 taken 26 times.
✓ Branch 2 taken 26 times.
 52 if (server_ == nullptr) {
922 26 return;
923 }
924 26 stop_requested_.store(true, std::memory_order_relaxed);
925 {
926
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 std::lock_guard<std::mutex> guard(mu_);
927 26 cv_.notify_all();
928 26 }
929 26 server_->Shutdown();
930 26 server_.reset();
931 26 service_.reset();
932 }
933
934 } // namespace petps
935