Add snapshot command to lttng UI
[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>
31
32#include "consumer.h"
7972aab2
DG
33#include "health.h"
34#include "ust-app.h"
0b2dc8df 35#include "utils.h"
00e2e675 36
f50f23d9
DG
37/*
38 * Receive a reply command status message from the consumer. Consumer socket
39 * lock MUST be acquired before calling this function.
40 *
41 * Return 0 on success, -1 on recv error or a negative lttng error code which
42 * was possibly returned by the consumer.
43 */
44int consumer_recv_status_reply(struct consumer_socket *sock)
45{
46 int ret;
47 struct lttcomm_consumer_status_msg reply;
48
49 assert(sock);
50
51 ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
a6cd2b97
DG
52 if (ret <= 0) {
53 if (ret == 0) {
54 /* Orderly shutdown. Don't return 0 which means success. */
55 ret = -1;
56 }
3448e266
DG
57 /* The above call will print a PERROR on error. */
58 DBG("Fail to receive status reply on sock %d", sock->fd);
f50f23d9
DG
59 goto end;
60 }
61
62 if (reply.ret_code == LTTNG_OK) {
63 /* All good. */
64 ret = 0;
65 } else {
66 ret = -reply.ret_code;
ffe60014 67 DBG("Consumer ret code %d", ret);
f50f23d9
DG
68 }
69
70end:
71 return ret;
72}
73
ffe60014
DG
74/*
75 * Once the ASK_CHANNEL command is sent to the consumer, the channel
76 * information are sent back. This call receives that data and populates key
77 * and stream_count.
78 *
79 * On success return 0 and both key and stream_count are set. On error, a
80 * negative value is sent back and both parameters are untouched.
81 */
82int consumer_recv_status_channel(struct consumer_socket *sock,
d88aee68 83 uint64_t *key, unsigned int *stream_count)
ffe60014
DG
84{
85 int ret;
86 struct lttcomm_consumer_status_channel reply;
87
88 assert(sock);
89 assert(stream_count);
90 assert(key);
91
92 ret = lttcomm_recv_unix_sock(sock->fd, &reply, sizeof(reply));
93 if (ret <= 0) {
94 if (ret == 0) {
95 /* Orderly shutdown. Don't return 0 which means success. */
96 ret = -1;
97 }
98 /* The above call will print a PERROR on error. */
99 DBG("Fail to receive status reply on sock %d", sock->fd);
100 goto end;
101 }
102
103 /* An error is possible so don't touch the key and stream_count. */
104 if (reply.ret_code != LTTNG_OK) {
105 ret = -1;
106 goto end;
107 }
108
109 *key = reply.key;
110 *stream_count = reply.stream_count;
111
112end:
113 return ret;
114}
115
2f77fc4b
DG
116/*
117 * Send destroy relayd command to consumer.
118 *
119 * On success return positive value. On error, negative value.
120 */
121int consumer_send_destroy_relayd(struct consumer_socket *sock,
122 struct consumer_output *consumer)
123{
124 int ret;
125 struct lttcomm_consumer_msg msg;
126
127 assert(consumer);
128 assert(sock);
129
ffe60014 130 DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
2f77fc4b
DG
131
132 /* Bail out if consumer is disabled */
133 if (!consumer->enabled) {
f73fabfd 134 ret = LTTNG_OK;
2f77fc4b
DG
135 DBG3("Consumer is disabled");
136 goto error;
137 }
138
139 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
140 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
141
142 pthread_mutex_lock(sock->lock);
143 ret = lttcomm_send_unix_sock(sock->fd, &msg, sizeof(msg));
2f77fc4b 144 if (ret < 0) {
c5c45efa
DG
145 /* Indicate that the consumer is probably closing at this point. */
146 DBG("send consumer destroy relayd command");
f50f23d9 147 goto error_send;
2f77fc4b
DG
148 }
149
f50f23d9
DG
150 /* Don't check the return value. The caller will do it. */
151 ret = consumer_recv_status_reply(sock);
152
2f77fc4b
DG
153 DBG2("Consumer send destroy relayd command done");
154
f50f23d9
DG
155error_send:
156 pthread_mutex_unlock(sock->lock);
2f77fc4b
DG
157error:
158 return ret;
159}
160
161/*
162 * For each consumer socket in the consumer output object, send a destroy
163 * relayd command.
164 */
165void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
166{
2f77fc4b
DG
167 struct lttng_ht_iter iter;
168 struct consumer_socket *socket;
169
170 assert(consumer);
171
172 /* Destroy any relayd connection */
173 if (consumer && consumer->type == CONSUMER_DST_NET) {
174 rcu_read_lock();
175 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
176 node.node) {
c617c0c6
MD
177 int ret;
178
2f77fc4b
DG
179 /* Send destroy relayd command */
180 ret = consumer_send_destroy_relayd(socket, consumer);
181 if (ret < 0) {
c5c45efa 182 DBG("Unable to send destroy relayd command to consumer");
2f77fc4b
DG
183 /* Continue since we MUST delete everything at this point. */
184 }
185 }
186 rcu_read_unlock();
187 }
188}
189
a4b92340
DG
190/*
191 * From a consumer_data structure, allocate and add a consumer socket to the
192 * consumer output.
193 *
194 * Return 0 on success, else negative value on error
195 */
196int consumer_create_socket(struct consumer_data *data,
197 struct consumer_output *output)
198{
199 int ret = 0;
200 struct consumer_socket *socket;
201
202 assert(data);
203
204 if (output == NULL || data->cmd_sock < 0) {
205 /*
206 * Not an error. Possible there is simply not spawned consumer or it's
207 * disabled for the tracing session asking the socket.
208 */
209 goto error;
210 }
211
212 rcu_read_lock();
213 socket = consumer_find_socket(data->cmd_sock, output);
214 rcu_read_unlock();
215 if (socket == NULL) {
216 socket = consumer_allocate_socket(data->cmd_sock);
217 if (socket == NULL) {
218 ret = -1;
219 goto error;
220 }
221
2f77fc4b 222 socket->registered = 0;
a4b92340
DG
223 socket->lock = &data->lock;
224 rcu_read_lock();
225 consumer_add_socket(socket, output);
226 rcu_read_unlock();
227 }
228
229 DBG3("Consumer socket created (fd: %d) and added to output",
230 data->cmd_sock);
231
232error:
233 return ret;
234}
235
7972aab2
DG
236/*
237 * Return the consumer socket from the given consumer output with the right
238 * bitness. On error, returns NULL.
239 *
240 * The caller MUST acquire a rcu read side lock and keep it until the socket
241 * object reference is not needed anymore.
242 */
243struct consumer_socket *consumer_find_socket_by_bitness(int bits,
244 struct consumer_output *consumer)
245{
246 int consumer_fd;
247 struct consumer_socket *socket = NULL;
248
249 switch (bits) {
250 case 64:
251 consumer_fd = uatomic_read(&ust_consumerd64_fd);
252 break;
253 case 32:
254 consumer_fd = uatomic_read(&ust_consumerd32_fd);
255 break;
256 default:
257 assert(0);
258 goto end;
259 }
260
261 socket = consumer_find_socket(consumer_fd, consumer);
262 if (!socket) {
263 ERR("Consumer socket fd %d not found in consumer obj %p",
264 consumer_fd, consumer);
265 }
266
267end:
268 return socket;
269}
270
173af62f
DG
271/*
272 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
273 * be acquired before calling this function and across use of the
274 * returned consumer_socket.
275 */
276struct consumer_socket *consumer_find_socket(int key,
277 struct consumer_output *consumer)
278{
279 struct lttng_ht_iter iter;
280 struct lttng_ht_node_ulong *node;
281 struct consumer_socket *socket = NULL;
282
283 /* Negative keys are lookup failures */
a4b92340 284 if (key < 0 || consumer == NULL) {
173af62f
DG
285 return NULL;
286 }
287
288 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
289 &iter);
290 node = lttng_ht_iter_get_node_ulong(&iter);
291 if (node != NULL) {
292 socket = caa_container_of(node, struct consumer_socket, node);
293 }
294
295 return socket;
296}
297
298/*
299 * Allocate a new consumer_socket and return the pointer.
300 */
301struct consumer_socket *consumer_allocate_socket(int fd)
302{
303 struct consumer_socket *socket = NULL;
304
305 socket = zmalloc(sizeof(struct consumer_socket));
306 if (socket == NULL) {
307 PERROR("zmalloc consumer socket");
308 goto error;
309 }
310
311 socket->fd = fd;
312 lttng_ht_node_init_ulong(&socket->node, fd);
313
314error:
315 return socket;
316}
317
318/*
319 * Add consumer socket to consumer output object. Read side lock must be
320 * acquired before calling this function.
321 */
322void consumer_add_socket(struct consumer_socket *sock,
323 struct consumer_output *consumer)
324{
325 assert(sock);
326 assert(consumer);
327
328 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
329}
330
331/*
332 * Delte consumer socket to consumer output object. Read side lock must be
333 * acquired before calling this function.
334 */
335void consumer_del_socket(struct consumer_socket *sock,
336 struct consumer_output *consumer)
337{
338 int ret;
339 struct lttng_ht_iter iter;
340
341 assert(sock);
342 assert(consumer);
343
344 iter.iter.node = &sock->node.node;
345 ret = lttng_ht_del(consumer->socks, &iter);
346 assert(!ret);
347}
348
349/*
350 * RCU destroy call function.
351 */
352static void destroy_socket_rcu(struct rcu_head *head)
353{
354 struct lttng_ht_node_ulong *node =
355 caa_container_of(head, struct lttng_ht_node_ulong, head);
356 struct consumer_socket *socket =
357 caa_container_of(node, struct consumer_socket, node);
358
359 free(socket);
360}
361
362/*
363 * Destroy and free socket pointer in a call RCU. Read side lock must be
364 * acquired before calling this function.
365 */
366void consumer_destroy_socket(struct consumer_socket *sock)
367{
368 assert(sock);
369
370 /*
371 * We DO NOT close the file descriptor here since it is global to the
2f77fc4b
DG
372 * session daemon and is closed only if the consumer dies or a custom
373 * consumer was registered,
173af62f 374 */
2f77fc4b
DG
375 if (sock->registered) {
376 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
377 lttcomm_close_unix_sock(sock->fd);
378 }
173af62f
DG
379
380 call_rcu(&sock->node.head, destroy_socket_rcu);
381}
382
00e2e675
DG
383/*
384 * Allocate and assign data to a consumer_output object.
385 *
386 * Return pointer to structure.
387 */
388struct consumer_output *consumer_create_output(enum consumer_dst_type type)
389{
390 struct consumer_output *output = NULL;
391
392 output = zmalloc(sizeof(struct consumer_output));
393 if (output == NULL) {
394 PERROR("zmalloc consumer_output");
395 goto error;
396 }
397
398 /* By default, consumer output is enabled */
399 output->enabled = 1;
400 output->type = type;
d88aee68 401 output->net_seq_index = (uint64_t) -1ULL;
173af62f
DG
402
403 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
404
405error:
406 return output;
407}
408
409/*
410 * Delete the consumer_output object from the list and free the ptr.
36b588ed
MD
411 *
412 * Should *NOT* be called with RCU read-side lock held.
00e2e675
DG
413 */
414void consumer_destroy_output(struct consumer_output *obj)
415{
416 if (obj == NULL) {
417 return;
418 }
419
173af62f
DG
420 if (obj->socks) {
421 struct lttng_ht_iter iter;
422 struct consumer_socket *socket;
423
2f77fc4b 424 rcu_read_lock();
173af62f 425 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
2f77fc4b 426 consumer_del_socket(socket, obj);
173af62f
DG
427 consumer_destroy_socket(socket);
428 }
2f77fc4b
DG
429 rcu_read_unlock();
430
431 /* Finally destroy HT */
0b2dc8df 432 ht_cleanup_push(obj->socks);
00e2e675 433 }
173af62f 434
00e2e675
DG
435 free(obj);
436}
437
438/*
439 * Copy consumer output and returned the newly allocated copy.
36b588ed
MD
440 *
441 * Should *NOT* be called with RCU read-side lock held.
00e2e675
DG
442 */
443struct consumer_output *consumer_copy_output(struct consumer_output *obj)
444{
09a90bcd 445 struct lttng_ht *tmp_ht_ptr;
173af62f
DG
446 struct lttng_ht_iter iter;
447 struct consumer_socket *socket, *copy_sock;
00e2e675
DG
448 struct consumer_output *output;
449
450 assert(obj);
451
452 output = consumer_create_output(obj->type);
453 if (output == NULL) {
454 goto error;
455 }
09a90bcd
DG
456 /* Avoid losing the HT reference after the memcpy() */
457 tmp_ht_ptr = output->socks;
00e2e675
DG
458
459 memcpy(output, obj, sizeof(struct consumer_output));
460
09a90bcd
DG
461 /* Putting back the HT pointer and start copying socket(s). */
462 output->socks = tmp_ht_ptr;
173af62f 463
b82c5c4d 464 rcu_read_lock();
173af62f
DG
465 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
466 /* Create new socket object. */
467 copy_sock = consumer_allocate_socket(socket->fd);
468 if (copy_sock == NULL) {
b82c5c4d 469 rcu_read_unlock();
173af62f
DG
470 goto malloc_error;
471 }
472
09a90bcd 473 copy_sock->registered = socket->registered;
173af62f
DG
474 copy_sock->lock = socket->lock;
475 consumer_add_socket(copy_sock, output);
476 }
b82c5c4d 477 rcu_read_unlock();
173af62f 478
00e2e675
DG
479error:
480 return output;
173af62f
DG
481
482malloc_error:
483 consumer_destroy_output(output);
484 return NULL;
00e2e675
DG
485}
486
487/*
488 * Set network URI to the consumer output object.
489 *
ad20f474
DG
490 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
491 * error.
00e2e675
DG
492 */
493int consumer_set_network_uri(struct consumer_output *obj,
494 struct lttng_uri *uri)
495{
496 int ret;
497 char tmp_path[PATH_MAX];
498 char hostname[HOST_NAME_MAX];
499 struct lttng_uri *dst_uri = NULL;
500
501 /* Code flow error safety net. */
502 assert(obj);
503 assert(uri);
504
505 switch (uri->stype) {
506 case LTTNG_STREAM_CONTROL:
507 dst_uri = &obj->dst.net.control;
508 obj->dst.net.control_isset = 1;
509 if (uri->port == 0) {
510 /* Assign default port. */
511 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
a74934ba
DG
512 } else {
513 if (obj->dst.net.data_isset && uri->port ==
514 obj->dst.net.data.port) {
515 ret = -LTTNG_ERR_INVALID;
516 goto error;
517 }
00e2e675 518 }
ad20f474 519 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
520 break;
521 case LTTNG_STREAM_DATA:
522 dst_uri = &obj->dst.net.data;
523 obj->dst.net.data_isset = 1;
524 if (uri->port == 0) {
525 /* Assign default port. */
526 uri->port = DEFAULT_NETWORK_DATA_PORT;
a74934ba
DG
527 } else {
528 if (obj->dst.net.control_isset && uri->port ==
529 obj->dst.net.control.port) {
530 ret = -LTTNG_ERR_INVALID;
531 goto error;
532 }
00e2e675 533 }
ad20f474 534 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
535 break;
536 default:
537 ERR("Set network uri type unknown %d", uri->stype);
a74934ba 538 ret = -LTTNG_ERR_INVALID;
00e2e675
DG
539 goto error;
540 }
541
542 ret = uri_compare(dst_uri, uri);
543 if (!ret) {
544 /* Same URI, don't touch it and return success. */
545 DBG3("URI network compare are the same");
ad20f474 546 goto equal;
00e2e675
DG
547 }
548
549 /* URIs were not equal, replacing it. */
550 memset(dst_uri, 0, sizeof(struct lttng_uri));
551 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
552 obj->type = CONSUMER_DST_NET;
553
554 /* Handle subdir and add hostname in front. */
555 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
556 /* Get hostname to append it in the pathname */
557 ret = gethostname(hostname, sizeof(hostname));
558 if (ret < 0) {
559 PERROR("gethostname. Fallback on default localhost");
560 strncpy(hostname, "localhost", sizeof(hostname));
561 }
562 hostname[sizeof(hostname) - 1] = '\0';
563
564 /* Setup consumer subdir if none present in the control URI */
565 if (strlen(dst_uri->subdir) == 0) {
566 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
567 hostname, obj->subdir);
568 } else {
569 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
570 hostname, dst_uri->subdir);
571 }
572 if (ret < 0) {
573 PERROR("snprintf set consumer uri subdir");
a74934ba 574 ret = -LTTNG_ERR_NOMEM;
00e2e675
DG
575 goto error;
576 }
577
578 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
579 DBG3("Consumer set network uri subdir path %s", tmp_path);
580 }
581
00e2e675 582 return 0;
ad20f474
DG
583equal:
584 return 1;
00e2e675 585error:
a74934ba 586 return ret;
00e2e675
DG
587}
588
589/*
590 * Send file descriptor to consumer via sock.
591 */
f50f23d9 592int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
00e2e675
DG
593{
594 int ret;
595
596 assert(fds);
f50f23d9 597 assert(sock);
00e2e675
DG
598 assert(nb_fd > 0);
599
f50f23d9 600 ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
00e2e675 601 if (ret < 0) {
3448e266
DG
602 /* The above call will print a PERROR on error. */
603 DBG("Error when sending consumer fds on sock %d", sock->fd);
00e2e675
DG
604 goto error;
605 }
606
f50f23d9
DG
607 ret = consumer_recv_status_reply(sock);
608
00e2e675
DG
609error:
610 return ret;
611}
612
ffe60014
DG
613/*
614 * Consumer send communication message structure to consumer.
615 */
616int consumer_send_msg(struct consumer_socket *sock,
617 struct lttcomm_consumer_msg *msg)
618{
619 int ret;
620
621 assert(msg);
622 assert(sock);
623 assert(sock->fd >= 0);
624
625 ret = lttcomm_send_unix_sock(sock->fd, msg,
626 sizeof(struct lttcomm_consumer_msg));
627 if (ret < 0) {
628 /* The above call will print a PERROR on error. */
629 DBG("Error when sending consumer channel on sock %d", sock->fd);
630 goto error;
631 }
632
633 ret = consumer_recv_status_reply(sock);
634
635error:
636 return ret;
637}
638
00e2e675
DG
639/*
640 * Consumer send channel communication message structure to consumer.
641 */
f50f23d9
DG
642int consumer_send_channel(struct consumer_socket *sock,
643 struct lttcomm_consumer_msg *msg)
00e2e675
DG
644{
645 int ret;
646
647 assert(msg);
f50f23d9
DG
648 assert(sock);
649 assert(sock->fd >= 0);
00e2e675 650
f50f23d9 651 ret = lttcomm_send_unix_sock(sock->fd, msg,
00e2e675
DG
652 sizeof(struct lttcomm_consumer_msg));
653 if (ret < 0) {
3448e266
DG
654 /* The above call will print a PERROR on error. */
655 DBG("Error when sending consumer channel on sock %d", sock->fd);
00e2e675
DG
656 goto error;
657 }
658
f50f23d9
DG
659 ret = consumer_recv_status_reply(sock);
660
00e2e675
DG
661error:
662 return ret;
663}
664
ffe60014
DG
665/*
666 * Populate the given consumer msg structure with the ask_channel command
667 * information.
668 */
669void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
670 uint64_t subbuf_size,
671 uint64_t num_subbuf,
672 int overwrite,
673 unsigned int switch_timer_interval,
674 unsigned int read_timer_interval,
675 int output,
676 int type,
677 uint64_t session_id,
678 const char *pathname,
679 const char *name,
680 uid_t uid,
681 gid_t gid,
d88aee68
DG
682 uint64_t relayd_id,
683 uint64_t key,
7972aab2 684 unsigned char *uuid,
1624d5b7
JD
685 uint32_t chan_id,
686 uint64_t tracefile_size,
687 uint64_t tracefile_count)
ffe60014
DG
688{
689 assert(msg);
690
691 /* Zeroed structure */
692 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
693
694 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
695 msg->u.ask_channel.subbuf_size = subbuf_size;
696 msg->u.ask_channel.num_subbuf = num_subbuf ;
697 msg->u.ask_channel.overwrite = overwrite;
698 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
699 msg->u.ask_channel.read_timer_interval = read_timer_interval;
700 msg->u.ask_channel.output = output;
701 msg->u.ask_channel.type = type;
702 msg->u.ask_channel.session_id = session_id;
703 msg->u.ask_channel.uid = uid;
704 msg->u.ask_channel.gid = gid;
705 msg->u.ask_channel.relayd_id = relayd_id;
706 msg->u.ask_channel.key = key;
7972aab2 707 msg->u.ask_channel.chan_id = chan_id;
1624d5b7
JD
708 msg->u.ask_channel.tracefile_size = tracefile_size;
709 msg->u.ask_channel.tracefile_count = tracefile_count;
ffe60014
DG
710
711 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
712
713 strncpy(msg->u.ask_channel.pathname, pathname,
714 sizeof(msg->u.ask_channel.pathname));
715 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
716
717 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
718 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
719}
720
00e2e675
DG
721/*
722 * Init channel communication message structure.
723 */
724void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
725 enum lttng_consumer_command cmd,
d88aee68 726 uint64_t channel_key,
ffe60014
DG
727 uint64_t session_id,
728 const char *pathname,
729 uid_t uid,
730 gid_t gid,
d88aee68 731 uint64_t relayd_id,
c30aaa51 732 const char *name,
ffe60014
DG
733 unsigned int nb_init_streams,
734 enum lttng_event_output output,
1624d5b7
JD
735 int type,
736 uint64_t tracefile_size,
737 uint64_t tracefile_count)
00e2e675
DG
738{
739 assert(msg);
740
00e2e675
DG
741 /* Zeroed structure */
742 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
743
744 /* Send channel */
745 msg->cmd_type = cmd;
746 msg->u.channel.channel_key = channel_key;
ffe60014
DG
747 msg->u.channel.session_id = session_id;
748 msg->u.channel.uid = uid;
749 msg->u.channel.gid = gid;
750 msg->u.channel.relayd_id = relayd_id;
c30aaa51 751 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
752 msg->u.channel.output = output;
753 msg->u.channel.type = type;
1624d5b7
JD
754 msg->u.channel.tracefile_size = tracefile_size;
755 msg->u.channel.tracefile_count = tracefile_count;
ffe60014
DG
756
757 strncpy(msg->u.channel.pathname, pathname,
758 sizeof(msg->u.channel.pathname));
759 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
760
761 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
762 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
00e2e675
DG
763}
764
765/*
766 * Init stream communication message structure.
767 */
768void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
769 enum lttng_consumer_command cmd,
d88aee68
DG
770 uint64_t channel_key,
771 uint64_t stream_key,
ffe60014 772 int cpu)
00e2e675
DG
773{
774 assert(msg);
775
776 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
777
00e2e675
DG
778 msg->cmd_type = cmd;
779 msg->u.stream.channel_key = channel_key;
780 msg->u.stream.stream_key = stream_key;
ffe60014 781 msg->u.stream.cpu = cpu;
00e2e675
DG
782}
783
784/*
785 * Send stream communication structure to the consumer.
786 */
f50f23d9
DG
787int consumer_send_stream(struct consumer_socket *sock,
788 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
789 int *fds, size_t nb_fd)
00e2e675
DG
790{
791 int ret;
792
793 assert(msg);
794 assert(dst);
f50f23d9 795 assert(sock);
ffe60014 796 assert(fds);
00e2e675
DG
797
798 /* Send on socket */
f50f23d9 799 ret = lttcomm_send_unix_sock(sock->fd, msg,
00e2e675
DG
800 sizeof(struct lttcomm_consumer_msg));
801 if (ret < 0) {
3448e266
DG
802 /* The above call will print a PERROR on error. */
803 DBG("Error when sending consumer stream on sock %d", sock->fd);
00e2e675
DG
804 goto error;
805 }
806
f50f23d9
DG
807 ret = consumer_recv_status_reply(sock);
808 if (ret < 0) {
809 goto error;
810 }
811
00e2e675
DG
812 ret = consumer_send_fds(sock, fds, nb_fd);
813 if (ret < 0) {
814 goto error;
815 }
816
817error:
818 return ret;
819}
37278a1e
DG
820
821/*
822 * Send relayd socket to consumer associated with a session name.
823 *
824 * On success return positive value. On error, negative value.
825 */
f50f23d9 826int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
6151a90f 827 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
d88aee68 828 enum lttng_stream_type type, uint64_t session_id)
37278a1e
DG
829{
830 int ret;
831 struct lttcomm_consumer_msg msg;
832
833 /* Code flow error. Safety net. */
6151a90f 834 assert(rsock);
37278a1e 835 assert(consumer);
f50f23d9 836 assert(consumer_sock);
37278a1e
DG
837
838 /* Bail out if consumer is disabled */
839 if (!consumer->enabled) {
f73fabfd 840 ret = LTTNG_OK;
37278a1e
DG
841 goto error;
842 }
843
844 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
845 /*
846 * Assign network consumer output index using the temporary consumer since
847 * this call should only be made from within a set_consumer_uri() function
848 * call in the session daemon.
849 */
850 msg.u.relayd_sock.net_index = consumer->net_seq_index;
851 msg.u.relayd_sock.type = type;
46e6455f 852 msg.u.relayd_sock.session_id = session_id;
6151a90f 853 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
37278a1e 854
f50f23d9
DG
855 DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
856 ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
37278a1e 857 if (ret < 0) {
3448e266 858 /* The above call will print a PERROR on error. */
6151a90f 859 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
37278a1e
DG
860 goto error;
861 }
862
f50f23d9
DG
863 ret = consumer_recv_status_reply(consumer_sock);
864 if (ret < 0) {
865 goto error;
866 }
867
37278a1e 868 DBG3("Sending relayd socket file descriptor to consumer");
6151a90f 869 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
37278a1e
DG
870 if (ret < 0) {
871 goto error;
872 }
873
874 DBG2("Consumer relayd socket sent");
875
876error:
877 return ret;
878}
173af62f
DG
879
880/*
2f77fc4b
DG
881 * Set consumer subdirectory using the session name and a generated datetime if
882 * needed. This is appended to the current subdirectory.
173af62f 883 */
2f77fc4b
DG
884int consumer_set_subdir(struct consumer_output *consumer,
885 const char *session_name)
173af62f 886{
2f77fc4b
DG
887 int ret = 0;
888 unsigned int have_default_name = 0;
889 char datetime[16], tmp_path[PATH_MAX];
890 time_t rawtime;
891 struct tm *timeinfo;
173af62f
DG
892
893 assert(consumer);
2f77fc4b
DG
894 assert(session_name);
895
896 memset(tmp_path, 0, sizeof(tmp_path));
897
898 /* Flag if we have a default session. */
899 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
900 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
901 have_default_name = 1;
902 } else {
903 /* Get date and time for session path */
904 time(&rawtime);
905 timeinfo = localtime(&rawtime);
906 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
173af62f
DG
907 }
908
2f77fc4b
DG
909 if (have_default_name) {
910 ret = snprintf(tmp_path, sizeof(tmp_path),
911 "%s/%s", consumer->subdir, session_name);
912 } else {
913 ret = snprintf(tmp_path, sizeof(tmp_path),
914 "%s/%s-%s/", consumer->subdir, session_name, datetime);
915 }
173af62f 916 if (ret < 0) {
2f77fc4b 917 PERROR("snprintf session name date");
173af62f
DG
918 goto error;
919 }
920
2f77fc4b
DG
921 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
922 DBG2("Consumer subdir set to %s", consumer->subdir);
173af62f
DG
923
924error:
925 return ret;
926}
806e2684
DG
927
928/*
6d805429 929 * Ask the consumer if the data is ready to read (NOT pending) for the specific
806e2684
DG
930 * session id.
931 *
932 * This function has a different behavior with the consumer i.e. that it waits
6d805429 933 * for a reply from the consumer if yes or no the data is pending.
806e2684 934 */
d88aee68 935int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
936 struct consumer_output *consumer)
937{
938 int ret;
6d805429 939 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
940 struct consumer_socket *socket;
941 struct lttng_ht_iter iter;
942 struct lttcomm_consumer_msg msg;
943
944 assert(consumer);
945
6d805429 946 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
806e2684 947
d88aee68 948 msg.u.data_pending.session_id = session_id;
806e2684 949
d88aee68 950 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 951
c8f59ee5 952 /* Send command for each consumer */
b82c5c4d 953 rcu_read_lock();
806e2684
DG
954 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
955 node.node) {
956 /* Code flow error */
957 assert(socket->fd >= 0);
958
959 pthread_mutex_lock(socket->lock);
960
961 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
962 if (ret < 0) {
3448e266
DG
963 /* The above call will print a PERROR on error. */
964 DBG("Error on consumer is data pending on sock %d", socket->fd);
806e2684 965 pthread_mutex_unlock(socket->lock);
b82c5c4d 966 goto error_unlock;
806e2684
DG
967 }
968
f50f23d9
DG
969 /*
970 * No need for a recv reply status because the answer to the command is
971 * the reply status message.
972 */
973
806e2684 974 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
a6cd2b97
DG
975 if (ret <= 0) {
976 if (ret == 0) {
977 /* Orderly shutdown. Don't return 0 which means success. */
978 ret = -1;
979 }
3448e266
DG
980 /* The above call will print a PERROR on error. */
981 DBG("Error on recv consumer is data pending on sock %d", socket->fd);
806e2684 982 pthread_mutex_unlock(socket->lock);
b82c5c4d 983 goto error_unlock;
806e2684
DG
984 }
985
986 pthread_mutex_unlock(socket->lock);
987
6d805429 988 if (ret_code == 1) {
806e2684
DG
989 break;
990 }
991 }
b82c5c4d 992 rcu_read_unlock();
806e2684 993
d88aee68
DG
994 DBG("Consumer data is %s pending for session id %" PRIu64,
995 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
996 return ret_code;
997
b82c5c4d
DG
998error_unlock:
999 rcu_read_unlock();
806e2684
DG
1000 return -1;
1001}
7972aab2
DG
1002
1003/*
1004 * Send a flush command to consumer using the given channel key.
1005 *
1006 * Return 0 on success else a negative value.
1007 */
1008int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1009{
1010 int ret;
1011 struct lttcomm_consumer_msg msg;
1012
1013 assert(socket);
1014 assert(socket->fd >= 0);
1015
1016 DBG2("Consumer flush channel key %" PRIu64, key);
1017
1018 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1019 msg.u.flush_channel.key = key;
1020
1021 pthread_mutex_lock(socket->lock);
1022 health_code_update();
1023
1024 ret = consumer_send_msg(socket, &msg);
1025 if (ret < 0) {
1026 goto end;
1027 }
1028
1029end:
1030 health_code_update();
1031 pthread_mutex_unlock(socket->lock);
1032 return ret;
1033}
1034
1035/*
1036 * Send a close metdata command to consumer using the given channel key.
1037 *
1038 * Return 0 on success else a negative value.
1039 */
1040int consumer_close_metadata(struct consumer_socket *socket,
1041 uint64_t metadata_key)
1042{
1043 int ret;
1044 struct lttcomm_consumer_msg msg;
1045
1046 assert(socket);
1047 assert(socket->fd >= 0);
1048
1049 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1050
1051 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1052 msg.u.close_metadata.key = metadata_key;
1053
1054 pthread_mutex_lock(socket->lock);
1055 health_code_update();
1056
1057 ret = consumer_send_msg(socket, &msg);
1058 if (ret < 0) {
1059 goto end;
1060 }
1061
1062end:
1063 health_code_update();
1064 pthread_mutex_unlock(socket->lock);
1065 return ret;
1066}
1067
1068/*
1069 * Send a setup metdata command to consumer using the given channel key.
1070 *
1071 * Return 0 on success else a negative value.
1072 */
1073int consumer_setup_metadata(struct consumer_socket *socket,
1074 uint64_t metadata_key)
1075{
1076 int ret;
1077 struct lttcomm_consumer_msg msg;
1078
1079 assert(socket);
1080 assert(socket->fd >= 0);
1081
1082 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1083
1084 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1085 msg.u.setup_metadata.key = metadata_key;
1086
1087 pthread_mutex_lock(socket->lock);
1088 health_code_update();
1089
1090 ret = consumer_send_msg(socket, &msg);
1091 if (ret < 0) {
1092 goto end;
1093 }
1094
1095end:
1096 health_code_update();
1097 pthread_mutex_unlock(socket->lock);
1098 return ret;
1099}
1100
1101/*
331744e3 1102 * Send metadata string to consumer. Socket lock MUST be acquired.
7972aab2
DG
1103 *
1104 * Return 0 on success else a negative value.
1105 */
1106int consumer_push_metadata(struct consumer_socket *socket,
1107 uint64_t metadata_key, char *metadata_str, size_t len,
1108 size_t target_offset)
1109{
1110 int ret;
1111 struct lttcomm_consumer_msg msg;
1112
1113 assert(socket);
1114 assert(socket->fd >= 0);
1115
1116 DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1117
1118 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1119 msg.u.push_metadata.key = metadata_key;
1120 msg.u.push_metadata.target_offset = target_offset;
1121 msg.u.push_metadata.len = len;
1122
7972aab2
DG
1123 health_code_update();
1124 ret = consumer_send_msg(socket, &msg);
331744e3 1125 if (ret < 0 || len == 0) {
7972aab2
DG
1126 goto end;
1127 }
1128
8fd623e0 1129 DBG3("Consumer pushing metadata on sock %d of len %zu", socket->fd, len);
7972aab2
DG
1130
1131 ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1132 if (ret < 0) {
1133 goto end;
1134 }
1135
1136 health_code_update();
1137 ret = consumer_recv_status_reply(socket);
1138 if (ret < 0) {
1139 goto end;
1140 }
1141
1142end:
1143 health_code_update();
7972aab2
DG
1144 return ret;
1145}
This page took 0.145673 seconds and 4 git commands to generate.