GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/nfs4/rpc.cc
Date: 2026-11-01 02:41:19
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 155 void PutReplyHeader(uint32_t xid, XdrWriter *w) {
29 155 w->PutU32(xid);
30 155 w->PutU32(kRpcReply);
31 155 }
32
33 155 void PutAccepted(uint32_t xid, AcceptStat stat, XdrWriter *w) {
34 155 PutReplyHeader(xid, w);
35 155 w->PutU32(kMsgAccepted);
36 155 w->PutU32(kAuthNone); // verifier
37 155 w->PutU32(0);
38 155 w->PutU32(stat);
39 155 }
40
41 62 bool ParseAuthSys(const std::string &body, Credentials *creds) {
42 62 XdrReader r(reinterpret_cast<const unsigned char *>(body.data()),
43 124 body.size());
44 uint32_t stamp, ngids;
45 62 std::string machine;
46
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 r.GetU32(&stamp);
47
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 r.GetString(&machine, 255);
48
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 r.GetU32(&creds->uid);
49
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 r.GetU32(&creds->gid);
50
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 r.GetU32(&ngids);
51
3/6
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 62 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 62 times.
 62 if (!r.ok() || ngids > kMaxAuthSysGids)
52 ✗ return false;
53
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 creds->gids.resize(ngids);
54
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 62 times.
 62 for (uint32_t i = 0; i < ngids; ++i)
55 ✗ r.GetU32(&creds->gids[i]);
56 62 return r.ok();
57 62 }
58
59 310 bool ReadFull(int fd, void *buf, size_t len) {
60 310 char *p = static_cast<char *>(buf);
61
2/2
✓ Branch 0 taken 310 times.
✓ Branch 1 taken 248 times.
 558 while (len > 0) {
62 310 const ssize_t n = read(fd, p, len);
63
2/2
✓ Branch 0 taken 62 times.
✓ Branch 1 taken 248 times.
 310 if (n == 0)
64 62 return false;
65
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 248 times.
 248 if (n < 0) {
66 ✗ if (errno == EINTR)
67 ✗ continue;
68 ✗ return false;
69 }
70 248 p += n;
71 248 len -= n;
72 }
73 248 return true;
74 }
75
76 /** Record marker and record in one system call where possible */
77 93 bool WriteRecord(int fd, uint32_t marker, const std::string &record) {
78 struct iovec iov[2];
79 93 iov[0].iov_base = &marker;
80 93 iov[0].iov_len = sizeof(marker);
81 93 iov[1].iov_base = const_cast<char *>(record.data());
82 93 iov[1].iov_len = record.size();
83 93 int flags = 0;
84 #ifdef MSG_NOSIGNAL
85 93 flags = MSG_NOSIGNAL;
86 #endif
87 93 int iovcnt = 2;
88 93 struct iovec *v = iov;
89
2/2
✓ Branch 0 taken 93 times.
✓ Branch 1 taken 93 times.
 186 while (iovcnt > 0) {
90 struct msghdr msg;
91 93 memset(&msg, 0, sizeof(msg));
92 93 msg.msg_iov = v;
93 93 msg.msg_iovlen = iovcnt;
94
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 const ssize_t n = sendmsg(fd, &msg, flags);
95
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 93 times.
 93 if (n < 0) {
96 ✗ if (errno == EINTR)
97 ✗ continue;
98 ✗ return false;
99 }
100 93 size_t done = n;
101
3/4
✓ Branch 0 taken 186 times.
✓ Branch 1 taken 93 times.
✓ Branch 2 taken 186 times.
✗ Branch 3 not taken.
 279 while (iovcnt > 0 && done >= v->iov_len) {
102 186 done -= v->iov_len;
103 186 v++;
104 186 iovcnt--;
105 }
106
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 93 times.
 93 if (iovcnt > 0) {
107 ✗ v->iov_base = static_cast<char *>(v->iov_base) + done;
108 ✗ v->iov_len -= done;
109 }
110 }
111 93 return true;
112 }
113
114 62 bool IsLoopback(const struct sockaddr_storage &addr) {
115
1/2
✓ Branch 0 taken 62 times.
✗ Branch 1 not taken.
 62 if (addr.ss_family == AF_INET) {
116 62 const struct sockaddr_in *in = reinterpret_cast<const struct sockaddr_in *>(
117 &addr);
118 62 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 186 bool HandleRpcRecord(RpcProgram *program, const std::string &record,
132 std::string *reply) {
133 186 XdrReader r(reinterpret_cast<const unsigned char *>(record.data()),
134 372 record.size());
135 186 RpcCall call;
136 uint32_t msg_type, rpc_version;
137
8/12
✓ Branch 1 taken 186 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 155 times.
✓ Branch 4 taken 31 times.
✓ Branch 6 taken 155 times.
✗ Branch 7 not taken.
✓ Branch 8 taken 155 times.
✗ Branch 9 not taken.
✗ Branch 10 not taken.
✓ Branch 11 taken 155 times.
✓ Branch 12 taken 31 times.
✓ Branch 13 taken 155 times.
 186 if (!r.GetU32(&call.xid) || !r.GetU32(&msg_type) || msg_type != kRpcCall)
138 31 return false;
139
140 155 XdrWriter w;
141
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetU32(&rpc_version);
142
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetU32(&call.program);
143
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetU32(&call.version);
144
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetU32(&call.procedure);
145 155 std::string cred_body, verf_body;
146 uint32_t verf_flavor;
147
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetU32(&call.cred_flavor);
148
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetOpaque(&cred_body, kMaxAuthBody);
149
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetU32(&verf_flavor);
150
1/2
✓ Branch 1 taken 155 times.
✗ Branch 2 not taken.
 155 r.GetOpaque(&verf_body, kMaxAuthBody);
151
2/2
✓ Branch 1 taken 31 times.
✓ Branch 2 taken 124 times.
 155 if (!r.ok()) {
152
1/2
✓ Branch 1 taken 31 times.
✗ Branch 2 not taken.
 31 Log(kLogLevelWarn, "garbage RPC header (xid %u)", call.xid);
153
1/2
✓ Branch 1 taken 31 times.
✗ Branch 2 not taken.
 31 PutAccepted(call.xid, kAcceptGarbageArgs, &w);
154 31 reply->swap(*w.mutable_buffer());
155 31 return true;
156 }
157
158
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 124 times.
 124 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 124 bool auth_ok = false;
169
2/2
✓ Branch 0 taken 62 times.
✓ Branch 1 taken 62 times.
 124 if (call.cred_flavor == kAuthSys)
170
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 auth_ok = ParseAuthSys(cred_body, &call.creds);
171
1/2
✓ Branch 0 taken 62 times.
✗ Branch 1 not taken.
 62 else if (call.cred_flavor == kAuthNone)
172 62 auth_ok = true;
173
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 124 times.
 124 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 124 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 124 times.
 124 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 124 times.
✗ Branch 2 not taken.
 124 if (call.version < program->min_version()
190
6/8
✓ Branch 0 taken 93 times.
✓ Branch 1 taken 31 times.
✓ Branch 3 taken 93 times.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 93 times.
✓ Branch 7 taken 31 times.
✓ Branch 8 taken 93 times.
 124 || call.version > program->max_version()) {
191
1/2
✓ Branch 1 taken 31 times.
✗ Branch 2 not taken.
 31 PutAccepted(call.xid, kAcceptProgMismatch, &w);
192
2/4
✓ Branch 1 taken 31 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 31 times.
✗ Branch 5 not taken.
 31 w.PutU32(program->min_version());
193
2/4
✓ Branch 1 taken 31 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 31 times.
✗ Branch 5 not taken.
 31 w.PutU32(program->max_version());
194 31 reply->swap(*w.mutable_buffer());
195 31 return true;
196 }
197
198
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 Log(kLogLevelDebug, "call xid %u proc %u", call.xid, call.procedure);
199 93 call.args = r.cursor();
200 93 call.args_size = r.remaining();
201 93 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 93 times.
✗ Branch 2 not taken.
 93 PutAccepted(call.xid, kAcceptSuccess, &w);
204 93 const size_t stat_offset = w.size() - 4;
205
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 const AcceptStat stat = program->Dispatch(call, &args, &w);
206
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 93 times.
 93 if (stat != kAcceptSuccess) {
207 ✗ w.Truncate(stat_offset + 4);
208 ✗ w.PatchU32(stat_offset, stat);
209 }
210 93 reply->swap(*w.mutable_buffer());
211 93 return true;
212 186 }
213
214
215 62 RpcServer::Connection::~Connection() { close(fd); }
216
217
218 62 RpcServer::RpcServer(RpcProgram *program, const Options &options)
219 62 : program_(program)
220 62 , options_(options)
221 62 , listen_fd_(-1)
222 62 , port_(0)
223 62 , stopping_(false)
224
1/2
✓ Branch 8 taken 62 times.
✗ Branch 9 not taken.
 62 , active_readers_(0) { }
225
226
227 62 RpcServer::~RpcServer() {
228 62 Stop();
229
1/2
✓ Branch 0 taken 62 times.
✗ Branch 1 not taken.
 62 if (listen_fd_ >= 0)
230 62 close(listen_fd_);
231 62 }
232
233
234 62 bool RpcServer::Listen(const std::string &address, uint16_t port) {
235 struct sockaddr_in addr;
236 62 memset(&addr, 0, sizeof(addr));
237 62 addr.sin_family = AF_INET;
238 62 addr.sin_port = htons(port);
239
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 62 times.
 62 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 62 listen_fd_ = socket(AF_INET, SOCK_STREAM, 0);
246
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 62 times.
 62 if (listen_fd_ < 0)
247 ✗ return false;
248 62 const int on = 1;
249 62 setsockopt(listen_fd_, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
250 62 if (bind(listen_fd_, reinterpret_cast<struct sockaddr *>(&addr), sizeof(addr))
251 != 0
252
3/6
✓ Branch 0 taken 62 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 62 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 62 times.
 62 || 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 62 socklen_t len = sizeof(addr);
260 62 getsockname(listen_fd_, reinterpret_cast<struct sockaddr *>(&addr), &len);
261 62 port_ = ntohs(addr.sin_port);
262 62 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 62 void RpcServer::Start() {
280 62 stopping_ = false;
281
2/2
✓ Branch 0 taken 558 times.
✓ Branch 1 taken 62 times.
 620 for (unsigned i = 0; i < options_.workers; ++i)
282
2/4
✓ Branch 1 taken 558 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 558 times.
✗ Branch 5 not taken.
 558 workers_.push_back(std::thread(&RpcServer::WorkerLoop, this));
283
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 accept_thread_ = std::thread(&RpcServer::AcceptLoop, this);
284 62 }
285
286
287 124 void RpcServer::Stop() {
288
2/2
✓ Branch 1 taken 62 times.
✓ Branch 2 taken 62 times.
 124 if (stopping_.exchange(true))
289 62 return;
290
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 if (accept_thread_.joinable())
291 62 accept_thread_.join();
292
293 {
294
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 std::unique_lock<std::mutex> lock(connections_lock_);
295
2/2
✓ Branch 1 taken 62 times.
✓ Branch 2 taken 62 times.
 124 for (unsigned i = 0; i < connections_.size(); ++i) {
296 62 ConnectionPtr const c = connections_[i].lock();
297
2/2
✓ Branch 1 taken 31 times.
✓ Branch 2 taken 31 times.
 62 if (c)
298 31 shutdown(c->fd, SHUT_RDWR);
299 62 }
300 155 readers_cond_.wait(lock, [this] { return active_readers_ == 0; });
301 62 connections_.clear();
302 62 }
303
304 62 queue_cond_.notify_all();
305
2/2
✓ Branch 1 taken 558 times.
✓ Branch 2 taken 62 times.
 620 for (unsigned i = 0; i < workers_.size(); ++i)
306 558 workers_[i].join();
307 62 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 62 void RpcServer::AcceptLoop() {
323
2/2
✓ Branch 1 taken 403 times.
✓ Branch 2 taken 62 times.
 465 while (!stopping_) {
324 struct pollfd pfd;
325 403 pfd.fd = listen_fd_;
326 403 pfd.events = POLLIN;
327 403 pfd.revents = 0;
328
1/2
✓ Branch 1 taken 403 times.
✗ Branch 2 not taken.
 403 const int retval = poll(&pfd, 1, 200);
329
2/2
✓ Branch 0 taken 341 times.
✓ Branch 1 taken 62 times.
 403 if (retval <= 0)
330 341 continue;
331
332 struct sockaddr_storage peer;
333 62 socklen_t peer_len = sizeof(peer);
334
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 const int fd = accept(
335 listen_fd_, reinterpret_cast<struct sockaddr *>(&peer), &peer_len);
336
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 62 times.
 62 if (fd < 0)
337 ✗ continue;
338
3/6
✓ Branch 0 taken 62 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 62 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 62 times.
 62 if (!options_.allow_remote && !IsLoopback(peer)) {
339 ✗ Log(kLogLevelWarn, "rejected NFS connection from non-loopback peer");
340 ✗ close(fd);
341 ✗ continue;
342 }
343 62 const int on = 1;
344 62 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 62 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 62 times.
✗ Branch 6 not taken.
 62 ConnectionPtr const connection(new Connection(fd));
350
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 std::lock_guard<std::mutex> const lock(connections_lock_);
351 // Drop entries of closed connections
352 62 std::vector<std::weak_ptr<Connection> > alive;
353
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 62 times.
 62 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 62 times.
✗ Branch 3 not taken.
 62 alive.push_back(connection);
358 62 connections_.swap(alive);
359 62 active_readers_++;
360
2/4
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 62 times.
✗ Branch 5 not taken.
 62 std::thread(&RpcServer::ReadLoop, this, connection).detach();
361 62 }
362 62 }
363
364
365 62 void RpcServer::ReadLoop(ConnectionPtr connection) {
366 62 std::string record;
367
1/2
✓ Branch 1 taken 186 times.
✗ Branch 2 not taken.
 186 while (!stopping_) {
368 uint32_t marker;
369
3/4
✓ Branch 2 taken 186 times.
✗ Branch 3 not taken.
✓ Branch 4 taken 62 times.
✓ Branch 5 taken 124 times.
 186 if (!ReadFull(connection->fd, &marker, sizeof(marker)))
370 62 break;
371 124 marker = ntohl(marker);
372 124 const uint32_t len = marker & ~kLastFragment;
373
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 124 times.
 124 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 124 const size_t offset = record.size();
379
1/2
✓ Branch 1 taken 124 times.
✗ Branch 2 not taken.
 124 record.resize(offset + len);
380
5/10
✓ Branch 0 taken 124 times.
✗ Branch 1 not taken.
✓ Branch 3 taken 124 times.
✗ Branch 4 not taken.
✓ Branch 7 taken 124 times.
✗ Branch 8 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 124 times.
✗ Branch 11 not taken.
✓ Branch 12 taken 124 times.
 124 if (len > 0 && !ReadFull(connection->fd, &record[offset], len))
381 ✗ break;
382
2/2
✓ Branch 0 taken 31 times.
✓ Branch 1 taken 93 times.
 124 if (!(marker & kLastFragment))
383 31 continue;
384
385 93 Job job;
386 93 job.connection = connection;
387 93 job.xid = 0;
388
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 if (record.size() >= 4) {
389 93 memcpy(&job.xid, record.data(), 4);
390 93 job.xid = ntohl(job.xid);
391 }
392
1/2
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
 93 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 93 times.
✗ Branch 3 not taken.
 93 std::lock_guard<std::mutex> const lock(connection->inflight_lock);
397
2/4
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 93 times.
 93 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 93 times.
✗ Branch 2 not taken.
 93 }
403 93 job.record.swap(record);
404 {
405
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 std::lock_guard<std::mutex> const lock(queue_lock_);
406
1/2
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
 93 queue_.push_back(Job());
407 93 queue_.back().connection.swap(job.connection);
408 93 queue_.back().xid = job.xid;
409 93 queue_.back().record.swap(job.record);
410 93 }
411 93 queue_cond_.notify_one();
412
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 }
413
414 62 shutdown(connection->fd, SHUT_RDWR);
415
1/2
✓ Branch 1 taken 62 times.
✗ Branch 2 not taken.
 62 std::lock_guard<std::mutex> const lock(connections_lock_);
416 62 active_readers_--;
417 62 readers_cond_.notify_all();
418 62 }
419
420
421 651 void RpcServer::WorkerLoop() {
422 while (true) {
423 651 Job job;
424 {
425
1/2
✓ Branch 1 taken 651 times.
✗ Branch 2 not taken.
 651 std::unique_lock<std::mutex> lock(queue_lock_);
426
4/4
✓ Branch 2 taken 744 times.
✓ Branch 3 taken 558 times.
✓ Branch 5 taken 93 times.
✓ Branch 6 taken 651 times.
 1953 queue_cond_.wait(lock, [this] { return stopping_ || !queue_.empty(); });
427
2/2
✓ Branch 1 taken 558 times.
✓ Branch 2 taken 93 times.
 651 if (queue_.empty())
428 1054 return;
429 93 job.connection.swap(queue_.front().connection);
430 93 job.xid = queue_.front().xid;
431 93 job.record.swap(queue_.front().record);
432 93 queue_.pop_front();
433
2/2
✓ Branch 1 taken 93 times.
✓ Branch 2 taken 558 times.
 651 }
434
435 93 std::string reply;
436
2/4
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 93 times.
✗ Branch 4 not taken.
 93 if (HandleRpcRecord(program_, job.record, &reply)) {
437
1/2
✓ Branch 1 taken 93 times.
✗ Branch 2 not taken.
 93 Send(job.connection, reply);
438
1/2
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
 93 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 93 times.
✗ Branch 3 not taken.
 93 std::lock_guard<std::mutex> const lock(job.connection->inflight_lock);
444
1/2
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
 93 job.connection->inflight.erase(job.xid);
445
2/2
✓ Branch 3 taken 93 times.
✓ Branch 4 taken 496 times.
 744 }
446 }
447
448
449 93 void RpcServer::Send(const ConnectionPtr &connection,
450 const std::string &reply) {
451 93 const uint32_t marker = htonl(kLastFragment
452 93 | static_cast<uint32_t>(reply.size()));
453
1/2
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
 93 std::lock_guard<std::mutex> const lock(connection->write_lock);
454
2/4
✓ Branch 2 taken 93 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 93 times.
 93 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 93 }
459
460 } // namespace nfs4
461