Line data Source code
1 : // Sep 14, 2016. "TERMS AND CONDITIONS" governing this file are in the README
2 : // or COPYING file. If you do not have such a file, one can be obtained by
3 : // contacting Ron or Fermi Lab in Batavia IL, 60510, phone: 630-840-3000.
4 : // $RCSfile: .emacs.gnu,v $
5 : // rev="$Revision: 1.30 $$Date: 2016/03/01 14:27:27 $";
6 :
7 : // C Includes
8 : #include <arpa/inet.h> // ntohl, ntohs
9 : #include <linux/sockios.h> // SIOCOUTQ
10 : #include <poll.h> // struct pollfd
11 : #include <sys/ioctl.h> // ioctl
12 : #include <sys/socket.h> // socket, socklen_t
13 : #include <sys/types.h> // size_t
14 : #include <sys/un.h> // sockaddr_un
15 : #include <cstdlib> // atoi, strtoul
16 :
17 : // C++ Includes
18 : #include <atomic>
19 : #include <fstream>
20 : #include <memory>
21 : #include <mutex>
22 : #include <stdexcept>
23 : #include <string>
24 :
25 : // product Includes
26 : #include "artdaq/DAQdata/Globals.hh"
27 : #define TRACE_NAME (app_name + "_TCPSocketTransfer").c_str()
28 :
29 : // artdaq Includes
30 : #include <iomanip>
31 : #include "artdaq-core/Data/Fragment.hh"
32 : #include "artdaq-core/Utilities/TimeUtils.hh"
33 : #include "artdaq/DAQdata/TCPConnect.hh"
34 : #include "artdaq/DAQdata/TCP_listen_fd.hh"
35 : #include "artdaq/TransferPlugins/TCPSocketTransfer.hh"
36 : #include "artdaq/TransferPlugins/detail/SRSockets.hh"
37 : #include "artdaq/TransferPlugins/detail/Timeout.hh"
38 :
39 : #define USE_SENDMSG 1
40 :
41 : std::atomic<int> artdaq::TCPSocketTransfer::listen_thread_refcount_(0);
42 : std::unique_ptr<boost::thread> artdaq::TCPSocketTransfer::listen_thread_ = nullptr;
43 : std::map<int, std::set<int>> artdaq::TCPSocketTransfer::connected_fds_ = std::map<int, std::set<int>>();
44 : std::map<int, std::set<int>> artdaq::TCPSocketTransfer::first_fragment_received_ = std::map<int, std::set<int>>();
45 : std::mutex artdaq::TCPSocketTransfer::listen_thread_mutex_;
46 : std::mutex artdaq::TCPSocketTransfer::fd_mutex_;
47 :
48 3 : artdaq::TCPSocketTransfer::
49 3 : TCPSocketTransfer(fhicl::ParameterSet const& pset, TransferInterface::Role role)
50 : : TransferInterface(pset, role)
51 3 : , send_fd_(-1)
52 6 : , port_(pset.get<int>("port", portMan->GetTCPSocketTransferPort(destination_rank())))
53 6 : , rcvbuf_(pset.get<size_t>("tcp_receive_buffer_size", 0))
54 6 : , sndbuf_(pset.get<size_t>("tcp_send_buffer_size", max_fragment_size_words_ * sizeof(artdaq::RawDataType) * buffer_count_))
55 6 : , send_retry_timeout_us_(pset.get<size_t>("send_retry_timeout_us", 1000000))
56 3 : , timeoutMessageArmed_(true)
57 6 : , receive_disconnected_wait_s_(pset.get<double>("receive_socket_disconnected_wait_s", 10.0))
58 6 : , receive_err_wait_us_(pset.get<size_t>("receive_socket_disconnected_wait_us", 10000))
59 3 : , receive_socket_has_been_connected_(false)
60 9 : , send_ack_diff_(0)
61 : {
62 6 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << " Constructor: pset=" << pset.to_string() << ", role=" << (role == TransferInterface::Role::kReceive ? "kReceive" : "kSend");
63 3 : connection_was_lost_ = false;
64 :
65 3 : if (role == TransferInterface::Role::kReceive)
66 : {
67 : // Wait for sender to connect...
68 4 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Listening for connections";
69 2 : start_listen_thread_();
70 4 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Done Listening";
71 : }
72 : else
73 : {
74 1 : hostMap_ = MakeHostMap(pset);
75 2 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Connecting to destination";
76 1 : connect_();
77 2 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Done Connecting";
78 : }
79 6 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "End of Constructor";
80 3 : }
81 :
82 5 : artdaq::TCPSocketTransfer::~TCPSocketTransfer() noexcept
83 : {
84 6 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Shutting down TCPSocketTransfer";
85 :
86 3 : if (role() == TransferInterface::Role::kSend)
87 : {
88 : // close all open connections (send stop_v0) first
89 1 : MessHead mh = {0, MessHead::stop_v0, htons(TransferInterface::source_rank()), {0}};
90 1 : if (send_fd_ != -1)
91 : {
92 : // should be blocking with modest timeo
93 1 : timeval tv = {0, 100000};
94 1 : socklen_t len = sizeof(tv);
95 1 : setsockopt(send_fd_, SOL_SOCKET, SO_SNDTIMEO, &tv, len);
96 1 : write(send_fd_, &mh, sizeof(mh));
97 : }
98 1 : close(send_fd_);
99 1 : send_fd_ = -1;
100 : }
101 : else
102 : {
103 2 : TCPSocketTransfer::flush_buffers();
104 : try
105 : {
106 2 : if (ack_listen_thread_ && ack_listen_thread_->joinable())
107 : {
108 0 : ack_listen_thread_->join();
109 : }
110 : }
111 0 : catch (...)
112 : {
113 : // IGNORED
114 0 : }
115 :
116 2 : std::lock_guard<std::mutex> lk(listen_thread_mutex_);
117 2 : listen_thread_refcount_--;
118 : try
119 : {
120 2 : if (listen_thread_refcount_ <= 0 && listen_thread_ && listen_thread_->joinable())
121 : {
122 2 : listen_thread_->join();
123 : }
124 : }
125 0 : catch (...)
126 : {
127 : // IGNORED
128 0 : }
129 2 : }
130 :
131 6 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "End of Destructor";
132 5 : }
133 :
134 12 : int artdaq::TCPSocketTransfer::receiveFragmentHeader(detail::RawFragmentHeader& header, size_t timeout_usec)
135 : {
136 24 : TLOG(TLVL_DEBUG + 34) << GetTraceName() << "receiveFragmentHeader BEGIN";
137 12 : int ret_rank = RECV_TIMEOUT;
138 :
139 : // Don't bomb out until received at least one connection...
140 12 : if (getConnectedFDCount_(source_rank()) == 0)
141 : { // what if just listen_fd???
142 : // if (receive_socket_has_been_connected_ && TimeUtils::GetElapsedTime(last_recv_time_) > receive_disconnected_wait_s_)
143 : // {
144 : // TLOG(TLVL_ERROR) << GetTraceName() << "receiveFragmentHeader: senders have been disconnected for "
145 : // << TimeUtils::GetElapsedTime(last_recv_time_) << " s (receive_socket_disconnected_wait_s = " << receive_disconnected_wait_s_ << " s). RETURNING DATA_END!";
146 : // return DATA_END;
147 : // }
148 : // if (++not_connected_count_ > receive_err_threshold_) { return DATA_END; }
149 0 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "Receive socket not connected, returning RECV_TIMEOUT";
150 0 : usleep(receive_err_wait_us_);
151 0 : return RECV_TIMEOUT;
152 : }
153 12 : receive_socket_has_been_connected_ = true;
154 12 : last_recv_time_ = std::chrono::steady_clock::now();
155 :
156 24 : TLOG(TLVL_DEBUG + 34) << GetTraceName() << "timeout_usec=" << timeout_usec;
157 : // void* buff=alloca(max_fragment_size_words_*8);
158 12 : size_t byte_cnt = 0;
159 : int sts;
160 12 : int offset = 0;
161 12 : SocketState state = SocketState::Metadata;
162 12 : int target_bytes = sizeof(MessHead);
163 12 : uint64_t start_time_us = TimeUtils::gettimeofday_us();
164 :
165 : // while (active_receive_fd_ != -1)
166 : //{
167 : // TLOG(TLVL_DEBUG + 33) << GetTraceName() << "Currently receiving from fd " << active_receive_fd_ << ", waiting!";
168 : // usleep(1000);
169 : // }
170 :
171 : uint8_t* buff;
172 :
173 : int timeout_ms;
174 12 : if (timeout_usec == 0)
175 : {
176 0 : timeout_ms = 0;
177 : }
178 : else
179 : {
180 12 : timeout_ms = (timeout_usec + 999) / 1000; // want at least 1 ms
181 : }
182 :
183 12 : bool done = false;
184 12 : bool noDataWarningSent = false;
185 12 : int loop_guard = 0;
186 :
187 16 : while (!done && getConnectedFDCount_(source_rank()) > 0)
188 : {
189 14 : if (getActiveFD_(source_rank()) == -1)
190 : {
191 12 : loop_guard = 0;
192 12 : size_t fd_count = 0;
193 12 : std::vector<pollfd> pollfds;
194 : {
195 12 : std::lock_guard<std::mutex> lk(fd_mutex_);
196 12 : fd_count = connected_fds_[source_rank()].size();
197 12 : bool first = false;
198 12 : if (first_fragment_received_[source_rank()].size() < connected_fds_[source_rank()].size())
199 : {
200 1 : fd_count -= first_fragment_received_[source_rank()].size();
201 1 : first = true;
202 : }
203 12 : pollfds.resize(fd_count);
204 12 : auto iter = connected_fds_[source_rank()].begin();
205 24 : for (size_t ii = 0; ii < fd_count; ++ii)
206 : {
207 12 : while (first && first_fragment_received_[source_rank()].count(*iter))
208 : {
209 0 : ++iter;
210 : }
211 12 : pollfds[ii].events = POLLIN | POLLPRI | POLLERR;
212 12 : pollfds[ii].fd = *iter;
213 12 : ++iter;
214 : }
215 12 : }
216 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "receiveFragment: Polling fd to see if there's data" ;
217 12 : int num_fds_ready = poll(&pollfds[0], fd_count, timeout_ms);
218 12 : if (num_fds_ready <= 0)
219 : {
220 18 : TLOG(TLVL_DEBUG + 34) << GetTraceName() << "No data on receive socket, returning RECV_TIMEOUT";
221 9 : return RECV_TIMEOUT;
222 : }
223 :
224 3 : size_t index = 0;
225 3 : if (getLastActiveFD_(source_rank()) != -1)
226 : {
227 2 : for (auto& pollfd : pollfds)
228 : {
229 2 : index++;
230 2 : if (pollfd.fd == getLastActiveFD_(source_rank()))
231 : {
232 2 : break;
233 : }
234 : }
235 : }
236 :
237 3 : int active_index = -1;
238 3 : int16_t anomolous_events = 0;
239 3 : for (size_t ii = index; ii < index + pollfds.size(); ++ii)
240 : {
241 3 : auto pollfd_index = (ii + index) % pollfds.size();
242 3 : setActiveFD_(source_rank(), pollfds[pollfd_index].fd);
243 3 : if ((pollfds[pollfd_index].revents & (POLLIN | POLLPRI)) != 0)
244 : {
245 3 : active_index = pollfd_index;
246 3 : break;
247 : }
248 0 : if ((pollfds[pollfd_index].revents & (POLLHUP | POLLERR)) != 0)
249 : {
250 0 : disconnect_receive_socket_("Poll returned POLLHUP or POLLERR, indicating problems with the sender.");
251 0 : continue;
252 : }
253 0 : else if ((pollfds[pollfd_index].revents & (POLLNVAL)) != 0)
254 : {
255 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "FD is closed, most likely because the peer went away. Removing from fd list.";
256 0 : disconnect_receive_socket_("FD is closed, most likely because the peer went away.");
257 0 : continue;
258 0 : }
259 0 : else if (pollfds[pollfd_index].revents != 0)
260 : {
261 0 : anomolous_events |= pollfds[pollfd_index].revents;
262 : }
263 : }
264 :
265 3 : if (active_index == -1)
266 : {
267 0 : if (anomolous_events != 0)
268 : {
269 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Wrong event received from a pollfd. Mask: " << static_cast<int>(anomolous_events);
270 : }
271 0 : setActiveFD_(source_rank(), -1);
272 0 : continue;
273 0 : }
274 :
275 3 : if (!done && timeout_usec > 0)
276 : {
277 : // calc next timeout_ms (unless timed out)
278 3 : size_t delta_us = TimeUtils::gettimeofday_us() - start_time_us;
279 3 : if (delta_us > timeout_usec)
280 : {
281 0 : return RECV_TIMEOUT;
282 : }
283 3 : timeout_ms = ((timeout_usec - delta_us) + 999) / 1000; // want at least 1 ms
284 : }
285 12 : }
286 5 : if (loop_guard > 10) { usleep(1000); }
287 5 : if (++loop_guard > 10010)
288 : {
289 0 : TLOG(TLVL_WARNING) << GetTraceName() << "loop guard triggered, returning RECV_TIMEOUT";
290 0 : usleep(receive_err_wait_us_);
291 0 : setActiveFD_(source_rank(), -1);
292 0 : return RECV_TIMEOUT;
293 : }
294 :
295 5 : if (state == SocketState::Metadata)
296 : {
297 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Reading Message Header" ;
298 3 : buff = &(mha[offset]); // NOLINT(cppcoreguidelines-pro-bounds-constant-array-index)
299 3 : byte_cnt = sizeof(MessHead) - offset;
300 : }
301 : else
302 : {
303 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Reading data" ;
304 2 : buff = reinterpret_cast<uint8_t*>(&header) + offset; // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast,cppcoreguidelines-pro-bounds-pointer-arithmetic)
305 2 : byte_cnt = target_bytes - offset;
306 : }
307 : // if (byte_cnt > sizeof(MessHead))
308 : // {
309 : // TLOG(TLVL_ERROR) << "Invalid byte count for read (count=" << byte_cnt
310 : // << ",offset=" << offset << ",mh.byte_count=" << mh.byte_count
311 : // << "), skipping read and returning RECV_TIMEOUT";
312 : // return RECV_TIMEOUT;
313 : // }
314 :
315 5 : auto fd = getActiveFD_(source_rank());
316 5 : if (byte_cnt > 0)
317 : {
318 10 : TLOG(TLVL_DEBUG + 35) << GetTraceName() << "Reading " << byte_cnt << " bytes from socket " << fd;
319 5 : sts = read(fd, buff, byte_cnt);
320 10 : TLOG(TLVL_DEBUG + 35) << GetTraceName() << "Done with read sts=" << sts;
321 : }
322 5 : if (sts > 0)
323 : {
324 5 : loop_guard = 0;
325 5 : last_recv_time_ = std::chrono::steady_clock::now();
326 : }
327 :
328 10 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "state=" << static_cast<int>(state) << " read=" << sts;
329 5 : if (sts < 0 && errno != EAGAIN)
330 : {
331 0 : TLOG(TLVL_WARNING) << GetTraceName() << "Error on receive, closing socket " << fd
332 0 : << " (errno=" << errno << ": " << strerror(errno) << ")";
333 0 : disconnect_receive_socket_("Error on receive");
334 0 : }
335 5 : else if (sts <= 0 && errno == EAGAIN)
336 : {
337 0 : if (!noDataWarningSent)
338 : {
339 0 : TLOG(TLVL_WARNING) << GetTraceName() << "No data received, is the sender still sending?!?";
340 0 : noDataWarningSent = true;
341 : }
342 0 : if (TimeUtils::GetElapsedTime(last_recv_time_) > receive_disconnected_wait_s_)
343 : {
344 0 : TLOG(TLVL_ERROR) << GetTraceName() << "No data received within timeout, aborting!";
345 0 : return RECV_TIMEOUT;
346 : }
347 0 : }
348 : else
349 : {
350 : // see if we're done (with this state)
351 10 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "Checking for complete sts=" << sts << ", offset=" << offset << ", target_bytes=" << target_bytes;
352 5 : sts = offset += sts;
353 5 : if (sts >= target_bytes)
354 : {
355 10 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "Target read bytes reached. Changing state";
356 5 : offset = 0;
357 5 : if (state == SocketState::Metadata)
358 : {
359 3 : state = SocketState::Data;
360 3 : mh.byte_count = ntohl(mh.byte_count);
361 3 : mh.source_id = ntohs(mh.source_id);
362 3 : target_bytes = mh.byte_count;
363 6 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "Expected header size = " << target_bytes << ", sizeof(RawFragmentHeader) = " << sizeof(artdaq::detail::RawFragmentHeader);
364 : // assert(target_bytes == sizeof(artdaq::detail::RawFragmentHeader) || target_bytes == 0);
365 :
366 3 : if (mh.message_type == MessHead::stop_v0)
367 : {
368 2 : disconnect_receive_socket_("Stop Message received.");
369 : }
370 2 : else if (mh.message_type == MessHead::data_v0 || mh.message_type == MessHead::data_more_v0)
371 : {
372 0 : TLOG(TLVL_WARNING) << GetTraceName() << "Message header indicates that Fragment data follows when I was expecting a Fragment header!";
373 0 : disconnect_receive_socket_("Desync detected");
374 : }
375 :
376 3 : if (target_bytes == 0)
377 : {
378 : // Probably a stop_v0, return timeout so we can try again.
379 1 : return RECV_TIMEOUT;
380 : }
381 : }
382 : else
383 : {
384 2 : ret_rank = source_rank();
385 4 : TLOG(TLVL_DEBUG + 35) << GetTraceName() << "done sts=" << sts << " src=" << ret_rank;
386 4 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "Done receiving fragment header. Moving into output.";
387 :
388 2 : done = true; // no more polls
389 : // break; // no more read of ready fds
390 :
391 : {
392 2 : std::lock_guard<std::mutex> lk(fd_mutex_);
393 2 : first_fragment_received_[source_rank()].insert(fd);
394 2 : }
395 : }
396 : }
397 : }
398 :
399 : } // while(!done)...poll
400 :
401 4 : TLOG(TLVL_DEBUG + 34) << GetTraceName() << "Returning " << ret_rank;
402 2 : return ret_rank;
403 : }
404 :
405 1 : void artdaq::TCPSocketTransfer::disconnect_receive_socket_(const std::string& msg)
406 : {
407 1 : std::lock_guard<std::mutex> lk(fd_mutex_);
408 1 : auto fd = active_receive_fds_[source_rank()];
409 4 : TLOG(TLVL_WARNING) << GetTraceName() << "disconnect_receive_socket_: " << msg << " Closing socket " << fd << " for rank " << source_rank();
410 1 : close(fd);
411 1 : if (connected_fds_.count(source_rank()) != 0u)
412 : {
413 1 : connected_fds_[source_rank()].erase(fd);
414 : }
415 1 : if (first_fragment_received_.count(source_rank()) != 0)
416 : {
417 1 : first_fragment_received_[source_rank()].erase(fd);
418 : }
419 1 : active_receive_fds_[source_rank()] = -1;
420 2 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "disconnect_receive_socket_: There are now " << connected_fds_[source_rank()].size() << " active senders.";
421 1 : }
422 :
423 2 : int artdaq::TCPSocketTransfer::receiveFragmentData(RawDataType* destination, size_t /*wordCount*/)
424 : {
425 4 : TLOG(TLVL_DEBUG + 39) << GetTraceName() << "receiveFragmentData: BEGIN";
426 2 : int ret_rank = RECV_TIMEOUT;
427 2 : if (getActiveFD_(source_rank()) == -1)
428 : { // what if just listen_fd???
429 0 : TLOG(TLVL_ERROR) << GetTraceName() << "receiveFragmentData: Receive socket not connected, returning RECV_TIMEOUT (Will result in \"Unexpected return code error\")";
430 0 : return RECV_TIMEOUT;
431 : }
432 :
433 : // void* buff=alloca(max_fragment_size_words_*8);
434 : uint8_t* buff;
435 2 : size_t byte_cnt = 0;
436 : int sts;
437 2 : size_t offset = 0;
438 2 : SocketState state = SocketState::Metadata;
439 2 : size_t target_bytes = sizeof(MessHead);
440 :
441 : pollfd pollfd_s;
442 2 : pollfd_s.events = POLLIN | POLLPRI | POLLERR;
443 2 : pollfd_s.fd = getActiveFD_(source_rank());
444 :
445 2 : int loop_guard = 0;
446 2 : bool done = false;
447 2 : bool noDataWarningSent = false;
448 2 : last_recv_time_ = std::chrono::steady_clock::now();
449 6 : while (!done)
450 : {
451 8 : TLOG(TLVL_DEBUG + 33) << GetTraceName() << "receiveFragmentData: Polling fd to see if there's data";
452 4 : int num_fds_ready = poll(&pollfd_s, 1, 1000);
453 8 : TLOG(TLVL_DEBUG + 33) << GetTraceName() << "receiveFragmentData: Polled fd to see if there's data"
454 4 : << ", num_fds_ready = " << num_fds_ready;
455 4 : if (num_fds_ready <= 0)
456 : {
457 0 : if (num_fds_ready == 0)
458 : {
459 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: No data from " << source_rank() << " in " << TimeUtils::GetElapsedTimeMilliseconds(last_recv_time_) << " ms!"
460 0 : << " State = " << (state == SocketState::Metadata ? "Metadata" : "Data") << ", recvd/total=" << offset << "/" << target_bytes << " (delta=" << target_bytes - offset << ")";
461 :
462 0 : if (TimeUtils::GetElapsedTime(last_recv_time_) > receive_disconnected_wait_s_)
463 : {
464 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: No data received within timeout (" << TimeUtils::GetElapsedTime(last_recv_time_) << " / " << receive_disconnected_wait_s_ << " ), returning RECV_TIMEOUT";
465 0 : disconnect_receive_socket_("No data on this socket within timeout");
466 0 : return RECV_TIMEOUT;
467 : }
468 0 : continue;
469 0 : }
470 :
471 0 : TLOG(TLVL_ERROR) << "Error in poll: errno=" << errno;
472 0 : break;
473 : }
474 :
475 4 : last_recv_time_ = std::chrono::steady_clock::now();
476 :
477 4 : if ((pollfd_s.revents & (POLLIN | POLLPRI)) != 0)
478 : {
479 : // Expected, don't have to check revents any further
480 : }
481 0 : else if ((pollfd_s.revents & (POLLNVAL)) != 0)
482 : {
483 0 : disconnect_receive_socket_("FD is closed, most likely because the peer went away.");
484 0 : break;
485 : }
486 0 : else if ((pollfd_s.revents & (POLLHUP | POLLERR)) != 0)
487 : {
488 0 : disconnect_receive_socket_("Poll returned POLLHUP or POLLERR, indicating problems with the sender.");
489 0 : break;
490 : }
491 : else
492 : {
493 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: Wrong event received from pollfd: " << pollfd_s.revents;
494 0 : disconnect_receive_socket_("Wrong event received from pollfd.");
495 0 : break;
496 : }
497 :
498 4 : if (state == SocketState::Metadata)
499 : {
500 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "receiveFragmentData: Reading Message Header" ;
501 2 : buff = &(mha[offset]); // NOLINT(cppcoreguidelines-pro-bounds-constant-array-index)
502 2 : byte_cnt = sizeof(MessHead) - offset;
503 : }
504 : else
505 : {
506 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "receiveFragmentData: Reading data" ;
507 2 : buff = reinterpret_cast<uint8_t*>(destination) + offset; // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic,cppcoreguidelines-pro-type-reinterpret-cast)
508 2 : byte_cnt = mh.byte_count - offset;
509 : }
510 :
511 8 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "receiveFragmentData: Reading " << byte_cnt << " bytes from socket into " << static_cast<void*>(buff);
512 4 : sts = read(getActiveFD_(source_rank()), buff, byte_cnt);
513 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "receiveFragmentData: Done with read" ;
514 :
515 8 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "recvFragment state=" << static_cast<int>(state) << " read=" << sts;
516 :
517 4 : if (sts == 0 || (sts < 0 && errno == EAGAIN))
518 : {
519 0 : sts = 0; // Treat EAGAIN as receiving no data
520 0 : if (loop_guard > 10) { usleep(1000); }
521 0 : if (++loop_guard > 10010)
522 : {
523 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: loop guard triggered, returning RECV_TIMEOUT";
524 0 : setActiveFD_(source_rank(), -1);
525 0 : return RECV_TIMEOUT;
526 : }
527 0 : }
528 4 : else if (sts > 0)
529 : {
530 4 : loop_guard = 0;
531 4 : last_recv_time_ = std::chrono::steady_clock::now();
532 : }
533 :
534 4 : if (sts < 0)
535 : {
536 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: Error on receive, closing socket"
537 0 : << " (errno=" << errno << ": " << strerror(errno) << ")";
538 0 : disconnect_receive_socket_("Error on receive");
539 : }
540 4 : else if (sts == 0)
541 : {
542 0 : if (!noDataWarningSent)
543 : {
544 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: No data received, is the sender still sending?!?";
545 0 : noDataWarningSent = true;
546 : }
547 0 : if (TimeUtils::GetElapsedTime(last_recv_time_) > receive_disconnected_wait_s_)
548 : {
549 0 : TLOG(TLVL_ERROR) << GetTraceName() << "receiveFragmentData: No data received within timeout, aborting!";
550 0 : return RECV_TIMEOUT;
551 : }
552 : }
553 : else
554 : {
555 : // see if we're done (with this state)
556 4 : sts = offset += sts;
557 4 : if (static_cast<size_t>(sts) >= target_bytes)
558 : {
559 8 : TLOG(TLVL_DEBUG + 42) << GetTraceName() << "receiveFragmentData: Target read bytes reached. Changing state";
560 4 : offset = 0;
561 4 : if (state == SocketState::Metadata)
562 : {
563 2 : state = SocketState::Data;
564 2 : mh.byte_count = ntohl(mh.byte_count);
565 2 : mh.source_id = ntohs(mh.source_id);
566 2 : target_bytes = mh.byte_count;
567 :
568 2 : if (mh.message_type == MessHead::header_v0)
569 : {
570 0 : TLOG(TLVL_WARNING) << GetTraceName() << "receiveFragmentData: Message header indicates that a Fragment header follows when I was expecting Fragment data!";
571 0 : disconnect_receive_socket_("Desync detected");
572 : }
573 : }
574 : else
575 : {
576 2 : ret_rank = source_rank();
577 4 : TLOG(TLVL_DEBUG + 41) << GetTraceName() << "receiveFragmentData done sts=" << sts << " src=" << ret_rank;
578 4 : TLOG(TLVL_DEBUG + 39) << GetTraceName() << "receiveFragmentData: Done receiving fragment. Moving into output.";
579 :
580 : #if USE_ACKS
581 : send_ack_(active_receive_fd_);
582 : #endif
583 :
584 2 : done = true; // no more polls
585 : // break; // no more read of ready fds
586 : }
587 : }
588 : }
589 :
590 : // Check if we were asked to do a 0-size receive
591 4 : if (target_bytes == 0 && state == SocketState::Data)
592 : {
593 0 : ret_rank = source_rank();
594 0 : TLOG(TLVL_DEBUG + 41) << GetTraceName() << "receiveFragmentData done sts=" << sts << " src=" << ret_rank;
595 0 : TLOG(TLVL_DEBUG + 39) << GetTraceName() << "receiveFragmentData: Done receiving fragment. Moving into output.";
596 :
597 : #if USE_ACKS
598 : send_ack_(active_receive_fd_);
599 : #endif
600 :
601 0 : done = true; // no more polls
602 : }
603 :
604 : } // while(!done)...poll
605 :
606 2 : setLastActiveFD_(source_rank(), getActiveFD_(source_rank()));
607 2 : setActiveFD_(source_rank(), -1);
608 :
609 4 : TLOG(TLVL_DEBUG + 39) << GetTraceName() << "receiveFragmentData: Returning rank " << ret_rank;
610 2 : return ret_rank;
611 : }
612 :
613 2 : bool artdaq::TCPSocketTransfer::isRunning()
614 : {
615 2 : switch (role())
616 : {
617 0 : case TransferInterface::Role::kSend:
618 0 : return send_fd_ != -1;
619 2 : case TransferInterface::Role::kReceive:
620 2 : auto count = getConnectedFDCount_(source_rank());
621 4 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "isRunning: There are " << count << " fds connected.";
622 2 : return count > 0;
623 : }
624 0 : return false;
625 : }
626 :
627 3 : void artdaq::TCPSocketTransfer::flush_buffers()
628 : {
629 3 : std::set<int> fds;
630 3 : auto rank = TransferInterface::source_rank();
631 : {
632 3 : std::lock_guard<std::mutex> lk(fd_mutex_);
633 3 : if (connected_fds_.count(rank) != 0)
634 : {
635 1 : fds = connected_fds_[rank];
636 1 : connected_fds_.erase(rank);
637 : }
638 3 : if (first_fragment_received_.count(rank))
639 : {
640 1 : first_fragment_received_.erase(rank);
641 : }
642 3 : }
643 :
644 : char discard_buf[0x1000];
645 3 : for (auto& fd : fds)
646 : {
647 0 : TLOG(TLVL_INFO) << GetTraceName() << "flush_buffers: Checking for data in socket " << fd << " for rank " << rank;
648 0 : size_t bytes_read = 0;
649 0 : while (int sts = static_cast<int>(read(fd, discard_buf, sizeof(discard_buf)) > 0))
650 : {
651 0 : bytes_read += sts;
652 0 : }
653 0 : if (bytes_read > 0)
654 : {
655 0 : TLOG(TLVL_WARNING) << GetTraceName() << "flush_buffers: Flushed " << bytes_read << " bytes from socket " << fd << " for rank " << rank;
656 : }
657 0 : TLOG(TLVL_INFO) << GetTraceName() << "flush_buffers: Closing socket " << fd << " for rank " << rank;
658 0 : close(fd);
659 : }
660 3 : active_receive_fds_[rank] = -1;
661 3 : last_active_receive_fds_[rank] = -1;
662 3 : }
663 :
664 : // Send the given Fragment. Return the rank of the destination to which
665 : // the Fragment was sent OR -1 if to none.
666 2 : artdaq::TransferInterface::CopyStatus artdaq::TCPSocketTransfer::sendFragment_(Fragment&& frag, size_t send_timeout_usec)
667 : {
668 4 : TLOG(TLVL_DEBUG + 42) << GetTraceName() << "sendFragment begin send of fragment with sequenceID=" << frag.sequenceID();
669 2 : artdaq::Fragment grab_ownership_frag = std::move(frag);
670 :
671 2 : reconnect_();
672 2 : if (send_fd_ == -1 && connection_was_lost_)
673 : {
674 0 : TLOG(TLVL_INFO) << GetTraceName() << "reconnection attempt failed, returning quickly.";
675 0 : return TransferInterface::CopyStatus::kErrorNotRequiringException;
676 : }
677 :
678 : // Report the kernel TCP send queue occupancy (bytes accepted by the socket but not yet ACKed by the receiver), at most once per second
679 4 : if (metricMan && send_fd_ != -1 &&
680 4 : std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::steady_clock::now() - last_outq_metric_time_).count() >= 1000)
681 : {
682 1 : last_outq_metric_time_ = std::chrono::steady_clock::now();
683 1 : int outq_bytes = 0;
684 1 : if (ioctl(send_fd_, SIOCOUTQ, &outq_bytes) == 0)
685 : {
686 5 : metricMan->sendMetric("TCP Send Queue Bytes to Rank " + std::to_string(destination_rank()), static_cast<unsigned long>(outq_bytes), "B", 3, MetricMode::LastPoint);
687 : }
688 : }
689 :
690 : // Send Fragment Header
691 :
692 : #if USE_ACKS
693 : // Wait for fragments to be received
694 : while (static_cast<size_t>(send_ack_diff_) > buffer_count_) usleep(10000);
695 : #endif
696 :
697 2 : iovec iov = {static_cast<void*>(grab_ownership_frag.headerAddress()),
698 2 : detail::RawFragmentHeader::num_words() * sizeof(RawDataType)};
699 :
700 2 : auto sts = sendData_(&iov, 1, send_retry_timeout_us_, true);
701 2 : auto start_time = std::chrono::steady_clock::now();
702 : // If it takes more than 10 seconds to write a Fragment header, give up
703 2 : while (sts == CopyStatus::kTimeout && (send_timeout_usec == 0 || TimeUtils::GetElapsedTimeMicroseconds(start_time) < send_timeout_usec) && TimeUtils::GetElapsedTimeMicroseconds(start_time) < 10000000)
704 : {
705 0 : TLOG(TLVL_DEBUG + 43) << GetTraceName() << "sendFragment: Timeout sending fragment";
706 0 : sts = sendData_(&iov, 1, send_retry_timeout_us_, true);
707 0 : usleep(1000);
708 : }
709 2 : if (sts != CopyStatus::kSuccess)
710 : {
711 0 : return sts;
712 : }
713 :
714 : // Send Fragment Data
715 :
716 2 : iov = {static_cast<void*>(grab_ownership_frag.headerAddress() + detail::RawFragmentHeader::num_words()), // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
717 2 : grab_ownership_frag.sizeBytes() - detail::RawFragmentHeader::num_words() * sizeof(RawDataType)};
718 2 : sts = sendData_(&iov, 1, send_retry_timeout_us_);
719 2 : start_time = std::chrono::steady_clock::now();
720 2 : while (sts == CopyStatus::kTimeout && (send_timeout_usec == 0 || TimeUtils::GetElapsedTimeMicroseconds(start_time) < send_timeout_usec) && TimeUtils::GetElapsedTimeMicroseconds(start_time) < 10000000)
721 : {
722 0 : TLOG(TLVL_DEBUG + 43) << GetTraceName() << "sendFragment: Timeout sending fragment";
723 0 : sts = sendData_(&iov, 1, send_retry_timeout_us_);
724 0 : usleep(1000);
725 : }
726 :
727 : #if USE_ACKS
728 : send_ack_diff_++;
729 : #endif
730 :
731 4 : TLOG(TLVL_DEBUG + 42) << GetTraceName() << "sendFragment returning " << CopyStatusToString(sts);
732 2 : return sts;
733 2 : }
734 :
735 0 : artdaq::TransferInterface::CopyStatus artdaq::TCPSocketTransfer::sendData_(const void* buf, size_t bytes, size_t send_timeout_usec, bool isHeader)
736 : {
737 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "sendData_ Converting buf to iovec";
738 0 : iovec iov = {const_cast<void*>(buf), bytes}; // NOLINT(cppcoreguidelines-pro-type-const-cast)
739 0 : return sendData_(&iov, 1, send_timeout_usec, isHeader);
740 : }
741 :
742 4 : artdaq::TransferInterface::CopyStatus artdaq::TCPSocketTransfer::sendData_(const struct iovec* iov, int iovcnt, size_t send_timeout_usec, bool isHeader)
743 : {
744 : // check all connected??? -- currently just check fd!=-1
745 4 : if (send_fd_ == -1)
746 : {
747 0 : if (timeoutMessageArmed_)
748 : {
749 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "sendData_: Send fd is not open. Returning kTimeout";
750 0 : timeoutMessageArmed_ = false;
751 : }
752 0 : return CopyStatus::kTimeout;
753 : }
754 4 : timeoutMessageArmed_ = true;
755 8 : TLOG(TLVL_DEBUG + 44) << GetTraceName() << "send_timeout_usec is " << send_timeout_usec << ", currently unused.";
756 :
757 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "sendData_: Determining write size" ;
758 4 : uint32_t total_to_write_bytes = 0;
759 8 : std::vector<iovec> iov_in(iovcnt + 1); // need contiguous (for the unlike case that only partial MH
760 4 : std::vector<iovec> iovv(iovcnt + 2); // 1 more for mh and another one for any partial
761 : int ii;
762 8 : for (ii = 0; ii < iovcnt; ++ii)
763 : {
764 4 : iov_in[ii + 1] = iov[ii]; // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
765 4 : total_to_write_bytes += iov[ii].iov_len; // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
766 : }
767 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << "sendData_: Constructing Message Header" ;
768 4 : MessHead mh = {0, isHeader ? MessHead::header_v0 : MessHead::data_v0, htons(source_rank()), {htonl(total_to_write_bytes)}};
769 4 : iov_in[0].iov_base = &mh;
770 4 : iov_in[0].iov_len = sizeof(mh);
771 4 : total_to_write_bytes += sizeof(mh);
772 :
773 4 : ssize_t sts = 0;
774 4 : ssize_t total_written_bytes = 0;
775 4 : ssize_t per_write_max_bytes = (32 * 1024);
776 :
777 4 : size_t in_iov_idx = 0; // only increment this when we know the associated data has been xferred
778 4 : size_t out_iov_idx = 0;
779 4 : ssize_t this_write_bytes = 0;
780 :
781 : do
782 : {
783 : // The first out_iov may be set at the end of the previous loop.
784 : // iov looping from below (b/c of the latter, we need to check this_write_bytes)
785 4 : for (;
786 12 : (in_iov_idx + out_iov_idx) < iov_in.size() && this_write_bytes < per_write_max_bytes;
787 : ++out_iov_idx)
788 : {
789 8 : this_write_bytes += iov_in[in_iov_idx + out_iov_idx].iov_len;
790 8 : iovv[out_iov_idx] = iov_in[in_iov_idx + out_iov_idx];
791 : }
792 4 : if (this_write_bytes > per_write_max_bytes)
793 : {
794 0 : iovv[out_iov_idx - 1].iov_len -= this_write_bytes - per_write_max_bytes;
795 0 : this_write_bytes = per_write_max_bytes;
796 : }
797 :
798 : // need to do blocking algorithm -- including throttled block notifications
799 4 : do_again:
800 : #ifndef __OPTIMIZE__ // This can be an expensive TRACE call (even if disabled) due to multiplicity of calls
801 8 : TLOG(TLVL_DEBUG + 44) << GetTraceName() << "sendFragment b4 writev " << std::setw(7) << total_written_bytes << " total_written_bytes send_fd_=" << send_fd_ << " in_idx=" << in_iov_idx
802 4 : << " iovcnt=" << out_iov_idx << " 1st.len=" << iovv[0].iov_len;
803 : #endif
804 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << " calling writev" ;
805 : #if USE_SENDMSG
806 : msghdr msg;
807 4 : memset(&msg, 0, sizeof(msghdr));
808 4 : msg.msg_iov = &(iovv[0]);
809 4 : msg.msg_iovlen = out_iov_idx; // at this point out_iov_idx is really the count (not an idx per se)
810 4 : sts = sendmsg(send_fd_, &msg, MSG_NOSIGNAL | (blocking != 0u ? 0 : MSG_DONTWAIT));
811 : #else
812 : sts = writev(send_fd_, &(iovv[0]), out_iov_idx); // SIGPIPE may occur -- need signal handler or mask/ignore
813 : #endif
814 : // TLOG(TLVL_DEBUG + 32) << GetTraceName() << " done with writev" ;
815 :
816 4 : if (sts == -1)
817 : {
818 0 : if (errno == EAGAIN /* same as EWOULDBLOCK */)
819 : {
820 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "sendFragment EWOULDBLOCK";
821 0 : blocking = 1u;
822 :
823 0 : fcntl(send_fd_, F_SETFL, 0); // clear O_NONBLOCK
824 :
825 : // NOTE: YES -- could drop here
826 0 : goto do_again;
827 : }
828 0 : TLOG(TLVL_WARNING) << GetTraceName() << "sendFragment_: WRITE ERROR " << errno << ": " << strerror(errno);
829 0 : connect_state = 0; // any write error closes
830 0 : close(send_fd_);
831 0 : send_fd_ = -1;
832 0 : connection_was_lost_ = true;
833 0 : return TransferInterface::CopyStatus::kErrorNotRequiringException;
834 : }
835 4 : if (sts != this_write_bytes)
836 : {
837 : // we'll loop around -- with
838 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "sendFragment writev sts(" << sts << ")!=requested_send_bytes(" << this_write_bytes << ")";
839 0 : total_written_bytes += sts; // add sts to total_written_bytes now as sts is adjusted next
840 : // find which iovs are done
841 0 : for (ii = 0; static_cast<size_t>(sts) >= iovv[ii].iov_len; ++ii)
842 : {
843 0 : sts -= iovv[ii].iov_len;
844 : }
845 0 : in_iov_idx += ii; // done with these in_iovs
846 0 : iovv[ii].iov_len -= sts; // adjust partial iov
847 0 : iovv[ii].iov_base = static_cast<uint8_t*>(iovv[ii].iov_base) + sts; // adjust partial iov // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
848 :
849 : // add more to get up to per_write_max_bytes
850 0 : out_iov_idx = 0;
851 0 : if (ii != 0)
852 : {
853 0 : iovv[out_iov_idx] = iovv[ii];
854 : }
855 : // starting over
856 0 : this_write_bytes = iovv[out_iov_idx].iov_len;
857 : // add any left over from appropriate in_iov_idx --
858 : // i.e. match this out_iov with the in_iov that was used to
859 : // initialize it; see how close the out base+len is to in base+len
860 : // check !>per_write_max_bytes
861 0 : auto additional = (reinterpret_cast<uintptr_t>(iov_in[in_iov_idx].iov_base) + iov_in[in_iov_idx].iov_len) - (reinterpret_cast<uintptr_t>(iovv[out_iov_idx].iov_base) + iovv[out_iov_idx].iov_len); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
862 0 : if (additional != 0u)
863 : {
864 0 : iovv[out_iov_idx].iov_len += additional;
865 0 : this_write_bytes += additional;
866 0 : if (this_write_bytes > per_write_max_bytes)
867 : {
868 0 : iovv[out_iov_idx].iov_len -= this_write_bytes - per_write_max_bytes;
869 0 : this_write_bytes = per_write_max_bytes;
870 : }
871 : }
872 0 : ++out_iov_idx; // done with
873 0 : TLOG(TLVL_DEBUG + 33) << GetTraceName() << "sendFragment writev sts!=: this_write_bytes=" << this_write_bytes
874 0 : << " out_iov_idx=" << out_iov_idx
875 0 : << " additional=" << additional
876 0 : << " ii=" << ii;
877 : }
878 : else
879 : {
880 : #ifndef __OPTIMIZE__ // This can be an expensive TRACE call (even if disabled) due to multiplicity of calls
881 8 : TLOG(TLVL_DEBUG + 33) << GetTraceName() << "sendFragment writev sts(" << sts << ")==requested_send_bytes(" << this_write_bytes << ")";
882 : #endif
883 4 : total_written_bytes += sts;
884 4 : --out_iov_idx; // make it the index of the last iovv
885 4 : iovv[out_iov_idx].iov_base = static_cast<uint8_t*>(iovv[out_iov_idx].iov_base) + iovv[out_iov_idx].iov_len; // NOLINT(cppcoreguidelines-pro-bounds-pointer-arithmetic)
886 4 : iovv[out_iov_idx].iov_len = 0;
887 4 : in_iov_idx += out_iov_idx; // at least this many complete (one more if "last iovv" is complete
888 4 : this_write_bytes = 0;
889 : // need to check last iovv against appropriate iov_in
890 4 : auto additional = (reinterpret_cast<uintptr_t>(iov_in[in_iov_idx].iov_base) + iov_in[in_iov_idx].iov_len) - (reinterpret_cast<uintptr_t>(iovv[out_iov_idx].iov_base) + iovv[out_iov_idx].iov_len); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
891 4 : if (additional != 0u)
892 : {
893 0 : iovv[out_iov_idx].iov_len += additional;
894 0 : this_write_bytes += additional;
895 0 : if (this_write_bytes > per_write_max_bytes)
896 : {
897 0 : iovv[out_iov_idx].iov_len -= this_write_bytes - per_write_max_bytes;
898 0 : this_write_bytes = per_write_max_bytes;
899 : }
900 0 : if (out_iov_idx != 0)
901 : {
902 0 : iovv[0] = iovv[out_iov_idx];
903 : }
904 0 : out_iov_idx = 1;
905 : }
906 : else
907 : {
908 4 : ++in_iov_idx;
909 4 : out_iov_idx = 0;
910 : }
911 : }
912 4 : } while (total_written_bytes < total_to_write_bytes);
913 4 : if (total_written_bytes > total_to_write_bytes)
914 : {
915 0 : TLOG(TLVL_ERROR) << GetTraceName() << "sendFragment program error: too many bytes transferred";
916 : }
917 :
918 4 : if (blocking != 0u)
919 : {
920 0 : blocking = 0u;
921 0 : fcntl(send_fd_, F_SETFL, O_NONBLOCK); // set O_NONBLOCK
922 : }
923 4 : sts = total_written_bytes - sizeof(MessHead);
924 :
925 8 : TLOG(TLVL_DEBUG + 44) << GetTraceName() << "sendFragment sts=" << sts;
926 4 : return TransferInterface::CopyStatus::kSuccess;
927 4 : }
928 :
929 1 : void artdaq::TCPSocketTransfer::connect_()
930 : {
931 1 : auto start_time = std::chrono::steady_clock::now();
932 :
933 : // Retry a few times if we can't connect
934 2 : while (send_fd_ == -1 && TimeUtils::GetElapsedTimeMicroseconds(start_time) < send_retry_timeout_us_ * 10)
935 : {
936 2 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Connecting sender socket";
937 1 : int sndbuf_bytes = static_cast<int>(sndbuf_);
938 1 : if (sndbuf_ > INT_MAX)
939 : {
940 0 : sndbuf_bytes = INT_MAX;
941 0 : TLOG(TLVL_WARNING) << "Requested SNDBUF " << sndbuf_ << " too large, setting to INT_MAX: " << INT_MAX;
942 : }
943 2 : TLOG(TLVL_DEBUG + 32) << "Requested SNDBUF is " << sndbuf_bytes;
944 :
945 1 : send_fd_ = TCPConnect(hostMap_[destination_rank()].c_str(), port_, O_NONBLOCK, sndbuf_bytes);
946 1 : if (send_fd_ == -1)
947 : {
948 0 : if (connection_was_lost_) { break; }
949 :
950 0 : usleep(send_retry_timeout_us_);
951 : }
952 : }
953 1 : connect_state = 0;
954 1 : blocking = 0;
955 2 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "connect_ " + hostMap_[destination_rank()] + ":" << port_ << " send_fd_=" << send_fd_;
956 1 : if (send_fd_ != -1)
957 : {
958 : // write connect msg
959 2 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "connect_: Writing connect message";
960 1 : MessHead mh = {0, MessHead::connect_v0, htons(source_rank()), {htonl(CONN_MAGIC)}};
961 1 : ssize_t sts = write(send_fd_, &mh, sizeof(mh));
962 1 : if (sts == -1)
963 : {
964 0 : TLOG(TLVL_ERROR) << GetTraceName() << "connect_: Error writing connect message!";
965 : // a write error here is completely unexpected!
966 0 : connect_state = 0;
967 0 : close(send_fd_);
968 0 : send_fd_ = -1;
969 : }
970 : else
971 : {
972 4 : TLOG(TLVL_INFO) << GetTraceName() << "connect_: Successfully connected";
973 : // consider it all connected/established
974 1 : connect_state = 1;
975 1 : connection_was_lost_ = false;
976 : }
977 :
978 : #if USE_ACKS
979 : if (ack_listen_thread_ && ack_listen_thread_->joinable()) ack_listen_thread_->join();
980 : TLOG(TLVL_INFO) << GetTraceName() << "Starting Ack Listener Thread";
981 :
982 : try
983 : {
984 : ack_listen_thread_ = std::make_unique<boost::thread>(&TCPSocketTransfer::receive_acks_, this);
985 : char tname[16]; // Size 16 - see man page pthread_setname_np(3) and/or prctl(2)
986 : snprintf(tname, sizeof(tname) - 1, "%d-AckListen", my_rank); // NOLINT
987 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
988 : auto handle = ack_listen_thread_.native_handle();
989 : pthread_setname_np(handle, tname);
990 : }
991 : catch (const boost::exception& e)
992 : {
993 : TLOG(TLVL_ERROR) << "Caught boost::exception starting TCP Socket Ack Listen thread: " << boost::diagnostic_information(e) << ", errno=" << errno;
994 : std::cerr << "Caught boost::exception starting TCP Socket Ack Listen thread: " << boost::diagnostic_information(e) << ", errno=" << errno << std::endl;
995 : exit(5);
996 : }
997 : #endif
998 : }
999 1 : }
1000 :
1001 2 : void artdaq::TCPSocketTransfer::reconnect_()
1002 : {
1003 2 : if (send_fd_ == -1 && role() == TransferInterface::Role::kSend)
1004 : {
1005 0 : TLOG(TLVL_DEBUG + 33) << GetTraceName() << "check/reconnect";
1006 0 : return connect_();
1007 : }
1008 : }
1009 :
1010 2 : void artdaq::TCPSocketTransfer::start_listen_thread_()
1011 : {
1012 2 : std::lock_guard<std::mutex> start_lock(listen_thread_mutex_);
1013 2 : if (listen_thread_refcount_ == 0)
1014 : {
1015 2 : if (listen_thread_ && listen_thread_->joinable())
1016 : {
1017 0 : listen_thread_->join();
1018 : }
1019 2 : listen_thread_refcount_ = 1;
1020 8 : TLOG(TLVL_INFO) << GetTraceName() << "Starting Listener Thread";
1021 :
1022 : try
1023 : {
1024 2 : listen_thread_ = std::make_unique<boost::thread>(&TCPSocketTransfer::listen_, port_, rcvbuf_);
1025 : char tname[16]; // Size 16 - see man page pthread_setname_np(3) and/or prctl(2)
1026 2 : snprintf(tname, sizeof(tname) - 1, "%d-Listen", my_rank); // NOLINT
1027 2 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
1028 2 : auto handle = listen_thread_->native_handle();
1029 2 : pthread_setname_np(handle, tname);
1030 : }
1031 0 : catch (const boost::exception& e)
1032 : {
1033 0 : TLOG(TLVL_ERROR) << "Caught boost::exception starting TCP Socket Listen thread: " << boost::diagnostic_information(e) << ", errno=" << errno;
1034 0 : std::cerr << "Caught boost::exception starting TCP Socket Listen thread: " << boost::diagnostic_information(e) << ", errno=" << errno << std::endl;
1035 0 : exit(5);
1036 0 : }
1037 : }
1038 : else
1039 : {
1040 0 : listen_thread_refcount_++;
1041 : }
1042 2 : }
1043 :
1044 : #if USE_ACKS
1045 : void artdaq::TCPSocketTransfer::receive_acks_()
1046 : {
1047 : while (send_fd_ >= 0)
1048 : {
1049 : pollfd pollfd_s;
1050 : pollfd_s.events = POLLIN | POLLPRI;
1051 : pollfd_s.fd = send_fd_;
1052 :
1053 : TLOG(TLVL_DEBUG + 48) << GetTraceName() << "receive_acks_: Polling fd to see if there's data";
1054 : int num_fds_ready = poll(&pollfd_s, 1, 1000);
1055 : if (num_fds_ready <= 0)
1056 : {
1057 : if (num_fds_ready == 0)
1058 : {
1059 : TLOG(TLVL_DEBUG + 48) << GetTraceName() << "receive_acks_: No data on receive socket";
1060 : continue;
1061 : }
1062 :
1063 : TLOG(TLVL_ERROR) << "Error in poll: errno=" << errno;
1064 : break;
1065 : }
1066 :
1067 : if (pollfd_s.revents & (POLLIN | POLLPRI))
1068 : {
1069 : // Expected, don't have to check revents any further
1070 : }
1071 : else
1072 : {
1073 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "receive_acks_: Wrong event received from pollfd: " << pollfd_s.revents;
1074 : break;
1075 : }
1076 :
1077 : MessHead mh;
1078 : auto sts = read(send_fd_, &mh, sizeof(mh));
1079 :
1080 : if (sts != sizeof(mh))
1081 : {
1082 : TLOG(TLVL_ERROR) << GetTraceName() << "receive_ack_: Wrong message header length received! (actual " << sts << " != " << sizeof(mh) << " expected)";
1083 : continue;
1084 : }
1085 :
1086 : // check for "magic" and valid source_id(aka rank)
1087 : mh.source_id = ntohs(mh.source_id); // convert here as it is reference several times
1088 : if (mh.source_id != my_rank)
1089 : {
1090 : TLOG(TLVL_ERROR) << GetTraceName() << "receive_ack_: Received ack for different sender! Rank=" << my_rank << ", hdr=" << mh.source_id;
1091 : continue;
1092 : }
1093 : if (ntohl(mh.conn_magic) != ACK_MAGIC || !(mh.message_type == MessHead::ack_v0)) // Allow for future connect message versions
1094 : {
1095 : TLOG(TLVL_ERROR) << GetTraceName() << "receive_ack_: Wrong magic bytes in header!";
1096 : continue;
1097 : }
1098 :
1099 : TLOG(TLVL_DEBUG + 47) << GetTraceName() << "receive_acks_: Received ack message, diff is now " << (send_ack_diff_.load() - 1);
1100 : send_ack_diff_--;
1101 : }
1102 : }
1103 :
1104 : void artdaq::TCPSocketTransfer::send_ack_(int fd)
1105 : {
1106 : MessHead mh = {0, MessHead::ack_v0, htons(source_rank()), {htonl(ACK_MAGIC)}};
1107 : write(fd, &mh, sizeof(mh));
1108 : }
1109 : #endif
1110 :
1111 2 : void artdaq::TCPSocketTransfer::listen_(int port, size_t rcvbuf)
1112 : {
1113 2 : int listen_fd = -1;
1114 6 : while (listen_thread_refcount_ > 0)
1115 : {
1116 8 : TLOG(TLVL_DEBUG + 33) << "listen_: Listening/accepting new connections on port " << port;
1117 4 : if (listen_fd == -1)
1118 : {
1119 4 : TLOG(TLVL_DEBUG + 32) << "listen_: Opening listener";
1120 2 : listen_fd = TCP_listen_fd(port, rcvbuf);
1121 : }
1122 4 : if (listen_fd == -1)
1123 : {
1124 0 : TLOG(TLVL_DEBUG + 32) << "listen_: Error creating listen_fd!";
1125 0 : break;
1126 : }
1127 :
1128 : int res;
1129 4 : timeval tv = {2, 0}; // maybe increase of some global "debugging" flag set???
1130 : fd_set rfds;
1131 68 : FD_ZERO(&rfds);
1132 4 : FD_SET(listen_fd, &rfds); // NOLINT
1133 :
1134 4 : res = select(listen_fd + 1, &rfds, static_cast<fd_set*>(nullptr), static_cast<fd_set*>(nullptr), &tv);
1135 4 : if (res > 0)
1136 : {
1137 : int sts;
1138 : sockaddr_un un;
1139 1 : socklen_t arglen = sizeof(un);
1140 : int fd;
1141 2 : TLOG(TLVL_DEBUG + 32) << "listen_: Calling accept";
1142 1 : fd = accept(listen_fd, reinterpret_cast<sockaddr*>(&un), &arglen); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
1143 2 : TLOG(TLVL_DEBUG + 32) << "listen_: Done with accept";
1144 :
1145 2 : TLOG(TLVL_DEBUG + 32) << "listen_: Reading connect message";
1146 1 : socklen_t lenlen = sizeof(tv);
1147 : /*sts=*/
1148 1 : setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, lenlen); // see man 7 socket.
1149 : MessHead mh;
1150 1 : uint64_t mark_us = TimeUtils::gettimeofday_us();
1151 1 : sts = read(fd, &mh, sizeof(mh));
1152 1 : uint64_t delta_us = TimeUtils::gettimeofday_us() - mark_us;
1153 2 : TLOG(TLVL_DEBUG + 32) << "listen_: Read of connect message took " << delta_us << " microseconds.";
1154 1 : if (sts != sizeof(mh))
1155 : {
1156 0 : TLOG(TLVL_DEBUG + 32) << "listen_: Wrong message header length received!";
1157 0 : close(fd);
1158 0 : continue;
1159 0 : }
1160 :
1161 : // check for "magic" and valid source_id(aka rank)
1162 1 : mh.source_id = ntohs(mh.source_id); // convert here as it is reference several times
1163 1 : if (ntohl(mh.conn_magic) != CONN_MAGIC || !(mh.message_type == MessHead::connect_v0)) // Allow for future connect message versions
1164 : {
1165 0 : TLOG(TLVL_DEBUG + 32) << "listen_: Wrong magic bytes in header!";
1166 0 : close(fd);
1167 0 : continue;
1168 0 : }
1169 :
1170 : // now add (new) connection
1171 1 : std::lock_guard<std::mutex> lk(fd_mutex_);
1172 1 : connected_fds_[mh.source_id].insert(fd);
1173 :
1174 3 : TLOG(TLVL_INFO) << "listen_: New fd is " << fd << " for source rank " << mh.source_id;
1175 1 : }
1176 : else
1177 : {
1178 6 : TLOG(TLVL_DEBUG + 46) << "listen_: No connections in timeout interval!";
1179 : }
1180 : }
1181 :
1182 6 : TLOG(TLVL_INFO) << "listen_: Shutting down connection listener";
1183 2 : if (listen_fd != -1)
1184 : {
1185 2 : close(listen_fd);
1186 : }
1187 2 : std::lock_guard<std::mutex> lk(fd_mutex_);
1188 2 : auto it = connected_fds_.begin();
1189 2 : while (it != connected_fds_.end())
1190 : {
1191 0 : auto& fd_set = it->second;
1192 0 : auto rank_it = fd_set.begin();
1193 0 : while (rank_it != fd_set.end())
1194 : {
1195 0 : close(*rank_it);
1196 0 : rank_it = fd_set.erase(rank_it);
1197 : }
1198 0 : it = connected_fds_.erase(it);
1199 : }
1200 2 : first_fragment_received_.clear();
1201 :
1202 2 : } // do_connect_
1203 :
1204 28 : size_t artdaq::TCPSocketTransfer::getConnectedFDCount_(int source_rank)
1205 : {
1206 28 : std::lock_guard<std::mutex> lk(fd_mutex_);
1207 : #ifndef __OPTIMIZE__
1208 56 : TLOG(TLVL_DEBUG + 45) << GetTraceName() << "getConnectedFDCount_: count is " << (connected_fds_.count(source_rank) != 0u ? connected_fds_[source_rank].size() : 0);
1209 : #endif
1210 56 : return connected_fds_.count(source_rank) != 0u ? connected_fds_[source_rank].size() : 0;
1211 28 : }
1212 :
1213 29 : int artdaq::TCPSocketTransfer::getActiveFD_(int source_rank)
1214 : {
1215 29 : std::lock_guard<std::mutex> lk(fd_mutex_);
1216 : #ifndef __OPTIMIZE__
1217 58 : TLOG(TLVL_DEBUG + 45) << GetTraceName() << "getActiveFD_: fd is " << (active_receive_fds_.count(source_rank) != 0u ? active_receive_fds_[source_rank] : -1);
1218 : #endif
1219 58 : return active_receive_fds_.count(source_rank) != 0u ? active_receive_fds_[source_rank] : -1;
1220 29 : }
1221 5 : void artdaq::TCPSocketTransfer::setActiveFD_(int source_rank, int fd)
1222 : {
1223 5 : std::lock_guard<std::mutex> lk(fd_mutex_);
1224 : #ifndef __OPTIMIZE__
1225 10 : TLOG(TLVL_DEBUG + 45) << GetTraceName() << "setActiveFD_: setting active fd for rank " << source_rank << " to " << fd;
1226 : #endif
1227 5 : active_receive_fds_[source_rank] = fd;
1228 5 : }
1229 5 : int artdaq::TCPSocketTransfer::getLastActiveFD_(int source_rank)
1230 : {
1231 5 : std::lock_guard<std::mutex> lk(fd_mutex_);
1232 : #ifndef __OPTIMIZE__
1233 10 : TLOG(TLVL_DEBUG + 45) << GetTraceName() << "getLastActiveFD_: fd is " << (last_active_receive_fds_.count(source_rank) != 0u ? last_active_receive_fds_[source_rank] : -1);
1234 : #endif
1235 10 : return last_active_receive_fds_.count(source_rank) != 0u ? last_active_receive_fds_[source_rank] : -1;
1236 5 : }
1237 2 : void artdaq::TCPSocketTransfer::setLastActiveFD_(int source_rank, int fd)
1238 : {
1239 2 : std::lock_guard<std::mutex> lk(fd_mutex_);
1240 : #ifndef __OPTIMIZE__
1241 4 : TLOG(TLVL_DEBUG + 45) << GetTraceName() << "setLastActiveFD_: setting last active fd for rank " << source_rank << " to " << fd;
1242 : #endif
1243 2 : last_active_receive_fds_[source_rank] = fd;
1244 2 : }
|
