Rename LTTng index in CTF index
[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
18#define _GNU_SOURCE
19#include <assert.h>
20#include <stdio.h>
21#include <stdlib.h>
22#include <string.h>
23#include <sys/stat.h>
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
2f77fc4b
DG
210 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
211 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
212
213 pthread_mutex_lock(sock->lock);
52898cb1 214 ret = consumer_socket_send(sock, &msg, sizeof(msg));
2f77fc4b 215 if (ret < 0) {
52898cb1 216 goto error;
2f77fc4b
DG
217 }
218
f50f23d9
DG
219 /* Don't check the return value. The caller will do it. */
220 ret = consumer_recv_status_reply(sock);
221
2f77fc4b
DG
222 DBG2("Consumer send destroy relayd command done");
223
224error:
52898cb1 225 pthread_mutex_unlock(sock->lock);
2f77fc4b
DG
226 return ret;
227}
228
229/*
230 * For each consumer socket in the consumer output object, send a destroy
231 * relayd command.
232 */
233void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
234{
2f77fc4b
DG
235 struct lttng_ht_iter iter;
236 struct consumer_socket *socket;
237
238 assert(consumer);
239
240 /* Destroy any relayd connection */
6dc3064a 241 if (consumer->type == CONSUMER_DST_NET) {
2f77fc4b
DG
242 rcu_read_lock();
243 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
244 node.node) {
c617c0c6
MD
245 int ret;
246
2f77fc4b
DG
247 /* Send destroy relayd command */
248 ret = consumer_send_destroy_relayd(socket, consumer);
249 if (ret < 0) {
c5c45efa 250 DBG("Unable to send destroy relayd command to consumer");
2f77fc4b
DG
251 /* Continue since we MUST delete everything at this point. */
252 }
253 }
254 rcu_read_unlock();
255 }
256}
257
a4b92340
DG
258/*
259 * From a consumer_data structure, allocate and add a consumer socket to the
260 * consumer output.
261 *
262 * Return 0 on success, else negative value on error
263 */
264int consumer_create_socket(struct consumer_data *data,
265 struct consumer_output *output)
266{
267 int ret = 0;
268 struct consumer_socket *socket;
269
270 assert(data);
271
272 if (output == NULL || data->cmd_sock < 0) {
273 /*
274 * Not an error. Possible there is simply not spawned consumer or it's
275 * disabled for the tracing session asking the socket.
276 */
277 goto error;
278 }
279
280 rcu_read_lock();
281 socket = consumer_find_socket(data->cmd_sock, output);
282 rcu_read_unlock();
283 if (socket == NULL) {
4ce514c4 284 socket = consumer_allocate_socket(&data->cmd_sock);
a4b92340
DG
285 if (socket == NULL) {
286 ret = -1;
287 goto error;
288 }
289
2f77fc4b 290 socket->registered = 0;
a4b92340
DG
291 socket->lock = &data->lock;
292 rcu_read_lock();
293 consumer_add_socket(socket, output);
294 rcu_read_unlock();
295 }
296
6dc3064a
DG
297 socket->type = data->type;
298
a4b92340
DG
299 DBG3("Consumer socket created (fd: %d) and added to output",
300 data->cmd_sock);
301
302error:
303 return ret;
304}
305
7972aab2
DG
306/*
307 * Return the consumer socket from the given consumer output with the right
308 * bitness. On error, returns NULL.
309 *
310 * The caller MUST acquire a rcu read side lock and keep it until the socket
311 * object reference is not needed anymore.
312 */
313struct consumer_socket *consumer_find_socket_by_bitness(int bits,
314 struct consumer_output *consumer)
315{
316 int consumer_fd;
317 struct consumer_socket *socket = NULL;
318
319 switch (bits) {
320 case 64:
321 consumer_fd = uatomic_read(&ust_consumerd64_fd);
322 break;
323 case 32:
324 consumer_fd = uatomic_read(&ust_consumerd32_fd);
325 break;
326 default:
327 assert(0);
328 goto end;
329 }
330
331 socket = consumer_find_socket(consumer_fd, consumer);
332 if (!socket) {
333 ERR("Consumer socket fd %d not found in consumer obj %p",
334 consumer_fd, consumer);
335 }
336
337end:
338 return socket;
339}
340
173af62f
DG
341/*
342 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
343 * be acquired before calling this function and across use of the
344 * returned consumer_socket.
345 */
346struct consumer_socket *consumer_find_socket(int key,
347 struct consumer_output *consumer)
348{
349 struct lttng_ht_iter iter;
350 struct lttng_ht_node_ulong *node;
351 struct consumer_socket *socket = NULL;
352
353 /* Negative keys are lookup failures */
a4b92340 354 if (key < 0 || consumer == NULL) {
173af62f
DG
355 return NULL;
356 }
357
358 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
359 &iter);
360 node = lttng_ht_iter_get_node_ulong(&iter);
361 if (node != NULL) {
362 socket = caa_container_of(node, struct consumer_socket, node);
363 }
364
365 return socket;
366}
367
368/*
369 * Allocate a new consumer_socket and return the pointer.
370 */
4ce514c4 371struct consumer_socket *consumer_allocate_socket(int *fd)
173af62f
DG
372{
373 struct consumer_socket *socket = NULL;
374
4ce514c4
DG
375 assert(fd);
376
173af62f
DG
377 socket = zmalloc(sizeof(struct consumer_socket));
378 if (socket == NULL) {
379 PERROR("zmalloc consumer socket");
380 goto error;
381 }
382
9363801e 383 socket->fd_ptr = fd;
4ce514c4 384 lttng_ht_node_init_ulong(&socket->node, *fd);
173af62f
DG
385
386error:
387 return socket;
388}
389
390/*
391 * Add consumer socket to consumer output object. Read side lock must be
392 * acquired before calling this function.
393 */
394void consumer_add_socket(struct consumer_socket *sock,
395 struct consumer_output *consumer)
396{
397 assert(sock);
398 assert(consumer);
399
400 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
401}
402
403/*
404 * Delte consumer socket to consumer output object. Read side lock must be
405 * acquired before calling this function.
406 */
407void consumer_del_socket(struct consumer_socket *sock,
408 struct consumer_output *consumer)
409{
410 int ret;
411 struct lttng_ht_iter iter;
412
413 assert(sock);
414 assert(consumer);
415
416 iter.iter.node = &sock->node.node;
417 ret = lttng_ht_del(consumer->socks, &iter);
418 assert(!ret);
419}
420
421/*
422 * RCU destroy call function.
423 */
424static void destroy_socket_rcu(struct rcu_head *head)
425{
426 struct lttng_ht_node_ulong *node =
427 caa_container_of(head, struct lttng_ht_node_ulong, head);
428 struct consumer_socket *socket =
429 caa_container_of(node, struct consumer_socket, node);
430
431 free(socket);
432}
433
434/*
435 * Destroy and free socket pointer in a call RCU. Read side lock must be
436 * acquired before calling this function.
437 */
438void consumer_destroy_socket(struct consumer_socket *sock)
439{
440 assert(sock);
441
442 /*
443 * We DO NOT close the file descriptor here since it is global to the
2f77fc4b
DG
444 * session daemon and is closed only if the consumer dies or a custom
445 * consumer was registered,
173af62f 446 */
2f77fc4b 447 if (sock->registered) {
9363801e
DG
448 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
449 lttcomm_close_unix_sock(*sock->fd_ptr);
2f77fc4b 450 }
173af62f
DG
451
452 call_rcu(&sock->node.head, destroy_socket_rcu);
453}
454
00e2e675
DG
455/*
456 * Allocate and assign data to a consumer_output object.
457 *
458 * Return pointer to structure.
459 */
460struct consumer_output *consumer_create_output(enum consumer_dst_type type)
461{
462 struct consumer_output *output = NULL;
463
464 output = zmalloc(sizeof(struct consumer_output));
465 if (output == NULL) {
466 PERROR("zmalloc consumer_output");
467 goto error;
468 }
469
470 /* By default, consumer output is enabled */
471 output->enabled = 1;
472 output->type = type;
d88aee68 473 output->net_seq_index = (uint64_t) -1ULL;
173af62f
DG
474
475 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
476
477error:
478 return output;
479}
480
af706bb7
DG
481/*
482 * Iterate over the consumer output socket hash table and destroy them. The
483 * socket file descriptor are only closed if the consumer output was
484 * registered meaning it's an external consumer.
485 */
486void consumer_destroy_output_sockets(struct consumer_output *obj)
487{
488 struct lttng_ht_iter iter;
489 struct consumer_socket *socket;
490
491 if (!obj->socks) {
492 return;
493 }
494
495 rcu_read_lock();
496 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
497 consumer_del_socket(socket, obj);
498 consumer_destroy_socket(socket);
499 }
500 rcu_read_unlock();
501}
502
00e2e675
DG
503/*
504 * Delete the consumer_output object from the list and free the ptr.
36b588ed
MD
505 *
506 * Should *NOT* be called with RCU read-side lock held.
00e2e675
DG
507 */
508void consumer_destroy_output(struct consumer_output *obj)
509{
510 if (obj == NULL) {
511 return;
512 }
513
af706bb7 514 consumer_destroy_output_sockets(obj);
2f77fc4b 515
af706bb7 516 if (obj->socks) {
2f77fc4b 517 /* Finally destroy HT */
0b2dc8df 518 ht_cleanup_push(obj->socks);
00e2e675 519 }
173af62f 520
00e2e675
DG
521 free(obj);
522}
523
524/*
525 * Copy consumer output and returned the newly allocated copy.
36b588ed
MD
526 *
527 * Should *NOT* be called with RCU read-side lock held.
00e2e675
DG
528 */
529struct consumer_output *consumer_copy_output(struct consumer_output *obj)
530{
6dc3064a 531 int ret;
09a90bcd 532 struct lttng_ht *tmp_ht_ptr;
00e2e675
DG
533 struct consumer_output *output;
534
535 assert(obj);
536
537 output = consumer_create_output(obj->type);
538 if (output == NULL) {
539 goto error;
540 }
09a90bcd
DG
541 /* Avoid losing the HT reference after the memcpy() */
542 tmp_ht_ptr = output->socks;
00e2e675
DG
543
544 memcpy(output, obj, sizeof(struct consumer_output));
545
09a90bcd
DG
546 /* Putting back the HT pointer and start copying socket(s). */
547 output->socks = tmp_ht_ptr;
173af62f 548
6dc3064a
DG
549 ret = consumer_copy_sockets(output, obj);
550 if (ret < 0) {
551 goto malloc_error;
552 }
553
554error:
555 return output;
556
557malloc_error:
558 consumer_destroy_output(output);
559 return NULL;
560}
561
562/*
563 * Copy consumer sockets from src to dst.
564 *
565 * Return 0 on success or else a negative value.
566 */
567int consumer_copy_sockets(struct consumer_output *dst,
568 struct consumer_output *src)
569{
570 int ret = 0;
571 struct lttng_ht_iter iter;
572 struct consumer_socket *socket, *copy_sock;
573
574 assert(dst);
575 assert(src);
576
b82c5c4d 577 rcu_read_lock();
6dc3064a
DG
578 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
579 /* Ignore socket that are already there. */
9363801e 580 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
6dc3064a
DG
581 if (copy_sock) {
582 continue;
583 }
584
173af62f 585 /* Create new socket object. */
9363801e 586 copy_sock = consumer_allocate_socket(socket->fd_ptr);
173af62f 587 if (copy_sock == NULL) {
b82c5c4d 588 rcu_read_unlock();
6dc3064a
DG
589 ret = -ENOMEM;
590 goto error;
173af62f
DG
591 }
592
09a90bcd 593 copy_sock->registered = socket->registered;
6dc3064a
DG
594 /*
595 * This is valid because this lock is shared accross all consumer
596 * object being the global lock of the consumer data structure of the
597 * session daemon.
598 */
173af62f 599 copy_sock->lock = socket->lock;
6dc3064a 600 consumer_add_socket(copy_sock, dst);
173af62f 601 }
b82c5c4d 602 rcu_read_unlock();
173af62f 603
00e2e675 604error:
6dc3064a 605 return ret;
00e2e675
DG
606}
607
608/*
609 * Set network URI to the consumer output object.
610 *
ad20f474
DG
611 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
612 * error.
00e2e675
DG
613 */
614int consumer_set_network_uri(struct consumer_output *obj,
615 struct lttng_uri *uri)
616{
617 int ret;
618 char tmp_path[PATH_MAX];
619 char hostname[HOST_NAME_MAX];
620 struct lttng_uri *dst_uri = NULL;
621
622 /* Code flow error safety net. */
623 assert(obj);
624 assert(uri);
625
626 switch (uri->stype) {
627 case LTTNG_STREAM_CONTROL:
628 dst_uri = &obj->dst.net.control;
629 obj->dst.net.control_isset = 1;
630 if (uri->port == 0) {
631 /* Assign default port. */
632 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
a74934ba
DG
633 } else {
634 if (obj->dst.net.data_isset && uri->port ==
635 obj->dst.net.data.port) {
636 ret = -LTTNG_ERR_INVALID;
637 goto error;
638 }
00e2e675 639 }
ad20f474 640 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
641 break;
642 case LTTNG_STREAM_DATA:
643 dst_uri = &obj->dst.net.data;
644 obj->dst.net.data_isset = 1;
645 if (uri->port == 0) {
646 /* Assign default port. */
647 uri->port = DEFAULT_NETWORK_DATA_PORT;
a74934ba
DG
648 } else {
649 if (obj->dst.net.control_isset && uri->port ==
650 obj->dst.net.control.port) {
651 ret = -LTTNG_ERR_INVALID;
652 goto error;
653 }
00e2e675 654 }
ad20f474 655 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
656 break;
657 default:
658 ERR("Set network uri type unknown %d", uri->stype);
a74934ba 659 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
660 goto error;
661 }
662
663 ret = uri_compare(dst_uri, uri);
664 if (!ret) {
665 /* Same URI, don't touch it and return success. */
666 DBG3("URI network compare are the same");
ad20f474 667 goto equal;
00e2e675
DG
668 }
669
670 /* URIs were not equal, replacing it. */
671 memset(dst_uri, 0, sizeof(struct lttng_uri));
672 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
673 obj->type = CONSUMER_DST_NET;
674
675 /* Handle subdir and add hostname in front. */
676 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
677 /* Get hostname to append it in the pathname */
678 ret = gethostname(hostname, sizeof(hostname));
679 if (ret < 0) {
680 PERROR("gethostname. Fallback on default localhost");
681 strncpy(hostname, "localhost", sizeof(hostname));
682 }
683 hostname[sizeof(hostname) - 1] = '\0';
684
685 /* Setup consumer subdir if none present in the control URI */
686 if (strlen(dst_uri->subdir) == 0) {
687 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
688 hostname, obj->subdir);
689 } else {
690 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
691 hostname, dst_uri->subdir);
692 }
693 if (ret < 0) {
694 PERROR("snprintf set consumer uri subdir");
a74934ba 695 ret = -LTTNG_ERR_NOMEM;
00e2e675
DG
696 goto error;
697 }
698
699 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
700 DBG3("Consumer set network uri subdir path %s", tmp_path);
701 }
702
00e2e675 703 return 0;
ad20f474
DG
704equal:
705 return 1;
00e2e675 706error:
a74934ba 707 return ret;
00e2e675
DG
708}
709
710/*
711 * Send file descriptor to consumer via sock.
712 */
f50f23d9 713int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
00e2e675
DG
714{
715 int ret;
716
717 assert(fds);
f50f23d9 718 assert(sock);
00e2e675
DG
719 assert(nb_fd > 0);
720
9363801e 721 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
00e2e675 722 if (ret < 0) {
3448e266 723 /* The above call will print a PERROR on error. */
9363801e 724 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
00e2e675
DG
725 goto error;
726 }
727
f50f23d9
DG
728 ret = consumer_recv_status_reply(sock);
729
00e2e675
DG
730error:
731 return ret;
732}
733
ffe60014
DG
734/*
735 * Consumer send communication message structure to consumer.
736 */
737int consumer_send_msg(struct consumer_socket *sock,
738 struct lttcomm_consumer_msg *msg)
739{
740 int ret;
741
742 assert(msg);
743 assert(sock);
ffe60014 744
52898cb1 745 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
ffe60014 746 if (ret < 0) {
ffe60014
DG
747 goto error;
748 }
749
750 ret = consumer_recv_status_reply(sock);
751
752error:
753 return ret;
754}
755
00e2e675
DG
756/*
757 * Consumer send channel communication message structure to consumer.
758 */
f50f23d9
DG
759int consumer_send_channel(struct consumer_socket *sock,
760 struct lttcomm_consumer_msg *msg)
00e2e675
DG
761{
762 int ret;
763
764 assert(msg);
f50f23d9 765 assert(sock);
00e2e675 766
52898cb1 767 ret = consumer_send_msg(sock, msg);
00e2e675 768 if (ret < 0) {
00e2e675
DG
769 goto error;
770 }
771
772error:
773 return ret;
774}
775
ffe60014
DG
776/*
777 * Populate the given consumer msg structure with the ask_channel command
778 * information.
779 */
780void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
781 uint64_t subbuf_size,
782 uint64_t num_subbuf,
783 int overwrite,
784 unsigned int switch_timer_interval,
785 unsigned int read_timer_interval,
ecc48a90 786 unsigned int live_timer_interval,
ffe60014
DG
787 int output,
788 int type,
789 uint64_t session_id,
790 const char *pathname,
791 const char *name,
792 uid_t uid,
793 gid_t gid,
d88aee68
DG
794 uint64_t relayd_id,
795 uint64_t key,
7972aab2 796 unsigned char *uuid,
1624d5b7
JD
797 uint32_t chan_id,
798 uint64_t tracefile_size,
2bba9e53 799 uint64_t tracefile_count,
1950109e 800 uint64_t session_id_per_pid,
567eb353
DG
801 unsigned int monitor,
802 uint32_t ust_app_uid)
ffe60014
DG
803{
804 assert(msg);
805
806 /* Zeroed structure */
807 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
808
809 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
810 msg->u.ask_channel.subbuf_size = subbuf_size;
811 msg->u.ask_channel.num_subbuf = num_subbuf ;
812 msg->u.ask_channel.overwrite = overwrite;
813 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
814 msg->u.ask_channel.read_timer_interval = read_timer_interval;
ecc48a90 815 msg->u.ask_channel.live_timer_interval = live_timer_interval;
ffe60014
DG
816 msg->u.ask_channel.output = output;
817 msg->u.ask_channel.type = type;
818 msg->u.ask_channel.session_id = session_id;
1950109e 819 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
ffe60014
DG
820 msg->u.ask_channel.uid = uid;
821 msg->u.ask_channel.gid = gid;
822 msg->u.ask_channel.relayd_id = relayd_id;
823 msg->u.ask_channel.key = key;
7972aab2 824 msg->u.ask_channel.chan_id = chan_id;
1624d5b7
JD
825 msg->u.ask_channel.tracefile_size = tracefile_size;
826 msg->u.ask_channel.tracefile_count = tracefile_count;
2bba9e53 827 msg->u.ask_channel.monitor = monitor;
567eb353 828 msg->u.ask_channel.ust_app_uid = ust_app_uid;
ffe60014
DG
829
830 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
831
10a50311
JD
832 if (pathname) {
833 strncpy(msg->u.ask_channel.pathname, pathname,
834 sizeof(msg->u.ask_channel.pathname));
835 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
836 }
ffe60014
DG
837
838 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
839 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
840}
841
00e2e675
DG
842/*
843 * Init channel communication message structure.
844 */
845void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
846 enum lttng_consumer_command cmd,
d88aee68 847 uint64_t channel_key,
ffe60014
DG
848 uint64_t session_id,
849 const char *pathname,
850 uid_t uid,
851 gid_t gid,
d88aee68 852 uint64_t relayd_id,
c30aaa51 853 const char *name,
ffe60014
DG
854 unsigned int nb_init_streams,
855 enum lttng_event_output output,
1624d5b7
JD
856 int type,
857 uint64_t tracefile_size,
2bba9e53 858 uint64_t tracefile_count,
ecc48a90
JD
859 unsigned int monitor,
860 unsigned int live_timer_interval)
00e2e675
DG
861{
862 assert(msg);
863
00e2e675
DG
864 /* Zeroed structure */
865 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
866
867 /* Send channel */
868 msg->cmd_type = cmd;
869 msg->u.channel.channel_key = channel_key;
ffe60014
DG
870 msg->u.channel.session_id = session_id;
871 msg->u.channel.uid = uid;
872 msg->u.channel.gid = gid;
873 msg->u.channel.relayd_id = relayd_id;
c30aaa51 874 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
875 msg->u.channel.output = output;
876 msg->u.channel.type = type;
1624d5b7
JD
877 msg->u.channel.tracefile_size = tracefile_size;
878 msg->u.channel.tracefile_count = tracefile_count;
2bba9e53 879 msg->u.channel.monitor = monitor;
ecc48a90 880 msg->u.channel.live_timer_interval = live_timer_interval;
ffe60014
DG
881
882 strncpy(msg->u.channel.pathname, pathname,
883 sizeof(msg->u.channel.pathname));
884 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
885
886 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
887 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
00e2e675
DG
888}
889
890/*
891 * Init stream communication message structure.
892 */
893void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
894 enum lttng_consumer_command cmd,
d88aee68
DG
895 uint64_t channel_key,
896 uint64_t stream_key,
ffe60014 897 int cpu)
00e2e675
DG
898{
899 assert(msg);
900
901 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
902
00e2e675
DG
903 msg->cmd_type = cmd;
904 msg->u.stream.channel_key = channel_key;
905 msg->u.stream.stream_key = stream_key;
ffe60014 906 msg->u.stream.cpu = cpu;
00e2e675
DG
907}
908
909/*
910 * Send stream communication structure to the consumer.
911 */
f50f23d9
DG
912int consumer_send_stream(struct consumer_socket *sock,
913 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
914 int *fds, size_t nb_fd)
00e2e675
DG
915{
916 int ret;
917
918 assert(msg);
919 assert(dst);
f50f23d9 920 assert(sock);
ffe60014 921 assert(fds);
00e2e675 922
52898cb1 923 ret = consumer_send_msg(sock, msg);
f50f23d9
DG
924 if (ret < 0) {
925 goto error;
926 }
927
00e2e675
DG
928 ret = consumer_send_fds(sock, fds, nb_fd);
929 if (ret < 0) {
930 goto error;
931 }
932
933error:
934 return ret;
935}
37278a1e
DG
936
937/*
938 * Send relayd socket to consumer associated with a session name.
939 *
940 * On success return positive value. On error, negative value.
941 */
f50f23d9 942int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
6151a90f 943 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
d3e2ba59
JD
944 enum lttng_stream_type type, uint64_t session_id,
945 char *session_name, char *hostname, int session_live_timer)
37278a1e
DG
946{
947 int ret;
948 struct lttcomm_consumer_msg msg;
949
950 /* Code flow error. Safety net. */
6151a90f 951 assert(rsock);
37278a1e 952 assert(consumer);
f50f23d9 953 assert(consumer_sock);
37278a1e
DG
954
955 /* Bail out if consumer is disabled */
956 if (!consumer->enabled) {
f73fabfd 957 ret = LTTNG_OK;
37278a1e
DG
958 goto error;
959 }
960
d3e2ba59
JD
961 if (type == LTTNG_STREAM_CONTROL) {
962 ret = relayd_create_session(rsock,
963 &msg.u.relayd_sock.relayd_session_id,
7d2f7452
DG
964 session_name, hostname, session_live_timer,
965 consumer->snapshot);
d3e2ba59
JD
966 if (ret < 0) {
967 /* Close the control socket. */
968 (void) relayd_close(rsock);
969 goto error;
970 }
971 }
972
37278a1e
DG
973 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
974 /*
975 * Assign network consumer output index using the temporary consumer since
976 * this call should only be made from within a set_consumer_uri() function
977 * call in the session daemon.
978 */
979 msg.u.relayd_sock.net_index = consumer->net_seq_index;
980 msg.u.relayd_sock.type = type;
46e6455f 981 msg.u.relayd_sock.session_id = session_id;
6151a90f 982 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
37278a1e 983
9363801e 984 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
52898cb1 985 ret = consumer_send_msg(consumer_sock, &msg);
f50f23d9
DG
986 if (ret < 0) {
987 goto error;
988 }
989
37278a1e 990 DBG3("Sending relayd socket file descriptor to consumer");
6151a90f 991 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
37278a1e
DG
992 if (ret < 0) {
993 goto error;
994 }
995
996 DBG2("Consumer relayd socket sent");
997
998error:
999 return ret;
1000}
173af62f
DG
1001
1002/*
2f77fc4b
DG
1003 * Set consumer subdirectory using the session name and a generated datetime if
1004 * needed. This is appended to the current subdirectory.
173af62f 1005 */
2f77fc4b
DG
1006int consumer_set_subdir(struct consumer_output *consumer,
1007 const char *session_name)
173af62f 1008{
2f77fc4b
DG
1009 int ret = 0;
1010 unsigned int have_default_name = 0;
1011 char datetime[16], tmp_path[PATH_MAX];
1012 time_t rawtime;
1013 struct tm *timeinfo;
173af62f
DG
1014
1015 assert(consumer);
2f77fc4b
DG
1016 assert(session_name);
1017
1018 memset(tmp_path, 0, sizeof(tmp_path));
1019
1020 /* Flag if we have a default session. */
1021 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
1022 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
1023 have_default_name = 1;
1024 } else {
1025 /* Get date and time for session path */
1026 time(&rawtime);
1027 timeinfo = localtime(&rawtime);
1028 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
173af62f
DG
1029 }
1030
2f77fc4b
DG
1031 if (have_default_name) {
1032 ret = snprintf(tmp_path, sizeof(tmp_path),
1033 "%s/%s", consumer->subdir, session_name);
1034 } else {
1035 ret = snprintf(tmp_path, sizeof(tmp_path),
1036 "%s/%s-%s/", consumer->subdir, session_name, datetime);
1037 }
173af62f 1038 if (ret < 0) {
2f77fc4b 1039 PERROR("snprintf session name date");
173af62f
DG
1040 goto error;
1041 }
1042
2f77fc4b
DG
1043 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
1044 DBG2("Consumer subdir set to %s", consumer->subdir);
173af62f
DG
1045
1046error:
1047 return ret;
1048}
806e2684
DG
1049
1050/*
6d805429 1051 * Ask the consumer if the data is ready to read (NOT pending) for the specific
806e2684
DG
1052 * session id.
1053 *
1054 * This function has a different behavior with the consumer i.e. that it waits
6d805429 1055 * for a reply from the consumer if yes or no the data is pending.
806e2684 1056 */
d88aee68 1057int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
1058 struct consumer_output *consumer)
1059{
1060 int ret;
6d805429 1061 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
1062 struct consumer_socket *socket;
1063 struct lttng_ht_iter iter;
1064 struct lttcomm_consumer_msg msg;
1065
1066 assert(consumer);
1067
6d805429 1068 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
806e2684 1069
d88aee68 1070 msg.u.data_pending.session_id = session_id;
806e2684 1071
d88aee68 1072 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 1073
c8f59ee5 1074 /* Send command for each consumer */
b82c5c4d 1075 rcu_read_lock();
806e2684
DG
1076 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1077 node.node) {
806e2684 1078 pthread_mutex_lock(socket->lock);
52898cb1 1079 ret = consumer_socket_send(socket, &msg, sizeof(msg));
806e2684 1080 if (ret < 0) {
806e2684 1081 pthread_mutex_unlock(socket->lock);
b82c5c4d 1082 goto error_unlock;
806e2684
DG
1083 }
1084
f50f23d9
DG
1085 /*
1086 * No need for a recv reply status because the answer to the command is
1087 * the reply status message.
1088 */
1089
52898cb1
DG
1090 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1091 if (ret < 0) {
806e2684 1092 pthread_mutex_unlock(socket->lock);
b82c5c4d 1093 goto error_unlock;
806e2684 1094 }
806e2684
DG
1095 pthread_mutex_unlock(socket->lock);
1096
6d805429 1097 if (ret_code == 1) {
806e2684
DG
1098 break;
1099 }
1100 }
b82c5c4d 1101 rcu_read_unlock();
806e2684 1102
d88aee68
DG
1103 DBG("Consumer data is %s pending for session id %" PRIu64,
1104 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
1105 return ret_code;
1106
b82c5c4d
DG
1107error_unlock:
1108 rcu_read_unlock();
806e2684
DG
1109 return -1;
1110}
7972aab2
DG
1111
1112/*
1113 * Send a flush command to consumer using the given channel key.
1114 *
1115 * Return 0 on success else a negative value.
1116 */
1117int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1118{
1119 int ret;
1120 struct lttcomm_consumer_msg msg;
1121
1122 assert(socket);
7972aab2
DG
1123
1124 DBG2("Consumer flush channel key %" PRIu64, key);
1125
1126 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1127 msg.u.flush_channel.key = key;
1128
1129 pthread_mutex_lock(socket->lock);
1130 health_code_update();
1131
1132 ret = consumer_send_msg(socket, &msg);
1133 if (ret < 0) {
1134 goto end;
1135 }
1136
1137end:
1138 health_code_update();
1139 pthread_mutex_unlock(socket->lock);
1140 return ret;
1141}
1142
1143/*
1144 * Send a close metdata command to consumer using the given channel key.
1145 *
1146 * Return 0 on success else a negative value.
1147 */
1148int consumer_close_metadata(struct consumer_socket *socket,
1149 uint64_t metadata_key)
1150{
1151 int ret;
1152 struct lttcomm_consumer_msg msg;
1153
1154 assert(socket);
7972aab2
DG
1155
1156 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1157
1158 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1159 msg.u.close_metadata.key = metadata_key;
1160
1161 pthread_mutex_lock(socket->lock);
1162 health_code_update();
1163
1164 ret = consumer_send_msg(socket, &msg);
1165 if (ret < 0) {
1166 goto end;
1167 }
1168
1169end:
1170 health_code_update();
1171 pthread_mutex_unlock(socket->lock);
1172 return ret;
1173}
1174
1175/*
1176 * Send a setup metdata command to consumer using the given channel key.
1177 *
1178 * Return 0 on success else a negative value.
1179 */
1180int consumer_setup_metadata(struct consumer_socket *socket,
1181 uint64_t metadata_key)
1182{
1183 int ret;
1184 struct lttcomm_consumer_msg msg;
1185
1186 assert(socket);
7972aab2
DG
1187
1188 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1189
1190 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1191 msg.u.setup_metadata.key = metadata_key;
1192
1193 pthread_mutex_lock(socket->lock);
1194 health_code_update();
1195
1196 ret = consumer_send_msg(socket, &msg);
1197 if (ret < 0) {
1198 goto end;
1199 }
1200
1201end:
1202 health_code_update();
1203 pthread_mutex_unlock(socket->lock);
1204 return ret;
1205}
1206
1207/*
331744e3 1208 * Send metadata string to consumer. Socket lock MUST be acquired.
7972aab2
DG
1209 *
1210 * Return 0 on success else a negative value.
1211 */
1212int consumer_push_metadata(struct consumer_socket *socket,
1213 uint64_t metadata_key, char *metadata_str, size_t len,
1214 size_t target_offset)
1215{
1216 int ret;
1217 struct lttcomm_consumer_msg msg;
1218
1219 assert(socket);
7972aab2 1220
9363801e 1221 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
7972aab2
DG
1222
1223 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1224 msg.u.push_metadata.key = metadata_key;
1225 msg.u.push_metadata.target_offset = target_offset;
1226 msg.u.push_metadata.len = len;
1227
7972aab2
DG
1228 health_code_update();
1229 ret = consumer_send_msg(socket, &msg);
331744e3 1230 if (ret < 0 || len == 0) {
7972aab2
DG
1231 goto end;
1232 }
1233
9363801e
DG
1234 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1235 len);
7972aab2 1236
52898cb1 1237 ret = consumer_socket_send(socket, metadata_str, len);
7972aab2
DG
1238 if (ret < 0) {
1239 goto end;
1240 }
1241
1242 health_code_update();
1243 ret = consumer_recv_status_reply(socket);
1244 if (ret < 0) {
1245 goto end;
1246 }
1247
1248end:
1249 health_code_update();
7972aab2
DG
1250 return ret;
1251}
6dc3064a
DG
1252
1253/*
1254 * Ask the consumer to snapshot a specific channel using the key.
1255 *
1256 * Return 0 on success or else a negative error.
1257 */
1258int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1259 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
5c786ded 1260 const char *session_path, int wait, int max_stream_size)
6dc3064a
DG
1261{
1262 int ret;
1263 struct lttcomm_consumer_msg msg;
1264
1265 assert(socket);
6dc3064a
DG
1266 assert(output);
1267 assert(output->consumer);
1268
1269 DBG("Consumer snapshot channel key %" PRIu64, key);
1270
ee91bab2 1271 memset(&msg, 0, sizeof(msg));
6dc3064a
DG
1272 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1273 msg.u.snapshot_channel.key = key;
5c786ded 1274 msg.u.snapshot_channel.max_stream_size = max_stream_size;
6dc3064a
DG
1275 msg.u.snapshot_channel.metadata = metadata;
1276
1277 if (output->consumer->type == CONSUMER_DST_NET) {
1278 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1279 msg.u.snapshot_channel.use_relayd = 1;
1280 ret = snprintf(msg.u.snapshot_channel.pathname,
1bfe7328
DG
1281 sizeof(msg.u.snapshot_channel.pathname),
1282 "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir,
1283 output->name, output->datetime, output->nb_snapshot,
ee91bab2 1284 session_path);
6dc3064a
DG
1285 if (ret < 0) {
1286 ret = -LTTNG_ERR_NOMEM;
1287 goto error;
1288 }
1289 } else {
1290 ret = snprintf(msg.u.snapshot_channel.pathname,
1bfe7328
DG
1291 sizeof(msg.u.snapshot_channel.pathname),
1292 "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path,
1293 output->name, output->datetime, output->nb_snapshot,
1294 session_path);
6dc3064a
DG
1295 if (ret < 0) {
1296 ret = -LTTNG_ERR_NOMEM;
1297 goto error;
1298 }
07b86b52 1299 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
6dc3064a
DG
1300
1301 /* Create directory. Ignore if exist. */
1302 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1303 S_IRWXU | S_IRWXG, uid, gid);
1304 if (ret < 0) {
1305 if (ret != -EEXIST) {
1306 ERR("Trace directory creation error");
1307 goto error;
1308 }
1309 }
1310 }
1311
1312 health_code_update();
1313 ret = consumer_send_msg(socket, &msg);
1314 if (ret < 0) {
1315 goto error;
1316 }
1317
1318error:
1319 health_code_update();
1320 return ret;
1321}
This page took 0.095548 seconds and 4 git commands to generate.