GCC Code Coverage Report

Directory: ./
File: src/PZmqBackend.cpp
Date: 2026-06-23 10:44:39
Exec Total Coverage
Lines: 95 123 77.2%
Functions: 19 22 86.4%
Branches: 31 73 42.5%
Line Branch Exec Source
1 /***************************************
2 Auteur : Pierre Aubert
3 Mail : pierre.aubert@lapp.in2p3.fr
4 Licence : CeCILL-C
5 ****************************************/
6
7 #include "PZmqBackend.h"
8
9 ///Check the send result and convert it into PSendStatus
10 /**
11 * @param res : result of the zmq send
12 * @return corresponding PSendStatus
13 *
14 * ZMQ socket send has a bit of a strange implementation:
15 * @li if sending works: the send_result_t has a value
16 * @li if sending doesn't work due to EAGAIN error: the send_result_t is empty (no value) and error should be EAGAIN
17 * @li if sending doesn't work for another reason: an exception is thrown
18 * This function only checks the behaviour due to the send result, not the thrown error.
19 */
20 21 PSendStatus::PSendStatus checkSendStatus(zmq::send_result_t res){
21
2/2
✓ Branch 0 (3→4) taken 20 times.
✓ Branch 1 (3→5) taken 1 times.
 21 if(res.has_value()){
22 20 return PSendStatus::OK;
23 }else{
24 1 int err = zmq_errno();
25
1/2
✓ Branch 0 (6→7) taken 1 times.
✗ Branch 1 (6→8) not taken.
 1 if(err == EAGAIN){
26 1 return PSendStatus::SOCKET_NOT_AVAILABLE;
27 }else{
28 ✗ std::cerr << "Unknown ZMQ error in send: '" << err << "'"<< std::endl;
29 ✗ return PSendStatus::BROKEN_BACKEND;
30 }
31 }
32 }
33
34 ///Check the recv result and convert it into PRecvStatus
35 /** @param res : result of the zmq recv
36 * @return corresponding PRecvStatus
37 * ZMQ socket recv has a bit of a strange implementation:
38 * @li if recv works: the recv_result_t has a value
39 * @li if recv doesn't work due to EAGAIN error: the recv_result_t is empty (no value) and error should be EAGAIN
40 * @li if recv doesn't work for another reason: an exception is thrown
41 * This function only checks the behaviour due to the recv result, not the thrown error.
42 */
43 22 PRecvStatus::PRecvStatus checkRecvStatus(zmq::recv_result_t res){
44
2/2
✓ Branch 0 (3→4) taken 20 times.
✓ Branch 1 (3→5) taken 2 times.
 22 if(res.has_value()){
45 20 return PRecvStatus::OK;
46 }else{
47 2 int err = zmq_errno();
48
1/2
✓ Branch 0 (6→7) taken 2 times.
✗ Branch 1 (6→8) not taken.
 2 if(err == EAGAIN){
49 2 return PRecvStatus::NO_MESSAGE_RECEIVED;
50 }else{
51 ✗ return PRecvStatus::BROKEN_SOCKET;
52 }
53 }
54 }
55
56 ///Convert a send flag into zmq flag
57 /** @param flag : generic PSendFlag
58 * @return corresponding zmq::send_flags
59 */
60 22 zmq::send_flags convertToSendFlag(PSendFlag::PSendFlag flag){
61
2/2
✓ Branch 0 (2→3) taken 12 times.
✓ Branch 1 (2→4) taken 10 times.
 22 if(flag == PSendFlag::NON_BLOCK){return zmq::send_flags::dontwait;}
62 10 else{return zmq::send_flags::none;}
63 }
64
65 ///Convert a recv flag into zmq flag
66 /** @param flag : generic PRecvFlag
67 * @return corresponding zmq::recv_flags
68 */
69 24 zmq::recv_flags convertToRecvFlag(PRecvFlag::PRecvFlag flag){
70
2/2
✓ Branch 0 (2→3) taken 14 times.
✓ Branch 1 (2→4) taken 10 times.
 24 if(flag == PRecvFlag::NON_BLOCK){return zmq::recv_flags::dontwait;}
71 10 else{return zmq::recv_flags::none;}
72 }
73
74 ///Create param for a client socket
75 /** @param context : zmq context where to create socket
76 * @param type : type of the connection (ZMQ_PULL, ZMQ_PUSH, etc)
77 * @param nbBufferMessage : number of messages to be buffered
78 * @param bufferSizeByte : size of the zmq buffer in bytes
79 * @param threadAffinity : bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
80 * @param dataRate : expected data rate (in kilobytes per second)
81 * @param linger : linger period for socket shutdown
82 * @return corresponding PZmqParam
83 */
84 7 PZmqParam pzmq_createParamClient(int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate, int linger, int immediate){
85 PZmqParam param;
86 7 param.type = type;
87 7 param.nbBufferMessage = nbBufferMessage;
88 7 param.bufferSizeByte = bufferSizeByte;
89 7 param.threadAffinity = threadAffinity;
90 7 param.dataRate = dataRate;
91 7 param.linger = linger;
92 7 param.immediate = immediate; //EAGAIN if no peer connected
93 7 return param;
94 }
95
96 ///Create param for a client socket
97 /** @param type : type of the connection (ZMQ_PULL, ZMQ_PUSH, etc)
98 * @param nbBufferMessage : number of messages to be buffered
99 * @param bufferSizeByte : size of the zmq buffer in bytes
100 * @param threadAffinity : bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
101 * @param dataRate : expected data rate (in kilobytes per second)
102 * @param linger : linger period for socket shutdown
103 * @return corresponding PZmqParam
104 */
105 6 PZmqParam pzmq_createParamServer(int type, int nbBufferMessage, int bufferSizeByte, size_t threadAffinity, ssize_t dataRate, int linger, int immediate){
106 PZmqParam param;
107 6 param.type = type;
108 6 param.nbBufferMessage = nbBufferMessage;
109 6 param.bufferSizeByte = bufferSizeByte;
110 6 param.threadAffinity = threadAffinity;
111 6 param.dataRate = dataRate;
112 6 param.linger = linger;
113 6 param.immediate = immediate; //EAGAIN if no peer connected
114 6 return param;
115 }
116
117 ///Default constructor of PZmqSocket
118 13 PZmqSocket::PZmqSocket()
119 13 :p_socket(NULL)
120 {
121
122 13 }
123
124 ///Destructor of PZmqSocket
125 13 PZmqSocket::~PZmqSocket(){
126 13 close();
127 13 }
128
129 ///Create a client socket
130 /** @param context : zmq context where to create socket
131 * @param socketParam : parameters of the server (hostname, port), the client has to connect to
132 * @param extraParam : extra customisable parameters for the creation of the socket (depends on the backend)
133 * @return true if the socket has been created, false otherwise
134 */
135 7 bool PZmqSocket::createClientSocket(zmq::context_t & context, const PSocketParam & socketParam, const Param & extraParam){
136 14 p_socket = pzmq_createClientSocket(context, socketParam.hostname, socketParam.port, extraParam.type, extraParam.nbBufferMessage,
137 7 extraParam.bufferSizeByte, extraParam.threadAffinity, extraParam.dataRate, extraParam.immediate);
138 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
139 7 p_socket->set(zmq::sockopt::rcvtimeo, socketParam.recvTimeOut);
140 7 p_socket->set(zmq::sockopt::sndtimeo, socketParam.sendTimeOut);
141 7 p_socket->set(zmq::sockopt::linger, extraParam.linger); //number of ms to stop
142 7 p_socket->set(zmq::sockopt::immediate, extraParam.immediate); //EAGAIN if no peer connected
143 #else
144 p_socket->setsockopt(ZMQ_RCVTIMEO, &socketParam.recvTimeOut, sizeof(int));
145 p_socket->setsockopt(ZMQ_SNDTIMEO, &socketParam.sendTimeOut, sizeof(int));
146 p_socket->setsockopt(ZMQ_LINGER, &extraParam.linger, sizeof(int)); //number of ms to stop
147 p_socket->setsockopt(ZMQ_IMMEDIATE, &extraParam.immediate, sizeof(int)); //EAGAIN if no peer connected
148 #endif
149 7 return p_socket != NULL;
150 }
151
152 ///Create a server socket
153 /** @param context : zmq context where to create socket
154 * @param socketParam : parameters of the server (hostname, port), the client has to connect to
155 * @param extraParam : extra customisable parameters for the creation of the socket (depends on the backend)
156 * @return true if the socket has been created, false otherwise
157 */
158 6 bool PZmqSocket::createServerSocket(zmq::context_t & context, const PSocketParam & socketParam, const Param & extraParam){
159 12 p_socket = pzmq_createServerSocket(context, socketParam.port, extraParam.type, extraParam.nbBufferMessage, extraParam.bufferSizeByte,
160 6 extraParam.threadAffinity, extraParam.dataRate);
161 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
162 6 p_socket->set(zmq::sockopt::rcvtimeo, socketParam.recvTimeOut);
163 6 p_socket->set(zmq::sockopt::sndtimeo, socketParam.sendTimeOut);
164 6 p_socket->set(zmq::sockopt::linger, extraParam.linger); //number of ms to stop
165 6 p_socket->set(zmq::sockopt::immediate, extraParam.immediate); //EAGAIN if no peer connected
166 #else
167 p_socket->setsockopt(ZMQ_RCVTIMEO, &socketParam.recvTimeOut, sizeof(int));
168 p_socket->setsockopt(ZMQ_SNDTIMEO, &socketParam.sendTimeOut, sizeof(int));
169 p_socket->setsockopt(ZMQ_LINGER, &extraParam.linger, sizeof(int)); //number of ms to stop
170 p_socket->setsockopt(ZMQ_IMMEDIATE, &extraParam.immediate, sizeof(int)); //EAGAIN if no peer connected
171 #endif
172 6 return p_socket != NULL;
173 }
174
175 ///Send data with the socket
176 /** @param msg : message to be sent with the socket
177 * @param flag : sending flag (BLOCK, NON_BLOCK)
178 * @return status of the send
179 */
180 22 PSendStatus::PSendStatus PZmqSocket::sendMsg(Message & msg, PSendFlag::PSendFlag flag){
181 try {
182
2/2
✓ Branch 0 (2→3) taken 22 times.
✓ Branch 2 (3→4) taken 21 times.
 22 zmq::send_result_t result = p_socket->send(msg, convertToSendFlag(flag));
183
1/1
✓ Branch 0 (4→5) taken 21 times.
 21 return checkSendStatus(result);
184
1/2
✗ Branch 0 (9→10) not taken.
✓ Branch 1 (9→11) taken 1 times.
 1 }catch(const zmq::error_t& e) {
185
1/2
✗ Branch 0 (13→14) not taken.
✓ Branch 1 (13→15) taken 1 times.
 1 if(e.num() == ENOTSOCK){
186 ✗ return PSendStatus::BROKEN_SOCKET;
187
2/10
✗ Branch 0 (16→17) not taken.
✓ Branch 1 (16→23) taken 1 times.
✗ Branch 2 (18→19) not taken.
✗ Branch 3 (18→23) not taken.
✗ Branch 4 (20→21) not taken.
✗ Branch 5 (20→23) not taken.
✗ Branch 6 (22→23) not taken.
✗ Branch 7 (22→24) not taken.
✓ Branch 8 (25→26) taken 1 times.
✗ Branch 9 (25→27) not taken.
 1 }else if(e.num() == ENOTSUP || e.num() == EFSM || e.num() == ETERM || e.num() == EINVAL){
188 1 return PSendStatus::BROKEN_BACKEND;
189 ✗ }else if(e.num() == EINTR){
190 ✗ return PSendStatus::SIGNAL_INTERRUPTION;
191 ✗ }else if(e.num() == EHOSTUNREACH){
192 ✗ return PSendStatus::NO_ROUTE_TO_RECEIVER;
193 }else{
194 ✗ std::cerr << "Unknown ZMQ error in send: " << e.what() << std::endl;
195 ✗ return PSendStatus::BROKEN_BACKEND;
196 }
197 1 }
198 }
199
200 ///Receive data with the socket
201 /** @param msg : message to be received with the socket
202 * @param flag : receiving flag (BLOCK, NON_BLOCK)
203 * @return status of the recv
204 */
205 24 PRecvStatus::PRecvStatus PZmqSocket::recvMsg(Message & msg, PRecvFlag::PRecvFlag flag){
206 try {
207
2/2
✓ Branch 0 (2→3) taken 24 times.
✓ Branch 2 (3→4) taken 22 times.
 24 zmq::recv_result_t result = p_socket->recv(msg, convertToRecvFlag(flag));
208
1/1
✓ Branch 0 (4→5) taken 22 times.
 22 return checkRecvStatus(result);
209
1/2
✗ Branch 0 (9→10) not taken.
✓ Branch 1 (9→11) taken 2 times.
 2 }catch(const zmq::error_t& e){
210
1/2
✗ Branch 0 (13→14) not taken.
✓ Branch 1 (13→19) taken 2 times.
 2 if(e.num() == ENOTSOCK){
211 ✗ std::cerr << "ZMQ error in recv: " << e.what() << std::endl;
212 ✗ return PRecvStatus::BROKEN_SOCKET;
213 }
214
2/10
✗ Branch 0 (20→21) not taken.
✓ Branch 1 (20→27) taken 2 times.
✗ Branch 2 (22→23) not taken.
✗ Branch 3 (22→27) not taken.
✗ Branch 4 (24→25) not taken.
✗ Branch 5 (24→27) not taken.
✗ Branch 6 (26→27) not taken.
✗ Branch 7 (26→28) not taken.
✓ Branch 8 (29→30) taken 2 times.
✗ Branch 9 (29→35) not taken.
 2 else if(e.num() == ENOTSUP || e.num() == EFSM || e.num() == ETERM || e.num() == EINVAL){
215
3/3
✓ Branch 0 (30→31) taken 2 times.
✓ Branch 2 (32→33) taken 2 times.
✓ Branch 4 (33→34) taken 2 times.
 2 std::cerr << "ZMQ error in recv: " << e.what() << std::endl;
216 2 return PRecvStatus::BROKEN_BACKEND;
217 }
218 ✗ else if(e.num() == EINTR){
219 ✗ std::cerr << "ZMQ error in recv: " << e.what() << std::endl;
220 ✗ return PRecvStatus::SIGNAL_INTERRUPTION;
221 }
222 else{
223 ✗ std::cerr << "Unknown ZMQ error in recv: " << e.what() << std::endl;
224 ✗ return PRecvStatus::BROKEN_BACKEND;
225 }
226 2 }
227 }
228
229 ///Check if the socket is connected
230 /** @return true if the socket is connected, false otherwise
231 */
232 1 bool PZmqSocket::isConnected() const{
233
1/2
✓ Branch 0 (2→3) taken 1 times.
✗ Branch 1 (2→5) not taken.
 1 if(p_socket != NULL){
234 1 return p_socket->handle() != NULL;
235 }
236 ✗ return false;
237 }
238
239 ///Close the socket
240 38 void PZmqSocket::close(){
241
2/2
✓ Branch 0 (2→3) taken 13 times.
✓ Branch 1 (2→8) taken 25 times.
 38 if(p_socket != NULL){
242 13 p_socket->close();
243
1/2
✓ Branch 0 (4→5) taken 13 times.
✗ Branch 1 (4→7) not taken.
 13 delete p_socket;
244 13 p_socket = NULL;
245 }
246 38 }
247
248 ///Default constructor of PZmqSocketGenerator setting the number of threads for zmq I/O to 1
249 11 PZmqSocketGenerator::PZmqSocketGenerator()
250 11 :p_context(1)
251 {
252
253 11 }
254
255 ///Create a client parameter
256 /** @return corresponding PZmqSocketGenerator::Param (or PZmqParam)
257 */
258 7 PZmqSocketGenerator::Param PZmqSocketGenerator::client(){
259 7 return pzmq_createParamClient(ZMQ_PUSH);
260 }
261
262 ///Create a server parameter
263 /** @return corresponding PZmqSocketGenerator::Param (or PZmqParam)
264 */
265 6 PZmqSocketGenerator::Param PZmqSocketGenerator::server(){
266 6 return pzmq_createParamServer(ZMQ_PULL);
267 }
268
269 ///Create a client socket
270 /** @param[out] socket : socket to be created
271 * @param socketParam : parameters of the server (hostname, port), the client has to connect to
272 * @param port : port to be used for the connection
273 * @param param : extra customisable parameters for the creation of the socket (depends on the backend)
274 * @return true if the socket has been created, false otherwise
275 */
276 7 bool PZmqSocketGenerator::createClientSocket(PZmqSocketGenerator::Socket & socket, const PSocketParam & socketParam, const PZmqParam & param){
277 7 return socket.createClientSocket(p_context, socketParam, param);
278 }
279
280 ///Create a server socket
281 /** @param[out] socket : socket to be created
282 * @param socketParam : parameters of the server (hostname, port), the client has to connect to
283 * @param param : extra customisable parameters for the creation of the socket (depends on the backend)
284 * @return true if the socket has been created, false otherwise
285 */
286 6 bool PZmqSocketGenerator::createServerSocket(PZmqSocketGenerator::Socket & socket, const PSocketParam & socketParam, const PZmqParam & param){
287 6 return socket.createServerSocket(p_context, socketParam, param);
288 }
289
290 ///Copy current backend message data into mock message
291 /** @param[out] mockMsg : mock message
292 * @param msg : message of the current backend to be converted
293 */
294 ✗ void PZmqSocketGenerator::msgToMock(DataStreamMsg & mockMsg, const PZmqSocketGenerator::Message & msg){
295 ✗ size_t dataSize(msg.size());
296 ✗ mockMsg.resize(dataSize);
297 ✗ memcpy(mockMsg.data(), (const void*)msg.data(), dataSize);
298 ✗ }
299
300 ///Copy mock message data into current backend message
301 /** @param[out] msg : message of the current backend to be converted
302 * @param mockMsg : mock message
303 */
304 ✗ void PZmqSocketGenerator::mockToMsg(PZmqSocketGenerator::Message & msg, DataStreamMsg & mockMsg){
305 ✗ size_t dataSize(mockMsg.size());
306 ✗ msg.rebuild(dataSize);
307 ✗ memcpy((void*)msg.data(), mockMsg.data(), dataSize);
308 ✗ }
309