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