GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/nfs4/rpc.cc
Date: 2026-10-18 02:40:11
Exec Total Coverage
Lines: 245 315 77.8%
Branches: 150 326 46.0%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #include "nfs4/rpc.h"
6
7 #include <arpa/inet.h>
8 #include <errno.h>
9 #include <netinet/in.h>
10 #include <netinet/tcp.h>
11 #include <poll.h>
12 #include <sys/socket.h>
13 #include <sys/uio.h>
14 #include <unistd.h>
15
16 #include <cstring>
17
18 #include "nfs4/logging.h"
19
20 namespace nfs4 {
21
22 namespace {
23
24 const uint32_t kMaxAuthBody = 400;
25 const uint32_t kMaxAuthSysGids = 16;
26 const uint32_t kLastFragment = 0x80000000u;
27
28 65 void PutReplyHeader(uint32_t xid, XdrWriter *w) {
29 65 w->PutU32(xid);
30 65 w->PutU32(kRpcReply);
31 65 }
32
33 65 void PutAccepted(uint32_t xid, AcceptStat stat, XdrWriter *w) {
34 65 PutReplyHeader(xid, w);
35 65 w->PutU32(kMsgAccepted);
36 65 w->PutU32(kAuthNone); // verifier
37 65 w->PutU32(0);
38 65 w->PutU32(stat);
39 65 }
40
41 26 bool ParseAuthSys(const std::string &body, Credentials *creds) {
42 26 XdrReader r(reinterpret_cast<const unsigned char *>(body.data()),
43 52 body.size());
44 uint32_t stamp, ngids;
45 26 std::string machine;
46
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 r.GetU32(&stamp);
47
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 r.GetString(&machine, 255);
48
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 r.GetU32(&creds->uid);
49
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 r.GetU32(&creds->gid);
50
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 r.GetU32(&ngids);
51
3/6
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 26 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 26 times.
 26 if (!r.ok() || ngids > kMaxAuthSysGids)
52 ✗ return false;
53
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 creds->gids.resize(ngids);
54
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26 times.
 26 for (uint32_t i = 0; i < ngids; ++i)
55 ✗ r.GetU32(&creds->gids[i]);
56 26 return r.ok();
57 26 }
58
59 130 bool ReadFull(int fd, void *buf, size_t len) {
60 130 char *p = static_cast<char *>(buf);
61
2/2
✓ Branch 0 taken 130 times.
✓ Branch 1 taken 104 times.
 234 while (len > 0) {
62 130 const ssize_t n = read(fd, p, len);
63
2/2
✓ Branch 0 taken 26 times.
✓ Branch 1 taken 104 times.
 130 if (n == 0)
64 26 return false;
65
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 104 times.
 104 if (n < 0) {
66 ✗ if (errno == EINTR)
67 ✗ continue;
68 ✗ return false;
69 }
70 104 p += n;
71 104 len -= n;
72 }
73 104 return true;
74 }
75
76 /** Record marker and record in one system call where possible */
77 39 bool WriteRecord(int fd, uint32_t marker, const std::string &record) {
78 struct iovec iov[2];
79 39 iov[0].iov_base = &marker;
80 39 iov[0].iov_len = sizeof(marker);
81 39 iov[1].iov_base = const_cast<char *>(record.data());
82 39 iov[1].iov_len = record.size();
83 39 int flags = 0;
84 #ifdef MSG_NOSIGNAL
85 39 flags = MSG_NOSIGNAL;
86 #endif
87 39 int iovcnt = 2;
88 39 struct iovec *v = iov;
89
2/2
✓ Branch 0 taken 39 times.
✓ Branch 1 taken 39 times.
 78 while (iovcnt > 0) {
90 struct msghdr msg;
91 39 memset(&msg, 0, sizeof(msg));
92 39 msg.msg_iov = v;
93 39 msg.msg_iovlen = iovcnt;
94
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 const ssize_t n = sendmsg(fd, &msg, flags);
95
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 39 times.
 39 if (n < 0) {
96 ✗ if (errno == EINTR)
97 ✗ continue;
98 ✗ return false;
99 }
100 39 size_t done = n;
101
3/4
✓ Branch 0 taken 78 times.
✓ Branch 1 taken 39 times.
✓ Branch 2 taken 78 times.
✗ Branch 3 not taken.
 117 while (iovcnt > 0 && done >= v->iov_len) {
102 78 done -= v->iov_len;
103 78 v++;
104 78 iovcnt--;
105 }
106
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 39 times.
 39 if (iovcnt > 0) {
107 ✗ v->iov_base = static_cast<char *>(v->iov_base) + done;
108 ✗ v->iov_len -= done;
109 }
110 }
111 39 return true;
112 }
113
114 26 bool IsLoopback(const struct sockaddr_storage &addr) {
115
1/2
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
 26 if (addr.ss_family == AF_INET) {
116 26 const struct sockaddr_in *in = reinterpret_cast<const struct sockaddr_in *>(
117 &addr);
118 26 return (ntohl(in->sin_addr.s_addr) >> 24) == 127;
119 }
120 ✗ if (addr.ss_family == AF_INET6) {
121 const struct sockaddr_in6
122 ✗ *in6 = reinterpret_cast<const struct sockaddr_in6 *>(&addr);
123 ✗ return IN6_IS_ADDR_LOOPBACK(&in6->sin6_addr);
124 }
125 ✗ return false;
126 }
127
128 } // anonymous namespace
129
130
131 78 bool HandleRpcRecord(RpcProgram *program, const std::string &record,
132 std::string *reply) {
133 78 XdrReader r(reinterpret_cast<const unsigned char *>(record.data()),
134 156 record.size());
135 78 RpcCall call;
136 uint32_t msg_type, rpc_version;
137
8/12
✓ Branch 1 taken 78 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 65 times.
✓ Branch 4 taken 13 times.
✓ Branch 6 taken 65 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 65 times.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✓ Branch 11 taken 65 times.
✓ Branch 12 taken 13 times.
✓ Branch 13 taken 65 times.
 78 if (!r.GetU32(&call.xid) || !r.GetU32(&msg_type) || msg_type != kRpcCall)
138 13 return false;
139
140 65 XdrWriter w;
141
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetU32(&rpc_version);
142
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetU32(&call.program);
143
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetU32(&call.version);
144
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetU32(&call.procedure);
145 65 std::string cred_body, verf_body;
146 uint32_t verf_flavor;
147
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetU32(&call.cred_flavor);
148
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetOpaque(&cred_body, kMaxAuthBody);
149
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetU32(&verf_flavor);
150
1/2
✓ Branch 1 taken 65 times.
✗ Branch 2 not taken.
 65 r.GetOpaque(&verf_body, kMaxAuthBody);
151
2/2
✓ Branch 1 taken 13 times.
✓ Branch 2 taken 52 times.
 65 if (!r.ok()) {
152
1/2
✓ Branch 1 taken 13 times.
✗ Branch 2 not taken.
 13 Log(kLogLevelWarn, "garbage RPC header (xid %u)", call.xid);
153
1/2
✓ Branch 1 taken 13 times.
✗ Branch 2 not taken.
 13 PutAccepted(call.xid, kAcceptGarbageArgs, &w);
154 13 reply->swap(*w.mutable_buffer());
155 13 return true;
156 }
157
158
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 52 times.
 52 if (rpc_version != kRpcVersion) {
159 ✗ PutReplyHeader(call.xid, &w);
160 ✗ w.PutU32(kMsgDenied);
161 ✗ w.PutU32(kRpcMismatch);
162 ✗ w.PutU32(kRpcVersion);
163 ✗ w.PutU32(kRpcVersion);
164 ✗ reply->swap(*w.mutable_buffer());
165 ✗ return true;
166 }
167
168 52 bool auth_ok = false;
169
2/2
✓ Branch 0 taken 26 times.
✓ Branch 1 taken 26 times.
 52 if (call.cred_flavor == kAuthSys)
170
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 auth_ok = ParseAuthSys(cred_body, &call.creds);
171
1/2
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
 26 else if (call.cred_flavor == kAuthNone)
172 26 auth_ok = true;
173
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 52 times.
 52 if (!auth_ok) {
174 ✗ PutReplyHeader(call.xid, &w);
175 ✗ w.PutU32(kMsgDenied);
176 ✗ w.PutU32(kAuthError);
177 ✗ w.PutU32(call.cred_flavor == kAuthSys ? kAuthBadCred : kAuthTooWeak);
178 ✗ reply->swap(*w.mutable_buffer());
179 ✗ return true;
180 }
181
182
2/4
✓ Branch 1 taken 52 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 52 times.
 52 if (call.program != program->program()) {
183 ✗ Log(kLogLevelWarn, "call for unknown program %u (xid %u)", call.program,
184 call.xid);
185 ✗ PutAccepted(call.xid, kAcceptProgUnavail, &w);
186 ✗ reply->swap(*w.mutable_buffer());
187 ✗ return true;
188 }
189
1/2
✓ Branch 1 taken 52 times.
✗ Branch 2 not taken.
 52 if (call.version < program->min_version()
190
6/8
✓ Branch 0 taken 39 times.
✓ Branch 1 taken 13 times.
✓ Branch 3 taken 39 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 39 times.
✓ Branch 7 taken 13 times.
✓ Branch 8 taken 39 times.
 52 || call.version > program->max_version()) {
191
1/2
✓ Branch 1 taken 13 times.
✗ Branch 2 not taken.
 13 PutAccepted(call.xid, kAcceptProgMismatch, &w);
192
2/4
✓ Branch 1 taken 13 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 13 times.
✗ Branch 5 not taken.
 13 w.PutU32(program->min_version());
193
2/4
✓ Branch 1 taken 13 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 13 times.
✗ Branch 5 not taken.
 13 w.PutU32(program->max_version());
194 13 reply->swap(*w.mutable_buffer());
195 13 return true;
196 }
197
198
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 Log(kLogLevelDebug, "call xid %u proc %u", call.xid, call.procedure);
199 39 call.args = r.cursor();
200 39 call.args_size = r.remaining();
201 39 XdrReader args(call.args, call.args_size);
202 // Results go right behind the header, the accept status is patched in
203
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 PutAccepted(call.xid, kAcceptSuccess, &w);
204 39 const size_t stat_offset = w.size() - 4;
205
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 const AcceptStat stat = program->Dispatch(call, &args, &w);
206
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 39 times.
 39 if (stat != kAcceptSuccess) {
207 ✗ w.Truncate(stat_offset + 4);
208 ✗ w.PatchU32(stat_offset, stat);
209 }
210 39 reply->swap(*w.mutable_buffer());
211 39 return true;
212 78 }
213
214
215 26 RpcServer::Connection::~Connection() { close(fd); }
216
217
218 26 RpcServer::RpcServer(RpcProgram *program, const Options &options)
219 26 : program_(program)
220 26 , options_(options)
221 26 , listen_fd_(-1)
222 26 , port_(0)
223 26 , stopping_(false)
224
1/2
✓ Branch 8 taken 26 times.
✗ Branch 9 not taken.
 26 , active_readers_(0) { }
225
226
227 26 RpcServer::~RpcServer() {
228 26 Stop();
229
1/2
✓ Branch 0 taken 26 times.
✗ Branch 1 not taken.
 26 if (listen_fd_ >= 0)
230 26 close(listen_fd_);
231 26 }
232
233
234 26 bool RpcServer::Listen(const std::string &address, uint16_t port) {
235 struct sockaddr_in addr;
236 26 memset(&addr, 0, sizeof(addr));
237 26 addr.sin_family = AF_INET;
238 26 addr.sin_port = htons(port);
239
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 26 times.
 26 if (inet_pton(AF_INET, address.c_str(), &addr.sin_addr) != 1) {
240 ✗ Log(kLogLevelError, "invalid listen address %s",
241 address.c_str());
242 ✗ return false;
243 }
244
245 26 listen_fd_ = socket(AF_INET, SOCK_STREAM, 0);
246
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26 times.
 26 if (listen_fd_ < 0)
247 ✗ return false;
248 26 const int on = 1;
249 26 setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
250 26 if (bind(listen_fd_, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr))
251 != 0
252
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.
 26 || listen(listen_fd_, 64) != 0) {
253 ✗ Log(kLogLevelError, "failed to listen on %s:%u (%d)", address.c_str(), port, errno);
254 ✗ close(listen_fd_);
255 ✗ listen_fd_ = -1;
256 ✗ return false;
257 }
258
259 26 socklen_t len = sizeof(addr);
260 26 getsockname(listen_fd_, reinterpret_cast<struct sockaddr *>(&addr), &len);
261 26 port_ = ntohs(addr.sin_port);
262 26 return true;
263 }
264
265
266 ✗ bool RpcServer::Adopt(int listen_fd) {
267 struct sockaddr_in addr;
268 ✗ socklen_t len = sizeof(addr);
269 ✗ if (getsockname(listen_fd, reinterpret_cast<struct sockaddr *>(&addr), &len)
270 ✗ != 0) {
271 ✗ return false;
272 }
273 ✗ listen_fd_ = listen_fd;
274 ✗ port_ = ntohs(addr.sin_port);
275 ✗ return true;
276 }
277
278
279 26 void RpcServer::Start() {
280 26 stopping_ = false;
281
2/2
✓ Branch 0 taken 234 times.
✓ Branch 1 taken 26 times.
 260 for (unsigned i = 0; i < options_.workers; ++i)
282
2/4
✓ Branch 1 taken 234 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 234 times.
✗ Branch 5 not taken.
 234 workers_.push_back(std::thread(&RpcServer::WorkerLoop, this));
283
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 accept_thread_ = std::thread(&RpcServer::AcceptLoop, this);
284 26 }
285
286
287 52 void RpcServer::Stop() {
288
2/2
✓ Branch 1 taken 26 times.
✓ Branch 2 taken 26 times.
 52 if (stopping_.exchange(true))
289 26 return;
290
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 if (accept_thread_.joinable())
291 26 accept_thread_.join();
292
293 {
294
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 std::unique_lock<std::mutex> lock(connections_lock_);
295
2/2
✓ Branch 1 taken 26 times.
✓ Branch 2 taken 26 times.
 52 for (unsigned i = 0; i < connections_.size(); ++i) {
296 26 ConnectionPtr const c = connections_[i].lock();
297
2/2
✓ Branch 1 taken 13 times.
✓ Branch 2 taken 13 times.
 26 if (c)
298 13 shutdown(c->fd, SHUT_RDWR);
299 26 }
300 65 readers_cond_.wait(lock, [this] { return active_readers_ == 0; });
301 26 connections_.clear();
302 26 }
303
304 26 queue_cond_.notify_all();
305
2/2
✓ Branch 1 taken 234 times.
✓ Branch 2 taken 26 times.
 260 for (unsigned i = 0; i < workers_.size(); ++i)
306 234 workers_[i].join();
307 26 workers_.clear();
308 }
309
310
311 ✗ unsigned RpcServer::NumConnections() {
312 ✗ std::lock_guard<std::mutex> const lock(connections_lock_);
313 ✗ unsigned n = 0;
314 ✗ for (unsigned i = 0; i < connections_.size(); ++i) {
315 ✗ if (!connections_[i].expired())
316 ✗ n++;
317 }
318 ✗ return n;
319 }
320
321
322 26 void RpcServer::AcceptLoop() {
323
2/2
✓ Branch 1 taken 169 times.
✓ Branch 2 taken 26 times.
 195 while (!stopping_) {
324 struct pollfd pfd;
325 169 pfd.fd = listen_fd_;
326 169 pfd.events = POLLIN;
327 169 pfd.revents = 0;
328
1/2
✓ Branch 1 taken 169 times.
✗ Branch 2 not taken.
 169 const int retval = poll(&pfd, 1, 200);
329
2/2
✓ Branch 0 taken 143 times.
✓ Branch 1 taken 26 times.
 169 if (retval <= 0)
330 143 continue;
331
332 struct sockaddr_storage peer;
333 26 socklen_t peer_len = sizeof(peer);
334
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 const int fd = accept(
335 listen_fd_, reinterpret_cast<struct sockaddr *>(&peer), &peer_len);
336
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26 times.
 26 if (fd < 0)
337 ✗ continue;
338
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.
 26 if (!options_.allow_remote && !IsLoopback(peer)) {
339 ✗ Log(kLogLevelWarn, "rejected NFS connection from non-loopback peer");
340 ✗ close(fd);
341 ✗ continue;
342 }
343 26 const int on = 1;
344 26 setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &on, sizeof(on));
345 #ifdef SO_NOSIGPIPE
346 setsockopt(fd, SOL_SOCKET, SO_NOSIGPIPE, &on, sizeof(on));
347 #endif
348
349
2/4
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 26 times.
✗ Branch 6 not taken.
 26 ConnectionPtr const connection(new Connection(fd));
350
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 std::lock_guard<std::mutex> const lock(connections_lock_);
351 // Drop entries of closed connections
352 26 std::vector<std::weak_ptr<Connection> > alive;
353
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 26 times.
 26 for (unsigned i = 0; i < connections_.size(); ++i) {
354 ✗ if (!connections_[i].expired())
355 ✗ alive.push_back(connections_[i]);
356 }
357
1/2
✓ Branch 2 taken 26 times.
✗ Branch 3 not taken.
 26 alive.push_back(connection);
358 26 connections_.swap(alive);
359 26 active_readers_++;
360
2/4
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 26 times.
✗ Branch 5 not taken.
 26 std::thread(&RpcServer::ReadLoop, this, connection).detach();
361 26 }
362 26 }
363
364
365 26 void RpcServer::ReadLoop(ConnectionPtr connection) {
366 26 std::string record;
367
1/2
✓ Branch 1 taken 78 times.
✗ Branch 2 not taken.
 78 while (!stopping_) {
368 uint32_t marker;
369
3/4
✓ Branch 2 taken 78 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 26 times.
✓ Branch 5 taken 52 times.
 78 if (!ReadFull(connection->fd, &marker, sizeof(marker)))
370 26 break;
371 52 marker = ntohl(marker);
372 52 const uint32_t len = marker & ~kLastFragment;
373
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 52 times.
 52 if (record.size() + len > options_.max_record) {
374 ✗ Log(kLogLevelWarn, "RPC record exceeds %u bytes, closing connection",
375 options_.max_record);
376 ✗ break;
377 }
378 52 const size_t offset = record.size();
379
1/2
✓ Branch 1 taken 52 times.
✗ Branch 2 not taken.
 52 record.resize(offset + len);
380
5/10
✓ Branch 0 taken 52 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 52 times.
✗ Branch 4 not taken.
✓ Branch 7 taken 52 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 52 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 52 times.
 52 if (len > 0 && !ReadFull(connection->fd, &record[offset], len))
381 ✗ break;
382
2/2
✓ Branch 0 taken 13 times.
✓ Branch 1 taken 39 times.
 52 if (!(marker & kLastFragment))
383 13 continue;
384
385 39 Job job;
386 39 job.connection = connection;
387 39 job.xid = 0;
388
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 if (record.size() >= 4) {
389 39 memcpy(&job.xid, record.data(), 4);
390 39 job.xid = ntohl(job.xid);
391 }
392
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 Log(kLogLevelDebug, "recv xid %u (%zu bytes)", job.xid, record.size());
393 {
394 // A retransmission of a call that is still being processed is
395 // answered by the original
396
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 std::lock_guard<std::mutex> const lock(connection->inflight_lock);
397
2/4
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 39 times.
 39 if (!connection->inflight.insert(job.xid).second) {
398 ✗ Log(kLogLevelDebug, "dropping retransmission of xid %u", job.xid);
399 ✗ record.clear();
400 ✗ continue;
401 }
402
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 }
403 39 job.record.swap(record);
404 {
405
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 std::lock_guard<std::mutex> const lock(queue_lock_);
406
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 queue_.push_back(Job());
407 39 queue_.back().connection.swap(job.connection);
408 39 queue_.back().xid = job.xid;
409 39 queue_.back().record.swap(job.record);
410 39 }
411 39 queue_cond_.notify_one();
412
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 }
413
414 26 shutdown(connection->fd, SHUT_RDWR);
415
1/2
✓ Branch 1 taken 26 times.
✗ Branch 2 not taken.
 26 std::lock_guard<std::mutex> const lock(connections_lock_);
416 26 active_readers_--;
417 26 readers_cond_.notify_all();
418 26 }
419
420
421 273 void RpcServer::WorkerLoop() {
422 while (true) {
423 273 Job job;
424 {
425
1/2
✓ Branch 1 taken 273 times.
✗ Branch 2 not taken.
 273 std::unique_lock<std::mutex> lock(queue_lock_);
426
4/4
✓ Branch 2 taken 312 times.
✓ Branch 3 taken 234 times.
✓ Branch 5 taken 39 times.
✓ Branch 6 taken 273 times.
 819 queue_cond_.wait(lock, [this] { return stopping_ || !queue_.empty(); });
427
2/2
✓ Branch 1 taken 234 times.
✓ Branch 2 taken 39 times.
 273 if (queue_.empty())
428 468 return;
429 39 job.connection.swap(queue_.front().connection);
430 39 job.xid = queue_.front().xid;
431 39 job.record.swap(queue_.front().record);
432 39 queue_.pop_front();
433
2/2
✓ Branch 1 taken 39 times.
✓ Branch 2 taken 234 times.
 273 }
434
435 39 std::string reply;
436
2/4
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 39 times.
✗ Branch 4 not taken.
 39 if (HandleRpcRecord(program_, job.record, &reply)) {
437
1/2
✓ Branch 1 taken 39 times.
✗ Branch 2 not taken.
 39 Send(job.connection, reply);
438
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 Log(kLogLevelDebug, "sent xid %u (%zu bytes)", job.xid, reply.size());
439 } else {
440 ✗ Log(kLogLevelWarn, "ignoring record that is not a call (xid %u)",
441 job.xid);
442 }
443
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 std::lock_guard<std::mutex> const lock(job.connection->inflight_lock);
444
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 job.connection->inflight.erase(job.xid);
445
2/2
✓ Branch 3 taken 39 times.
✓ Branch 4 taken 234 times.
 312 }
446 }
447
448
449 39 void RpcServer::Send(const ConnectionPtr &connection,
450 const std::string &reply) {
451 39 const uint32_t marker = htonl(kLastFragment
452 39 | static_cast<uint32_t>(reply.size()));
453
1/2
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
 39 std::lock_guard<std::mutex> const lock(connection->write_lock);
454
2/4
✓ Branch 2 taken 39 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 39 times.
 39 if (!WriteRecord(connection->fd, marker, reply)) {
455 ✗ Log(kLogLevelWarn, "failed to send reply (%d), closing connection", errno);
456 ✗ shutdown(connection->fd, SHUT_RDWR);
457 }
458 39 }
459
460 } // namespace nfs4
461