Cleanup: remove dead increment of pointer
[lttng-tools.git] / src / common / relayd / relayd.c
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 _LGPL_SOURCE
19 #include <assert.h>
20 #include <stdio.h>
21 #include <stdlib.h>
22 #include <string.h>
23 #include <sys/stat.h>
24 #include <inttypes.h>
25
26 #include <common/common.h>
27 #include <common/defaults.h>
28 #include <common/compat/endian.h>
29 #include <common/sessiond-comm/relayd.h>
30 #include <common/index/ctf-index.h>
31
32 #include "relayd.h"
33
34 /*
35 * Send command. Fill up the header and append the data.
36 */
37 static int send_command(struct lttcomm_relayd_sock *rsock,
38 enum lttcomm_relayd_command cmd, void *data, size_t size,
39 int flags)
40 {
41 int ret;
42 struct lttcomm_relayd_hdr header;
43 char *buf;
44 uint64_t buf_size = sizeof(header);
45
46 if (rsock->sock.fd < 0) {
47 return -ECONNRESET;
48 }
49
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
61 memset(&header, 0, sizeof(header));
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
75 ret = rsock->sock.ops->sendmsg(&rsock->sock, buf, buf_size, flags);
76 if (ret < 0) {
77 ret = -errno;
78 goto error;
79 }
80
81 DBG3("Relayd sending command %d of size %" PRIu64, cmd, buf_size);
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 */
93 static int recv_reply(struct lttcomm_relayd_sock *rsock, void *data, size_t size)
94 {
95 int ret;
96
97 if (rsock->sock.fd < 0) {
98 return -ECONNRESET;
99 }
100
101 DBG3("Relayd waiting for reply of size %zu", size);
102
103 ret = rsock->sock.ops->recvmsg(&rsock->sock, data, size, 0);
104 if (ret <= 0 || ret != size) {
105 if (ret == 0) {
106 /* Orderly shutdown. */
107 DBG("Socket %d has performed an orderly shutdown", rsock->sock.fd);
108 } else {
109 DBG("Receiving reply failed on sock %d for size %zu with ret %d",
110 rsock->sock.fd, size, ret);
111 }
112 /* Always return -1 here and the caller can use errno. */
113 ret = -1;
114 goto error;
115 }
116
117 error:
118 return ret;
119 }
120
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,
127 int session_live_timer, unsigned int snapshot)
128 {
129 int ret;
130 struct lttcomm_relayd_create_session_2_4 msg;
131
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 }
141 msg.live_timer = htobe32(session_live_timer);
142 msg.snapshot = htobe32(snapshot);
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
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 *
176 * On success, return 0 else a negative value which is either an errno error or
177 * a lttng error code from the relayd.
178 */
179 int relayd_create_session(struct lttcomm_relayd_sock *rsock, uint64_t *session_id,
180 char *session_name, char *hostname, int session_live_timer,
181 unsigned int snapshot)
182 {
183 int ret;
184 struct lttcomm_relayd_status_session reply;
185
186 assert(rsock);
187 assert(session_id);
188
189 DBG("Relayd create session");
190
191 switch(rsock->minor) {
192 case 1:
193 case 2:
194 case 3:
195 ret = relayd_create_session_2_1(rsock, session_id);
196 break;
197 case 4:
198 default:
199 ret = relayd_create_session_2_4(rsock, session_id, session_name,
200 hostname, session_live_timer, snapshot);
201 break;
202 }
203
204 if (ret < 0) {
205 goto error;
206 }
207
208 /* Receive response */
209 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
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) {
219 ret = -1;
220 ERR("Relayd create session replied error %d", reply.ret_code);
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
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 */
238 int relayd_add_stream(struct lttcomm_relayd_sock *rsock, const char *channel_name,
239 const char *pathname, uint64_t *stream_id,
240 uint64_t tracefile_size, uint64_t tracefile_count)
241 {
242 int ret;
243 struct lttcomm_relayd_add_stream msg;
244 struct lttcomm_relayd_add_stream_2_2 msg_2_2;
245 struct lttcomm_relayd_status_stream reply;
246
247 /* Code flow error. Safety net. */
248 assert(rsock);
249 assert(channel_name);
250 assert(pathname);
251
252 DBG("Relayd adding stream for channel name %s", channel_name);
253
254 /* Compat with relayd 2.1 */
255 if (rsock->minor == 1) {
256 memset(&msg, 0, sizeof(msg));
257 if (lttng_strncpy(msg.channel_name, channel_name,
258 sizeof(msg.channel_name))) {
259 ret = -1;
260 goto error;
261 }
262 if (lttng_strncpy(msg.pathname, pathname,
263 sizeof(msg.pathname))) {
264 ret = -1;
265 goto error;
266 }
267
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 {
274 memset(&msg_2_2, 0, sizeof(msg_2_2));
275 /* Compat with relayd 2.2+ */
276 if (lttng_strncpy(msg_2_2.channel_name, channel_name,
277 sizeof(msg_2_2.channel_name))) {
278 ret = -1;
279 goto error;
280 }
281 if (lttng_strncpy(msg_2_2.pathname, pathname,
282 sizeof(msg_2_2.pathname))) {
283 ret = -1;
284 goto error;
285 }
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 }
294 }
295
296 /* Waiting for reply */
297 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
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. */
307 if (reply.ret_code != LTTNG_OK) {
308 ret = -1;
309 ERR("Relayd add stream replied error %d", reply.ret_code);
310 } else {
311 /* Success */
312 ret = 0;
313 *stream_id = reply.handle;
314 }
315
316 DBG("Relayd stream added successfully with handle %" PRIu64,
317 reply.handle);
318
319 error:
320 return ret;
321 }
322
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 */
328 int 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
371 error:
372 end:
373 return ret;
374 }
375
376 /*
377 * Check version numbers on the relayd.
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.
380 *
381 * Return 0 if compatible else negative value.
382 */
383 int relayd_version_check(struct lttcomm_relayd_sock *rsock)
384 {
385 int ret;
386 struct lttcomm_relayd_version msg;
387
388 /* Code flow error. Safety net. */
389 assert(rsock);
390
391 DBG("Relayd version check for major.minor %u.%u", rsock->major,
392 rsock->minor);
393
394 memset(&msg, 0, sizeof(msg));
395 /* Prepare network byte order before transmission. */
396 msg.major = htobe32(rsock->major);
397 msg.minor = htobe32(rsock->minor);
398
399 /* Send command */
400 ret = send_command(rsock, RELAYD_VERSION, (void *) &msg, sizeof(msg), 0);
401 if (ret < 0) {
402 goto error;
403 }
404
405 /* Receive response */
406 ret = recv_reply(rsock, (void *) &msg, sizeof(msg));
407 if (ret < 0) {
408 goto error;
409 }
410
411 /* Set back to host bytes order */
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.
420 */
421 if (msg.major != rsock->major) {
422 /* Not compatible */
423 ret = -1;
424 DBG2("Relayd version is NOT compatible. Relayd version %u != %u (us)",
425 msg.major, rsock->major);
426 goto error;
427 }
428
429 /*
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.
433 */
434 if (rsock->minor > msg.minor) {
435 rsock->minor = msg.minor;
436 }
437
438 /* Version number compatible */
439 DBG2("Relayd version is compatible, using protocol version %u.%u",
440 rsock->major, rsock->minor);
441 ret = 0;
442
443 error:
444 return ret;
445 }
446
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 */
452 int relayd_send_metadata(struct lttcomm_relayd_sock *rsock, size_t len)
453 {
454 int ret;
455
456 /* Code flow error. Safety net. */
457 assert(rsock);
458
459 DBG("Relayd sending metadata of size %zu", len);
460
461 /* Send command */
462 ret = send_command(rsock, RELAYD_SEND_METADATA, NULL, len, 0);
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
472 * header. This is why we don't wait for a reply here.
473 */
474
475 error:
476 return ret;
477 }
478
479 /*
480 * Connect to relay daemon with an allocated lttcomm_relayd_sock.
481 */
482 int relayd_connect(struct lttcomm_relayd_sock *rsock)
483 {
484 /* Code flow error. Safety net. */
485 assert(rsock);
486
487 if (!rsock->sock.ops) {
488 /*
489 * Attempting a connect on a non-initialized socket.
490 */
491 return -ECONNRESET;
492 }
493
494 DBG3("Relayd connect ...");
495
496 return rsock->sock.ops->connect(&rsock->sock);
497 }
498
499 /*
500 * Close relayd socket with an allocated lttcomm_relayd_sock.
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).
511 */
512 int relayd_close(struct lttcomm_relayd_sock *rsock)
513 {
514 int ret;
515
516 /* Code flow error. Safety net. */
517 assert(rsock);
518
519 /* An invalid fd is fine, return success. */
520 if (rsock->sock.fd < 0) {
521 ret = 0;
522 goto end;
523 }
524
525 DBG3("Relayd closing socket %d", rsock->sock.fd);
526
527 if (rsock->sock.ops) {
528 ret = rsock->sock.ops->close(&rsock->sock);
529 } else {
530 /* Default call if no specific ops found. */
531 ret = close(rsock->sock.fd);
532 if (ret < 0) {
533 PERROR("relayd_close default close");
534 }
535 }
536 rsock->sock.fd = -1;
537
538 end:
539 return ret;
540 }
541
542 /*
543 * Send data header structure to the relayd.
544 */
545 int relayd_send_data_hdr(struct lttcomm_relayd_sock *rsock,
546 struct lttcomm_relayd_data_hdr *hdr, size_t size)
547 {
548 int ret;
549
550 /* Code flow error. Safety net. */
551 assert(rsock);
552 assert(hdr);
553
554 if (rsock->sock.fd < 0) {
555 return -ECONNRESET;
556 }
557
558 DBG3("Relayd sending data header of size %zu", size);
559
560 /* Again, safety net */
561 if (size == 0) {
562 size = sizeof(struct lttcomm_relayd_data_hdr);
563 }
564
565 /* Only send data header. */
566 ret = rsock->sock.ops->sendmsg(&rsock->sock, hdr, size, 0);
567 if (ret < 0) {
568 ret = -errno;
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
577 error:
578 return ret;
579 }
580
581 /*
582 * Send close stream command to the relayd.
583 */
584 int relayd_send_close_stream(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
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. */
592 assert(rsock);
593
594 DBG("Relayd closing stream id %" PRIu64, stream_id);
595
596 memset(&msg, 0, sizeof(msg));
597 msg.stream_id = htobe64(stream_id);
598 msg.last_net_seq_num = htobe64(last_net_seq_num);
599
600 /* Send command */
601 ret = send_command(rsock, RELAYD_CLOSE_STREAM, (void *) &msg, sizeof(msg), 0);
602 if (ret < 0) {
603 goto error;
604 }
605
606 /* Receive response */
607 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
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. */
615 if (reply.ret_code != LTTNG_OK) {
616 ret = -1;
617 ERR("Relayd close stream replied error %d", reply.ret_code);
618 } else {
619 /* Success */
620 ret = 0;
621 }
622
623 DBG("Relayd close stream id %" PRIu64 " successfully", stream_id);
624
625 error:
626 return ret;
627 }
628
629 /*
630 * Check for data availability for a given stream id.
631 *
632 * Return 0 if NOT pending, 1 if so and a negative value on error.
633 */
634 int relayd_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t stream_id,
635 uint64_t last_net_seq_num)
636 {
637 int ret;
638 struct lttcomm_relayd_data_pending msg;
639 struct lttcomm_relayd_generic_reply reply;
640
641 /* Code flow error. Safety net. */
642 assert(rsock);
643
644 DBG("Relayd data pending for stream id %" PRIu64, stream_id);
645
646 memset(&msg, 0, sizeof(msg));
647 msg.stream_id = htobe64(stream_id);
648 msg.last_net_seq_num = htobe64(last_net_seq_num);
649
650 /* Send command */
651 ret = send_command(rsock, RELAYD_DATA_PENDING, (void *) &msg,
652 sizeof(msg), 0);
653 if (ret < 0) {
654 goto error;
655 }
656
657 /* Receive response */
658 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
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) {
667 ERR("Relayd data pending replied error %d", reply.ret_code);
668 }
669
670 /* At this point, the ret code is either 1 or 0 */
671 ret = reply.ret_code;
672
673 DBG("Relayd data is %s pending for stream id %" PRIu64,
674 ret == 1 ? "" : "NOT", stream_id);
675
676 error:
677 return ret;
678 }
679
680 /*
681 * Check on the relayd side for a quiescent state on the control socket.
682 */
683 int relayd_quiescent_control(struct lttcomm_relayd_sock *rsock,
684 uint64_t metadata_stream_id)
685 {
686 int ret;
687 struct lttcomm_relayd_quiescent_control msg;
688 struct lttcomm_relayd_generic_reply reply;
689
690 /* Code flow error. Safety net. */
691 assert(rsock);
692
693 DBG("Relayd checking quiescent control state");
694
695 memset(&msg, 0, sizeof(msg));
696 msg.stream_id = htobe64(metadata_stream_id);
697
698 /* Send command */
699 ret = send_command(rsock, RELAYD_QUIESCENT_CONTROL, &msg, sizeof(msg), 0);
700 if (ret < 0) {
701 goto error;
702 }
703
704 /* Receive response */
705 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
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) {
714 ret = -1;
715 ERR("Relayd quiescent control replied error %d", reply.ret_code);
716 goto error;
717 }
718
719 /* Control socket is quiescent */
720 return 0;
721
722 error:
723 return ret;
724 }
725
726 /*
727 * Begin a data pending command for a specific session id.
728 */
729 int relayd_begin_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id)
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. */
736 assert(rsock);
737
738 DBG("Relayd begin data pending");
739
740 memset(&msg, 0, sizeof(msg));
741 msg.session_id = htobe64(id);
742
743 /* Send command */
744 ret = send_command(rsock, RELAYD_BEGIN_DATA_PENDING, &msg, sizeof(msg), 0);
745 if (ret < 0) {
746 goto error;
747 }
748
749 /* Receive response */
750 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
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) {
759 ret = -1;
760 ERR("Relayd begin data pending replied error %d", reply.ret_code);
761 goto error;
762 }
763
764 return 0;
765
766 error:
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 */
776 int relayd_end_data_pending(struct lttcomm_relayd_sock *rsock, uint64_t id,
777 unsigned int *is_data_inflight)
778 {
779 int ret, recv_ret;
780 struct lttcomm_relayd_end_data_pending msg;
781 struct lttcomm_relayd_generic_reply reply;
782
783 /* Code flow error. Safety net. */
784 assert(rsock);
785
786 DBG("Relayd end data pending");
787
788 memset(&msg, 0, sizeof(msg));
789 msg.session_id = htobe64(id);
790
791 /* Send command */
792 ret = send_command(rsock, RELAYD_END_DATA_PENDING, &msg, sizeof(msg), 0);
793 if (ret < 0) {
794 goto error;
795 }
796
797 /* Receive response */
798 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
799 if (ret < 0) {
800 goto error;
801 }
802
803 recv_ret = be32toh(reply.ret_code);
804 if (recv_ret < 0) {
805 ret = recv_ret;
806 goto error;
807 }
808
809 *is_data_inflight = recv_ret;
810
811 DBG("Relayd end data pending is data inflight: %d", recv_ret);
812
813 return 0;
814
815 error:
816 return ret;
817 }
818
819 /*
820 * Send index to the relayd.
821 */
822 int relayd_send_index(struct lttcomm_relayd_sock *rsock,
823 struct ctf_packet_index *index, uint64_t relay_stream_id,
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
841 memset(&msg, 0, sizeof(msg));
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
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
858 /* Send command */
859 ret = send_command(rsock, RELAYD_SEND_INDEX, &msg,
860 lttcomm_relayd_index_len(lttng_to_index_major(rsock->major,
861 rsock->minor),
862 lttng_to_index_minor(rsock->major, rsock->minor)),
863 0);
864 if (ret < 0) {
865 goto error;
866 }
867
868 /* Receive response */
869 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
870 if (ret < 0) {
871 goto error;
872 }
873
874 reply.ret_code = be32toh(reply.ret_code);
875
876 /* Return session id or negative ret code. */
877 if (reply.ret_code != LTTNG_OK) {
878 ret = -1;
879 ERR("Relayd send index replied error %d", reply.ret_code);
880 } else {
881 /* Success */
882 ret = 0;
883 }
884
885 error:
886 return ret;
887 }
888
889 /*
890 * Ask the relay to reset the metadata trace file (regeneration).
891 */
892 int relayd_reset_metadata(struct lttcomm_relayd_sock *rsock,
893 uint64_t stream_id, uint64_t version)
894 {
895 int ret;
896 struct lttcomm_relayd_reset_metadata msg;
897 struct lttcomm_relayd_generic_reply reply;
898
899 /* Code flow error. Safety net. */
900 assert(rsock);
901
902 /* Should have been prevented by the sessiond. */
903 if (rsock->minor < 8) {
904 ERR("Metadata regeneration unsupported before 2.8");
905 ret = -1;
906 goto error;
907 }
908
909 DBG("Relayd reset metadata stream id %" PRIu64, stream_id);
910
911 memset(&msg, 0, sizeof(msg));
912 msg.stream_id = htobe64(stream_id);
913 msg.version = htobe64(version);
914
915 /* Send command */
916 ret = send_command(rsock, RELAYD_RESET_METADATA, (void *) &msg, sizeof(msg), 0);
917 if (ret < 0) {
918 goto error;
919 }
920
921 /* Receive response */
922 ret = recv_reply(rsock, (void *) &reply, sizeof(reply));
923 if (ret < 0) {
924 goto error;
925 }
926
927 reply.ret_code = be32toh(reply.ret_code);
928
929 /* Return session id or negative ret code. */
930 if (reply.ret_code != LTTNG_OK) {
931 ret = -1;
932 ERR("Relayd reset metadata replied error %d", reply.ret_code);
933 } else {
934 /* Success */
935 ret = 0;
936 }
937
938 DBG("Relayd reset metadata stream id %" PRIu64 " successfully", stream_id);
939
940 error:
941 return ret;
942 }
This page took 0.061959 seconds and 4 git commands to generate.