Fix: illegal memory access in relayd_create_session_2_4
[lttng-tools.git] / src / common / relayd / relayd.c
CommitLineData
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
6c1c0768 18#define _LGPL_SOURCE
00e2e675
DG
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 37static 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
83error:
84 free(buf);
85alloc_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 93static 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
117error:
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 */
125static 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
3a13ffd5
MD
132 if (lttng_strncpy(msg.session_name, session_name,
133 sizeof(msg.session_name))) {
246777db
MD
134 ret = -1;
135 goto error;
136 }
3a13ffd5 137 if (lttng_strncpy(msg.hostname, hostname, sizeof(msg.hostname))) {
246777db
MD
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
150error:
151 return ret;
152}
153
154/*
155 * RELAYD_CREATE_SESSION from 2.1 to 2.3.
156 */
157static 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
168error:
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 179int 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
229error:
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 238int 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));
246777db
MD
257 if (strlen(channel_name) >= sizeof(msg.channel_name)) {
258 ret = -1;
259 goto error;
260 }
0f907de1 261 strncpy(msg.channel_name, channel_name, sizeof(msg.channel_name));
246777db
MD
262 if (strlen(pathname) >= sizeof(msg.pathname)) {
263 ret = -1;
264 goto error;
265 }
0f907de1 266 strncpy(msg.pathname, pathname, sizeof(msg.pathname));
00e2e675 267
0f907de1
JD
268 /* Send command */
269 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg, sizeof(msg), 0);
270 if (ret < 0) {
271 goto error;
272 }
273 } else {
53efb85a 274 memset(&msg_2_2, 0, sizeof(msg_2_2));
0f907de1 275 /* Compat with relayd 2.2+ */
246777db
MD
276 if (strlen(channel_name) >= sizeof(msg_2_2.channel_name)) {
277 ret = -1;
278 goto error;
279 }
0f907de1 280 strncpy(msg_2_2.channel_name, channel_name, sizeof(msg_2_2.channel_name));
246777db
MD
281 if (strlen(pathname) >= sizeof(msg_2_2.pathname)) {
282 ret = -1;
283 goto error;
284 }
0f907de1
JD
285 strncpy(msg_2_2.pathname, pathname, sizeof(msg_2_2.pathname));
286 msg_2_2.tracefile_size = htobe64(tracefile_size);
287 msg_2_2.tracefile_count = htobe64(tracefile_count);
288
289 /* Send command */
290 ret = send_command(rsock, RELAYD_ADD_STREAM, (void *) &msg_2_2, sizeof(msg_2_2), 0);
291 if (ret < 0) {
292 goto error;
293 }
00e2e675
DG
294 }
295
633d0084 296 /* Waiting for reply */
6151a90f 297 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
00e2e675
DG
298 if (ret < 0) {
299 goto error;
300 }
301
302 /* Back to host bytes order. */
303 reply.handle = be64toh(reply.handle);
304 reply.ret_code = be32toh(reply.ret_code);
305
306 /* Return session id or negative ret code. */
f73fabfd 307 if (reply.ret_code != LTTNG_OK) {
bb63afd9
DG
308 ret = -1;
309 ERR("Relayd add stream replied error %d", reply.ret_code);
00e2e675
DG
310 } else {
311 /* Success */
312 ret = 0;
313 *stream_id = reply.handle;
314 }
315
77c7c900
MD
316 DBG("Relayd stream added successfully with handle %" PRIu64,
317 reply.handle);
00e2e675
DG
318
319error:
320 return ret;
321}
322
a4baae1b
JD
323/*
324 * Inform the relay that all the streams for the current channel has been sent.
325 *
326 * On success return 0 else return ret_code negative value.
327 */
328int relayd_streams_sent(struct lttcomm_relayd_sock *rsock)
329{
330 int ret;
331 struct lttcomm_relayd_generic_reply reply;
332
333 /* Code flow error. Safety net. */
334 assert(rsock);
335
336 DBG("Relayd sending streams sent.");
337
338 /* This feature was introduced in 2.4, ignore it for earlier versions. */
339 if (rsock->minor < 4) {
340 ret = 0;
341 goto end;
342 }
343
344 /* Send command */
345 ret = send_command(rsock, RELAYD_STREAMS_SENT, NULL, 0, 0);
346 if (ret < 0) {
347 goto error;
348 }
349
350 /* Waiting for reply */
351 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
352 if (ret < 0) {
353 goto error;
354 }
355
356 /* Back to host bytes order. */
357 reply.ret_code = be32toh(reply.ret_code);
358
359 /* Return session id or negative ret code. */
360 if (reply.ret_code != LTTNG_OK) {
361 ret = -1;
362 ERR("Relayd streams sent replied error %d", reply.ret_code);
363 goto error;
364 } else {
365 /* Success */
366 ret = 0;
367 }
368
369 DBG("Relayd streams sent success");
370
371error:
372end:
373 return ret;
374}
375
00e2e675
DG
376/*
377 * Check version numbers on the relayd.
d4519fa3
JD
378 * If major versions are compatible, we assign minor_to_use to the
379 * minor version of the procotol we are going to use for this session.
00e2e675
DG
380 *
381 * Return 0 if compatible else negative value.
382 */
6151a90f 383int relayd_version_check(struct lttcomm_relayd_sock *rsock)
00e2e675
DG
384{
385 int ret;
092b6259 386 struct lttcomm_relayd_version msg;
00e2e675
DG
387
388 /* Code flow error. Safety net. */
6151a90f 389 assert(rsock);
00e2e675 390
6151a90f
JD
391 DBG("Relayd version check for major.minor %u.%u", rsock->major,
392 rsock->minor);
00e2e675 393
53efb85a 394 memset(&msg, 0, sizeof(msg));
092b6259 395 /* Prepare network byte order before transmission. */
6151a90f
JD
396 msg.major = htobe32(rsock->major);
397 msg.minor = htobe32(rsock->minor);
092b6259 398
00e2e675 399 /* Send command */
6151a90f 400 ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
00e2e675
DG
401 if (ret < 0) {
402 goto error;
403 }
404
20275fe8 405 /* Receive response */
6151a90f 406 ret = recv_reply(rsock, (void *) &msg, sizeof(msg));
00e2e675
DG
407 if (ret < 0) {
408 goto error;
409 }
410
411 /* Set back to host bytes order */
092b6259
DG
412 msg.major = be32toh(msg.major);
413 msg.minor = be32toh(msg.minor);
414
415 /*
416 * Only validate the major version. If the other side is higher,
417 * communication is not possible. Only major version equal can talk to each
418 * other. If the minor version differs, the lowest version is used by both
419 * sides.
092b6259 420 */
6151a90f 421 if (msg.major != rsock->major) {
d4519fa3
JD
422 /* Not compatible */
423 ret = -1;
424 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
6151a90f 425 msg.major, rsock->major);
092b6259 426 goto error;
00e2e675
DG
427 }
428
092b6259 429 /*
6151a90f
JD
430 * If the relayd's minor version is higher, it will adapt to our version so
431 * we can continue to use the latest relayd communication data structure.
432 * If the received minor version is higher, the relayd should adapt to us.
092b6259 433 */
6151a90f
JD
434 if (rsock->minor > msg.minor) {
435 rsock->minor = msg.minor;
d4519fa3 436 }
092b6259 437
d4519fa3
JD
438 /* Version number compatible */
439 DBG2("Relayd version is compatible, using protocol version %u.%u",
6151a90f 440 rsock->major, rsock->minor);
d4519fa3 441 ret = 0;
00e2e675
DG
442
443error:
444 return ret;
445}
446
00e2e675
DG
447/*
448 * Add stream on the relayd and assign stream handle to the stream_id argument.
449 *
450 * On success return 0 else return ret_code negative value.
451 */
6151a90f 452int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len)
00e2e675
DG
453{
454 int ret;
455
456 /* Code flow error. Safety net. */
6151a90f 457 assert(rsock);
00e2e675 458
77c7c900 459 DBG("Relayd sending metadata of size %zu", len);
00e2e675
DG
460
461 /* Send command */
6151a90f 462 ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0);
00e2e675
DG
463 if (ret < 0) {
464 goto error;
465 }
466
467 DBG2("Relayd metadata added successfully");
468
469 /*
470 * After that call, the metadata data MUST be sent to the relayd so the
471 * receive size on the other end matches the len of the metadata packet
633d0084 472 * header. This is why we don't wait for a reply here.
00e2e675
DG
473 */
474
475error:
476 return ret;
477}
478
479/*
6151a90f 480 * Connect to relay daemon with an allocated lttcomm_relayd_sock.
00e2e675 481 */
6151a90f 482int relayd_connect(struct lttcomm_relayd_sock *rsock)
00e2e675
DG
483{
484 /* Code flow error. Safety net. */
6151a90f 485 assert(rsock);
00e2e675 486
f96e4545
MD
487 if (!rsock->sock.ops) {
488 /*
489 * Attempting a connect on a non-initialized socket.
490 */
491 return -ECONNRESET;
492 }
493
00e2e675
DG
494 DBG3("Relayd connect ...");
495
6151a90f 496 return rsock->sock.ops->connect(&rsock->sock);
00e2e675
DG
497}
498
499/*
6151a90f 500 * Close relayd socket with an allocated lttcomm_relayd_sock.
ffe60014
DG
501 *
502 * If no socket operations are found, simply return 0 meaning that everything
503 * is fine. Without operations, the socket can not possibly be opened or used.
504 * This is possible if the socket was allocated but not created. However, the
505 * caller could simply use it to store a valid file descriptor for instance
506 * passed over a Unix socket and call this to cleanup but still without a valid
507 * ops pointer.
508 *
509 * Return the close returned value. On error, a negative value is usually
510 * returned back from close(2).
00e2e675 511 */
6151a90f 512int relayd_close(struct lttcomm_relayd_sock *rsock)
00e2e675 513{
ffe60014
DG
514 int ret;
515
00e2e675 516 /* Code flow error. Safety net. */
6151a90f 517 assert(rsock);
00e2e675 518
ffe60014 519 /* An invalid fd is fine, return success. */
6151a90f 520 if (rsock->sock.fd < 0) {
ffe60014
DG
521 ret = 0;
522 goto end;
523 }
524
6151a90f 525 DBG3("Relayd closing socket %d", rsock->sock.fd);
00e2e675 526
6151a90f
JD
527 if (rsock->sock.ops) {
528 ret = rsock->sock.ops->close(&rsock->sock);
ffe60014
DG
529 } else {
530 /* Default call if no specific ops found. */
6151a90f 531 ret = close(rsock->sock.fd);
ffe60014
DG
532 if (ret < 0) {
533 PERROR("relayd_close default close");
534 }
535 }
f96e4545 536 rsock->sock.fd = -1;
ffe60014
DG
537
538end:
539 return ret;
00e2e675
DG
540}
541
542/*
543 * Send data header structure to the relayd.
544 */
6151a90f 545int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock,
00e2e675
DG
546 struct lttcomm_relayd_data_hdr *hdr, size_t size)
547{
548 int ret;
549
550 /* Code flow error. Safety net. */
6151a90f 551 assert(rsock);
00e2e675
DG
552 assert(hdr);
553
f96e4545
MD
554 if (rsock->sock.fd < 0) {
555 return -ECONNRESET;
556 }
557
8fd623e0 558 DBG3("Relayd sending data header of size %zu", size);
00e2e675
DG
559
560 /* Again, safety net */
561 if (size == 0) {
562 size = sizeof(struct lttcomm_relayd_data_hdr);
563 }
564
565 /* Only send data header. */
6151a90f 566 ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0);
00e2e675 567 if (ret < 0) {
8994307f 568 ret = -errno;
00e2e675
DG
569 goto error;
570 }
571
572 /*
573 * The data MUST be sent right after that command for the receive on the
574 * other end to match the size in the header.
575 */
576
577error:
578 return ret;
579}
173af62f
DG
580
581/*
582 * Send close stream command to the relayd.
583 */
6151a90f 584int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
173af62f
DG
585 uint64_t last_net_seq_num)
586{
587 int ret;
588 struct lttcomm_relayd_close_stream msg;
589 struct lttcomm_relayd_generic_reply reply;
590
591 /* Code flow error. Safety net. */
6151a90f 592 assert(rsock);
173af62f 593
77c7c900 594 DBG("Relayd closing stream id %" PRIu64, stream_id);
173af62f 595
53efb85a 596 memset(&msg, 0, sizeof(msg));
173af62f
DG
597 msg.stream_id = htobe64(stream_id);
598 msg.last_net_seq_num = htobe64(last_net_seq_num);
599
600 /* Send command */
6151a90f 601 ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
173af62f
DG
602 if (ret < 0) {
603 goto error;
604 }
605
20275fe8 606 /* Receive response */
6151a90f 607 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
173af62f
DG
608 if (ret < 0) {
609 goto error;
610 }
611
612 reply.ret_code = be32toh(reply.ret_code);
613
614 /* Return session id or negative ret code. */
f73fabfd 615 if (reply.ret_code != LTTNG_OK) {
bb63afd9
DG
616 ret = -1;
617 ERR("Relayd close stream replied error %d", reply.ret_code);
173af62f
DG
618 } else {
619 /* Success */
620 ret = 0;
621 }
622
77c7c900 623 DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
173af62f
DG
624
625error:
626 return ret;
627}
c8f59ee5
DG
628
629/*
630 * Check for data availability for a given stream id.
631 *
6d805429 632 * Return 0 if NOT pending, 1 if so and a negative value on error.
c8f59ee5 633 */
6151a90f 634int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
c8f59ee5
DG
635 uint64_t last_net_seq_num)
636{
637 int ret;
6d805429 638 struct lttcomm_relayd_data_pending msg;
c8f59ee5
DG
639 struct lttcomm_relayd_generic_reply reply;
640
641 /* Code flow error. Safety net. */
6151a90f 642 assert(rsock);
c8f59ee5 643
6d805429 644 DBG("Relayd data pending for stream id %" PRIu64, stream_id);
c8f59ee5 645
53efb85a 646 memset(&msg, 0, sizeof(msg));
c8f59ee5
DG
647 msg.stream_id = htobe64(stream_id);
648 msg.last_net_seq_num = htobe64(last_net_seq_num);
649
650 /* Send command */
6151a90f 651 ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg,
c8f59ee5
DG
652 sizeof(msg), 0);
653 if (ret < 0) {
654 goto error;
655 }
656
20275fe8 657 /* Receive response */
6151a90f 658 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
c8f59ee5
DG
659 if (ret < 0) {
660 goto error;
661 }
662
663 reply.ret_code = be32toh(reply.ret_code);
664
665 /* Return session id or negative ret code. */
666 if (reply.ret_code >= LTTNG_OK) {
bb63afd9 667 ERR("Relayd data pending replied error %d", reply.ret_code);
c8f59ee5
DG
668 }
669
670 /* At this point, the ret code is either 1 or 0 */
671 ret = reply.ret_code;
672
6d805429 673 DBG("Relayd data is %s pending for stream id %" PRIu64,
9dd26bb9 674 ret == 1 ? "" : "NOT", stream_id);
c8f59ee5
DG
675
676error:
677 return ret;
678}
679
680/*
681 * Check on the relayd side for a quiescent state on the control socket.
682 */
6151a90f 683int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock,
ad7051c0 684 uint64_t metadata_stream_id)
c8f59ee5
DG
685{
686 int ret;
ad7051c0 687 struct lttcomm_relayd_quiescent_control msg;
c8f59ee5
DG
688 struct lttcomm_relayd_generic_reply reply;
689
690 /* Code flow error. Safety net. */
6151a90f 691 assert(rsock);
c8f59ee5
DG
692
693 DBG("Relayd checking quiescent control state");
694
53efb85a 695 memset(&msg, 0, sizeof(msg));
ad7051c0
DG
696 msg.stream_id = htobe64(metadata_stream_id);
697
c8f59ee5 698 /* Send command */
6151a90f 699 ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0);
c8f59ee5
DG
700 if (ret < 0) {
701 goto error;
702 }
703
20275fe8 704 /* Receive response */
6151a90f 705 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
c8f59ee5
DG
706 if (ret < 0) {
707 goto error;
708 }
709
710 reply.ret_code = be32toh(reply.ret_code);
711
712 /* Return session id or negative ret code. */
713 if (reply.ret_code != LTTNG_OK) {
bb63afd9
DG
714 ret = -1;
715 ERR("Relayd quiescent control replied error %d", reply.ret_code);
c8f59ee5
DG
716 goto error;
717 }
718
719 /* Control socket is quiescent */
6d805429 720 return 0;
c8f59ee5
DG
721
722error:
723 return ret;
724}
f7079f67
DG
725
726/*
727 * Begin a data pending command for a specific session id.
728 */
6151a90f 729int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id)
f7079f67
DG
730{
731 int ret;
732 struct lttcomm_relayd_begin_data_pending msg;
733 struct lttcomm_relayd_generic_reply reply;
734
735 /* Code flow error. Safety net. */
6151a90f 736 assert(rsock);
f7079f67
DG
737
738 DBG("Relayd begin data pending");
739
53efb85a 740 memset(&msg, 0, sizeof(msg));
f7079f67
DG
741 msg.session_id = htobe64(id);
742
743 /* Send command */
6151a90f 744 ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
f7079f67
DG
745 if (ret < 0) {
746 goto error;
747 }
748
20275fe8 749 /* Receive response */
6151a90f 750 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
f7079f67
DG
751 if (ret < 0) {
752 goto error;
753 }
754
755 reply.ret_code = be32toh(reply.ret_code);
756
757 /* Return session id or negative ret code. */
758 if (reply.ret_code != LTTNG_OK) {
bb63afd9
DG
759 ret = -1;
760 ERR("Relayd begin data pending replied error %d", reply.ret_code);
f7079f67
DG
761 goto error;
762 }
763
764 return 0;
765
766error:
767 return ret;
768}
769
770/*
771 * End a data pending command for a specific session id.
772 *
773 * Return 0 on success and set is_data_inflight to 0 if no data is being
774 * streamed or 1 if it is the case.
775 */
6151a90f 776int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id,
f7079f67
DG
777 unsigned int *is_data_inflight)
778{
af6c30b5 779 int ret, recv_ret;
f7079f67
DG
780 struct lttcomm_relayd_end_data_pending msg;
781 struct lttcomm_relayd_generic_reply reply;
782
783 /* Code flow error. Safety net. */
6151a90f 784 assert(rsock);
f7079f67
DG
785
786 DBG("Relayd end data pending");
787
53efb85a 788 memset(&msg, 0, sizeof(msg));
f7079f67
DG
789 msg.session_id = htobe64(id);
790
791 /* Send command */
6151a90f 792 ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
f7079f67
DG
793 if (ret < 0) {
794 goto error;
795 }
796
20275fe8 797 /* Receive response */
6151a90f 798 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
f7079f67
DG
799 if (ret < 0) {
800 goto error;
801 }
802
af6c30b5
DG
803 recv_ret = be32toh(reply.ret_code);
804 if (recv_ret < 0) {
805 ret = recv_ret;
f7079f67
DG
806 goto error;
807 }
808
af6c30b5 809 *is_data_inflight = recv_ret;
f7079f67 810
af6c30b5 811 DBG("Relayd end data pending is data inflight: %d", recv_ret);
f7079f67
DG
812
813 return 0;
814
815error:
816 return ret;
817}
1c20f0e2
JD
818
819/*
820 * Send index to the relayd.
821 */
822int relayd_send_index(struct lttcomm_relayd_sock *rsock,
50adc264 823 struct ctf_packet_index *index, uint64_t relay_stream_id,
1c20f0e2
JD
824 uint64_t net_seq_num)
825{
826 int ret;
827 struct lttcomm_relayd_index msg;
828 struct lttcomm_relayd_generic_reply reply;
829
830 /* Code flow error. Safety net. */
831 assert(rsock);
832
833 if (rsock->minor < 4) {
834 DBG("Not sending indexes before protocol 2.4");
835 ret = 0;
836 goto error;
837 }
838
839 DBG("Relayd sending index for stream ID %" PRIu64, relay_stream_id);
840
53efb85a 841 memset(&msg, 0, sizeof(msg));
1c20f0e2
JD
842 msg.relay_stream_id = htobe64(relay_stream_id);
843 msg.net_seq_num = htobe64(net_seq_num);
844
845 /* The index is already in big endian. */
846 msg.packet_size = index->packet_size;
847 msg.content_size = index->content_size;
848 msg.timestamp_begin = index->timestamp_begin;
849 msg.timestamp_end = index->timestamp_end;
850 msg.events_discarded = index->events_discarded;
851 msg.stream_id = index->stream_id;
852
234cd636
JD
853 if (rsock->minor >= 8) {
854 msg.stream_instance_id = index->stream_instance_id;
855 msg.packet_seq_num = index->packet_seq_num;
856 }
857
1c20f0e2
JD
858 /* Send command */
859 ret = send_command(rsock, RELAYD_SEND_INDEX, &msg, sizeof(msg), 0);
860 if (ret < 0) {
861 goto error;
862 }
863
864 /* Receive response */
865 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
866 if (ret < 0) {
867 goto error;
868 }
869
870 reply.ret_code = be32toh(reply.ret_code);
871
872 /* Return session id or negative ret code. */
873 if (reply.ret_code != LTTNG_OK) {
874 ret = -1;
875 ERR("Relayd send index replied error %d", reply.ret_code);
876 } else {
877 /* Success */
878 ret = 0;
879 }
880
881error:
882 return ret;
883}
93ec662e
JD
884
885/*
886 * Ask the relay to reset the metadata trace file (regeneration).
887 */
888int relayd_reset_metadata(struct lttcomm_relayd_sock *rsock,
889 uint64_t stream_id, uint64_t version)
890{
891 int ret;
892 struct lttcomm_relayd_reset_metadata msg;
893 struct lttcomm_relayd_generic_reply reply;
894
895 /* Code flow error. Safety net. */
896 assert(rsock);
897
898 /* Should have been prevented by the sessiond. */
899 if (rsock->minor < 8) {
900 ERR("Metadata regeneration unsupported before 2.8");
901 ret = -1;
902 goto error;
903 }
904
905 DBG("Relayd reset metadata stream id %" PRIu64, stream_id);
906
907 memset(&msg, 0, sizeof(msg));
908 msg.stream_id = htobe64(stream_id);
909 msg.version = htobe64(version);
910
911 /* Send command */
912 ret = send_command(rsock, RELAYD_RESET_METADATA, (void *) &msg, sizeof(msg), 0);
913 if (ret < 0) {
914 goto error;
915 }
916
917 /* Receive response */
918 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
919 if (ret < 0) {
920 goto error;
921 }
922
923 reply.ret_code = be32toh(reply.ret_code);
924
925 /* Return session id or negative ret code. */
926 if (reply.ret_code != LTTNG_OK) {
927 ret = -1;
928 ERR("Relayd reset metadata replied error %d", reply.ret_code);
929 } else {
930 /* Success */
931 ret = 0;
932 }
933
934 DBG("Relayd reset metadata stream id %" PRIu64 " successfully", stream_id);
935
936error:
937 return ret;
938}
This page took 0.079543 seconds and 4 git commands to generate.