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