Fix: tests: update rotation ust number of tests
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
CommitLineData
00e2e675
DG
1/*
2 * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
92816cc3 3 * 2018 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
00e2e675
DG
4 *
5 * This program is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License, version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along with
15 * this program; if not, write to the Free Software Foundation, Inc., 51
16 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
6c1c0768 19#define _LGPL_SOURCE
00e2e675
DG
20#include <assert.h>
21#include <stdio.h>
22#include <stdlib.h>
23#include <string.h>
24#include <sys/stat.h>
25#include <sys/types.h>
26#include <unistd.h>
d88aee68 27#include <inttypes.h>
00e2e675
DG
28
29#include <common/common.h>
30#include <common/defaults.h>
31#include <common/uri.h>
d3e2ba59 32#include <common/relayd/relayd.h>
d2956687 33#include <common/string-utils/format.h>
00e2e675
DG
34
35#include "consumer.h"
8782cc74 36#include "health-sessiond.h"
7972aab2 37#include "ust-app.h"
0b2dc8df 38#include "utils.h"
658f12fa 39#include "lttng-sessiond.h"
00e2e675 40
52898cb1
DG
41/*
42 * Send a data payload using a given consumer socket of size len.
43 *
44 * The consumer socket lock MUST be acquired before calling this since this
45 * function can change the fd value.
46 *
47 * Return 0 on success else a negative value on error.
48 */
49int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
50{
51 int fd;
52 ssize_t size;
53
54 assert(socket);
9363801e 55 assert(socket->fd_ptr);
52898cb1
DG
56 assert(msg);
57
58 /* Consumer socket is invalid. Stopping. */
9363801e 59 fd = *socket->fd_ptr;
52898cb1
DG
60 if (fd < 0) {
61 goto error;
62 }
63
64 size = lttcomm_send_unix_sock(fd, msg, len);
65 if (size < 0) {
66 /* The above call will print a PERROR on error. */
67 DBG("Error when sending data to consumer on sock %d", fd);
68 /*
92db7cdc
DG
69 * At this point, the socket is not usable anymore thus closing it and
70 * setting the file descriptor to -1 so it is not reused.
52898cb1
DG
71 */
72
73 /* This call will PERROR on error. */
74 (void) lttcomm_close_unix_sock(fd);
9363801e 75 *socket->fd_ptr = -1;
52898cb1
DG
76 goto error;
77 }
78
79 return 0;
80
81error:
82 return -1;
83}
84
85/*
86 * Receive a data payload using a given consumer socket of size len.
87 *
88 * The consumer socket lock MUST be acquired before calling this since this
89 * function can change the fd value.
90 *
91 * Return 0 on success else a negative value on error.
92 */
93int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
94{
95 int fd;
96 ssize_t size;
97
98 assert(socket);
9363801e 99 assert(socket->fd_ptr);
52898cb1
DG
100 assert(msg);
101
102 /* Consumer socket is invalid. Stopping. */
9363801e 103 fd = *socket->fd_ptr;
52898cb1
DG
104 if (fd < 0) {
105 goto error;
106 }
107
108 size = lttcomm_recv_unix_sock(fd, msg, len);
109 if (size <= 0) {
110 /* The above call will print a PERROR on error. */
111 DBG("Error when receiving data from the consumer socket %d", fd);
112 /*
92db7cdc
DG
113 * At this point, the socket is not usable anymore thus closing it and
114 * setting the file descriptor to -1 so it is not reused.
52898cb1
DG
115 */
116
117 /* This call will PERROR on error. */
118 (void) lttcomm_close_unix_sock(fd);
9363801e 119 *socket->fd_ptr = -1;
52898cb1
DG
120 goto error;
121 }
122
123 return 0;
124
125error:
126 return -1;
127}
128
f50f23d9
DG
129/*
130 * Receive a reply command status message from the consumer. Consumer socket
131 * lock MUST be acquired before calling this function.
132 *
133 * Return 0 on success, -1 on recv error or a negative lttng error code which
134 * was possibly returned by the consumer.
135 */
136int consumer_recv_status_reply(struct consumer_socket *sock)
137{
138 int ret;
139 struct lttcomm_consumer_status_msg reply;
140
141 assert(sock);
142
52898cb1
DG
143 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
144 if (ret < 0) {
f50f23d9
DG
145 goto end;
146 }
147
0c759fc9 148 if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
f50f23d9
DG
149 /* All good. */
150 ret = 0;
151 } else {
152 ret = -reply.ret_code;
ffe60014 153 DBG("Consumer ret code %d", ret);
f50f23d9
DG
154 }
155
156end:
157 return ret;
158}
159
ffe60014
DG
160/*
161 * Once the ASK_CHANNEL command is sent to the consumer, the channel
162 * information are sent back. This call receives that data and populates key
163 * and stream_count.
164 *
165 * On success return 0 and both key and stream_count are set. On error, a
166 * negative value is sent back and both parameters are untouched.
167 */
168int consumer_recv_status_channel(struct consumer_socket *sock,
d88aee68 169 uint64_t *key, unsigned int *stream_count)
ffe60014
DG
170{
171 int ret;
172 struct lttcomm_consumer_status_channel reply;
173
174 assert(sock);
175 assert(stream_count);
176 assert(key);
177
52898cb1
DG
178 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
179 if (ret < 0) {
ffe60014
DG
180 goto end;
181 }
182
183 /* An error is possible so don't touch the key and stream_count. */
0c759fc9 184 if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
ffe60014
DG
185 ret = -1;
186 goto end;
187 }
188
189 *key = reply.key;
190 *stream_count = reply.stream_count;
0c759fc9 191 ret = 0;
ffe60014
DG
192
193end:
194 return ret;
195}
196
2f77fc4b
DG
197/*
198 * Send destroy relayd command to consumer.
199 *
200 * On success return positive value. On error, negative value.
201 */
202int consumer_send_destroy_relayd(struct consumer_socket *sock,
203 struct consumer_output *consumer)
204{
205 int ret;
206 struct lttcomm_consumer_msg msg;
207
208 assert(consumer);
209 assert(sock);
210
9363801e 211 DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
2f77fc4b 212
53efb85a 213 memset(&msg, 0, sizeof(msg));
2f77fc4b
DG
214 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
215 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
216
217 pthread_mutex_lock(sock->lock);
52898cb1 218 ret = consumer_socket_send(sock, &msg, sizeof(msg));
2f77fc4b 219 if (ret < 0) {
52898cb1 220 goto error;
2f77fc4b
DG
221 }
222
f50f23d9
DG
223 /* Don't check the return value. The caller will do it. */
224 ret = consumer_recv_status_reply(sock);
225
2f77fc4b
DG
226 DBG2("Consumer send destroy relayd command done");
227
228error:
52898cb1 229 pthread_mutex_unlock(sock->lock);
2f77fc4b
DG
230 return ret;
231}
232
233/*
234 * For each consumer socket in the consumer output object, send a destroy
235 * relayd command.
236 */
237void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
238{
2f77fc4b
DG
239 struct lttng_ht_iter iter;
240 struct consumer_socket *socket;
241
242 assert(consumer);
243
244 /* Destroy any relayd connection */
6dc3064a 245 if (consumer->type == CONSUMER_DST_NET) {
2f77fc4b
DG
246 rcu_read_lock();
247 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
248 node.node) {
c617c0c6
MD
249 int ret;
250
2f77fc4b
DG
251 /* Send destroy relayd command */
252 ret = consumer_send_destroy_relayd(socket, consumer);
253 if (ret < 0) {
c5c45efa 254 DBG("Unable to send destroy relayd command to consumer");
2f77fc4b
DG
255 /* Continue since we MUST delete everything at this point. */
256 }
257 }
258 rcu_read_unlock();
259 }
260}
261
a4b92340
DG
262/*
263 * From a consumer_data structure, allocate and add a consumer socket to the
264 * consumer output.
265 *
266 * Return 0 on success, else negative value on error
267 */
268int consumer_create_socket(struct consumer_data *data,
269 struct consumer_output *output)
270{
271 int ret = 0;
272 struct consumer_socket *socket;
273
274 assert(data);
275
276 if (output == NULL || data->cmd_sock < 0) {
277 /*
278 * Not an error. Possible there is simply not spawned consumer or it's
279 * disabled for the tracing session asking the socket.
280 */
281 goto error;
282 }
283
284 rcu_read_lock();
285 socket = consumer_find_socket(data->cmd_sock, output);
286 rcu_read_unlock();
287 if (socket == NULL) {
4ce514c4 288 socket = consumer_allocate_socket(&data->cmd_sock);
a4b92340
DG
289 if (socket == NULL) {
290 ret = -1;
291 goto error;
292 }
293
2f77fc4b 294 socket->registered = 0;
a4b92340
DG
295 socket->lock = &data->lock;
296 rcu_read_lock();
297 consumer_add_socket(socket, output);
298 rcu_read_unlock();
299 }
300
6dc3064a
DG
301 socket->type = data->type;
302
a4b92340
DG
303 DBG3("Consumer socket created (fd: %d) and added to output",
304 data->cmd_sock);
305
306error:
307 return ret;
308}
309
7972aab2
DG
310/*
311 * Return the consumer socket from the given consumer output with the right
312 * bitness. On error, returns NULL.
313 *
314 * The caller MUST acquire a rcu read side lock and keep it until the socket
315 * object reference is not needed anymore.
316 */
317struct consumer_socket *consumer_find_socket_by_bitness(int bits,
348a81dc 318 const struct consumer_output *consumer)
7972aab2
DG
319{
320 int consumer_fd;
321 struct consumer_socket *socket = NULL;
322
323 switch (bits) {
324 case 64:
325 consumer_fd = uatomic_read(&ust_consumerd64_fd);
326 break;
327 case 32:
328 consumer_fd = uatomic_read(&ust_consumerd32_fd);
329 break;
330 default:
331 assert(0);
332 goto end;
333 }
334
335 socket = consumer_find_socket(consumer_fd, consumer);
336 if (!socket) {
337 ERR("Consumer socket fd %d not found in consumer obj %p",
338 consumer_fd, consumer);
339 }
340
341end:
342 return socket;
343}
344
173af62f
DG
345/*
346 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
347 * be acquired before calling this function and across use of the
348 * returned consumer_socket.
349 */
350struct consumer_socket *consumer_find_socket(int key,
348a81dc 351 const struct consumer_output *consumer)
173af62f
DG
352{
353 struct lttng_ht_iter iter;
354 struct lttng_ht_node_ulong *node;
355 struct consumer_socket *socket = NULL;
356
357 /* Negative keys are lookup failures */
a4b92340 358 if (key < 0 || consumer == NULL) {
173af62f
DG
359 return NULL;
360 }
361
362 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
363 &iter);
364 node = lttng_ht_iter_get_node_ulong(&iter);
365 if (node != NULL) {
366 socket = caa_container_of(node, struct consumer_socket, node);
367 }
368
369 return socket;
370}
371
372/*
373 * Allocate a new consumer_socket and return the pointer.
374 */
4ce514c4 375struct consumer_socket *consumer_allocate_socket(int *fd)
173af62f
DG
376{
377 struct consumer_socket *socket = NULL;
378
4ce514c4
DG
379 assert(fd);
380
173af62f
DG
381 socket = zmalloc(sizeof(struct consumer_socket));
382 if (socket == NULL) {
383 PERROR("zmalloc consumer socket");
384 goto error;
385 }
386
9363801e 387 socket->fd_ptr = fd;
4ce514c4 388 lttng_ht_node_init_ulong(&socket->node, *fd);
173af62f
DG
389
390error:
391 return socket;
392}
393
394/*
395 * Add consumer socket to consumer output object. Read side lock must be
396 * acquired before calling this function.
397 */
398void consumer_add_socket(struct consumer_socket *sock,
399 struct consumer_output *consumer)
400{
401 assert(sock);
402 assert(consumer);
403
404 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
405}
406
407/*
348a81dc 408 * Delete consumer socket to consumer output object. Read side lock must be
173af62f
DG
409 * acquired before calling this function.
410 */
411void consumer_del_socket(struct consumer_socket *sock,
412 struct consumer_output *consumer)
413{
414 int ret;
415 struct lttng_ht_iter iter;
416
417 assert(sock);
418 assert(consumer);
419
420 iter.iter.node = &sock->node.node;
421 ret = lttng_ht_del(consumer->socks, &iter);
422 assert(!ret);
423}
424
425/*
426 * RCU destroy call function.
427 */
428static void destroy_socket_rcu(struct rcu_head *head)
429{
430 struct lttng_ht_node_ulong *node =
431 caa_container_of(head, struct lttng_ht_node_ulong, head);
432 struct consumer_socket *socket =
433 caa_container_of(node, struct consumer_socket, node);
434
435 free(socket);
436}
437
438/*
439 * Destroy and free socket pointer in a call RCU. Read side lock must be
440 * acquired before calling this function.
441 */
442void consumer_destroy_socket(struct consumer_socket *sock)
443{
444 assert(sock);
445
446 /*
447 * We DO NOT close the file descriptor here since it is global to the
2f77fc4b
DG
448 * session daemon and is closed only if the consumer dies or a custom
449 * consumer was registered,
173af62f 450 */
2f77fc4b 451 if (sock->registered) {
9363801e
DG
452 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
453 lttcomm_close_unix_sock(*sock->fd_ptr);
2f77fc4b 454 }
173af62f
DG
455
456 call_rcu(&sock->node.head, destroy_socket_rcu);
457}
458
00e2e675
DG
459/*
460 * Allocate and assign data to a consumer_output object.
461 *
462 * Return pointer to structure.
463 */
464struct consumer_output *consumer_create_output(enum consumer_dst_type type)
465{
466 struct consumer_output *output = NULL;
467
468 output = zmalloc(sizeof(struct consumer_output));
469 if (output == NULL) {
470 PERROR("zmalloc consumer_output");
471 goto error;
472 }
473
474 /* By default, consumer output is enabled */
475 output->enabled = 1;
476 output->type = type;
d88aee68 477 output->net_seq_index = (uint64_t) -1ULL;
6addfa37 478 urcu_ref_init(&output->ref);
173af62f
DG
479
480 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
481
482error:
483 return output;
484}
485
af706bb7
DG
486/*
487 * Iterate over the consumer output socket hash table and destroy them. The
488 * socket file descriptor are only closed if the consumer output was
489 * registered meaning it's an external consumer.
490 */
491void consumer_destroy_output_sockets(struct consumer_output *obj)
492{
493 struct lttng_ht_iter iter;
494 struct consumer_socket *socket;
495
496 if (!obj->socks) {
497 return;
498 }
499
500 rcu_read_lock();
501 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
502 consumer_del_socket(socket, obj);
503 consumer_destroy_socket(socket);
504 }
505 rcu_read_unlock();
506}
507
00e2e675
DG
508/*
509 * Delete the consumer_output object from the list and free the ptr.
36b588ed
MD
510 *
511 * Should *NOT* be called with RCU read-side lock held.
00e2e675 512 */
6addfa37 513static void consumer_release_output(struct urcu_ref *ref)
00e2e675 514{
6addfa37
MD
515 struct consumer_output *obj =
516 caa_container_of(ref, struct consumer_output, ref);
00e2e675 517
af706bb7 518 consumer_destroy_output_sockets(obj);
2f77fc4b 519
af706bb7 520 if (obj->socks) {
2f77fc4b 521 /* Finally destroy HT */
0b2dc8df 522 ht_cleanup_push(obj->socks);
00e2e675 523 }
173af62f 524
00e2e675
DG
525 free(obj);
526}
527
6addfa37
MD
528/*
529 * Get the consumer_output object.
530 */
531void consumer_output_get(struct consumer_output *obj)
532{
533 urcu_ref_get(&obj->ref);
534}
535
536/*
537 * Put the consumer_output object.
538 *
539 * Should *NOT* be called with RCU read-side lock held.
540 */
541void consumer_output_put(struct consumer_output *obj)
542{
543 if (!obj) {
544 return;
545 }
546 urcu_ref_put(&obj->ref, consumer_release_output);
547}
548
00e2e675
DG
549/*
550 * Copy consumer output and returned the newly allocated copy.
36b588ed
MD
551 *
552 * Should *NOT* be called with RCU read-side lock held.
00e2e675 553 */
b178f53e 554struct consumer_output *consumer_copy_output(struct consumer_output *src)
00e2e675 555{
6dc3064a 556 int ret;
00e2e675
DG
557 struct consumer_output *output;
558
b178f53e 559 assert(src);
00e2e675 560
b178f53e 561 output = consumer_create_output(src->type);
00e2e675 562 if (output == NULL) {
6addfa37 563 goto end;
00e2e675 564 }
b178f53e
JG
565 output->enabled = src->enabled;
566 output->net_seq_index = src->net_seq_index;
567 memcpy(output->domain_subdir, src->domain_subdir,
568 sizeof(output->domain_subdir));
569 output->snapshot = src->snapshot;
570 output->relay_major_version = src->relay_major_version;
571 output->relay_minor_version = src->relay_minor_version;
572 memcpy(&output->dst, &src->dst, sizeof(output->dst));
573 ret = consumer_copy_sockets(output, src);
6dc3064a 574 if (ret < 0) {
6addfa37 575 goto error_put;
6dc3064a 576 }
6addfa37 577end:
6dc3064a
DG
578 return output;
579
6addfa37
MD
580error_put:
581 consumer_output_put(output);
6dc3064a
DG
582 return NULL;
583}
584
585/*
586 * Copy consumer sockets from src to dst.
587 *
588 * Return 0 on success or else a negative value.
589 */
590int consumer_copy_sockets(struct consumer_output *dst,
591 struct consumer_output *src)
592{
593 int ret = 0;
594 struct lttng_ht_iter iter;
595 struct consumer_socket *socket, *copy_sock;
596
597 assert(dst);
598 assert(src);
599
b82c5c4d 600 rcu_read_lock();
6dc3064a
DG
601 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
602 /* Ignore socket that are already there. */
9363801e 603 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
6dc3064a
DG
604 if (copy_sock) {
605 continue;
606 }
607
173af62f 608 /* Create new socket object. */
9363801e 609 copy_sock = consumer_allocate_socket(socket->fd_ptr);
173af62f 610 if (copy_sock == NULL) {
b82c5c4d 611 rcu_read_unlock();
6dc3064a
DG
612 ret = -ENOMEM;
613 goto error;
173af62f
DG
614 }
615
09a90bcd 616 copy_sock->registered = socket->registered;
6dc3064a
DG
617 /*
618 * This is valid because this lock is shared accross all consumer
619 * object being the global lock of the consumer data structure of the
620 * session daemon.
621 */
173af62f 622 copy_sock->lock = socket->lock;
6dc3064a 623 consumer_add_socket(copy_sock, dst);
173af62f 624 }
b82c5c4d 625 rcu_read_unlock();
173af62f 626
00e2e675 627error:
6dc3064a 628 return ret;
00e2e675
DG
629}
630
631/*
b178f53e 632 * Set network URI to the consumer output.
00e2e675 633 *
ad20f474
DG
634 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
635 * error.
00e2e675 636 */
b178f53e
JG
637int consumer_set_network_uri(const struct ltt_session *session,
638 struct consumer_output *output,
00e2e675
DG
639 struct lttng_uri *uri)
640{
641 int ret;
00e2e675
DG
642 struct lttng_uri *dst_uri = NULL;
643
644 /* Code flow error safety net. */
b178f53e 645 assert(output);
00e2e675
DG
646 assert(uri);
647
648 switch (uri->stype) {
649 case LTTNG_STREAM_CONTROL:
b178f53e
JG
650 dst_uri = &output->dst.net.control;
651 output->dst.net.control_isset = 1;
00e2e675
DG
652 if (uri->port == 0) {
653 /* Assign default port. */
654 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
a74934ba 655 } else {
b178f53e
JG
656 if (output->dst.net.data_isset && uri->port ==
657 output->dst.net.data.port) {
a74934ba
DG
658 ret = -LTTNG_ERR_INVALID;
659 goto error;
660 }
00e2e675 661 }
ad20f474 662 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
663 break;
664 case LTTNG_STREAM_DATA:
b178f53e
JG
665 dst_uri = &output->dst.net.data;
666 output->dst.net.data_isset = 1;
00e2e675
DG
667 if (uri->port == 0) {
668 /* Assign default port. */
669 uri->port = DEFAULT_NETWORK_DATA_PORT;
a74934ba 670 } else {
b178f53e
JG
671 if (output->dst.net.control_isset && uri->port ==
672 output->dst.net.control.port) {
a74934ba
DG
673 ret = -LTTNG_ERR_INVALID;
674 goto error;
675 }
00e2e675 676 }
ad20f474 677 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
678 break;
679 default:
680 ERR("Set network uri type unknown %d", uri->stype);
a74934ba 681 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
682 goto error;
683 }
684
685 ret = uri_compare(dst_uri, uri);
686 if (!ret) {
687 /* Same URI, don't touch it and return success. */
688 DBG3("URI network compare are the same");
ad20f474 689 goto equal;
00e2e675
DG
690 }
691
692 /* URIs were not equal, replacing it. */
00e2e675 693 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
b178f53e
JG
694 output->type = CONSUMER_DST_NET;
695 if (dst_uri->stype != LTTNG_STREAM_CONTROL) {
696 /* Only the control uri needs to contain the path. */
697 goto end;
698 }
00e2e675 699
b178f53e
JG
700 /*
701 * If the user has specified a subdir as part of the control
702 * URL, the session's base output directory is:
703 * /RELAYD_OUTPUT_PATH/HOSTNAME/USER_SPECIFIED_DIR
704 *
705 * Hence, the "base_dir" from which all stream files and
706 * session rotation chunks are created takes the form
707 * /HOSTNAME/USER_SPECIFIED_DIR
708 *
709 * If the user has not specified an output directory as part of
710 * the control URL, the base output directory has the form:
711 * /RELAYD_OUTPUT_PATH/HOSTNAME/SESSION_NAME-CREATION_TIME
712 *
713 * Hence, the "base_dir" from which all stream files and
714 * session rotation chunks are created takes the form
715 * /HOSTNAME/SESSION_NAME-CREATION_TIME
716 *
717 * Note that automatically generated session names already
718 * contain the session's creation time. In that case, the
719 * creation time is omitted to prevent it from being duplicated
720 * in the final directory hierarchy.
721 */
722 if (*uri->subdir) {
723 if (strstr(uri->subdir, "../")) {
724 ERR("Network URI subdirs are not allowed to walk up the path hierarchy");
725 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
726 goto error;
727 }
b178f53e
JG
728 ret = snprintf(output->dst.net.base_dir,
729 sizeof(output->dst.net.base_dir),
730 "/%s/%s/", session->hostname, uri->subdir);
731 } else {
732 if (session->has_auto_generated_name) {
733 ret = snprintf(output->dst.net.base_dir,
734 sizeof(output->dst.net.base_dir),
735 "/%s/%s/", session->hostname,
736 session->name);
737 } else {
738 char session_creation_datetime[16];
739 size_t strftime_ret;
740 struct tm *timeinfo;
00e2e675 741
b178f53e
JG
742 timeinfo = localtime(&session->creation_time);
743 if (!timeinfo) {
744 ret = -LTTNG_ERR_FATAL;
745 goto error;
746 }
747 strftime_ret = strftime(session_creation_datetime,
748 sizeof(session_creation_datetime),
749 "%Y%m%d-%H%M%S", timeinfo);
750 if (strftime_ret == 0) {
751 ERR("Failed to format session creation timestamp while setting network URI");
752 ret = -LTTNG_ERR_FATAL;
753 goto error;
754 }
755 ret = snprintf(output->dst.net.base_dir,
756 sizeof(output->dst.net.base_dir),
757 "/%s/%s-%s/", session->hostname,
758 session->name,
759 session_creation_datetime);
bfc6eff0 760 }
00e2e675 761 }
b178f53e
JG
762 if (ret >= sizeof(output->dst.net.base_dir)) {
763 ret = -LTTNG_ERR_INVALID;
764 ERR("Truncation occurred while setting network output base directory");
765 goto error;
766 } else if (ret == -1) {
767 ret = -LTTNG_ERR_INVALID;
768 PERROR("Error occurred while setting network output base directory");
769 goto error;
770 }
771
772 DBG3("Consumer set network uri base_dir path %s",
773 output->dst.net.base_dir);
00e2e675 774
b178f53e 775end:
00e2e675 776 return 0;
ad20f474
DG
777equal:
778 return 1;
00e2e675 779error:
a74934ba 780 return ret;
00e2e675
DG
781}
782
783/*
784 * Send file descriptor to consumer via sock.
9a318688
JG
785 *
786 * The consumer socket lock must be held by the caller.
00e2e675 787 */
ac2f30af
JG
788int consumer_send_fds(struct consumer_socket *sock, const int *fds,
789 size_t nb_fd)
00e2e675
DG
790{
791 int ret;
792
793 assert(fds);
f50f23d9 794 assert(sock);
00e2e675 795 assert(nb_fd > 0);
05dac598 796 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
00e2e675 797
9363801e 798 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
00e2e675 799 if (ret < 0) {
3448e266 800 /* The above call will print a PERROR on error. */
9363801e 801 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
00e2e675
DG
802 goto error;
803 }
804
f50f23d9 805 ret = consumer_recv_status_reply(sock);
00e2e675
DG
806error:
807 return ret;
808}
809
ffe60014
DG
810/*
811 * Consumer send communication message structure to consumer.
9a318688
JG
812 *
813 * The consumer socket lock must be held by the caller.
ffe60014
DG
814 */
815int consumer_send_msg(struct consumer_socket *sock,
816 struct lttcomm_consumer_msg *msg)
817{
818 int ret;
819
820 assert(msg);
821 assert(sock);
211b734b 822 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
ffe60014 823
52898cb1 824 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
ffe60014 825 if (ret < 0) {
ffe60014
DG
826 goto error;
827 }
828
829 ret = consumer_recv_status_reply(sock);
830
831error:
832 return ret;
833}
834
00e2e675
DG
835/*
836 * Consumer send channel communication message structure to consumer.
9a318688
JG
837 *
838 * The consumer socket lock must be held by the caller.
00e2e675 839 */
f50f23d9
DG
840int consumer_send_channel(struct consumer_socket *sock,
841 struct lttcomm_consumer_msg *msg)
00e2e675
DG
842{
843 int ret;
844
845 assert(msg);
f50f23d9 846 assert(sock);
00e2e675 847
52898cb1 848 ret = consumer_send_msg(sock, msg);
00e2e675 849 if (ret < 0) {
00e2e675
DG
850 goto error;
851 }
852
853error:
854 return ret;
855}
856
ffe60014
DG
857/*
858 * Populate the given consumer msg structure with the ask_channel command
859 * information.
860 */
861void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
862 uint64_t subbuf_size,
863 uint64_t num_subbuf,
864 int overwrite,
865 unsigned int switch_timer_interval,
866 unsigned int read_timer_interval,
ecc48a90 867 unsigned int live_timer_interval,
e9404c27 868 unsigned int monitor_timer_interval,
ffe60014
DG
869 int output,
870 int type,
871 uint64_t session_id,
872 const char *pathname,
873 const char *name,
d88aee68
DG
874 uint64_t relayd_id,
875 uint64_t key,
7972aab2 876 unsigned char *uuid,
1624d5b7
JD
877 uint32_t chan_id,
878 uint64_t tracefile_size,
2bba9e53 879 uint64_t tracefile_count,
1950109e 880 uint64_t session_id_per_pid,
567eb353 881 unsigned int monitor,
d7ba1388 882 uint32_t ust_app_uid,
491d1539 883 int64_t blocking_timeout,
3d071855 884 const char *root_shm_path,
e098433c 885 const char *shm_path,
1b0bebcb
JG
886 struct lttng_trace_chunk *trace_chunk,
887 const struct lttng_credentials *buffer_credentials)
ffe60014
DG
888{
889 assert(msg);
890
d2956687 891 /* Zeroed structure */
ffe60014 892 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
d2956687
JG
893 msg->u.ask_channel.buffer_credentials.uid = UINT32_MAX;
894 msg->u.ask_channel.buffer_credentials.gid = UINT32_MAX;
895
d2956687
JG
896 if (trace_chunk) {
897 uint64_t chunk_id;
898 enum lttng_trace_chunk_status chunk_status;
d2956687
JG
899
900 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
901 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
902 LTTNG_OPTIONAL_SET(&msg->u.ask_channel.chunk_id, chunk_id);
d2956687 903 }
1b0bebcb
JG
904 msg->u.ask_channel.buffer_credentials.uid = buffer_credentials->uid;
905 msg->u.ask_channel.buffer_credentials.gid = buffer_credentials->gid;
ffe60014
DG
906
907 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
908 msg->u.ask_channel.subbuf_size = subbuf_size;
909 msg->u.ask_channel.num_subbuf = num_subbuf ;
910 msg->u.ask_channel.overwrite = overwrite;
911 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
912 msg->u.ask_channel.read_timer_interval = read_timer_interval;
ecc48a90 913 msg->u.ask_channel.live_timer_interval = live_timer_interval;
e9404c27 914 msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
ffe60014
DG
915 msg->u.ask_channel.output = output;
916 msg->u.ask_channel.type = type;
917 msg->u.ask_channel.session_id = session_id;
1950109e 918 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
ffe60014
DG
919 msg->u.ask_channel.relayd_id = relayd_id;
920 msg->u.ask_channel.key = key;
7972aab2 921 msg->u.ask_channel.chan_id = chan_id;
1624d5b7
JD
922 msg->u.ask_channel.tracefile_size = tracefile_size;
923 msg->u.ask_channel.tracefile_count = tracefile_count;
2bba9e53 924 msg->u.ask_channel.monitor = monitor;
567eb353 925 msg->u.ask_channel.ust_app_uid = ust_app_uid;
491d1539 926 msg->u.ask_channel.blocking_timeout = blocking_timeout;
ffe60014
DG
927
928 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
929
10a50311
JD
930 if (pathname) {
931 strncpy(msg->u.ask_channel.pathname, pathname,
932 sizeof(msg->u.ask_channel.pathname));
933 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
934 }
ffe60014
DG
935
936 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
937 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
d7ba1388 938
3d071855
MD
939 if (root_shm_path) {
940 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
941 sizeof(msg->u.ask_channel.root_shm_path));
942 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
943 }
d7ba1388
MD
944 if (shm_path) {
945 strncpy(msg->u.ask_channel.shm_path, shm_path,
946 sizeof(msg->u.ask_channel.shm_path));
947 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
948 }
ffe60014
DG
949}
950
00e2e675
DG
951/*
952 * Init channel communication message structure.
953 */
638e7b4e 954void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
d88aee68 955 uint64_t channel_key,
ffe60014
DG
956 uint64_t session_id,
957 const char *pathname,
958 uid_t uid,
959 gid_t gid,
d88aee68 960 uint64_t relayd_id,
c30aaa51 961 const char *name,
ffe60014
DG
962 unsigned int nb_init_streams,
963 enum lttng_event_output output,
1624d5b7
JD
964 int type,
965 uint64_t tracefile_size,
2bba9e53 966 uint64_t tracefile_count,
ecc48a90 967 unsigned int monitor,
e9404c27 968 unsigned int live_timer_interval,
d2956687
JG
969 unsigned int monitor_timer_interval,
970 struct lttng_trace_chunk *trace_chunk)
00e2e675
DG
971{
972 assert(msg);
973
00e2e675
DG
974 /* Zeroed structure */
975 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
976
d2956687
JG
977 if (trace_chunk) {
978 uint64_t chunk_id;
979 enum lttng_trace_chunk_status chunk_status;
980
981 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
982 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
983 LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
984 }
985
00e2e675 986 /* Send channel */
638e7b4e 987 msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
00e2e675 988 msg->u.channel.channel_key = channel_key;
ffe60014 989 msg->u.channel.session_id = session_id;
ffe60014 990 msg->u.channel.relayd_id = relayd_id;
c30aaa51 991 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
992 msg->u.channel.output = output;
993 msg->u.channel.type = type;
1624d5b7
JD
994 msg->u.channel.tracefile_size = tracefile_size;
995 msg->u.channel.tracefile_count = tracefile_count;
2bba9e53 996 msg->u.channel.monitor = monitor;
ecc48a90 997 msg->u.channel.live_timer_interval = live_timer_interval;
e9404c27 998 msg->u.channel.monitor_timer_interval = monitor_timer_interval;
ffe60014
DG
999
1000 strncpy(msg->u.channel.pathname, pathname,
1001 sizeof(msg->u.channel.pathname));
1002 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
1003
1004 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
1005 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
00e2e675
DG
1006}
1007
1008/*
1009 * Init stream communication message structure.
1010 */
e098433c 1011void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
d88aee68
DG
1012 uint64_t channel_key,
1013 uint64_t stream_key,
d2956687 1014 int32_t cpu)
00e2e675
DG
1015{
1016 assert(msg);
1017
1018 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1019
e098433c 1020 msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
00e2e675
DG
1021 msg->u.stream.channel_key = channel_key;
1022 msg->u.stream.stream_key = stream_key;
ffe60014 1023 msg->u.stream.cpu = cpu;
00e2e675
DG
1024}
1025
a4baae1b
JD
1026void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
1027 enum lttng_consumer_command cmd,
1028 uint64_t channel_key, uint64_t net_seq_idx)
1029{
1030 assert(msg);
1031
1032 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1033
1034 msg->cmd_type = cmd;
1035 msg->u.sent_streams.channel_key = channel_key;
1036 msg->u.sent_streams.net_seq_idx = net_seq_idx;
1037}
1038
00e2e675
DG
1039/*
1040 * Send stream communication structure to the consumer.
1041 */
f50f23d9
DG
1042int consumer_send_stream(struct consumer_socket *sock,
1043 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
ac2f30af 1044 const int *fds, size_t nb_fd)
00e2e675
DG
1045{
1046 int ret;
1047
1048 assert(msg);
1049 assert(dst);
f50f23d9 1050 assert(sock);
ffe60014 1051 assert(fds);
00e2e675 1052
52898cb1 1053 ret = consumer_send_msg(sock, msg);
f50f23d9
DG
1054 if (ret < 0) {
1055 goto error;
1056 }
1057
00e2e675
DG
1058 ret = consumer_send_fds(sock, fds, nb_fd);
1059 if (ret < 0) {
1060 goto error;
1061 }
1062
1063error:
1064 return ret;
1065}
37278a1e
DG
1066
1067/*
1068 * Send relayd socket to consumer associated with a session name.
1069 *
43fade62
JG
1070 * The consumer socket lock must be held by the caller.
1071 *
37278a1e
DG
1072 * On success return positive value. On error, negative value.
1073 */
f50f23d9 1074int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
6151a90f 1075 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
d3e2ba59 1076 enum lttng_stream_type type, uint64_t session_id,
fb9a95c4 1077 const char *session_name, const char *hostname,
db1da059
JG
1078 int session_live_timer, const uint64_t *current_chunk_id,
1079 time_t session_creation_time)
37278a1e
DG
1080{
1081 int ret;
1082 struct lttcomm_consumer_msg msg;
1083
1084 /* Code flow error. Safety net. */
6151a90f 1085 assert(rsock);
37278a1e 1086 assert(consumer);
f50f23d9 1087 assert(consumer_sock);
37278a1e 1088
53efb85a 1089 memset(&msg, 0, sizeof(msg));
37278a1e
DG
1090 /* Bail out if consumer is disabled */
1091 if (!consumer->enabled) {
f73fabfd 1092 ret = LTTNG_OK;
37278a1e
DG
1093 goto error;
1094 }
1095
d3e2ba59
JD
1096 if (type == LTTNG_STREAM_CONTROL) {
1097 ret = relayd_create_session(rsock,
1098 &msg.u.relayd_sock.relayd_session_id,
7d2f7452 1099 session_name, hostname, session_live_timer,
658f12fa 1100 consumer->snapshot, session_id,
db1da059
JG
1101 sessiond_uuid, current_chunk_id,
1102 session_creation_time);
d3e2ba59
JD
1103 if (ret < 0) {
1104 /* Close the control socket. */
1105 (void) relayd_close(rsock);
1106 goto error;
1107 }
1108 }
1109
37278a1e
DG
1110 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1111 /*
1112 * Assign network consumer output index using the temporary consumer since
1113 * this call should only be made from within a set_consumer_uri() function
1114 * call in the session daemon.
1115 */
1116 msg.u.relayd_sock.net_index = consumer->net_seq_index;
1117 msg.u.relayd_sock.type = type;
46e6455f 1118 msg.u.relayd_sock.session_id = session_id;
6151a90f 1119 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
37278a1e 1120
9363801e 1121 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
52898cb1 1122 ret = consumer_send_msg(consumer_sock, &msg);
f50f23d9
DG
1123 if (ret < 0) {
1124 goto error;
1125 }
1126
37278a1e 1127 DBG3("Sending relayd socket file descriptor to consumer");
6151a90f 1128 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
37278a1e
DG
1129 if (ret < 0) {
1130 goto error;
1131 }
1132
1133 DBG2("Consumer relayd socket sent");
1134
1135error:
1136 return ret;
1137}
173af62f 1138
62c43103
JD
1139static
1140int consumer_send_pipe(struct consumer_socket *consumer_sock,
1141 enum lttng_consumer_command cmd, int pipe)
e9404c27
JG
1142{
1143 int ret;
1144 struct lttcomm_consumer_msg msg;
62c43103
JD
1145 const char *pipe_name;
1146 const char *command_name;
1147
1148 switch (cmd) {
1149 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1150 pipe_name = "channel monitor";
1151 command_name = "SET_CHANNEL_MONITOR_PIPE";
1152 break;
62c43103
JD
1153 default:
1154 ERR("Unexpected command received in %s (cmd = %d)", __func__,
1155 (int) cmd);
1156 abort();
1157 }
e9404c27
JG
1158
1159 /* Code flow error. Safety net. */
1160
1161 memset(&msg, 0, sizeof(msg));
62c43103 1162 msg.cmd_type = cmd;
e9404c27 1163
3e4dc117 1164 pthread_mutex_lock(consumer_sock->lock);
62c43103 1165 DBG3("Sending %s command to consumer", command_name);
e9404c27
JG
1166 ret = consumer_send_msg(consumer_sock, &msg);
1167 if (ret < 0) {
1168 goto error;
1169 }
1170
62c43103
JD
1171 DBG3("Sending %s pipe %d to consumer on socket %d",
1172 pipe_name,
e9404c27
JG
1173 pipe, *consumer_sock->fd_ptr);
1174 ret = consumer_send_fds(consumer_sock, &pipe, 1);
1175 if (ret < 0) {
1176 goto error;
1177 }
1178
62c43103 1179 DBG2("%s pipe successfully sent", pipe_name);
e9404c27 1180error:
3e4dc117 1181 pthread_mutex_unlock(consumer_sock->lock);
e9404c27
JG
1182 return ret;
1183}
1184
62c43103
JD
1185int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock,
1186 int pipe)
1187{
1188 return consumer_send_pipe(consumer_sock,
1189 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1190}
1191
806e2684 1192/*
5e280d77
MD
1193 * Ask the consumer if the data is pending for the specific session id.
1194 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
806e2684 1195 */
d88aee68 1196int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
1197 struct consumer_output *consumer)
1198{
1199 int ret;
6d805429 1200 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
1201 struct consumer_socket *socket;
1202 struct lttng_ht_iter iter;
1203 struct lttcomm_consumer_msg msg;
1204
1205 assert(consumer);
1206
53efb85a 1207 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 1208
53efb85a
MD
1209 memset(&msg, 0, sizeof(msg));
1210 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
d88aee68 1211 msg.u.data_pending.session_id = session_id;
806e2684 1212
c8f59ee5 1213 /* Send command for each consumer */
b82c5c4d 1214 rcu_read_lock();
806e2684
DG
1215 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1216 node.node) {
806e2684 1217 pthread_mutex_lock(socket->lock);
52898cb1 1218 ret = consumer_socket_send(socket, &msg, sizeof(msg));
806e2684 1219 if (ret < 0) {
806e2684 1220 pthread_mutex_unlock(socket->lock);
b82c5c4d 1221 goto error_unlock;
806e2684
DG
1222 }
1223
f50f23d9
DG
1224 /*
1225 * No need for a recv reply status because the answer to the command is
1226 * the reply status message.
1227 */
1228
52898cb1
DG
1229 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1230 if (ret < 0) {
806e2684 1231 pthread_mutex_unlock(socket->lock);
b82c5c4d 1232 goto error_unlock;
806e2684 1233 }
806e2684
DG
1234 pthread_mutex_unlock(socket->lock);
1235
6d805429 1236 if (ret_code == 1) {
806e2684
DG
1237 break;
1238 }
1239 }
b82c5c4d 1240 rcu_read_unlock();
806e2684 1241
d88aee68
DG
1242 DBG("Consumer data is %s pending for session id %" PRIu64,
1243 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
1244 return ret_code;
1245
b82c5c4d
DG
1246error_unlock:
1247 rcu_read_unlock();
806e2684
DG
1248 return -1;
1249}
7972aab2
DG
1250
1251/*
1252 * Send a flush command to consumer using the given channel key.
1253 *
1254 * Return 0 on success else a negative value.
1255 */
1256int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1257{
1258 int ret;
1259 struct lttcomm_consumer_msg msg;
1260
1261 assert(socket);
7972aab2
DG
1262
1263 DBG2("Consumer flush channel key %" PRIu64, key);
1264
53efb85a 1265 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1266 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1267 msg.u.flush_channel.key = key;
1268
1269 pthread_mutex_lock(socket->lock);
1270 health_code_update();
1271
1272 ret = consumer_send_msg(socket, &msg);
1273 if (ret < 0) {
1274 goto end;
1275 }
1276
1277end:
1278 health_code_update();
1279 pthread_mutex_unlock(socket->lock);
1280 return ret;
1281}
1282
0dd01979
MD
1283/*
1284 * Send a clear quiescent command to consumer using the given channel key.
1285 *
1286 * Return 0 on success else a negative value.
1287 */
1288int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1289{
1290 int ret;
1291 struct lttcomm_consumer_msg msg;
1292
1293 assert(socket);
1294
1295 DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1296
1297 memset(&msg, 0, sizeof(msg));
1298 msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1299 msg.u.clear_quiescent_channel.key = key;
1300
1301 pthread_mutex_lock(socket->lock);
1302 health_code_update();
1303
1304 ret = consumer_send_msg(socket, &msg);
1305 if (ret < 0) {
1306 goto end;
1307 }
1308
1309end:
1310 health_code_update();
1311 pthread_mutex_unlock(socket->lock);
1312 return ret;
1313}
1314
7972aab2 1315/*
dc2bbdae
MD
1316 * Send a close metadata command to consumer using the given channel key.
1317 * Called with registry lock held.
7972aab2
DG
1318 *
1319 * Return 0 on success else a negative value.
1320 */
1321int consumer_close_metadata(struct consumer_socket *socket,
1322 uint64_t metadata_key)
1323{
1324 int ret;
1325 struct lttcomm_consumer_msg msg;
1326
1327 assert(socket);
7972aab2
DG
1328
1329 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1330
53efb85a 1331 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1332 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1333 msg.u.close_metadata.key = metadata_key;
1334
1335 pthread_mutex_lock(socket->lock);
1336 health_code_update();
1337
1338 ret = consumer_send_msg(socket, &msg);
1339 if (ret < 0) {
1340 goto end;
1341 }
1342
1343end:
1344 health_code_update();
1345 pthread_mutex_unlock(socket->lock);
1346 return ret;
1347}
1348
1349/*
1350 * Send a setup metdata command to consumer using the given channel key.
1351 *
1352 * Return 0 on success else a negative value.
1353 */
1354int consumer_setup_metadata(struct consumer_socket *socket,
1355 uint64_t metadata_key)
1356{
1357 int ret;
1358 struct lttcomm_consumer_msg msg;
1359
1360 assert(socket);
7972aab2
DG
1361
1362 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1363
53efb85a 1364 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1365 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1366 msg.u.setup_metadata.key = metadata_key;
1367
1368 pthread_mutex_lock(socket->lock);
1369 health_code_update();
1370
1371 ret = consumer_send_msg(socket, &msg);
1372 if (ret < 0) {
1373 goto end;
1374 }
1375
1376end:
1377 health_code_update();
1378 pthread_mutex_unlock(socket->lock);
1379 return ret;
1380}
1381
1382/*
dc2bbdae
MD
1383 * Send metadata string to consumer.
1384 * RCU read-side lock must be held to guarantee existence of socket.
7972aab2
DG
1385 *
1386 * Return 0 on success else a negative value.
1387 */
1388int consumer_push_metadata(struct consumer_socket *socket,
1389 uint64_t metadata_key, char *metadata_str, size_t len,
93ec662e 1390 size_t target_offset, uint64_t version)
7972aab2
DG
1391{
1392 int ret;
1393 struct lttcomm_consumer_msg msg;
1394
1395 assert(socket);
7972aab2 1396
9363801e 1397 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
7972aab2 1398
dc2bbdae
MD
1399 pthread_mutex_lock(socket->lock);
1400
53efb85a 1401 memset(&msg, 0, sizeof(msg));
7972aab2
DG
1402 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1403 msg.u.push_metadata.key = metadata_key;
1404 msg.u.push_metadata.target_offset = target_offset;
1405 msg.u.push_metadata.len = len;
93ec662e 1406 msg.u.push_metadata.version = version;
7972aab2 1407
7972aab2
DG
1408 health_code_update();
1409 ret = consumer_send_msg(socket, &msg);
331744e3 1410 if (ret < 0 || len == 0) {
7972aab2
DG
1411 goto end;
1412 }
1413
9363801e
DG
1414 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1415 len);
7972aab2 1416
52898cb1 1417 ret = consumer_socket_send(socket, metadata_str, len);
7972aab2
DG
1418 if (ret < 0) {
1419 goto end;
1420 }
1421
1422 health_code_update();
1423 ret = consumer_recv_status_reply(socket);
1424 if (ret < 0) {
1425 goto end;
1426 }
1427
1428end:
dc2bbdae 1429 pthread_mutex_unlock(socket->lock);
7972aab2 1430 health_code_update();
7972aab2
DG
1431 return ret;
1432}
6dc3064a
DG
1433
1434/*
1435 * Ask the consumer to snapshot a specific channel using the key.
1436 *
9a654598 1437 * Returns LTTNG_OK on success or else an LTTng error code.
6dc3064a 1438 */
9a654598 1439enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket,
348a81dc 1440 uint64_t key, const struct consumer_output *output, int metadata,
d2956687
JG
1441 uid_t uid, gid_t gid, const char *channel_path, int wait,
1442 uint64_t nb_packets_per_stream)
6dc3064a
DG
1443{
1444 int ret;
9a654598 1445 enum lttng_error_code status = LTTNG_OK;
6dc3064a
DG
1446 struct lttcomm_consumer_msg msg;
1447
1448 assert(socket);
6dc3064a 1449 assert(output);
6dc3064a
DG
1450
1451 DBG("Consumer snapshot channel key %" PRIu64, key);
1452
ee91bab2 1453 memset(&msg, 0, sizeof(msg));
6dc3064a
DG
1454 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1455 msg.u.snapshot_channel.key = key;
d07ceecd 1456 msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
6dc3064a
DG
1457 msg.u.snapshot_channel.metadata = metadata;
1458
348a81dc 1459 if (output->type == CONSUMER_DST_NET) {
d2956687 1460 msg.u.snapshot_channel.relayd_id =
348a81dc 1461 output->net_seq_index;
6dc3064a 1462 msg.u.snapshot_channel.use_relayd = 1;
6dc3064a 1463 } else {
07b86b52 1464 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
d2956687
JG
1465 }
1466 ret = lttng_strncpy(msg.u.snapshot_channel.pathname,
1467 channel_path,
1468 sizeof(msg.u.snapshot_channel.pathname));
1469 if (ret < 0) {
1470 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1471 sizeof(msg.u.snapshot_channel.pathname),
1472 strlen(channel_path),
1473 channel_path);
1474 status = LTTNG_ERR_SNAPSHOT_FAIL;
1475 goto error;
6dc3064a
DG
1476 }
1477
1478 health_code_update();
9d1103e6 1479 pthread_mutex_lock(socket->lock);
6dc3064a 1480 ret = consumer_send_msg(socket, &msg);
9d1103e6 1481 pthread_mutex_unlock(socket->lock);
6dc3064a 1482 if (ret < 0) {
9bbfb88c
MD
1483 switch (-ret) {
1484 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
9a654598 1485 status = LTTNG_ERR_CHAN_NOT_FOUND;
9bbfb88c
MD
1486 break;
1487 default:
9a654598 1488 status = LTTNG_ERR_SNAPSHOT_FAIL;
9bbfb88c
MD
1489 break;
1490 }
6dc3064a
DG
1491 goto error;
1492 }
1493
1494error:
1495 health_code_update();
9a654598 1496 return status;
6dc3064a 1497}
fb83fe64
JD
1498
1499/*
1500 * Ask the consumer the number of discarded events for a channel.
1501 */
1502int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1503 struct consumer_output *consumer, uint64_t *discarded)
1504{
1505 int ret;
1506 struct consumer_socket *socket;
1507 struct lttng_ht_iter iter;
1508 struct lttcomm_consumer_msg msg;
1509
1510 assert(consumer);
1511
1512 DBG3("Consumer discarded events id %" PRIu64, session_id);
1513
1514 memset(&msg, 0, sizeof(msg));
1515 msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1516 msg.u.discarded_events.session_id = session_id;
1517 msg.u.discarded_events.channel_key = channel_key;
1518
1519 *discarded = 0;
1520
1521 /* Send command for each consumer */
1522 rcu_read_lock();
1523 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1524 node.node) {
1525 uint64_t consumer_discarded = 0;
1526 pthread_mutex_lock(socket->lock);
1527 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1528 if (ret < 0) {
1529 pthread_mutex_unlock(socket->lock);
1530 goto end;
1531 }
1532
1533 /*
1534 * No need for a recv reply status because the answer to the
1535 * command is the reply status message.
1536 */
1537 ret = consumer_socket_recv(socket, &consumer_discarded,
1538 sizeof(consumer_discarded));
1539 if (ret < 0) {
1540 ERR("get discarded events");
1541 pthread_mutex_unlock(socket->lock);
1542 goto end;
1543 }
1544 pthread_mutex_unlock(socket->lock);
1545 *discarded += consumer_discarded;
1546 }
1547 ret = 0;
1548 DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1549 *discarded, session_id);
1550
1551end:
1552 rcu_read_unlock();
1553 return ret;
1554}
1555
1556/*
1557 * Ask the consumer the number of lost packets for a channel.
1558 */
1559int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1560 struct consumer_output *consumer, uint64_t *lost)
1561{
1562 int ret;
1563 struct consumer_socket *socket;
1564 struct lttng_ht_iter iter;
1565 struct lttcomm_consumer_msg msg;
1566
1567 assert(consumer);
1568
1569 DBG3("Consumer lost packets id %" PRIu64, session_id);
1570
1571 memset(&msg, 0, sizeof(msg));
1572 msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1573 msg.u.lost_packets.session_id = session_id;
1574 msg.u.lost_packets.channel_key = channel_key;
1575
1576 *lost = 0;
1577
1578 /* Send command for each consumer */
1579 rcu_read_lock();
1580 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1581 node.node) {
1582 uint64_t consumer_lost = 0;
1583 pthread_mutex_lock(socket->lock);
1584 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1585 if (ret < 0) {
1586 pthread_mutex_unlock(socket->lock);
1587 goto end;
1588 }
1589
1590 /*
1591 * No need for a recv reply status because the answer to the
1592 * command is the reply status message.
1593 */
1594 ret = consumer_socket_recv(socket, &consumer_lost,
1595 sizeof(consumer_lost));
1596 if (ret < 0) {
1597 ERR("get lost packets");
1598 pthread_mutex_unlock(socket->lock);
1599 goto end;
1600 }
1601 pthread_mutex_unlock(socket->lock);
1602 *lost += consumer_lost;
1603 }
1604 ret = 0;
1605 DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1606 *lost, session_id);
1607
1608end:
1609 rcu_read_unlock();
1610 return ret;
1611}
a1ae2ea5 1612
5c408ad8
JD
1613/*
1614 * Ask the consumer to rotate a channel.
5c408ad8
JD
1615 *
1616 * The new_chunk_id is the session->rotate_count that has been incremented
1617 * when the rotation started. On the relay, this allows to keep track in which
1618 * chunk each stream is currently writing to (for the rotate_pending operation).
1619 */
1620int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key,
1621 uid_t uid, gid_t gid, struct consumer_output *output,
d2956687 1622 bool is_metadata_channel)
5c408ad8
JD
1623{
1624 int ret;
1625 struct lttcomm_consumer_msg msg;
1626
1627 assert(socket);
1628
1629 DBG("Consumer rotate channel key %" PRIu64, key);
1630
1631 pthread_mutex_lock(socket->lock);
1632 memset(&msg, 0, sizeof(msg));
1633 msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1634 msg.u.rotate_channel.key = key;
1635 msg.u.rotate_channel.metadata = !!is_metadata_channel;
5c408ad8
JD
1636
1637 if (output->type == CONSUMER_DST_NET) {
1638 msg.u.rotate_channel.relayd_id = output->net_seq_index;
5c408ad8
JD
1639 } else {
1640 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
5c408ad8
JD
1641 }
1642
1643 health_code_update();
1644 ret = consumer_send_msg(socket, &msg);
1645 if (ret < 0) {
20f37cb4
MD
1646 switch (-ret) {
1647 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1648 ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1649 break;
1650 default:
1651 ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1652 break;
1653 }
5c408ad8
JD
1654 goto error;
1655 }
5c408ad8
JD
1656error:
1657 pthread_mutex_unlock(socket->lock);
1658 health_code_update();
1659 return ret;
1660}
1661
d2956687
JG
1662int consumer_init(struct consumer_socket *socket,
1663 const lttng_uuid sessiond_uuid)
00fb02ac
JD
1664{
1665 int ret;
d2956687
JG
1666 struct lttcomm_consumer_msg msg = {
1667 .cmd_type = LTTNG_CONSUMER_INIT,
1668 };
00fb02ac
JD
1669
1670 assert(socket);
00fb02ac 1671
d2956687
JG
1672 DBG("Sending consumer initialization command");
1673 lttng_uuid_copy(msg.u.init.sessiond_uuid, sessiond_uuid);
00fb02ac
JD
1674
1675 health_code_update();
1676 ret = consumer_send_msg(socket, &msg);
1677 if (ret < 0) {
1678 goto error;
1679 }
1680
d88744a4
JD
1681error:
1682 health_code_update();
1683 return ret;
1684}
1685
1686/*
d2956687 1687 * Ask the consumer to create a new chunk for a given session.
92816cc3 1688 *
d2956687 1689 * Called with the consumer socket lock held.
92816cc3 1690 */
d2956687
JG
1691int consumer_create_trace_chunk(struct consumer_socket *socket,
1692 uint64_t relayd_id, uint64_t session_id,
1693 struct lttng_trace_chunk *chunk)
92816cc3
JG
1694{
1695 int ret;
d2956687
JG
1696 enum lttng_trace_chunk_status chunk_status;
1697 struct lttng_credentials chunk_credentials;
1698 const struct lttng_directory_handle *chunk_directory_handle;
1699 int chunk_dirfd;
1700 const char *chunk_name;
1701 bool chunk_name_overriden;
1702 uint64_t chunk_id;
1703 time_t creation_timestamp;
1704 char creation_timestamp_buffer[ISO8601_STR_LEN];
1705 const char *creation_timestamp_str = "(none)";
1706 const bool chunk_has_local_output = relayd_id == -1ULL;
1707 struct lttcomm_consumer_msg msg = {
1708 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1709 .u.create_trace_chunk.session_id = session_id,
1710 };
92816cc3
JG
1711
1712 assert(socket);
d2956687 1713 assert(chunk);
92816cc3 1714
d2956687
JG
1715 if (relayd_id != -1ULL) {
1716 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id,
1717 relayd_id);
1718 }
92816cc3 1719
d2956687
JG
1720 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name,
1721 &chunk_name_overriden);
1722 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1723 chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1724 ERR("Failed to get name of trace chunk");
1725 ret = -LTTNG_ERR_FATAL;
92816cc3
JG
1726 goto error;
1727 }
d2956687
JG
1728 if (chunk_name_overriden) {
1729 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1730 chunk_name,
1731 sizeof(msg.u.create_trace_chunk.override_name));
1732 if (ret) {
1733 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1734 chunk_name);
1735 ret = -LTTNG_ERR_FATAL;
1736 goto error;
1737 }
1738 }
92816cc3 1739
d2956687
JG
1740 chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk,
1741 &creation_timestamp);
1742 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1743 ret = -LTTNG_ERR_FATAL;
92816cc3
JG
1744 goto error;
1745 }
d2956687
JG
1746 msg.u.create_trace_chunk.creation_timestamp =
1747 (uint64_t) creation_timestamp;
1748 /* Only used for logging purposes. */
1749 ret = time_to_iso8601_str(creation_timestamp,
1750 creation_timestamp_buffer,
1751 sizeof(creation_timestamp_buffer));
1752 creation_timestamp_str = !ret ? creation_timestamp_buffer :
1753 "(formatting error)";
1754
1755 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1756 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1757 /*
1758 * Anonymous trace chunks should never be transmitted
1759 * to remote peers (consumerd and relayd). They are used
1760 * internally for backward-compatibility purposes.
1761 */
1762 ret = -LTTNG_ERR_FATAL;
1763 goto error;
1764 }
1765 msg.u.create_trace_chunk.chunk_id = chunk_id;
92816cc3 1766
d2956687
JG
1767 if (chunk_has_local_output) {
1768 chunk_status = lttng_trace_chunk_get_chunk_directory_handle(
1769 chunk, &chunk_directory_handle);
1770 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1771 ret = -LTTNG_ERR_FATAL;
1772 goto error;
1773 }
d88744a4 1774
d2956687
JG
1775 /*
1776 * This will only compile on platforms that support
1777 * dirfd (POSIX.2008). This is fine as the session daemon
1778 * is only built for such platforms.
1779 *
1780 * The ownership of the chunk directory handle's is maintained
1781 * by the trace chunk.
1782 */
1783 chunk_dirfd = lttng_directory_handle_get_dirfd(
1784 chunk_directory_handle);
1785 assert(chunk_dirfd >= 0);
d88744a4 1786
e5add6d0
JG
1787 chunk_status = lttng_trace_chunk_get_credentials(
1788 chunk, &chunk_credentials);
1789 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1790 /*
1791 * Not associating credentials to a sessiond chunk is a
1792 * fatal internal error.
1793 */
1794 ret = -LTTNG_ERR_FATAL;
1795 goto error;
1796 }
1797 msg.u.create_trace_chunk.credentials.value.uid =
1798 chunk_credentials.uid;
1799 msg.u.create_trace_chunk.credentials.value.gid =
1800 chunk_credentials.gid;
1801 msg.u.create_trace_chunk.credentials.is_set = 1;
d2956687 1802 }
d88744a4 1803
d2956687
JG
1804 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1805 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1806 ", creation_timestamp = %s",
1807 relayd_id, session_id, chunk_id,
1808 creation_timestamp_str);
d88744a4
JD
1809 health_code_update();
1810 ret = consumer_send_msg(socket, &msg);
d2956687 1811 health_code_update();
d88744a4 1812 if (ret < 0) {
d2956687
JG
1813 ERR("Trace chunk creation error on consumer");
1814 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
d88744a4
JD
1815 goto error;
1816 }
1817
d2956687
JG
1818 if (chunk_has_local_output) {
1819 DBG("Sending trace chunk directory fd to consumer");
1820 health_code_update();
1821 ret = consumer_send_fds(socket, &chunk_dirfd, 1);
1822 health_code_update();
1823 if (ret < 0) {
1824 ERR("Trace chunk creation error on consumer");
1825 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1826 goto error;
1827 }
d88744a4 1828 }
00fb02ac 1829error:
00fb02ac
JD
1830 return ret;
1831}
1832
a1ae2ea5 1833/*
d2956687 1834 * Ask the consumer to close a trace chunk for a given session.
a1ae2ea5
JD
1835 *
1836 * Called with the consumer socket lock held.
1837 */
d2956687
JG
1838int consumer_close_trace_chunk(struct consumer_socket *socket,
1839 uint64_t relayd_id, uint64_t session_id,
1840 struct lttng_trace_chunk *chunk)
a1ae2ea5
JD
1841{
1842 int ret;
d2956687
JG
1843 enum lttng_trace_chunk_status chunk_status;
1844 struct lttcomm_consumer_msg msg = {
bbc4768c
JG
1845 .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1846 .u.close_trace_chunk.session_id = session_id,
d2956687
JG
1847 };
1848 uint64_t chunk_id;
1849 time_t close_timestamp;
bbc4768c
JG
1850 enum lttng_trace_chunk_command_type close_command;
1851 const char *close_command_name = "none";
a1ae2ea5
JD
1852
1853 assert(socket);
1854
d2956687 1855 if (relayd_id != -1ULL) {
bbc4768c
JG
1856 LTTNG_OPTIONAL_SET(
1857 &msg.u.close_trace_chunk.relayd_id, relayd_id);
1858 }
1859
1860 chunk_status = lttng_trace_chunk_get_close_command(
1861 chunk, &close_command);
1862 switch (chunk_status) {
1863 case LTTNG_TRACE_CHUNK_STATUS_OK:
1864 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
1865 (uint32_t) close_command);
1866 break;
1867 case LTTNG_TRACE_CHUNK_STATUS_NONE:
1868 break;
1869 default:
1870 ERR("Failed to get trace chunk close command");
1871 ret = -1;
1872 goto error;
a1ae2ea5
JD
1873 }
1874
d2956687
JG
1875 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1876 /*
1877 * Anonymous trace chunks should never be transmitted to remote peers
1878 * (consumerd and relayd). They are used internally for
1879 * backward-compatibility purposes.
1880 */
1881 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1882 msg.u.close_trace_chunk.chunk_id = chunk_id;
1883
1884 chunk_status = lttng_trace_chunk_get_close_timestamp(chunk,
1885 &close_timestamp);
1886 /*
1887 * A trace chunk should be closed locally before being closed remotely.
1888 * Otherwise, the close timestamp would never be transmitted to the
1889 * peers.
1890 */
1891 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1892 msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
1893
bbc4768c
JG
1894 if (msg.u.close_trace_chunk.close_command.is_set) {
1895 close_command_name = lttng_trace_chunk_command_type_get_name(
1896 close_command);
1897 }
d2956687 1898 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
bbc4768c
JG
1899 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1900 ", close command = \"%s\"",
1901 relayd_id, session_id, chunk_id, close_command_name);
a1ae2ea5
JD
1902
1903 health_code_update();
1904 ret = consumer_send_msg(socket, &msg);
1905 if (ret < 0) {
d2956687 1906 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
a1ae2ea5
JD
1907 goto error;
1908 }
1909
1910error:
1911 health_code_update();
1912 return ret;
1913}
3654ed19 1914
d2956687
JG
1915/*
1916 * Ask the consumer if a trace chunk exists.
1917 *
1918 * Called with the consumer socket lock held.
1919 * Returns 0 on success, or a negative value on error.
1920 */
1921int consumer_trace_chunk_exists(struct consumer_socket *socket,
1922 uint64_t relayd_id, uint64_t session_id,
1923 struct lttng_trace_chunk *chunk,
1924 enum consumer_trace_chunk_exists_status *result)
3654ed19
JG
1925{
1926 int ret;
d2956687 1927 enum lttng_trace_chunk_status chunk_status;
3654ed19 1928 struct lttcomm_consumer_msg msg = {
d2956687
JG
1929 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
1930 .u.trace_chunk_exists.session_id = session_id,
3654ed19 1931 };
d2956687
JG
1932 uint64_t chunk_id;
1933 const char *consumer_reply_str;
3654ed19
JG
1934
1935 assert(socket);
1936
d2956687
JG
1937 if (relayd_id != -1ULL) {
1938 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id,
1939 relayd_id);
1940 }
1941
1942 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1943 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1944 /*
1945 * Anonymous trace chunks should never be transmitted
1946 * to remote peers (consumerd and relayd). They are used
1947 * internally for backward-compatibility purposes.
1948 */
1949 ret = -LTTNG_ERR_FATAL;
1950 goto error;
1951 }
1952 msg.u.trace_chunk_exists.chunk_id = chunk_id;
1953
1954 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
1955 ", session_id = %" PRIu64
1956 ", chunk_id = %" PRIu64, relayd_id, session_id, chunk_id);
3654ed19
JG
1957
1958 health_code_update();
1959 ret = consumer_send_msg(socket, &msg);
d2956687
JG
1960 switch (-ret) {
1961 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
1962 consumer_reply_str = "unknown trace chunk";
1963 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
1964 break;
1965 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
1966 consumer_reply_str = "trace chunk exists locally";
1967 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
1968 break;
1969 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
1970 consumer_reply_str = "trace chunk exists on remote peer";
1971 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
1972 break;
1973 default:
1974 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
1975 ret = -1;
3654ed19
JG
1976 goto error;
1977 }
d2956687
JG
1978 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
1979 consumer_reply_str);
1980 ret = 0;
3654ed19
JG
1981error:
1982 health_code_update();
1983 return ret;
1984}
This page took 0.151802 seconds and 4 git commands to generate.