Line data Source code
1 : /**
2 : *
3 : * @file ZmqReceiver.cpp ZmqReceiver messaging class definitions
4 : *
5 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
6 : * Licensing/copyright details are in the COPYING file that you should have
7 : * received with this code.
8 : */
9 :
10 : #include "CallbackAdapter.hpp"
11 : #include "ipm/Receiver.hpp"
12 : #include "ipm/ZmqContext.hpp"
13 :
14 : #include "logging/Logging.hpp"
15 : #include "utilities/ZmqUri.hpp"
16 :
17 : #include <string>
18 : #include <vector>
19 :
20 : namespace dunedaq::ipm {
21 :
22 : class ZmqReceiver : public Receiver
23 : {
24 : public:
25 26 : ZmqReceiver()
26 26 : : m_socket(ZmqContext::instance().GetContext(), zmq::socket_type::pull)
27 : {
28 26 : }
29 :
30 52 : ~ZmqReceiver()
31 26 : {
32 26 : unregister_callback();
33 : // Probably (cpp)zmq does this in the socket dtor anyway, but I guess it doesn't hurt to be explicit
34 26 : if (m_connection_info.connection_string != "" && m_socket_connected) {
35 24 : try {
36 24 : m_socket.unbind(m_connection_info.connection_string);
37 24 : m_socket_connected = false;
38 0 : } catch (zmq::error_t const& err) {
39 0 : ers::error(ZmqOperationError(ERS_HERE,
40 0 : m_connection_info.connection_name,
41 : "unbind",
42 : "receive",
43 0 : err.what(),
44 0 : m_connection_info.connection_string));
45 0 : }
46 : }
47 26 : m_socket.close();
48 52 : }
49 :
50 27 : std::string connect_for_receives(const ConnectionInfo& connection_info) override
51 : {
52 27 : m_connection_info = connection_info;
53 27 : try {
54 27 : m_socket.set(zmq::sockopt::rcvtimeo, 0); // Return immediately if we can't receive
55 0 : } catch (zmq::error_t const& err) {
56 0 : throw ZmqOperationError(ERS_HERE,
57 0 : m_connection_info.connection_name,
58 : "set timeout",
59 : "receive",
60 0 : err.what(),
61 0 : m_connection_info.connection_string);
62 0 : }
63 :
64 27 : try {
65 27 : m_socket.set(zmq::sockopt::linger, 0); // Close connection immediately when close is called
66 0 : } catch (zmq::error_t const& err) {
67 0 : throw ZmqOperationError(ERS_HERE,
68 0 : m_connection_info.connection_name,
69 : "set linger",
70 : "receive",
71 0 : err.what(),
72 0 : m_connection_info.connection_string);
73 0 : }
74 :
75 27 : std::vector<std::string> resolved;
76 27 : try {
77 27 : utilities::ZmqUri this_uri(m_connection_info.connection_string);
78 26 : resolved = this_uri.get_uri_ip_addresses();
79 27 : } catch (utilities::InvalidUri const& err) {
80 1 : throw ZmqOperationError(ERS_HERE,
81 1 : m_connection_info.connection_name,
82 : "resolve connection_string",
83 : "receive",
84 : "An invalid URI was passed",
85 : m_connection_info.connection_string,
86 2 : err);
87 1 : }
88 :
89 26 : if (resolved.size() == 0) {
90 1 : throw ZmqOperationError(ERS_HERE,
91 1 : m_connection_info.connection_name,
92 : "resolve connection_string",
93 : "receive",
94 : "Unable to resolve connection_string",
95 2 : m_connection_info.connection_string);
96 : }
97 26 : for (auto& connection_string : resolved) {
98 25 : TLOG_DEBUG(TLVL_CONNECTIONSTRING) << m_connection_info.connection_name << ": Connection String is "
99 25 : << connection_string;
100 25 : try {
101 25 : m_socket.bind(connection_string);
102 24 : m_connection_info.connection_string = m_socket.get(zmq::sockopt::last_endpoint);
103 24 : m_socket_connected = true;
104 24 : break;
105 1 : } catch (zmq::error_t const& err) {
106 1 : ers::error(ZmqOperationError(
107 2 : ERS_HERE, m_connection_info.connection_name, "bind", "receive", err.what(), connection_string));
108 1 : }
109 : }
110 25 : if (!m_socket_connected) {
111 1 : throw ZmqOperationError(ERS_HERE,
112 1 : m_connection_info.connection_name,
113 : "bind",
114 : "receive",
115 : "Bind failed for all resolved connection strings",
116 2 : "");
117 : }
118 :
119 24 : m_callback_adapter.set_receiver(this);
120 :
121 48 : return m_connection_info.connection_string;
122 27 : }
123 :
124 20270 : bool can_receive() const noexcept override { return m_socket_connected; }
125 :
126 1 : void register_callback(std::function<void(Response&)> callback) override
127 : {
128 1 : m_callback_adapter.set_callback(callback);
129 1 : }
130 27 : void unregister_callback() override { m_callback_adapter.clear_callback(); }
131 :
132 4 : bool data_pending() override
133 : {
134 4 : try {
135 4 : auto events = m_socket.get(zmq::sockopt::events);
136 4 : return (events & ZMQ_POLLIN) != 0;
137 0 : } catch (zmq::error_t const& err) {
138 0 : ers::error(ZmqOperationError(ERS_HERE,
139 0 : m_connection_info.connection_name,
140 : "get events sockopt",
141 : "data_pending",
142 0 : err.what(),
143 0 : m_connection_info.connection_string));
144 0 : }
145 0 : return false;
146 : }
147 :
148 : protected:
149 20260 : Receiver::Response receive_(const duration_t& timeout, bool no_tmoexcept_mode) override
150 : {
151 20260 : Receiver::Response output;
152 20260 : zmq::message_t hdr;
153 20260 : zmq::message_t msg;
154 20260 : zmq::recv_result_t res{};
155 :
156 20260 : auto start_time = std::chrono::steady_clock::now();
157 25047 : do {
158 :
159 25047 : try {
160 25047 : TLOG_DEBUG(TLVL_ZMQRECEIVER_RECV_HDR) << m_connection_info.connection_name << ": Going to receive header";
161 25041 : res = m_socket.recv(hdr);
162 25045 : TLOG_DEBUG(TLVL_ZMQRECEIVER_RECV_HDR_2) << m_connection_info.connection_name << ": Recv res=" << res.value_or(0)
163 25043 : << " for header (hdr.size() == " << hdr.size() << ")";
164 0 : } catch (zmq::error_t const& err) {
165 0 : throw ZmqReceiveError(ERS_HERE, m_connection_info.connection_name, err.what(), "header");
166 0 : }
167 25045 : if (res || hdr.more()) {
168 20026 : TLOG_DEBUG(TLVL_ZMQRECEIVER_RECV_DATA) << m_connection_info.connection_name << ": Going to receive data";
169 20026 : output.metadata.resize(hdr.size());
170 20026 : memcpy(&output.metadata[0], hdr.data(), hdr.size());
171 :
172 : // ZMQ guarantees that the entire message has arrived
173 :
174 20026 : try {
175 20026 : res = m_socket.recv(msg);
176 0 : } catch (zmq::error_t const& err) {
177 0 : throw ZmqReceiveError(ERS_HERE, m_connection_info.connection_name, err.what(), "data");
178 0 : }
179 20026 : TLOG_DEBUG(TLVL_ZMQRECEIVER_RECV_DATA_2)
180 0 : << m_connection_info.connection_name << ": Recv res=" << res.value_or(0)
181 20026 : << " for data (msg.size() == " << msg.size() << ")";
182 20026 : output.data.resize(msg.size());
183 20026 : memcpy(&output.data[0], msg.data(), msg.size());
184 5010 : } else if (timeout > duration_t::zero()) {
185 5010 : usleep(1000);
186 : }
187 25041 : } while (std::chrono::duration_cast<duration_t>(std::chrono::steady_clock::now() - start_time) < timeout &&
188 45071 : res.value_or(0) == 0);
189 :
190 20260 : if (res.value_or(0) == 0 && !no_tmoexcept_mode) {
191 5 : throw ReceiveTimeoutExpired(ERS_HERE, m_connection_info.connection_name, timeout.count());
192 : }
193 :
194 20255 : TLOG_DEBUG(TLVL_ZMQRECEIVER_RECV_END)
195 0 : << m_connection_info.connection_name << ": Returning output with metadata size " << output.metadata.size()
196 20255 : << " and data size " << output.data.size();
197 20255 : return output;
198 20270 : }
199 :
200 : private:
201 : zmq::socket_t m_socket;
202 : bool m_socket_connected{ false };
203 : CallbackAdapter m_callback_adapter;
204 : };
205 : } // namespace dunedaq::ipm
206 :
207 26 : DEFINE_DUNE_IPM_RECEIVER(dunedaq::ipm::ZmqReceiver)
|
