Line data Source code
1 : #include "TRACE/tracemf.h"
2 : #include "artdaq/DAQdata/Globals.hh" // Before trace.h gets included in ConcurrentQueue (from GlobalQueue)
3 : #define TRACE_NAME (app_name + "_RequestSender").c_str()
4 : #include "artdaq/DAQrate/detail/RequestSender.hh"
5 :
6 : #include "artdaq/DAQdata/TCPConnect.hh"
7 :
8 : #include "fhiclcpp/ParameterSet.h"
9 :
10 : #include <boost/thread.hpp>
11 :
12 : #include <dlfcn.h>
13 : #include <chrono>
14 : #include <cstring>
15 : #include <fstream>
16 : #include <iomanip>
17 : #include <mutex>
18 : #include <sstream>
19 : #include <thread>
20 : #include <utility>
21 :
22 : namespace artdaq {
23 19 : RequestSender::RequestSender(const fhicl::ParameterSet& pset)
24 38 : : send_requests_(pset.get<bool>("send_requests", false))
25 57 : , request_address_(pset.get<std::string>("request_address", "227.128.12.26"))
26 38 : , request_port_(pset.get<int>("request_port", 3001))
27 57 : , request_shutdown_timeout_(pset.get<size_t>("request_shutdown_timeout_us", 100000))
28 95 : , multicast_out_addr_(pset.get<std::string>("multicast_interface_ip", pset.get<std::string>("output_address", "0.0.0.0")))
29 38 : , request_interval_(pset.get<size_t>("request_interval_ms", 100))
30 57 : , request_delay_(pset.get<size_t>("request_delay_ms", 0))
31 : {
32 57 : TLOG(TLVL_DEBUG) << "RequestSender CONSTRUCTOR pset=" << pset.to_string();
33 19 : setup_requests_();
34 :
35 38 : TLOG(TLVL_DEBUG + 35) << "artdaq::RequestSender::RequestSender ctor - reader_thread_ initialized";
36 19 : initialized_ = true;
37 19 : }
38 :
39 36 : RequestSender::~RequestSender()
40 : {
41 57 : TLOG(TLVL_INFO) << "Shutting down RequestSender: Waiting for requests to be sent (total sent: " << requests_sent_ << ")";
42 :
43 19 : stop_send_request_thread_();
44 : {
45 19 : std::lock_guard<std::mutex> lk(request_mutex_);
46 19 : }
47 57 : TLOG(TLVL_INFO) << "Shutting down RequestSender: request_socket_: " << request_socket_;
48 19 : if (request_socket_ != -1)
49 : {
50 1 : if (shutdown(request_socket_, 2) != 0 && errno == ENOTSOCK)
51 : {
52 0 : TLOG(TLVL_ERROR) << "Shutdown of request_socket_ resulted in ENOTSOCK. NOT Closing file descriptor!";
53 : }
54 : else
55 : {
56 1 : close(request_socket_);
57 : }
58 1 : request_socket_ = -1;
59 : }
60 36 : }
61 :
62 3 : void RequestSender::SetRequestMode(detail::RequestMessageMode const& mode)
63 : {
64 : {
65 3 : std::lock_guard<std::mutex> lk(request_mutex_);
66 3 : request_mode_ = mode;
67 3 : }
68 3 : sender_cv_.notify_all();
69 3 : }
70 :
71 19 : void RequestSender::setup_requests_()
72 : {
73 19 : if (send_requests_)
74 : {
75 1 : request_socket_ = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
76 1 : if (request_socket_ < 0)
77 : {
78 0 : TLOG(TLVL_ERROR) << "I failed to create the socket for sending Data Requests! err=" << strerror(errno);
79 0 : exit(1);
80 : }
81 1 : int sts = ResolveHost(request_address_.c_str(), request_port_, request_addr_);
82 1 : if (sts == -1)
83 : {
84 0 : TLOG(TLVL_ERROR) << "Unable to resolve Data Request address, err=" << strerror(errno);
85 0 : exit(1);
86 : }
87 :
88 : /* if (multicast_out_addr_ == "0.0.0.0")
89 : {
90 : char hostname[HOST_NAME_MAX];
91 : sts = gethostname(hostname, HOST_NAME_MAX);
92 : multicast_out_addr_ = std::string(hostname);
93 : if (sts < 0)
94 : {
95 : TLOG(TLVL_ERROR) << "Could not get current hostname, err=" << strerror(errno);
96 : exit(1);
97 : }
98 : }*/
99 :
100 : // For 0.0.0.0, use system-specified IP_MULTICAST_IF
101 1 : if (multicast_out_addr_ != "localhost" && multicast_out_addr_ != "0.0.0.0")
102 : {
103 : struct in_addr addr;
104 0 : sts = GetInterfaceForNetwork(multicast_out_addr_.c_str(), addr);
105 : // sts = ResolveHost(multicast_out_addr_.c_str(), addr);
106 0 : if (sts == -1)
107 : {
108 0 : TLOG(TLVL_ERROR) << "Unable to determine the multicast interface address for " << multicast_out_addr_ << ", err=" << strerror(errno);
109 0 : exit(1);
110 : }
111 : char addr_str[INET_ADDRSTRLEN];
112 0 : inet_ntop(AF_INET, &(addr), addr_str, INET_ADDRSTRLEN);
113 0 : TLOG(TLVL_INFO) << "Successfully determined the multicast network interface for " << multicast_out_addr_ << ": " << addr_str;
114 :
115 0 : if (setsockopt(request_socket_, IPPROTO_IP, IP_MULTICAST_IF, &addr, sizeof(addr)) == -1)
116 : {
117 0 : TLOG(TLVL_ERROR) << "Cannot set outgoing interface, err=" << strerror(errno);
118 0 : exit(1);
119 : }
120 : }
121 1 : int yes = 1;
122 1 : if (setsockopt(request_socket_, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(yes)) < 0)
123 : {
124 0 : TLOG(TLVL_ERROR) << "Unable to enable port reuse on request socket, err=" << strerror(errno);
125 0 : exit(1);
126 : }
127 1 : if (setsockopt(request_socket_, IPPROTO_IP, IP_MULTICAST_LOOP, &yes, sizeof(yes)) < 0)
128 : {
129 0 : TLOG(TLVL_ERROR) << "Unable to enable multicast loopback on request socket, err=" << strerror(errno);
130 0 : exit(1);
131 : }
132 1 : if (setsockopt(request_socket_, SOL_SOCKET, SO_BROADCAST, &yes, sizeof(yes)) == -1)
133 : {
134 0 : TLOG(TLVL_ERROR) << "Cannot set request socket to broadcast, err=" << strerror(errno);
135 0 : exit(1);
136 : }
137 : }
138 19 : }
139 :
140 1 : void RequestSender::start_send_request_thread_()
141 : {
142 1 : if (!send_requests_)
143 : {
144 0 : return;
145 : }
146 1 : stop_send_request_thread_();
147 3 : TLOG(TLVL_INFO) << "Starting Request Sender Thread";
148 1 : sender_thread_running_ = true;
149 : try
150 : {
151 1 : sender_thread_.reset(new boost::thread(&RequestSender::do_send_request_, this));
152 : char tname[16];
153 1 : snprintf(tname, 16, "%s", "RequestSend"); // NOLINT
154 1 : auto handle = sender_thread_->native_handle();
155 1 : pthread_setname_np(handle, tname);
156 : }
157 0 : catch (const boost::exception& e)
158 : {
159 0 : TLOG(TLVL_ERROR) << "Caught boost::exception starting Request Send thread: " << boost::diagnostic_information(e) << ", errno=" << errno;
160 0 : std::cerr << "Caught boost::exception starting Request Send thread: " << boost::diagnostic_information(e) << ", errno=" << errno << std::endl;
161 0 : exit(5);
162 0 : }
163 : }
164 :
165 20 : void RequestSender::stop_send_request_thread_()
166 : {
167 20 : if (!send_requests_)
168 : {
169 18 : return;
170 : }
171 6 : TLOG(TLVL_INFO) << "Stopping Request Sender Thread";
172 2 : auto start_time = std::chrono::steady_clock::now();
173 2 : sender_thread_running_ = false;
174 :
175 96 : while (!sender_thread_exited_ && std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::steady_clock::now() - start_time) < request_shutdown_timeout_)
176 : {
177 188 : TLOG(TLVL_DEBUG + 30) << "Waiting for Request Sender thread to exit...";
178 94 : std::this_thread::sleep_for(std::chrono::milliseconds(1));
179 : }
180 :
181 2 : if (sender_thread_ != nullptr && sender_thread_->joinable())
182 : {
183 1 : sender_thread_->join();
184 : }
185 : }
186 :
187 1 : void RequestSender::do_send_request_()
188 : {
189 1 : sender_thread_exited_ = false;
190 8 : while (sender_thread_running_)
191 : {
192 7 : if (request_socket_ == -1)
193 : {
194 0 : setup_requests_();
195 : }
196 :
197 14 : TLOG(TLVL_DEBUG + 33) << "Waiting for up to " << request_interval_ << " milliseconds.";
198 7 : std::unique_lock<std::mutex> lk(request_mutex_);
199 7 : sender_cv_.wait_for(lk, request_interval_);
200 :
201 14 : TLOG(TLVL_DEBUG + 33) << "Creating RequestMessage";
202 7 : detail::RequestMessage message;
203 7 : message.setRank(my_rank);
204 7 : message.setRunNumber(run_number_);
205 :
206 14 : for (auto& req : active_requests_)
207 : {
208 14 : TLOG(TLVL_DEBUG + 36) << "Adding a request with sequence ID " << req.first << ", timestamp " << req.second << " to request message";
209 7 : message.addRequest(req.first, req.second);
210 : }
211 14 : TLOG(TLVL_DEBUG + 33) << "Setting mode flag in Message Header to " << static_cast<int>(request_mode_);
212 7 : message.setMode(request_mode_);
213 :
214 7 : lk.unlock();
215 :
216 7 : if (message.size() == 0) continue;
217 :
218 : char str[INET_ADDRSTRLEN];
219 6 : inet_ntop(AF_INET, &(request_addr_.sin_addr), str, INET_ADDRSTRLEN);
220 12 : TLOG(TLVL_DEBUG + 33) << "Sending request for " << message.size() << " events to multicast group " << str
221 6 : << ", port " << request_port_ << ", interface " << multicast_out_addr_;
222 6 : auto buf = message.GetMessage();
223 6 : auto sts = sendto(request_socket_, &buf[0], buf.size(), 0, reinterpret_cast<struct sockaddr*>(&request_addr_), sizeof(request_addr_)); // NOLINT(cppcoreguidelines-pro-type-reinterpret-cast)
224 6 : if (sts < 0 || static_cast<size_t>(sts) != buf.size())
225 : {
226 0 : TLOG(TLVL_ERROR) << "Error sending request message err=" << strerror(errno) << "sts=" << sts;
227 0 : request_socket_ = -1;
228 0 : continue;
229 0 : }
230 12 : TLOG(TLVL_DEBUG + 33) << "Done sending request sts=" << sts;
231 6 : requests_sent_++;
232 8 : }
233 1 : sender_thread_exited_ = true;
234 1 : }
235 :
236 446 : void RequestSender::AddRequest(Fragment::sequence_id_t const& seqID, Fragment::timestamp_t const& timestamp)
237 : {
238 446 : if (!send_requests_)
239 : {
240 443 : return;
241 : }
242 6 : TLOG(TLVL_DEBUG + 37) << "Adding request for sequence ID " << seqID << " and timestamp " << timestamp << " to request list.";
243 3 : boost::thread request(&RequestSender::do_add_request_, this, seqID, timestamp);
244 3 : request.detach();
245 3 : }
246 :
247 3 : void RequestSender::do_add_request_(Fragment::sequence_id_t const& seqID, Fragment::timestamp_t const& timestamp)
248 : {
249 3 : while (!initialized_)
250 : {
251 0 : usleep(1000);
252 : }
253 :
254 3 : if (!sender_thread_running_)
255 : {
256 1 : start_send_request_thread_();
257 : }
258 :
259 3 : std::this_thread::sleep_for(request_delay_);
260 : {
261 3 : std::lock_guard<std::mutex> lk(request_mutex_);
262 3 : if (active_requests_.count(seqID) == 0u)
263 : {
264 6 : TLOG(TLVL_DEBUG + 37) << "Adding request for sequence ID " << seqID << " and timestamp " << timestamp << " to request list.";
265 3 : active_requests_[seqID] = timestamp;
266 : }
267 :
268 3 : while (active_requests_.size() > detail::RequestMessage::max_request_count())
269 : {
270 0 : TLOG(TLVL_WARNING) << "Erasing request with seqID " << active_requests_.begin()->first << " due to over-large request list size! (" << active_requests_.size() << " / " << detail::RequestMessage::max_request_count() << ")";
271 0 : active_requests_.erase(active_requests_.begin());
272 : }
273 3 : }
274 3 : sender_cv_.notify_all();
275 3 : }
276 :
277 445 : void RequestSender::RemoveRequest(Fragment::sequence_id_t const& seqID)
278 : {
279 445 : while (!initialized_)
280 : {
281 0 : usleep(1000);
282 : }
283 445 : std::lock_guard<std::mutex> lk(request_mutex_);
284 890 : TLOG(TLVL_DEBUG + 38) << "Removing request for sequence ID " << seqID << " from request list.";
285 445 : active_requests_.erase(seqID);
286 445 : }
287 : } // namespace artdaq
|
