Line data Source code
1 : #include <memory>
2 :
3 : #include "artdaq/DAQdata/Globals.hh"
4 : #define TRACE_NAME (app_name + "_BundleTransfer").c_str()
5 :
6 : #include "artdaq-core/Data/ContainerFragmentLoader.hh"
7 : #include "artdaq/TransferPlugins/TCPSocketTransfer.hh"
8 : #include "artdaq/TransferPlugins/TransferInterface.hh"
9 :
10 : #include <boost/thread.hpp>
11 :
12 : namespace artdaq {
13 : /**
14 : * \brief The BundleTransfer TransferInterface plugin automatically combines smaller Fragments to transfer a lower rate of larger Fragments, which TCP is better able to handle.
15 : */
16 : class BundleTransfer : public TransferInterface
17 : {
18 : public:
19 : /**
20 : * \brief BundleTransfer Constructor
21 : * \param pset ParameterSet used to configure BundleTransfer
22 : * \param role Role of this TransferInterface, either kReceive or kSend
23 : */
24 : BundleTransfer(const fhicl::ParameterSet& pset, Role role);
25 :
26 : /**
27 : * \brief BundleTransfer default Destructor
28 : */
29 : ~BundleTransfer() override;
30 :
31 : /**
32 : * \brief Receive a Fragment, using the underlying transfer plugin
33 : * \param fragment Output Fragment
34 : * \param receiveTimeout Time to wait before returning TransferInterface::RECV_TIMEOUT
35 : * \return Rank of sender
36 : */
37 0 : int receiveFragment(artdaq::Fragment& fragment,
38 : size_t receiveTimeout) override
39 : {
40 0 : if (bundle_fragment_ == nullptr)
41 : {
42 0 : receive_bundle_fragment_(receiveTimeout);
43 0 : if (current_rank_ < RECV_SUCCESS) return current_rank_;
44 : }
45 :
46 0 : ContainerFragment cf(*bundle_fragment_);
47 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Retrieving Fragment " << (current_block_index_ + 1) << " of " << cf.block_count();
48 0 : fragment.resizeBytes(cf.fragSize(current_block_index_) - sizeof(detail::RawFragmentHeader));
49 0 : memcpy(fragment.headerAddress(), static_cast<const uint8_t*>(cf.dataBegin()) + cf.fragmentIndex(current_block_index_), cf.fragSize(current_block_index_));
50 0 : current_block_index_++;
51 0 : if (current_block_index_ >= cf.block_count()) // Index vs. count!
52 : {
53 0 : bundle_fragment_.reset(nullptr);
54 : }
55 0 : return current_rank_;
56 0 : }
57 :
58 : /**
59 : * \brief Receive a Fragment Header from the transport mechanism
60 : * \param[out] header Received Fragment Header
61 : * \param receiveTimeout Timeout for receive
62 : * \return The rank the Fragment was received from (should be source_rank), or RECV_TIMEOUT
63 : */
64 0 : int receiveFragmentHeader(detail::RawFragmentHeader& header, size_t receiveTimeout) override
65 : {
66 0 : if (bundle_fragment_ == nullptr)
67 : {
68 0 : receive_bundle_fragment_(receiveTimeout);
69 0 : if (current_rank_ < RECV_SUCCESS) return current_rank_;
70 : }
71 0 : ContainerFragment cf(*bundle_fragment_);
72 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Retrieving Fragment Header " << (current_block_index_ + 1) << " of " << cf.block_count();
73 0 : memcpy(&header, static_cast<const uint8_t*>(cf.dataBegin()) + cf.fragmentIndex(current_block_index_), sizeof(detail::RawFragmentHeader));
74 0 : return current_rank_;
75 0 : }
76 :
77 : /**
78 : * \brief Receive the body of a Fragment to the given destination pointer
79 : * \param destination Pointer to memory region where Fragment data should be stored
80 : * \param wordCount Number of words of Fragment data to receive
81 : * \return The rank the Fragment was received from (should be source_rank), or RECV_TIMEOUT
82 : */
83 0 : int receiveFragmentData(RawDataType* destination, size_t /*wordCount*/) override
84 : {
85 0 : if (bundle_fragment_ == nullptr) // Should be impossible!
86 : {
87 0 : return RECV_TIMEOUT;
88 : }
89 0 : ContainerFragment cf(*bundle_fragment_);
90 0 : TLOG(TLVL_DEBUG + 32) << GetTraceName() << "Retrieving Fragment Data " << (current_block_index_ + 1) << " of " << cf.block_count();
91 0 : memcpy(destination, static_cast<const uint8_t*>(cf.dataBegin()) + cf.fragmentIndex(current_block_index_) + sizeof(detail::RawFragmentHeader), cf.fragSize(current_block_index_) - sizeof(detail::RawFragmentHeader));
92 0 : current_block_index_++;
93 0 : if (current_block_index_ >= cf.block_count()) // Index vs. count!
94 : {
95 0 : bundle_fragment_.reset(nullptr);
96 : }
97 0 : return current_rank_;
98 0 : }
99 :
100 : /**
101 : * \brief Send a Fragment in non-reliable mode, using the underlying transfer plugin
102 : * \param fragment The Fragment to send
103 : * \param send_timeout_usec How long to wait before aborting. Defaults to size_t::MAX_VALUE
104 : * \return A TransferInterface::CopyStatus result variable
105 : */
106 0 : CopyStatus transfer_fragment_min_blocking_mode(artdaq::Fragment const& fragment, size_t send_timeout_usec) override
107 : {
108 0 : TLOG(TLVL_DEBUG + 35) << GetTraceName() << "transfer_fragment_min_blocking_mode START";
109 0 : last_send_call_reliable_ = false;
110 0 : last_send_timeout_usec_ = send_timeout_usec;
111 : {
112 0 : std::unique_lock<std::mutex> lk(fragment_mutex_);
113 0 : if (current_buffer_size_bytes_ > max_hold_size_bytes_)
114 : {
115 0 : auto wait_start = std::chrono::steady_clock::now();
116 0 : fragment_cv_.wait_for(lk, std::chrono::microseconds(send_timeout_usec), [&] { return current_buffer_size_bytes_ < max_hold_size_bytes_; });
117 0 : report_wait_time_(std::chrono::duration<double>(std::chrono::steady_clock::now() - wait_start).count());
118 : }
119 :
120 0 : if (current_buffer_size_bytes_ > max_hold_size_bytes_)
121 : {
122 0 : TLOG(TLVL_WARNING) << GetTraceName() << "Dropping data due to timeout in min_blocking_mode";
123 0 : if (metricMan)
124 : {
125 0 : metricMan->sendMetric("Bundle Dropped Events to Rank " + std::to_string(destination_rank()), 1, "events/s", 2, MetricMode::Rate);
126 : }
127 0 : return CopyStatus::kTimeout;
128 : }
129 :
130 0 : TLOG(TLVL_DEBUG + 35) << GetTraceName() << "transfer_fragment_min_blocking_mode after wait for buffer";
131 : // Always send along Broadcast Fragments immediately
132 0 : if (Fragment::isBroadcastFragmentType(fragment.type()))
133 : {
134 0 : system_fragment_cached_ = true;
135 : }
136 :
137 0 : if (fragment_buffer_.empty())
138 : {
139 0 : buffer_oldest_time_ = std::chrono::steady_clock::now();
140 : }
141 0 : current_buffer_size_bytes_ += fragment.sizeBytes();
142 : // Eww, we have to copy
143 0 : fragment_buffer_.emplace_back(fragment);
144 0 : report_buffer_metrics_();
145 0 : }
146 0 : TLOG(TLVL_DEBUG + 35) << GetTraceName() << "transfer_fragment_min_blocking_mode END";
147 0 : return last_copy_status_; // Might be a lie, but we're going to send from the thread proc
148 : }
149 :
150 : /**
151 : * \brief Send a Fragment in reliable mode, using the underlying transfer plugin
152 : * \param fragment The Fragment to send
153 : * \return A TransferInterface::CopyStatus result variable
154 : */
155 0 : CopyStatus transfer_fragment_reliable_mode(artdaq::Fragment&& fragment) override
156 : {
157 0 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "transfer_fragment_reliable_mode START";
158 0 : last_send_call_reliable_ = true;
159 : {
160 0 : std::unique_lock<std::mutex> lk(fragment_mutex_);
161 0 : if (current_buffer_size_bytes_ > max_hold_size_bytes_)
162 : {
163 0 : auto wait_start = std::chrono::steady_clock::now();
164 0 : while (current_buffer_size_bytes_ > max_hold_size_bytes_)
165 : {
166 0 : fragment_cv_.wait(lk, [&] { return current_buffer_size_bytes_ < max_hold_size_bytes_; });
167 : }
168 0 : report_wait_time_(std::chrono::duration<double>(std::chrono::steady_clock::now() - wait_start).count());
169 : }
170 :
171 0 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "transfer_fragment_reliable_mode after wait for buffer";
172 :
173 : // Always send along Broadcast Fragments immediately
174 0 : if (Fragment::isBroadcastFragmentType(fragment.type()))
175 : {
176 0 : system_fragment_cached_ = true;
177 : }
178 :
179 0 : if (fragment_buffer_.empty())
180 : {
181 0 : buffer_oldest_time_ = std::chrono::steady_clock::now();
182 : }
183 0 : current_buffer_size_bytes_ += fragment.sizeBytes();
184 0 : fragment_buffer_.emplace_back(std::move(fragment));
185 0 : report_buffer_metrics_();
186 0 : }
187 0 : TLOG(TLVL_DEBUG + 36) << GetTraceName() << "transfer_fragment_reliable_mode END";
188 0 : return last_copy_status_; // Might be a lie, but we're going to send from the thread proc
189 : }
190 :
191 : /**
192 : * \brief Determine whether the TransferInterface plugin is able to send/receive data
193 : * \return True if the TransferInterface plugin is currently able to send/receive data
194 : */
195 0 : bool isRunning() override { return running_; }
196 :
197 : /**
198 : * \brief Flush any in-flight data. This should be used by the receiver after the receive loop has
199 : * ended.
200 : */
201 0 : void flush_buffers() override { theTransfer_->flush_buffers(); }
202 :
203 : private:
204 : BundleTransfer(BundleTransfer const&) = delete;
205 : BundleTransfer(BundleTransfer&&) = delete;
206 : BundleTransfer& operator=(BundleTransfer const&) = delete;
207 : BundleTransfer& operator=(BundleTransfer&&) = delete;
208 :
209 : private:
210 : std::unique_ptr<TransferInterface> theTransfer_;
211 : size_t send_threshold_bytes_;
212 : size_t max_hold_size_bytes_;
213 : int max_hold_time_us_;
214 : FragmentPtr bundle_fragment_{nullptr};
215 : Fragments fragment_buffer_;
216 : size_t current_block_index_{0};
217 : int current_rank_ = 0;
218 : CopyStatus last_copy_status_{CopyStatus::kSuccess};
219 :
220 : std::chrono::steady_clock::time_point send_fragment_started_;
221 : std::chrono::steady_clock::time_point buffer_oldest_time_{std::chrono::steady_clock::now()};
222 : std::atomic<size_t> current_buffer_size_bytes_{0};
223 : std::unique_ptr<boost::thread> send_timeout_thread_;
224 : std::atomic<bool> system_fragment_cached_{false};
225 : std::atomic<bool> send_timeout_thread_running_{false};
226 : std::atomic<bool> last_send_call_reliable_{true};
227 : std::atomic<size_t> last_send_timeout_usec_{1000000};
228 : std::atomic<bool> running_{true};
229 : std::mutex fragment_mutex_;
230 : std::condition_variable fragment_cv_;
231 :
232 : bool check_send_(bool force);
233 : void start_timeout_thread_();
234 : void send_timeout_thread_proc_();
235 : bool send_bundle_fragment_(bool forceSend = false);
236 : void receive_bundle_fragment_(size_t receiveTimeout);
237 :
238 : /// Report time spent blocked waiting for space in the bundle buffer (sender-side back-pressure)
239 0 : void report_wait_time_(double waitSeconds)
240 : {
241 0 : if (metricMan)
242 : {
243 0 : metricMan->sendMetric("Bundle Buffer Wait Time to Rank " + std::to_string(destination_rank()), waitSeconds, "s", 2, MetricMode::Accumulate | MetricMode::Maximum);
244 : }
245 0 : }
246 :
247 : /// Report bundle buffer occupancy metrics. Requires fragment_mutex_ to be held.
248 0 : void report_buffer_metrics_()
249 : {
250 0 : if (!metricMan) return;
251 0 : auto rankString = std::to_string(destination_rank());
252 0 : metricMan->sendMetric("Bundle Buffer Occupancy to Rank " + rankString, current_buffer_size_bytes_.load(), "B", 2, MetricMode::LastPoint);
253 0 : metricMan->sendMetric("Bundle Buffer Events to Rank " + rankString, static_cast<unsigned long>(fragment_buffer_.size()), "events", 2, MetricMode::LastPoint);
254 0 : double oldestAge = fragment_buffer_.empty() ? 0.0 : std::chrono::duration<double>(std::chrono::steady_clock::now() - buffer_oldest_time_).count();
255 0 : metricMan->sendMetric("Bundle Buffer Oldest Event Age to Rank " + rankString, oldestAge, "s", 2, MetricMode::LastPoint);
256 0 : }
257 : };
258 : } // namespace artdaq
259 :
260 0 : artdaq::BundleTransfer::BundleTransfer(const fhicl::ParameterSet& pset, Role role)
261 : : TransferInterface(pset, role)
262 0 : , send_threshold_bytes_(pset.get<size_t>("send_threshold_bytes", 10 * 0x100000)) // 10 MB
263 0 : , max_hold_size_bytes_(pset.get<size_t>("max_hold_size_bytes", 1000 * 0x100000)) // 1000 MB
264 0 : , max_hold_time_us_(pset.get<int>("max_hold_time_us", 100000)) // 0.1 s
265 : {
266 0 : TLOG(TLVL_INFO) << GetTraceName() << "Begin BundleTransfer constructor";
267 0 : TLOG(TLVL_INFO) << GetTraceName() << "Constructing TCPSocketTransfer";
268 0 : theTransfer_ = std::make_unique<TCPSocketTransfer>(pset, role);
269 :
270 0 : if (role == Role::kSend)
271 : {
272 0 : start_timeout_thread_();
273 : }
274 0 : }
275 :
276 0 : artdaq::BundleTransfer::~BundleTransfer()
277 : {
278 0 : if (role_ == Role::kSend)
279 : {
280 0 : send_timeout_thread_running_ = false;
281 0 : if (send_timeout_thread_ && send_timeout_thread_->joinable())
282 : {
283 0 : send_timeout_thread_->join();
284 : }
285 0 : send_bundle_fragment_(true);
286 : }
287 0 : running_ = false;
288 0 : }
289 :
290 0 : void artdaq::BundleTransfer::start_timeout_thread_()
291 : {
292 0 : if (send_timeout_thread_ && send_timeout_thread_->joinable())
293 : {
294 0 : send_timeout_thread_->join();
295 : }
296 0 : send_timeout_thread_running_ = true;
297 0 : TLOG(TLVL_INFO) << GetTraceName() << "Starting Send Timeout Thread";
298 :
299 : try
300 : {
301 0 : send_timeout_thread_ = std::make_unique<boost::thread>(&BundleTransfer::send_timeout_thread_proc_, this);
302 : char tname[16]; // Size 16 - see man page pthread_setname_np(3) and/or prctl(2)
303 0 : snprintf(tname, sizeof(tname) - 1, "%d-SNDTMO", my_rank); // NOLINT
304 0 : tname[sizeof(tname) - 1] = '\0'; // assure term. snprintf is not too evil :)
305 0 : auto handle = send_timeout_thread_->native_handle();
306 0 : pthread_setname_np(handle, tname);
307 : }
308 0 : catch (const boost::exception& e)
309 : {
310 0 : TLOG(TLVL_ERROR) << GetTraceName() << "Caught boost::exception starting Send Timeout thread: " << boost::diagnostic_information(e) << ", errno=" << errno;
311 0 : std::cerr << GetTraceName() << "Caught boost::exception starting Send Timeout thread: " << boost::diagnostic_information(e) << ", errno=" << errno << std::endl;
312 0 : exit(5);
313 0 : }
314 0 : }
315 :
316 0 : void artdaq::BundleTransfer::send_timeout_thread_proc_()
317 : {
318 0 : while (send_timeout_thread_running_)
319 : {
320 0 : if (!send_bundle_fragment_())
321 : {
322 0 : usleep(5000);
323 : }
324 : }
325 0 : }
326 :
327 0 : bool artdaq::BundleTransfer::check_send_(bool force)
328 : {
329 0 : if (force)
330 : {
331 0 : TLOG(TLVL_DEBUG + 37) << GetTraceName() << "check_send_: Send is forced, returning true";
332 0 : return true;
333 : }
334 :
335 0 : if (system_fragment_cached_.load())
336 : {
337 0 : TLOG(TLVL_DEBUG + 37) << GetTraceName() << "check_send_: System Fragment in cache, returning true";
338 0 : return true;
339 : }
340 :
341 0 : if (std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - send_fragment_started_).count() >= max_hold_time_us_)
342 : {
343 0 : TLOG(TLVL_DEBUG + 37) << GetTraceName() << "check_send_: Send timeout reached, returning true";
344 0 : return true;
345 : }
346 :
347 0 : if (current_buffer_size_bytes_ >= send_threshold_bytes_)
348 : {
349 0 : TLOG(TLVL_DEBUG + 37) << GetTraceName() << "check_send_: Buffer is full, returning true";
350 0 : return true;
351 : }
352 :
353 0 : TLOG(TLVL_DEBUG + 37) << GetTraceName() << "check_send_: returning false";
354 0 : return false;
355 : }
356 :
357 0 : bool artdaq::BundleTransfer::send_bundle_fragment_(bool forceSend)
358 : {
359 : {
360 0 : std::unique_lock<std::mutex> lk(fragment_mutex_);
361 :
362 0 : report_buffer_metrics_();
363 :
364 0 : bool send_fragment = check_send_(forceSend);
365 :
366 0 : if (send_fragment && fragment_buffer_.size() > 0)
367 : {
368 0 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "Swapping in new buffer";
369 0 : Fragments temp_buffer;
370 0 : size_t size = current_buffer_size_bytes_;
371 0 : size_t bundle_event_count = fragment_buffer_.size();
372 0 : fragment_buffer_.swap(temp_buffer);
373 0 : send_fragment_started_ = std::chrono::steady_clock::now();
374 0 : system_fragment_cached_ = false;
375 0 : current_buffer_size_bytes_ = 0;
376 0 : lk.unlock();
377 0 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "Notifying waiters";
378 0 : fragment_cv_.notify_one();
379 :
380 0 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "Setting up Bundle Fragment";
381 0 : bundle_fragment_.reset(new artdaq::Fragment(temp_buffer.front().sequenceID() + 1, temp_buffer.front().fragmentID(), artdaq::Fragment::ContainerFragmentType));
382 0 : bundle_fragment_->setTimestamp(temp_buffer.front().timestamp());
383 0 : bundle_fragment_->reserve(size / sizeof(artdaq::RawDataType));
384 :
385 0 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "Filling Bundle Fragment, sz = " << temp_buffer.size();
386 0 : ContainerFragmentLoader container_fragment(*bundle_fragment_);
387 0 : container_fragment.set_missing_data(false); // Buffer mode is never missing data, even if there IS no data.
388 0 : container_fragment.addFragments(temp_buffer, true);
389 0 : temp_buffer.clear();
390 :
391 0 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "Sending Fragment, reliable mode " << last_send_call_reliable_.load();
392 0 : CopyStatus sts = CopyStatus::kSuccess;
393 0 : if (last_send_call_reliable_)
394 : {
395 0 : sts = theTransfer_->transfer_fragment_reliable_mode(std::move(*bundle_fragment_.get()));
396 0 : bundle_fragment_.reset(nullptr);
397 : }
398 : else
399 : {
400 0 : while (sts != CopyStatus::kSuccess && send_timeout_thread_running_)
401 : {
402 0 : sts = theTransfer_->transfer_fragment_min_blocking_mode(*bundle_fragment_.get(), last_send_timeout_usec_);
403 : }
404 0 : bundle_fragment_.reset(nullptr);
405 : }
406 0 : last_copy_status_ = sts;
407 0 : if (sts != CopyStatus::kSuccess)
408 : {
409 0 : auto sts_string = sts == CopyStatus::kTimeout ? "timeout" : "other error";
410 0 : TLOG(TLVL_WARNING) << GetTraceName() << "Transfer of Bundle fragment returned status " << sts_string;
411 : }
412 0 : else if (metricMan)
413 : {
414 0 : auto rankString = std::to_string(destination_rank());
415 0 : metricMan->sendMetric("Bundle Output Event Rate to Rank " + rankString, bundle_event_count, "events/s", 2, MetricMode::Rate);
416 0 : metricMan->sendMetric("Bundle Output Data Rate to Rank " + rankString, size, "B/s", 2, MetricMode::Rate);
417 0 : metricMan->sendMetric("Bundle Size to Rank " + rankString, size, "B", 3, MetricMode::Average | MetricMode::Maximum);
418 0 : }
419 :
420 0 : TLOG(TLVL_DEBUG + 38) << GetTraceName() << "Done sending Bundle Fragment";
421 :
422 0 : return true; // Status of actual transfer
423 0 : }
424 0 : }
425 0 : return false; // Waiting on more data
426 : }
427 :
428 0 : void artdaq::BundleTransfer::receive_bundle_fragment_(size_t receiveTimeout)
429 : {
430 0 : std::lock_guard<std::mutex> lk(fragment_mutex_);
431 0 : bundle_fragment_.reset(new artdaq::Fragment(1));
432 :
433 0 : TLOG(TLVL_DEBUG + 34) << GetTraceName() << "Going to receive next bundle fragment";
434 0 : current_rank_ = theTransfer_->receiveFragment(*bundle_fragment_, receiveTimeout);
435 0 : TLOG(TLVL_DEBUG + 34) << GetTraceName() << "Done with receiveFragment, current_rank_ = " << current_rank_;
436 :
437 0 : if (current_rank_ < RECV_SUCCESS)
438 : {
439 0 : bundle_fragment_.reset(nullptr);
440 : }
441 0 : current_block_index_ = 0;
442 0 : }
443 :
444 0 : DEFINE_ARTDAQ_TRANSFER(artdaq::BundleTransfer)
|
