Commit | Line | Data |
---|---|---|
00e2e675 DG |
1 | /* |
2 | * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com> | |
3 | * | |
4 | * This program is free software; you can redistribute it and/or modify it | |
5 | * under the terms of the GNU General Public License, version 2 only, as | |
6 | * published by the Free Software Foundation. | |
7 | * | |
8 | * This program is distributed in the hope that it will be useful, but WITHOUT | |
9 | * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or | |
10 | * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for | |
11 | * more details. | |
12 | * | |
13 | * You should have received a copy of the GNU General Public License along with | |
14 | * this program; if not, write to the Free Software Foundation, Inc., 51 | |
15 | * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. | |
16 | */ | |
17 | ||
18 | #define _GNU_SOURCE | |
19 | #include <assert.h> | |
20 | #include <stdio.h> | |
21 | #include <stdlib.h> | |
22 | #include <string.h> | |
23 | #include <sys/stat.h> | |
77c7c900 | 24 | #include <inttypes.h> |
00e2e675 DG |
25 | |
26 | #include <common/common.h> | |
27 | #include <common/defaults.h> | |
f263b7fd | 28 | #include <common/compat/endian.h> |
00e2e675 | 29 | #include <common/sessiond-comm/relayd.h> |
50adc264 | 30 | #include <common/index/ctf-index.h> |
00e2e675 DG |
31 | |
32 | #include "relayd.h" | |
33 | ||
34 | /* | |
35 | * Send command. Fill up the header and append the data. | |
36 | */ | |
6151a90f | 37 | static int send_command(struct lttcomm_relayd_sock *rsock, |
7c9534d6 | 38 | enum lttcomm_relayd_command cmd, void *data, size_t size, |
00e2e675 DG |
39 | int flags) |
40 | { | |
41 | int ret; | |
42 | struct lttcomm_relayd_hdr header; | |
43 | char *buf; | |
44 | uint64_t buf_size = sizeof(header); | |
45 | ||
f96e4545 MD |
46 | if (rsock->sock.fd < 0) { |
47 | return -ECONNRESET; | |
48 | } | |
49 | ||
00e2e675 DG |
50 | if (data) { |
51 | buf_size += size; | |
52 | } | |
53 | ||
54 | buf = zmalloc(buf_size); | |
55 | if (buf == NULL) { | |
56 | PERROR("zmalloc relayd send command buf"); | |
57 | ret = -1; | |
58 | goto alloc_error; | |
59 | } | |
60 | ||
53efb85a | 61 | memset(&header, 0, sizeof(header)); |
00e2e675 DG |
62 | header.cmd = htobe32(cmd); |
63 | header.data_size = htobe64(size); | |
64 | ||
65 | /* Zeroed for now since not used. */ | |
66 | header.cmd_version = 0; | |
67 | header.circuit_id = 0; | |
68 | ||
69 | /* Prepare buffer to send. */ | |
70 | memcpy(buf, &header, sizeof(header)); | |
71 | if (data) { | |
72 | memcpy(buf + sizeof(header), data, size); | |
73 | } | |
74 | ||
6151a90f | 75 | ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags); |
00e2e675 | 76 | if (ret < 0) { |
8994307f | 77 | ret = -errno; |
00e2e675 DG |
78 | goto error; |
79 | } | |
80 | ||
633d0084 | 81 | DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size); |
00e2e675 DG |
82 | |
83 | error: | |
84 | free(buf); | |
85 | alloc_error: | |
86 | return ret; | |
87 | } | |
88 | ||
89 | /* | |
90 | * Receive reply data on socket. This MUST be call after send_command or else | |
91 | * could result in unexpected behavior(s). | |
92 | */ | |
6151a90f | 93 | static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size) |
00e2e675 DG |
94 | { |
95 | int ret; | |
96 | ||
f96e4545 MD |
97 | if (rsock->sock.fd < 0) { |
98 | return -ECONNRESET; | |
99 | } | |
100 | ||
8fd623e0 | 101 | DBG3("Relayd waiting for reply of size %zu", size); |
00e2e675 | 102 | |
6151a90f | 103 | ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0); |
20275fe8 DG |
104 | if (ret <= 0 || ret != size) { |
105 | if (ret == 0) { | |
106 | /* Orderly shutdown. */ | |
6151a90f | 107 | DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd); |
20275fe8 | 108 | } else { |
8fd623e0 | 109 | DBG("Receiving reply failed on sock %d for size %zu with ret %d", |
6151a90f | 110 | rsock->sock.fd, size, ret); |
20275fe8 DG |
111 | } |
112 | /* Always return -1 here and the caller can use errno. */ | |
113 | ret = -1; | |
00e2e675 DG |
114 | goto error; |
115 | } | |
116 | ||
117 | error: | |
118 | return ret; | |
119 | } | |
120 | ||
d3e2ba59 JD |
121 | /* |
122 | * Starting at 2.4, RELAYD_CREATE_SESSION takes additional parameters to | |
123 | * support the live reading capability. | |
124 | */ | |
125 | static int relayd_create_session_2_4(struct lttcomm_relayd_sock *rsock, | |
126 | uint64_t *session_id, char *session_name, char *hostname, | |
7d2f7452 | 127 | int session_live_timer, unsigned int snapshot) |
d3e2ba59 JD |
128 | { |
129 | int ret; | |
130 | struct lttcomm_relayd_create_session_2_4 msg; | |
131 | ||
66f9eb4e MD |
132 | if (lttng_strncpy(msg.session_name, session_name, |
133 | sizeof(msg.session_name))) { | |
134 | ret = -1; | |
135 | goto error; | |
136 | } | |
137 | if (lttng_strncpy(msg.hostname, hostname, sizeof(msg.hostname))) { | |
138 | ret = -1; | |
139 | goto error; | |
140 | } | |
d3e2ba59 | 141 | msg.live_timer = htobe32(session_live_timer); |
7d2f7452 | 142 | msg.snapshot = htobe32(snapshot); |
d3e2ba59 JD |
143 | |
144 | /* Send command */ | |
145 | ret = send_command(rsock, RELAYD_CREATE_SESSION, &msg, sizeof(msg), 0); | |
146 | if (ret < 0) { | |
147 | goto error; | |
148 | } | |
149 | ||
150 | error: | |
151 | return ret; | |
152 | } | |
153 | ||
154 | /* | |
155 | * RELAYD_CREATE_SESSION from 2.1 to 2.3. | |
156 | */ | |
157 | static int relayd_create_session_2_1(struct lttcomm_relayd_sock *rsock, | |
158 | uint64_t *session_id) | |
159 | { | |
160 | int ret; | |
161 | ||
162 | /* Send command */ | |
163 | ret = send_command(rsock, RELAYD_CREATE_SESSION, NULL, 0, 0); | |
164 | if (ret < 0) { | |
165 | goto error; | |
166 | } | |
167 | ||
168 | error: | |
169 | return ret; | |
170 | } | |
171 | ||
c5b6f4f0 DG |
172 | /* |
173 | * Send a RELAYD_CREATE_SESSION command to the relayd with the given socket and | |
174 | * set session_id of the relayd if we have a successful reply from the relayd. | |
175 | * | |
20275fe8 DG |
176 | * On success, return 0 else a negative value which is either an errno error or |
177 | * a lttng error code from the relayd. | |
c5b6f4f0 | 178 | */ |
d3e2ba59 | 179 | int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id, |
7d2f7452 DG |
180 | char *session_name, char *hostname, int session_live_timer, |
181 | unsigned int snapshot) | |
c5b6f4f0 DG |
182 | { |
183 | int ret; | |
184 | struct lttcomm_relayd_status_session reply; | |
185 | ||
6151a90f | 186 | assert(rsock); |
c5b6f4f0 DG |
187 | assert(session_id); |
188 | ||
189 | DBG("Relayd create session"); | |
190 | ||
d3e2ba59 JD |
191 | switch(rsock->minor) { |
192 | case 1: | |
193 | case 2: | |
194 | case 3: | |
195 | ret = relayd_create_session_2_1(rsock, session_id); | |
31d74dc3 | 196 | break; |
d3e2ba59 JD |
197 | case 4: |
198 | default: | |
7d2f7452 DG |
199 | ret = relayd_create_session_2_4(rsock, session_id, session_name, |
200 | hostname, session_live_timer, snapshot); | |
31d74dc3 | 201 | break; |
d3e2ba59 JD |
202 | } |
203 | ||
c5b6f4f0 DG |
204 | if (ret < 0) { |
205 | goto error; | |
206 | } | |
207 | ||
20275fe8 | 208 | /* Receive response */ |
6151a90f | 209 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
c5b6f4f0 DG |
210 | if (ret < 0) { |
211 | goto error; | |
212 | } | |
213 | ||
214 | reply.session_id = be64toh(reply.session_id); | |
215 | reply.ret_code = be32toh(reply.ret_code); | |
216 | ||
217 | /* Return session id or negative ret code. */ | |
218 | if (reply.ret_code != LTTNG_OK) { | |
bb63afd9 DG |
219 | ret = -1; |
220 | ERR("Relayd create session replied error %d", reply.ret_code); | |
c5b6f4f0 DG |
221 | goto error; |
222 | } else { | |
223 | ret = 0; | |
224 | *session_id = reply.session_id; | |
225 | } | |
226 | ||
227 | DBG("Relayd session created with id %" PRIu64, reply.session_id); | |
228 | ||
229 | error: | |
230 | return ret; | |
231 | } | |
232 | ||
00e2e675 DG |
233 | /* |
234 | * Add stream on the relayd and assign stream handle to the stream_id argument. | |
235 | * | |
236 | * On success return 0 else return ret_code negative value. | |
237 | */ | |
6151a90f | 238 | int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name, |
0f907de1 JD |
239 | const char *pathname, uint64_t *stream_id, |
240 | uint64_t tracefile_size, uint64_t tracefile_count) | |
00e2e675 DG |
241 | { |
242 | int ret; | |
243 | struct lttcomm_relayd_add_stream msg; | |
0f907de1 | 244 | struct lttcomm_relayd_add_stream_2_2 msg_2_2; |
00e2e675 DG |
245 | struct lttcomm_relayd_status_stream reply; |
246 | ||
247 | /* Code flow error. Safety net. */ | |
6151a90f | 248 | assert(rsock); |
00e2e675 DG |
249 | assert(channel_name); |
250 | assert(pathname); | |
251 | ||
252 | DBG("Relayd adding stream for channel name %s", channel_name); | |
253 | ||
0f907de1 JD |
254 | /* Compat with relayd 2.1 */ |
255 | if (rsock->minor == 1) { | |
53efb85a | 256 | memset(&msg, 0, sizeof(msg)); |
0f907de1 JD |
257 | strncpy(msg.channel_name, channel_name, sizeof(msg.channel_name)); |
258 | strncpy(msg.pathname, pathname, sizeof(msg.pathname)); | |
00e2e675 | 259 | |
0f907de1 JD |
260 | /* Send command */ |
261 | ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0); | |
262 | if (ret < 0) { | |
263 | goto error; | |
264 | } | |
265 | } else { | |
53efb85a | 266 | memset(&msg_2_2, 0, sizeof(msg_2_2)); |
0f907de1 JD |
267 | /* Compat with relayd 2.2+ */ |
268 | strncpy(msg_2_2.channel_name, channel_name, sizeof(msg_2_2.channel_name)); | |
269 | strncpy(msg_2_2.pathname, pathname, sizeof(msg_2_2.pathname)); | |
270 | msg_2_2.tracefile_size = htobe64(tracefile_size); | |
271 | msg_2_2.tracefile_count = htobe64(tracefile_count); | |
272 | ||
273 | /* Send command */ | |
274 | ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg_2_2, sizeof(msg_2_2), 0); | |
275 | if (ret < 0) { | |
276 | goto error; | |
277 | } | |
00e2e675 DG |
278 | } |
279 | ||
633d0084 | 280 | /* Waiting for reply */ |
6151a90f | 281 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
00e2e675 DG |
282 | if (ret < 0) { |
283 | goto error; | |
284 | } | |
285 | ||
286 | /* Back to host bytes order. */ | |
287 | reply.handle = be64toh(reply.handle); | |
288 | reply.ret_code = be32toh(reply.ret_code); | |
289 | ||
290 | /* Return session id or negative ret code. */ | |
f73fabfd | 291 | if (reply.ret_code != LTTNG_OK) { |
bb63afd9 DG |
292 | ret = -1; |
293 | ERR("Relayd add stream replied error %d", reply.ret_code); | |
00e2e675 DG |
294 | } else { |
295 | /* Success */ | |
296 | ret = 0; | |
297 | *stream_id = reply.handle; | |
298 | } | |
299 | ||
77c7c900 MD |
300 | DBG("Relayd stream added successfully with handle %" PRIu64, |
301 | reply.handle); | |
00e2e675 DG |
302 | |
303 | error: | |
304 | return ret; | |
305 | } | |
306 | ||
a4baae1b JD |
307 | /* |
308 | * Inform the relay that all the streams for the current channel has been sent. | |
309 | * | |
310 | * On success return 0 else return ret_code negative value. | |
311 | */ | |
312 | int relayd_streams_sent(struct lttcomm_relayd_sock *rsock) | |
313 | { | |
314 | int ret; | |
315 | struct lttcomm_relayd_generic_reply reply; | |
316 | ||
317 | /* Code flow error. Safety net. */ | |
318 | assert(rsock); | |
319 | ||
320 | DBG("Relayd sending streams sent."); | |
321 | ||
322 | /* This feature was introduced in 2.4, ignore it for earlier versions. */ | |
323 | if (rsock->minor < 4) { | |
324 | ret = 0; | |
325 | goto end; | |
326 | } | |
327 | ||
328 | /* Send command */ | |
329 | ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0); | |
330 | if (ret < 0) { | |
331 | goto error; | |
332 | } | |
333 | ||
334 | /* Waiting for reply */ | |
335 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
336 | if (ret < 0) { | |
337 | goto error; | |
338 | } | |
339 | ||
340 | /* Back to host bytes order. */ | |
341 | reply.ret_code = be32toh(reply.ret_code); | |
342 | ||
343 | /* Return session id or negative ret code. */ | |
344 | if (reply.ret_code != LTTNG_OK) { | |
345 | ret = -1; | |
346 | ERR("Relayd streams sent replied error %d", reply.ret_code); | |
347 | goto error; | |
348 | } else { | |
349 | /* Success */ | |
350 | ret = 0; | |
351 | } | |
352 | ||
353 | DBG("Relayd streams sent success"); | |
354 | ||
355 | error: | |
356 | end: | |
357 | return ret; | |
358 | } | |
359 | ||
00e2e675 DG |
360 | /* |
361 | * Check version numbers on the relayd. | |
d4519fa3 JD |
362 | * If major versions are compatible, we assign minor_to_use to the |
363 | * minor version of the procotol we are going to use for this session. | |
00e2e675 DG |
364 | * |
365 | * Return 0 if compatible else negative value. | |
366 | */ | |
6151a90f | 367 | int relayd_version_check(struct lttcomm_relayd_sock *rsock) |
00e2e675 DG |
368 | { |
369 | int ret; | |
092b6259 | 370 | struct lttcomm_relayd_version msg; |
00e2e675 DG |
371 | |
372 | /* Code flow error. Safety net. */ | |
6151a90f | 373 | assert(rsock); |
00e2e675 | 374 | |
6151a90f JD |
375 | DBG("Relayd version check for major.minor %u.%u", rsock->major, |
376 | rsock->minor); | |
00e2e675 | 377 | |
53efb85a | 378 | memset(&msg, 0, sizeof(msg)); |
092b6259 | 379 | /* Prepare network byte order before transmission. */ |
6151a90f JD |
380 | msg.major = htobe32(rsock->major); |
381 | msg.minor = htobe32(rsock->minor); | |
092b6259 | 382 | |
00e2e675 | 383 | /* Send command */ |
6151a90f | 384 | ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0); |
00e2e675 DG |
385 | if (ret < 0) { |
386 | goto error; | |
387 | } | |
388 | ||
20275fe8 | 389 | /* Receive response */ |
6151a90f | 390 | ret = recv_reply(rsock, (void *) &msg, sizeof(msg)); |
00e2e675 DG |
391 | if (ret < 0) { |
392 | goto error; | |
393 | } | |
394 | ||
395 | /* Set back to host bytes order */ | |
092b6259 DG |
396 | msg.major = be32toh(msg.major); |
397 | msg.minor = be32toh(msg.minor); | |
398 | ||
399 | /* | |
400 | * Only validate the major version. If the other side is higher, | |
401 | * communication is not possible. Only major version equal can talk to each | |
402 | * other. If the minor version differs, the lowest version is used by both | |
403 | * sides. | |
092b6259 | 404 | */ |
6151a90f | 405 | if (msg.major != rsock->major) { |
d4519fa3 JD |
406 | /* Not compatible */ |
407 | ret = -1; | |
408 | DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)", | |
6151a90f | 409 | msg.major, rsock->major); |
092b6259 | 410 | goto error; |
00e2e675 DG |
411 | } |
412 | ||
092b6259 | 413 | /* |
6151a90f JD |
414 | * If the relayd's minor version is higher, it will adapt to our version so |
415 | * we can continue to use the latest relayd communication data structure. | |
416 | * If the received minor version is higher, the relayd should adapt to us. | |
092b6259 | 417 | */ |
6151a90f JD |
418 | if (rsock->minor > msg.minor) { |
419 | rsock->minor = msg.minor; | |
d4519fa3 | 420 | } |
092b6259 | 421 | |
d4519fa3 JD |
422 | /* Version number compatible */ |
423 | DBG2("Relayd version is compatible, using protocol version %u.%u", | |
6151a90f | 424 | rsock->major, rsock->minor); |
d4519fa3 | 425 | ret = 0; |
00e2e675 DG |
426 | |
427 | error: | |
428 | return ret; | |
429 | } | |
430 | ||
00e2e675 DG |
431 | /* |
432 | * Add stream on the relayd and assign stream handle to the stream_id argument. | |
433 | * | |
434 | * On success return 0 else return ret_code negative value. | |
435 | */ | |
6151a90f | 436 | int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len) |
00e2e675 DG |
437 | { |
438 | int ret; | |
439 | ||
440 | /* Code flow error. Safety net. */ | |
6151a90f | 441 | assert(rsock); |
00e2e675 | 442 | |
77c7c900 | 443 | DBG("Relayd sending metadata of size %zu", len); |
00e2e675 DG |
444 | |
445 | /* Send command */ | |
6151a90f | 446 | ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0); |
00e2e675 DG |
447 | if (ret < 0) { |
448 | goto error; | |
449 | } | |
450 | ||
451 | DBG2("Relayd metadata added successfully"); | |
452 | ||
453 | /* | |
454 | * After that call, the metadata data MUST be sent to the relayd so the | |
455 | * receive size on the other end matches the len of the metadata packet | |
633d0084 | 456 | * header. This is why we don't wait for a reply here. |
00e2e675 DG |
457 | */ |
458 | ||
459 | error: | |
460 | return ret; | |
461 | } | |
462 | ||
463 | /* | |
6151a90f | 464 | * Connect to relay daemon with an allocated lttcomm_relayd_sock. |
00e2e675 | 465 | */ |
6151a90f | 466 | int relayd_connect(struct lttcomm_relayd_sock *rsock) |
00e2e675 DG |
467 | { |
468 | /* Code flow error. Safety net. */ | |
6151a90f | 469 | assert(rsock); |
00e2e675 | 470 | |
f96e4545 MD |
471 | if (!rsock->sock.ops) { |
472 | /* | |
473 | * Attempting a connect on a non-initialized socket. | |
474 | */ | |
475 | return -ECONNRESET; | |
476 | } | |
477 | ||
00e2e675 DG |
478 | DBG3("Relayd connect ..."); |
479 | ||
6151a90f | 480 | return rsock->sock.ops->connect(&rsock->sock); |
00e2e675 DG |
481 | } |
482 | ||
483 | /* | |
6151a90f | 484 | * Close relayd socket with an allocated lttcomm_relayd_sock. |
ffe60014 DG |
485 | * |
486 | * If no socket operations are found, simply return 0 meaning that everything | |
487 | * is fine. Without operations, the socket can not possibly be opened or used. | |
488 | * This is possible if the socket was allocated but not created. However, the | |
489 | * caller could simply use it to store a valid file descriptor for instance | |
490 | * passed over a Unix socket and call this to cleanup but still without a valid | |
491 | * ops pointer. | |
492 | * | |
493 | * Return the close returned value. On error, a negative value is usually | |
494 | * returned back from close(2). | |
00e2e675 | 495 | */ |
6151a90f | 496 | int relayd_close(struct lttcomm_relayd_sock *rsock) |
00e2e675 | 497 | { |
ffe60014 DG |
498 | int ret; |
499 | ||
00e2e675 | 500 | /* Code flow error. Safety net. */ |
6151a90f | 501 | assert(rsock); |
00e2e675 | 502 | |
ffe60014 | 503 | /* An invalid fd is fine, return success. */ |
6151a90f | 504 | if (rsock->sock.fd < 0) { |
ffe60014 DG |
505 | ret = 0; |
506 | goto end; | |
507 | } | |
508 | ||
6151a90f | 509 | DBG3("Relayd closing socket %d", rsock->sock.fd); |
00e2e675 | 510 | |
6151a90f JD |
511 | if (rsock->sock.ops) { |
512 | ret = rsock->sock.ops->close(&rsock->sock); | |
ffe60014 DG |
513 | } else { |
514 | /* Default call if no specific ops found. */ | |
6151a90f | 515 | ret = close(rsock->sock.fd); |
ffe60014 DG |
516 | if (ret < 0) { |
517 | PERROR("relayd_close default close"); | |
518 | } | |
519 | } | |
f96e4545 | 520 | rsock->sock.fd = -1; |
ffe60014 DG |
521 | |
522 | end: | |
523 | return ret; | |
00e2e675 DG |
524 | } |
525 | ||
526 | /* | |
527 | * Send data header structure to the relayd. | |
528 | */ | |
6151a90f | 529 | int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock, |
00e2e675 DG |
530 | struct lttcomm_relayd_data_hdr *hdr, size_t size) |
531 | { | |
532 | int ret; | |
533 | ||
534 | /* Code flow error. Safety net. */ | |
6151a90f | 535 | assert(rsock); |
00e2e675 DG |
536 | assert(hdr); |
537 | ||
f96e4545 MD |
538 | if (rsock->sock.fd < 0) { |
539 | return -ECONNRESET; | |
540 | } | |
541 | ||
8fd623e0 | 542 | DBG3("Relayd sending data header of size %zu", size); |
00e2e675 DG |
543 | |
544 | /* Again, safety net */ | |
545 | if (size == 0) { | |
546 | size = sizeof(struct lttcomm_relayd_data_hdr); | |
547 | } | |
548 | ||
549 | /* Only send data header. */ | |
6151a90f | 550 | ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0); |
00e2e675 | 551 | if (ret < 0) { |
8994307f | 552 | ret = -errno; |
00e2e675 DG |
553 | goto error; |
554 | } | |
555 | ||
556 | /* | |
557 | * The data MUST be sent right after that command for the receive on the | |
558 | * other end to match the size in the header. | |
559 | */ | |
560 | ||
561 | error: | |
562 | return ret; | |
563 | } | |
173af62f DG |
564 | |
565 | /* | |
566 | * Send close stream command to the relayd. | |
567 | */ | |
6151a90f | 568 | int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, |
173af62f DG |
569 | uint64_t last_net_seq_num) |
570 | { | |
571 | int ret; | |
572 | struct lttcomm_relayd_close_stream msg; | |
573 | struct lttcomm_relayd_generic_reply reply; | |
574 | ||
575 | /* Code flow error. Safety net. */ | |
6151a90f | 576 | assert(rsock); |
173af62f | 577 | |
77c7c900 | 578 | DBG("Relayd closing stream id %" PRIu64, stream_id); |
173af62f | 579 | |
53efb85a | 580 | memset(&msg, 0, sizeof(msg)); |
173af62f DG |
581 | msg.stream_id = htobe64(stream_id); |
582 | msg.last_net_seq_num = htobe64(last_net_seq_num); | |
583 | ||
584 | /* Send command */ | |
6151a90f | 585 | ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0); |
173af62f DG |
586 | if (ret < 0) { |
587 | goto error; | |
588 | } | |
589 | ||
20275fe8 | 590 | /* Receive response */ |
6151a90f | 591 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
173af62f DG |
592 | if (ret < 0) { |
593 | goto error; | |
594 | } | |
595 | ||
596 | reply.ret_code = be32toh(reply.ret_code); | |
597 | ||
598 | /* Return session id or negative ret code. */ | |
f73fabfd | 599 | if (reply.ret_code != LTTNG_OK) { |
bb63afd9 DG |
600 | ret = -1; |
601 | ERR("Relayd close stream replied error %d", reply.ret_code); | |
173af62f DG |
602 | } else { |
603 | /* Success */ | |
604 | ret = 0; | |
605 | } | |
606 | ||
77c7c900 | 607 | DBG("Relayd close stream id %" PRIu64 " successfully", stream_id); |
173af62f DG |
608 | |
609 | error: | |
610 | return ret; | |
611 | } | |
c8f59ee5 DG |
612 | |
613 | /* | |
614 | * Check for data availability for a given stream id. | |
615 | * | |
6d805429 | 616 | * Return 0 if NOT pending, 1 if so and a negative value on error. |
c8f59ee5 | 617 | */ |
6151a90f | 618 | int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id, |
c8f59ee5 DG |
619 | uint64_t last_net_seq_num) |
620 | { | |
621 | int ret; | |
6d805429 | 622 | struct lttcomm_relayd_data_pending msg; |
c8f59ee5 DG |
623 | struct lttcomm_relayd_generic_reply reply; |
624 | ||
625 | /* Code flow error. Safety net. */ | |
6151a90f | 626 | assert(rsock); |
c8f59ee5 | 627 | |
6d805429 | 628 | DBG("Relayd data pending for stream id %" PRIu64, stream_id); |
c8f59ee5 | 629 | |
53efb85a | 630 | memset(&msg, 0, sizeof(msg)); |
c8f59ee5 DG |
631 | msg.stream_id = htobe64(stream_id); |
632 | msg.last_net_seq_num = htobe64(last_net_seq_num); | |
633 | ||
634 | /* Send command */ | |
6151a90f | 635 | ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg, |
c8f59ee5 DG |
636 | sizeof(msg), 0); |
637 | if (ret < 0) { | |
638 | goto error; | |
639 | } | |
640 | ||
20275fe8 | 641 | /* Receive response */ |
6151a90f | 642 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
c8f59ee5 DG |
643 | if (ret < 0) { |
644 | goto error; | |
645 | } | |
646 | ||
647 | reply.ret_code = be32toh(reply.ret_code); | |
648 | ||
649 | /* Return session id or negative ret code. */ | |
650 | if (reply.ret_code >= LTTNG_OK) { | |
bb63afd9 | 651 | ERR("Relayd data pending replied error %d", reply.ret_code); |
c8f59ee5 DG |
652 | } |
653 | ||
654 | /* At this point, the ret code is either 1 or 0 */ | |
655 | ret = reply.ret_code; | |
656 | ||
6d805429 | 657 | DBG("Relayd data is %s pending for stream id %" PRIu64, |
9dd26bb9 | 658 | ret == 1 ? "" : "NOT", stream_id); |
c8f59ee5 DG |
659 | |
660 | error: | |
661 | return ret; | |
662 | } | |
663 | ||
664 | /* | |
665 | * Check on the relayd side for a quiescent state on the control socket. | |
666 | */ | |
6151a90f | 667 | int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock, |
ad7051c0 | 668 | uint64_t metadata_stream_id) |
c8f59ee5 DG |
669 | { |
670 | int ret; | |
ad7051c0 | 671 | struct lttcomm_relayd_quiescent_control msg; |
c8f59ee5 DG |
672 | struct lttcomm_relayd_generic_reply reply; |
673 | ||
674 | /* Code flow error. Safety net. */ | |
6151a90f | 675 | assert(rsock); |
c8f59ee5 DG |
676 | |
677 | DBG("Relayd checking quiescent control state"); | |
678 | ||
53efb85a | 679 | memset(&msg, 0, sizeof(msg)); |
ad7051c0 DG |
680 | msg.stream_id = htobe64(metadata_stream_id); |
681 | ||
c8f59ee5 | 682 | /* Send command */ |
6151a90f | 683 | ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0); |
c8f59ee5 DG |
684 | if (ret < 0) { |
685 | goto error; | |
686 | } | |
687 | ||
20275fe8 | 688 | /* Receive response */ |
6151a90f | 689 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
c8f59ee5 DG |
690 | if (ret < 0) { |
691 | goto error; | |
692 | } | |
693 | ||
694 | reply.ret_code = be32toh(reply.ret_code); | |
695 | ||
696 | /* Return session id or negative ret code. */ | |
697 | if (reply.ret_code != LTTNG_OK) { | |
bb63afd9 DG |
698 | ret = -1; |
699 | ERR("Relayd quiescent control replied error %d", reply.ret_code); | |
c8f59ee5 DG |
700 | goto error; |
701 | } | |
702 | ||
703 | /* Control socket is quiescent */ | |
6d805429 | 704 | return 0; |
c8f59ee5 DG |
705 | |
706 | error: | |
707 | return ret; | |
708 | } | |
f7079f67 DG |
709 | |
710 | /* | |
711 | * Begin a data pending command for a specific session id. | |
712 | */ | |
6151a90f | 713 | int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id) |
f7079f67 DG |
714 | { |
715 | int ret; | |
716 | struct lttcomm_relayd_begin_data_pending msg; | |
717 | struct lttcomm_relayd_generic_reply reply; | |
718 | ||
719 | /* Code flow error. Safety net. */ | |
6151a90f | 720 | assert(rsock); |
f7079f67 DG |
721 | |
722 | DBG("Relayd begin data pending"); | |
723 | ||
53efb85a | 724 | memset(&msg, 0, sizeof(msg)); |
f7079f67 DG |
725 | msg.session_id = htobe64(id); |
726 | ||
727 | /* Send command */ | |
6151a90f | 728 | ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0); |
f7079f67 DG |
729 | if (ret < 0) { |
730 | goto error; | |
731 | } | |
732 | ||
20275fe8 | 733 | /* Receive response */ |
6151a90f | 734 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
f7079f67 DG |
735 | if (ret < 0) { |
736 | goto error; | |
737 | } | |
738 | ||
739 | reply.ret_code = be32toh(reply.ret_code); | |
740 | ||
741 | /* Return session id or negative ret code. */ | |
742 | if (reply.ret_code != LTTNG_OK) { | |
bb63afd9 DG |
743 | ret = -1; |
744 | ERR("Relayd begin data pending replied error %d", reply.ret_code); | |
f7079f67 DG |
745 | goto error; |
746 | } | |
747 | ||
748 | return 0; | |
749 | ||
750 | error: | |
751 | return ret; | |
752 | } | |
753 | ||
754 | /* | |
755 | * End a data pending command for a specific session id. | |
756 | * | |
757 | * Return 0 on success and set is_data_inflight to 0 if no data is being | |
758 | * streamed or 1 if it is the case. | |
759 | */ | |
6151a90f | 760 | int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id, |
f7079f67 DG |
761 | unsigned int *is_data_inflight) |
762 | { | |
af6c30b5 | 763 | int ret, recv_ret; |
f7079f67 DG |
764 | struct lttcomm_relayd_end_data_pending msg; |
765 | struct lttcomm_relayd_generic_reply reply; | |
766 | ||
767 | /* Code flow error. Safety net. */ | |
6151a90f | 768 | assert(rsock); |
f7079f67 DG |
769 | |
770 | DBG("Relayd end data pending"); | |
771 | ||
53efb85a | 772 | memset(&msg, 0, sizeof(msg)); |
f7079f67 DG |
773 | msg.session_id = htobe64(id); |
774 | ||
775 | /* Send command */ | |
6151a90f | 776 | ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0); |
f7079f67 DG |
777 | if (ret < 0) { |
778 | goto error; | |
779 | } | |
780 | ||
20275fe8 | 781 | /* Receive response */ |
6151a90f | 782 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); |
f7079f67 DG |
783 | if (ret < 0) { |
784 | goto error; | |
785 | } | |
786 | ||
af6c30b5 DG |
787 | recv_ret = be32toh(reply.ret_code); |
788 | if (recv_ret < 0) { | |
789 | ret = recv_ret; | |
f7079f67 DG |
790 | goto error; |
791 | } | |
792 | ||
af6c30b5 | 793 | *is_data_inflight = recv_ret; |
f7079f67 | 794 | |
af6c30b5 | 795 | DBG("Relayd end data pending is data inflight: %d", recv_ret); |
f7079f67 DG |
796 | |
797 | return 0; | |
798 | ||
799 | error: | |
800 | return ret; | |
801 | } | |
1c20f0e2 JD |
802 | |
803 | /* | |
804 | * Send index to the relayd. | |
805 | */ | |
806 | int relayd_send_index(struct lttcomm_relayd_sock *rsock, | |
50adc264 | 807 | struct ctf_packet_index *index, uint64_t relay_stream_id, |
1c20f0e2 JD |
808 | uint64_t net_seq_num) |
809 | { | |
810 | int ret; | |
811 | struct lttcomm_relayd_index msg; | |
812 | struct lttcomm_relayd_generic_reply reply; | |
813 | ||
814 | /* Code flow error. Safety net. */ | |
815 | assert(rsock); | |
816 | ||
817 | if (rsock->minor < 4) { | |
818 | DBG("Not sending indexes before protocol 2.4"); | |
819 | ret = 0; | |
820 | goto error; | |
821 | } | |
822 | ||
823 | DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id); | |
824 | ||
53efb85a | 825 | memset(&msg, 0, sizeof(msg)); |
1c20f0e2 JD |
826 | msg.relay_stream_id = htobe64(relay_stream_id); |
827 | msg.net_seq_num = htobe64(net_seq_num); | |
828 | ||
829 | /* The index is already in big endian. */ | |
830 | msg.packet_size = index->packet_size; | |
831 | msg.content_size = index->content_size; | |
832 | msg.timestamp_begin = index->timestamp_begin; | |
833 | msg.timestamp_end = index->timestamp_end; | |
834 | msg.events_discarded = index->events_discarded; | |
835 | msg.stream_id = index->stream_id; | |
836 | ||
837 | /* Send command */ | |
838 | ret = send_command(rsock, RELAYD_SEND_INDEX, &msg, sizeof(msg), 0); | |
839 | if (ret < 0) { | |
840 | goto error; | |
841 | } | |
842 | ||
843 | /* Receive response */ | |
844 | ret = recv_reply(rsock, (void *) &reply, sizeof(reply)); | |
845 | if (ret < 0) { | |
846 | goto error; | |
847 | } | |
848 | ||
849 | reply.ret_code = be32toh(reply.ret_code); | |
850 | ||
851 | /* Return session id or negative ret code. */ | |
852 | if (reply.ret_code != LTTNG_OK) { | |
853 | ret = -1; | |
854 | ERR("Relayd send index replied error %d", reply.ret_code); | |
855 | } else { | |
856 | /* Success */ | |
857 | ret = 0; | |
858 | } | |
859 | ||
860 | error: | |
861 | return ret; | |
862 | } |