Fix: mark ltt_sessions_ht_destroy as static
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
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 _LGPL_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 #include <common/relayd/relayd.h>
32
33 #include "consumer.h"
34 #include "health-sessiond.h"
35 #include "ust-app.h"
36 #include "utils.h"
37
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 */
46 int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
47 {
48 int fd;
49 ssize_t size;
50
51 assert(socket);
52 assert(socket->fd_ptr);
53 assert(msg);
54
55 /* Consumer socket is invalid. Stopping. */
56 fd = *socket->fd_ptr;
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 /*
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.
68 */
69
70 /* This call will PERROR on error. */
71 (void) lttcomm_close_unix_sock(fd);
72 *socket->fd_ptr = -1;
73 goto error;
74 }
75
76 return 0;
77
78 error:
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 */
90 int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
91 {
92 int fd;
93 ssize_t size;
94
95 assert(socket);
96 assert(socket->fd_ptr);
97 assert(msg);
98
99 /* Consumer socket is invalid. Stopping. */
100 fd = *socket->fd_ptr;
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 /*
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.
112 */
113
114 /* This call will PERROR on error. */
115 (void) lttcomm_close_unix_sock(fd);
116 *socket->fd_ptr = -1;
117 goto error;
118 }
119
120 return 0;
121
122 error:
123 return -1;
124 }
125
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 */
133 int consumer_recv_status_reply(struct consumer_socket *sock)
134 {
135 int ret;
136 struct lttcomm_consumer_status_msg reply;
137
138 assert(sock);
139
140 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
141 if (ret < 0) {
142 goto end;
143 }
144
145 if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
146 /* All good. */
147 ret = 0;
148 } else {
149 ret = -reply.ret_code;
150 DBG("Consumer ret code %d", ret);
151 }
152
153 end:
154 return ret;
155 }
156
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 */
165 int consumer_recv_status_channel(struct consumer_socket *sock,
166 uint64_t *key, unsigned int *stream_count)
167 {
168 int ret;
169 struct lttcomm_consumer_status_channel reply;
170
171 assert(sock);
172 assert(stream_count);
173 assert(key);
174
175 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
176 if (ret < 0) {
177 goto end;
178 }
179
180 /* An error is possible so don't touch the key and stream_count. */
181 if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
182 ret = -1;
183 goto end;
184 }
185
186 *key = reply.key;
187 *stream_count = reply.stream_count;
188 ret = 0;
189
190 end:
191 return ret;
192 }
193
194 /*
195 * Send destroy relayd command to consumer.
196 *
197 * On success return positive value. On error, negative value.
198 */
199 int 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
208 DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
209
210 memset(&msg, 0, sizeof(msg));
211 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
212 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
213
214 pthread_mutex_lock(sock->lock);
215 ret = consumer_socket_send(sock, &msg, sizeof(msg));
216 if (ret < 0) {
217 goto error;
218 }
219
220 /* Don't check the return value. The caller will do it. */
221 ret = consumer_recv_status_reply(sock);
222
223 DBG2("Consumer send destroy relayd command done");
224
225 error:
226 pthread_mutex_unlock(sock->lock);
227 return ret;
228 }
229
230 /*
231 * For each consumer socket in the consumer output object, send a destroy
232 * relayd command.
233 */
234 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
235 {
236 struct lttng_ht_iter iter;
237 struct consumer_socket *socket;
238
239 assert(consumer);
240
241 /* Destroy any relayd connection */
242 if (consumer->type == CONSUMER_DST_NET) {
243 rcu_read_lock();
244 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
245 node.node) {
246 int ret;
247
248 /* Send destroy relayd command */
249 ret = consumer_send_destroy_relayd(socket, consumer);
250 if (ret < 0) {
251 DBG("Unable to send destroy relayd command to consumer");
252 /* Continue since we MUST delete everything at this point. */
253 }
254 }
255 rcu_read_unlock();
256 }
257 }
258
259 /*
260 * From a consumer_data structure, allocate and add a consumer socket to the
261 * consumer output.
262 *
263 * Return 0 on success, else negative value on error
264 */
265 int consumer_create_socket(struct consumer_data *data,
266 struct consumer_output *output)
267 {
268 int ret = 0;
269 struct consumer_socket *socket;
270
271 assert(data);
272
273 if (output == NULL || data->cmd_sock < 0) {
274 /*
275 * Not an error. Possible there is simply not spawned consumer or it's
276 * disabled for the tracing session asking the socket.
277 */
278 goto error;
279 }
280
281 rcu_read_lock();
282 socket = consumer_find_socket(data->cmd_sock, output);
283 rcu_read_unlock();
284 if (socket == NULL) {
285 socket = consumer_allocate_socket(&data->cmd_sock);
286 if (socket == NULL) {
287 ret = -1;
288 goto error;
289 }
290
291 socket->registered = 0;
292 socket->lock = &data->lock;
293 rcu_read_lock();
294 consumer_add_socket(socket, output);
295 rcu_read_unlock();
296 }
297
298 socket->type = data->type;
299
300 DBG3("Consumer socket created (fd: %d) and added to output",
301 data->cmd_sock);
302
303 error:
304 return ret;
305 }
306
307 /*
308 * Return the consumer socket from the given consumer output with the right
309 * bitness. On error, returns NULL.
310 *
311 * The caller MUST acquire a rcu read side lock and keep it until the socket
312 * object reference is not needed anymore.
313 */
314 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
315 struct consumer_output *consumer)
316 {
317 int consumer_fd;
318 struct consumer_socket *socket = NULL;
319
320 switch (bits) {
321 case 64:
322 consumer_fd = uatomic_read(&ust_consumerd64_fd);
323 break;
324 case 32:
325 consumer_fd = uatomic_read(&ust_consumerd32_fd);
326 break;
327 default:
328 assert(0);
329 goto end;
330 }
331
332 socket = consumer_find_socket(consumer_fd, consumer);
333 if (!socket) {
334 ERR("Consumer socket fd %d not found in consumer obj %p",
335 consumer_fd, consumer);
336 }
337
338 end:
339 return socket;
340 }
341
342 /*
343 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
344 * be acquired before calling this function and across use of the
345 * returned consumer_socket.
346 */
347 struct consumer_socket *consumer_find_socket(int key,
348 struct consumer_output *consumer)
349 {
350 struct lttng_ht_iter iter;
351 struct lttng_ht_node_ulong *node;
352 struct consumer_socket *socket = NULL;
353
354 /* Negative keys are lookup failures */
355 if (key < 0 || consumer == NULL) {
356 return NULL;
357 }
358
359 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
360 &iter);
361 node = lttng_ht_iter_get_node_ulong(&iter);
362 if (node != NULL) {
363 socket = caa_container_of(node, struct consumer_socket, node);
364 }
365
366 return socket;
367 }
368
369 /*
370 * Allocate a new consumer_socket and return the pointer.
371 */
372 struct consumer_socket *consumer_allocate_socket(int *fd)
373 {
374 struct consumer_socket *socket = NULL;
375
376 assert(fd);
377
378 socket = zmalloc(sizeof(struct consumer_socket));
379 if (socket == NULL) {
380 PERROR("zmalloc consumer socket");
381 goto error;
382 }
383
384 socket->fd_ptr = fd;
385 lttng_ht_node_init_ulong(&socket->node, *fd);
386
387 error:
388 return socket;
389 }
390
391 /*
392 * Add consumer socket to consumer output object. Read side lock must be
393 * acquired before calling this function.
394 */
395 void consumer_add_socket(struct consumer_socket *sock,
396 struct consumer_output *consumer)
397 {
398 assert(sock);
399 assert(consumer);
400
401 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
402 }
403
404 /*
405 * Delte consumer socket to consumer output object. Read side lock must be
406 * acquired before calling this function.
407 */
408 void consumer_del_socket(struct consumer_socket *sock,
409 struct consumer_output *consumer)
410 {
411 int ret;
412 struct lttng_ht_iter iter;
413
414 assert(sock);
415 assert(consumer);
416
417 iter.iter.node = &sock->node.node;
418 ret = lttng_ht_del(consumer->socks, &iter);
419 assert(!ret);
420 }
421
422 /*
423 * RCU destroy call function.
424 */
425 static void destroy_socket_rcu(struct rcu_head *head)
426 {
427 struct lttng_ht_node_ulong *node =
428 caa_container_of(head, struct lttng_ht_node_ulong, head);
429 struct consumer_socket *socket =
430 caa_container_of(node, struct consumer_socket, node);
431
432 free(socket);
433 }
434
435 /*
436 * Destroy and free socket pointer in a call RCU. Read side lock must be
437 * acquired before calling this function.
438 */
439 void consumer_destroy_socket(struct consumer_socket *sock)
440 {
441 assert(sock);
442
443 /*
444 * We DO NOT close the file descriptor here since it is global to the
445 * session daemon and is closed only if the consumer dies or a custom
446 * consumer was registered,
447 */
448 if (sock->registered) {
449 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
450 lttcomm_close_unix_sock(*sock->fd_ptr);
451 }
452
453 call_rcu(&sock->node.head, destroy_socket_rcu);
454 }
455
456 /*
457 * Allocate and assign data to a consumer_output object.
458 *
459 * Return pointer to structure.
460 */
461 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
462 {
463 struct consumer_output *output = NULL;
464
465 output = zmalloc(sizeof(struct consumer_output));
466 if (output == NULL) {
467 PERROR("zmalloc consumer_output");
468 goto error;
469 }
470
471 /* By default, consumer output is enabled */
472 output->enabled = 1;
473 output->type = type;
474 output->net_seq_index = (uint64_t) -1ULL;
475 urcu_ref_init(&output->ref);
476
477 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
478
479 error:
480 return output;
481 }
482
483 /*
484 * Iterate over the consumer output socket hash table and destroy them. The
485 * socket file descriptor are only closed if the consumer output was
486 * registered meaning it's an external consumer.
487 */
488 void consumer_destroy_output_sockets(struct consumer_output *obj)
489 {
490 struct lttng_ht_iter iter;
491 struct consumer_socket *socket;
492
493 if (!obj->socks) {
494 return;
495 }
496
497 rcu_read_lock();
498 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
499 consumer_del_socket(socket, obj);
500 consumer_destroy_socket(socket);
501 }
502 rcu_read_unlock();
503 }
504
505 /*
506 * Delete the consumer_output object from the list and free the ptr.
507 *
508 * Should *NOT* be called with RCU read-side lock held.
509 */
510 static void consumer_release_output(struct urcu_ref *ref)
511 {
512 struct consumer_output *obj =
513 caa_container_of(ref, struct consumer_output, ref);
514
515 consumer_destroy_output_sockets(obj);
516
517 if (obj->socks) {
518 /* Finally destroy HT */
519 ht_cleanup_push(obj->socks);
520 }
521
522 free(obj);
523 }
524
525 /*
526 * Get the consumer_output object.
527 */
528 void consumer_output_get(struct consumer_output *obj)
529 {
530 urcu_ref_get(&obj->ref);
531 }
532
533 /*
534 * Put the consumer_output object.
535 *
536 * Should *NOT* be called with RCU read-side lock held.
537 */
538 void consumer_output_put(struct consumer_output *obj)
539 {
540 if (!obj) {
541 return;
542 }
543 urcu_ref_put(&obj->ref, consumer_release_output);
544 }
545
546 /*
547 * Copy consumer output and returned the newly allocated copy.
548 *
549 * Should *NOT* be called with RCU read-side lock held.
550 */
551 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
552 {
553 int ret;
554 struct consumer_output *output;
555
556 assert(obj);
557
558 output = consumer_create_output(obj->type);
559 if (output == NULL) {
560 goto end;
561 }
562 output->enabled = obj->enabled;
563 output->net_seq_index = obj->net_seq_index;
564 memcpy(output->subdir, obj->subdir, PATH_MAX);
565 output->snapshot = obj->snapshot;
566 memcpy(&output->dst, &obj->dst, sizeof(output->dst));
567 ret = consumer_copy_sockets(output, obj);
568 if (ret < 0) {
569 goto error_put;
570 }
571 end:
572 return output;
573
574 error_put:
575 consumer_output_put(output);
576 return NULL;
577 }
578
579 /*
580 * Copy consumer sockets from src to dst.
581 *
582 * Return 0 on success or else a negative value.
583 */
584 int consumer_copy_sockets(struct consumer_output *dst,
585 struct consumer_output *src)
586 {
587 int ret = 0;
588 struct lttng_ht_iter iter;
589 struct consumer_socket *socket, *copy_sock;
590
591 assert(dst);
592 assert(src);
593
594 rcu_read_lock();
595 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
596 /* Ignore socket that are already there. */
597 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
598 if (copy_sock) {
599 continue;
600 }
601
602 /* Create new socket object. */
603 copy_sock = consumer_allocate_socket(socket->fd_ptr);
604 if (copy_sock == NULL) {
605 rcu_read_unlock();
606 ret = -ENOMEM;
607 goto error;
608 }
609
610 copy_sock->registered = socket->registered;
611 /*
612 * This is valid because this lock is shared accross all consumer
613 * object being the global lock of the consumer data structure of the
614 * session daemon.
615 */
616 copy_sock->lock = socket->lock;
617 consumer_add_socket(copy_sock, dst);
618 }
619 rcu_read_unlock();
620
621 error:
622 return ret;
623 }
624
625 /*
626 * Set network URI to the consumer output object.
627 *
628 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
629 * error.
630 */
631 int consumer_set_network_uri(struct consumer_output *obj,
632 struct lttng_uri *uri)
633 {
634 int ret;
635 char tmp_path[PATH_MAX];
636 char hostname[HOST_NAME_MAX];
637 struct lttng_uri *dst_uri = NULL;
638
639 /* Code flow error safety net. */
640 assert(obj);
641 assert(uri);
642
643 switch (uri->stype) {
644 case LTTNG_STREAM_CONTROL:
645 dst_uri = &obj->dst.net.control;
646 obj->dst.net.control_isset = 1;
647 if (uri->port == 0) {
648 /* Assign default port. */
649 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
650 } else {
651 if (obj->dst.net.data_isset && uri->port ==
652 obj->dst.net.data.port) {
653 ret = -LTTNG_ERR_INVALID;
654 goto error;
655 }
656 }
657 DBG3("Consumer control URI set with port %d", uri->port);
658 break;
659 case LTTNG_STREAM_DATA:
660 dst_uri = &obj->dst.net.data;
661 obj->dst.net.data_isset = 1;
662 if (uri->port == 0) {
663 /* Assign default port. */
664 uri->port = DEFAULT_NETWORK_DATA_PORT;
665 } else {
666 if (obj->dst.net.control_isset && uri->port ==
667 obj->dst.net.control.port) {
668 ret = -LTTNG_ERR_INVALID;
669 goto error;
670 }
671 }
672 DBG3("Consumer data URI set with port %d", uri->port);
673 break;
674 default:
675 ERR("Set network uri type unknown %d", uri->stype);
676 ret = -LTTNG_ERR_INVALID;
677 goto error;
678 }
679
680 ret = uri_compare(dst_uri, uri);
681 if (!ret) {
682 /* Same URI, don't touch it and return success. */
683 DBG3("URI network compare are the same");
684 goto equal;
685 }
686
687 /* URIs were not equal, replacing it. */
688 memset(dst_uri, 0, sizeof(struct lttng_uri));
689 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
690 obj->type = CONSUMER_DST_NET;
691
692 /* Handle subdir and add hostname in front. */
693 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
694 /* Get hostname to append it in the pathname */
695 ret = gethostname(hostname, sizeof(hostname));
696 if (ret < 0) {
697 PERROR("gethostname. Fallback on default localhost");
698 strncpy(hostname, "localhost", sizeof(hostname));
699 }
700 hostname[sizeof(hostname) - 1] = '\0';
701
702 /* Setup consumer subdir if none present in the control URI */
703 if (strlen(dst_uri->subdir) == 0) {
704 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
705 hostname, obj->subdir);
706 } else {
707 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
708 hostname, dst_uri->subdir);
709 }
710 if (ret < 0) {
711 PERROR("snprintf set consumer uri subdir");
712 ret = -LTTNG_ERR_NOMEM;
713 goto error;
714 }
715
716 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
717 DBG3("Consumer set network uri subdir path %s", tmp_path);
718 }
719
720 return 0;
721 equal:
722 return 1;
723 error:
724 return ret;
725 }
726
727 /*
728 * Send file descriptor to consumer via sock.
729 */
730 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
731 {
732 int ret;
733
734 assert(fds);
735 assert(sock);
736 assert(nb_fd > 0);
737
738 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
739 if (ret < 0) {
740 /* The above call will print a PERROR on error. */
741 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
742 goto error;
743 }
744
745 ret = consumer_recv_status_reply(sock);
746
747 error:
748 return ret;
749 }
750
751 /*
752 * Consumer send communication message structure to consumer.
753 */
754 int consumer_send_msg(struct consumer_socket *sock,
755 struct lttcomm_consumer_msg *msg)
756 {
757 int ret;
758
759 assert(msg);
760 assert(sock);
761
762 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
763 if (ret < 0) {
764 goto error;
765 }
766
767 ret = consumer_recv_status_reply(sock);
768
769 error:
770 return ret;
771 }
772
773 /*
774 * Consumer send channel communication message structure to consumer.
775 */
776 int consumer_send_channel(struct consumer_socket *sock,
777 struct lttcomm_consumer_msg *msg)
778 {
779 int ret;
780
781 assert(msg);
782 assert(sock);
783
784 ret = consumer_send_msg(sock, msg);
785 if (ret < 0) {
786 goto error;
787 }
788
789 error:
790 return ret;
791 }
792
793 /*
794 * Populate the given consumer msg structure with the ask_channel command
795 * information.
796 */
797 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
798 uint64_t subbuf_size,
799 uint64_t num_subbuf,
800 int overwrite,
801 unsigned int switch_timer_interval,
802 unsigned int read_timer_interval,
803 unsigned int live_timer_interval,
804 int output,
805 int type,
806 uint64_t session_id,
807 const char *pathname,
808 const char *name,
809 uid_t uid,
810 gid_t gid,
811 uint64_t relayd_id,
812 uint64_t key,
813 unsigned char *uuid,
814 uint32_t chan_id,
815 uint64_t tracefile_size,
816 uint64_t tracefile_count,
817 uint64_t session_id_per_pid,
818 unsigned int monitor,
819 uint32_t ust_app_uid,
820 const char *root_shm_path,
821 const char *shm_path)
822 {
823 assert(msg);
824
825 /* Zeroed structure */
826 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
827
828 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
829 msg->u.ask_channel.subbuf_size = subbuf_size;
830 msg->u.ask_channel.num_subbuf = num_subbuf ;
831 msg->u.ask_channel.overwrite = overwrite;
832 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
833 msg->u.ask_channel.read_timer_interval = read_timer_interval;
834 msg->u.ask_channel.live_timer_interval = live_timer_interval;
835 msg->u.ask_channel.output = output;
836 msg->u.ask_channel.type = type;
837 msg->u.ask_channel.session_id = session_id;
838 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
839 msg->u.ask_channel.uid = uid;
840 msg->u.ask_channel.gid = gid;
841 msg->u.ask_channel.relayd_id = relayd_id;
842 msg->u.ask_channel.key = key;
843 msg->u.ask_channel.chan_id = chan_id;
844 msg->u.ask_channel.tracefile_size = tracefile_size;
845 msg->u.ask_channel.tracefile_count = tracefile_count;
846 msg->u.ask_channel.monitor = monitor;
847 msg->u.ask_channel.ust_app_uid = ust_app_uid;
848
849 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
850
851 if (pathname) {
852 strncpy(msg->u.ask_channel.pathname, pathname,
853 sizeof(msg->u.ask_channel.pathname));
854 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
855 }
856
857 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
858 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
859
860 if (root_shm_path) {
861 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
862 sizeof(msg->u.ask_channel.root_shm_path));
863 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
864 }
865 if (shm_path) {
866 strncpy(msg->u.ask_channel.shm_path, shm_path,
867 sizeof(msg->u.ask_channel.shm_path));
868 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
869 }
870 }
871
872 /*
873 * Init channel communication message structure.
874 */
875 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
876 enum lttng_consumer_command cmd,
877 uint64_t channel_key,
878 uint64_t session_id,
879 const char *pathname,
880 uid_t uid,
881 gid_t gid,
882 uint64_t relayd_id,
883 const char *name,
884 unsigned int nb_init_streams,
885 enum lttng_event_output output,
886 int type,
887 uint64_t tracefile_size,
888 uint64_t tracefile_count,
889 unsigned int monitor,
890 unsigned int live_timer_interval)
891 {
892 assert(msg);
893
894 /* Zeroed structure */
895 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
896
897 /* Send channel */
898 msg->cmd_type = cmd;
899 msg->u.channel.channel_key = channel_key;
900 msg->u.channel.session_id = session_id;
901 msg->u.channel.uid = uid;
902 msg->u.channel.gid = gid;
903 msg->u.channel.relayd_id = relayd_id;
904 msg->u.channel.nb_init_streams = nb_init_streams;
905 msg->u.channel.output = output;
906 msg->u.channel.type = type;
907 msg->u.channel.tracefile_size = tracefile_size;
908 msg->u.channel.tracefile_count = tracefile_count;
909 msg->u.channel.monitor = monitor;
910 msg->u.channel.live_timer_interval = live_timer_interval;
911
912 strncpy(msg->u.channel.pathname, pathname,
913 sizeof(msg->u.channel.pathname));
914 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
915
916 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
917 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
918 }
919
920 /*
921 * Init stream communication message structure.
922 */
923 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
924 enum lttng_consumer_command cmd,
925 uint64_t channel_key,
926 uint64_t stream_key,
927 int cpu)
928 {
929 assert(msg);
930
931 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
932
933 msg->cmd_type = cmd;
934 msg->u.stream.channel_key = channel_key;
935 msg->u.stream.stream_key = stream_key;
936 msg->u.stream.cpu = cpu;
937 }
938
939 void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
940 enum lttng_consumer_command cmd,
941 uint64_t channel_key, uint64_t net_seq_idx)
942 {
943 assert(msg);
944
945 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
946
947 msg->cmd_type = cmd;
948 msg->u.sent_streams.channel_key = channel_key;
949 msg->u.sent_streams.net_seq_idx = net_seq_idx;
950 }
951
952 /*
953 * Send stream communication structure to the consumer.
954 */
955 int consumer_send_stream(struct consumer_socket *sock,
956 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
957 int *fds, size_t nb_fd)
958 {
959 int ret;
960
961 assert(msg);
962 assert(dst);
963 assert(sock);
964 assert(fds);
965
966 ret = consumer_send_msg(sock, msg);
967 if (ret < 0) {
968 goto error;
969 }
970
971 ret = consumer_send_fds(sock, fds, nb_fd);
972 if (ret < 0) {
973 goto error;
974 }
975
976 error:
977 return ret;
978 }
979
980 /*
981 * Send relayd socket to consumer associated with a session name.
982 *
983 * On success return positive value. On error, negative value.
984 */
985 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
986 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
987 enum lttng_stream_type type, uint64_t session_id,
988 char *session_name, char *hostname, int session_live_timer)
989 {
990 int ret;
991 struct lttcomm_consumer_msg msg;
992
993 /* Code flow error. Safety net. */
994 assert(rsock);
995 assert(consumer);
996 assert(consumer_sock);
997
998 memset(&msg, 0, sizeof(msg));
999 /* Bail out if consumer is disabled */
1000 if (!consumer->enabled) {
1001 ret = LTTNG_OK;
1002 goto error;
1003 }
1004
1005 if (type == LTTNG_STREAM_CONTROL) {
1006 ret = relayd_create_session(rsock,
1007 &msg.u.relayd_sock.relayd_session_id,
1008 session_name, hostname, session_live_timer,
1009 consumer->snapshot);
1010 if (ret < 0) {
1011 /* Close the control socket. */
1012 (void) relayd_close(rsock);
1013 goto error;
1014 }
1015 }
1016
1017 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1018 /*
1019 * Assign network consumer output index using the temporary consumer since
1020 * this call should only be made from within a set_consumer_uri() function
1021 * call in the session daemon.
1022 */
1023 msg.u.relayd_sock.net_index = consumer->net_seq_index;
1024 msg.u.relayd_sock.type = type;
1025 msg.u.relayd_sock.session_id = session_id;
1026 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
1027
1028 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
1029 ret = consumer_send_msg(consumer_sock, &msg);
1030 if (ret < 0) {
1031 goto error;
1032 }
1033
1034 DBG3("Sending relayd socket file descriptor to consumer");
1035 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
1036 if (ret < 0) {
1037 goto error;
1038 }
1039
1040 DBG2("Consumer relayd socket sent");
1041
1042 error:
1043 return ret;
1044 }
1045
1046 /*
1047 * Set consumer subdirectory using the session name and a generated datetime if
1048 * needed. This is appended to the current subdirectory.
1049 */
1050 int consumer_set_subdir(struct consumer_output *consumer,
1051 const char *session_name)
1052 {
1053 int ret = 0;
1054 unsigned int have_default_name = 0;
1055 char datetime[16], tmp_path[PATH_MAX];
1056 time_t rawtime;
1057 struct tm *timeinfo;
1058
1059 assert(consumer);
1060 assert(session_name);
1061
1062 memset(tmp_path, 0, sizeof(tmp_path));
1063
1064 /* Flag if we have a default session. */
1065 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
1066 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
1067 have_default_name = 1;
1068 } else {
1069 /* Get date and time for session path */
1070 time(&rawtime);
1071 timeinfo = localtime(&rawtime);
1072 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
1073 }
1074
1075 if (have_default_name) {
1076 ret = snprintf(tmp_path, sizeof(tmp_path),
1077 "%s/%s", consumer->subdir, session_name);
1078 } else {
1079 ret = snprintf(tmp_path, sizeof(tmp_path),
1080 "%s/%s-%s/", consumer->subdir, session_name, datetime);
1081 }
1082 if (ret < 0) {
1083 PERROR("snprintf session name date");
1084 goto error;
1085 }
1086
1087 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
1088 DBG2("Consumer subdir set to %s", consumer->subdir);
1089
1090 error:
1091 return ret;
1092 }
1093
1094 /*
1095 * Ask the consumer if the data is pending for the specific session id.
1096 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1097 */
1098 int consumer_is_data_pending(uint64_t session_id,
1099 struct consumer_output *consumer)
1100 {
1101 int ret;
1102 int32_t ret_code = 0; /* Default is that the data is NOT pending */
1103 struct consumer_socket *socket;
1104 struct lttng_ht_iter iter;
1105 struct lttcomm_consumer_msg msg;
1106
1107 assert(consumer);
1108
1109 DBG3("Consumer data pending for id %" PRIu64, session_id);
1110
1111 memset(&msg, 0, sizeof(msg));
1112 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1113 msg.u.data_pending.session_id = session_id;
1114
1115 /* Send command for each consumer */
1116 rcu_read_lock();
1117 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1118 node.node) {
1119 pthread_mutex_lock(socket->lock);
1120 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1121 if (ret < 0) {
1122 pthread_mutex_unlock(socket->lock);
1123 goto error_unlock;
1124 }
1125
1126 /*
1127 * No need for a recv reply status because the answer to the command is
1128 * the reply status message.
1129 */
1130
1131 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1132 if (ret < 0) {
1133 pthread_mutex_unlock(socket->lock);
1134 goto error_unlock;
1135 }
1136 pthread_mutex_unlock(socket->lock);
1137
1138 if (ret_code == 1) {
1139 break;
1140 }
1141 }
1142 rcu_read_unlock();
1143
1144 DBG("Consumer data is %s pending for session id %" PRIu64,
1145 ret_code == 1 ? "" : "NOT", session_id);
1146 return ret_code;
1147
1148 error_unlock:
1149 rcu_read_unlock();
1150 return -1;
1151 }
1152
1153 /*
1154 * Send a flush command to consumer using the given channel key.
1155 *
1156 * Return 0 on success else a negative value.
1157 */
1158 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1159 {
1160 int ret;
1161 struct lttcomm_consumer_msg msg;
1162
1163 assert(socket);
1164
1165 DBG2("Consumer flush channel key %" PRIu64, key);
1166
1167 memset(&msg, 0, sizeof(msg));
1168 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1169 msg.u.flush_channel.key = key;
1170
1171 pthread_mutex_lock(socket->lock);
1172 health_code_update();
1173
1174 ret = consumer_send_msg(socket, &msg);
1175 if (ret < 0) {
1176 goto end;
1177 }
1178
1179 end:
1180 health_code_update();
1181 pthread_mutex_unlock(socket->lock);
1182 return ret;
1183 }
1184
1185 /*
1186 * Send a close metadata command to consumer using the given channel key.
1187 * Called with registry lock held.
1188 *
1189 * Return 0 on success else a negative value.
1190 */
1191 int consumer_close_metadata(struct consumer_socket *socket,
1192 uint64_t metadata_key)
1193 {
1194 int ret;
1195 struct lttcomm_consumer_msg msg;
1196
1197 assert(socket);
1198
1199 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1200
1201 memset(&msg, 0, sizeof(msg));
1202 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1203 msg.u.close_metadata.key = metadata_key;
1204
1205 pthread_mutex_lock(socket->lock);
1206 health_code_update();
1207
1208 ret = consumer_send_msg(socket, &msg);
1209 if (ret < 0) {
1210 goto end;
1211 }
1212
1213 end:
1214 health_code_update();
1215 pthread_mutex_unlock(socket->lock);
1216 return ret;
1217 }
1218
1219 /*
1220 * Send a setup metdata command to consumer using the given channel key.
1221 *
1222 * Return 0 on success else a negative value.
1223 */
1224 int consumer_setup_metadata(struct consumer_socket *socket,
1225 uint64_t metadata_key)
1226 {
1227 int ret;
1228 struct lttcomm_consumer_msg msg;
1229
1230 assert(socket);
1231
1232 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1233
1234 memset(&msg, 0, sizeof(msg));
1235 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1236 msg.u.setup_metadata.key = metadata_key;
1237
1238 pthread_mutex_lock(socket->lock);
1239 health_code_update();
1240
1241 ret = consumer_send_msg(socket, &msg);
1242 if (ret < 0) {
1243 goto end;
1244 }
1245
1246 end:
1247 health_code_update();
1248 pthread_mutex_unlock(socket->lock);
1249 return ret;
1250 }
1251
1252 /*
1253 * Send metadata string to consumer.
1254 * RCU read-side lock must be held to guarantee existence of socket.
1255 *
1256 * Return 0 on success else a negative value.
1257 */
1258 int consumer_push_metadata(struct consumer_socket *socket,
1259 uint64_t metadata_key, char *metadata_str, size_t len,
1260 size_t target_offset)
1261 {
1262 int ret;
1263 struct lttcomm_consumer_msg msg;
1264
1265 assert(socket);
1266
1267 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1268
1269 pthread_mutex_lock(socket->lock);
1270
1271 memset(&msg, 0, sizeof(msg));
1272 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1273 msg.u.push_metadata.key = metadata_key;
1274 msg.u.push_metadata.target_offset = target_offset;
1275 msg.u.push_metadata.len = len;
1276
1277 health_code_update();
1278 ret = consumer_send_msg(socket, &msg);
1279 if (ret < 0 || len == 0) {
1280 goto end;
1281 }
1282
1283 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1284 len);
1285
1286 ret = consumer_socket_send(socket, metadata_str, len);
1287 if (ret < 0) {
1288 goto end;
1289 }
1290
1291 health_code_update();
1292 ret = consumer_recv_status_reply(socket);
1293 if (ret < 0) {
1294 goto end;
1295 }
1296
1297 end:
1298 pthread_mutex_unlock(socket->lock);
1299 health_code_update();
1300 return ret;
1301 }
1302
1303 /*
1304 * Ask the consumer to snapshot a specific channel using the key.
1305 *
1306 * Return 0 on success or else a negative error.
1307 */
1308 int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
1309 struct snapshot_output *output, int metadata, uid_t uid, gid_t gid,
1310 const char *session_path, int wait, uint64_t nb_packets_per_stream)
1311 {
1312 int ret;
1313 struct lttcomm_consumer_msg msg;
1314
1315 assert(socket);
1316 assert(output);
1317 assert(output->consumer);
1318
1319 DBG("Consumer snapshot channel key %" PRIu64, key);
1320
1321 memset(&msg, 0, sizeof(msg));
1322 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1323 msg.u.snapshot_channel.key = key;
1324 msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
1325 msg.u.snapshot_channel.metadata = metadata;
1326
1327 if (output->consumer->type == CONSUMER_DST_NET) {
1328 msg.u.snapshot_channel.relayd_id = output->consumer->net_seq_index;
1329 msg.u.snapshot_channel.use_relayd = 1;
1330 ret = snprintf(msg.u.snapshot_channel.pathname,
1331 sizeof(msg.u.snapshot_channel.pathname),
1332 "%s/%s-%s-%" PRIu64 "%s", output->consumer->subdir,
1333 output->name, output->datetime, output->nb_snapshot,
1334 session_path);
1335 if (ret < 0) {
1336 ret = -LTTNG_ERR_NOMEM;
1337 goto error;
1338 }
1339 } else {
1340 ret = snprintf(msg.u.snapshot_channel.pathname,
1341 sizeof(msg.u.snapshot_channel.pathname),
1342 "%s/%s-%s-%" PRIu64 "%s", output->consumer->dst.trace_path,
1343 output->name, output->datetime, output->nb_snapshot,
1344 session_path);
1345 if (ret < 0) {
1346 ret = -LTTNG_ERR_NOMEM;
1347 goto error;
1348 }
1349 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1350
1351 /* Create directory. Ignore if exist. */
1352 ret = run_as_mkdir_recursive(msg.u.snapshot_channel.pathname,
1353 S_IRWXU | S_IRWXG, uid, gid);
1354 if (ret < 0) {
1355 if (errno != EEXIST) {
1356 ERR("Trace directory creation error");
1357 goto error;
1358 }
1359 }
1360 }
1361
1362 health_code_update();
1363 ret = consumer_send_msg(socket, &msg);
1364 if (ret < 0) {
1365 goto error;
1366 }
1367
1368 error:
1369 health_code_update();
1370 return ret;
1371 }
1372
1373 /*
1374 * Ask the consumer the number of discarded events for a channel.
1375 */
1376 int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1377 struct consumer_output *consumer, uint64_t *discarded)
1378 {
1379 int ret;
1380 struct consumer_socket *socket;
1381 struct lttng_ht_iter iter;
1382 struct lttcomm_consumer_msg msg;
1383
1384 assert(consumer);
1385
1386 DBG3("Consumer discarded events id %" PRIu64, session_id);
1387
1388 memset(&msg, 0, sizeof(msg));
1389 msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1390 msg.u.discarded_events.session_id = session_id;
1391 msg.u.discarded_events.channel_key = channel_key;
1392
1393 *discarded = 0;
1394
1395 /* Send command for each consumer */
1396 rcu_read_lock();
1397 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1398 node.node) {
1399 uint64_t consumer_discarded = 0;
1400 pthread_mutex_lock(socket->lock);
1401 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1402 if (ret < 0) {
1403 pthread_mutex_unlock(socket->lock);
1404 goto end;
1405 }
1406
1407 /*
1408 * No need for a recv reply status because the answer to the
1409 * command is the reply status message.
1410 */
1411 ret = consumer_socket_recv(socket, &consumer_discarded,
1412 sizeof(consumer_discarded));
1413 if (ret < 0) {
1414 ERR("get discarded events");
1415 pthread_mutex_unlock(socket->lock);
1416 goto end;
1417 }
1418 pthread_mutex_unlock(socket->lock);
1419 *discarded += consumer_discarded;
1420 }
1421 ret = 0;
1422 DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1423 *discarded, session_id);
1424
1425 end:
1426 rcu_read_unlock();
1427 return ret;
1428 }
1429
1430 /*
1431 * Ask the consumer the number of lost packets for a channel.
1432 */
1433 int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1434 struct consumer_output *consumer, uint64_t *lost)
1435 {
1436 int ret;
1437 struct consumer_socket *socket;
1438 struct lttng_ht_iter iter;
1439 struct lttcomm_consumer_msg msg;
1440
1441 assert(consumer);
1442
1443 DBG3("Consumer lost packets id %" PRIu64, session_id);
1444
1445 memset(&msg, 0, sizeof(msg));
1446 msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1447 msg.u.lost_packets.session_id = session_id;
1448 msg.u.lost_packets.channel_key = channel_key;
1449
1450 *lost = 0;
1451
1452 /* Send command for each consumer */
1453 rcu_read_lock();
1454 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1455 node.node) {
1456 uint64_t consumer_lost = 0;
1457 pthread_mutex_lock(socket->lock);
1458 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1459 if (ret < 0) {
1460 pthread_mutex_unlock(socket->lock);
1461 goto end;
1462 }
1463
1464 /*
1465 * No need for a recv reply status because the answer to the
1466 * command is the reply status message.
1467 */
1468 ret = consumer_socket_recv(socket, &consumer_lost,
1469 sizeof(consumer_lost));
1470 if (ret < 0) {
1471 ERR("get lost packets");
1472 pthread_mutex_unlock(socket->lock);
1473 goto end;
1474 }
1475 pthread_mutex_unlock(socket->lock);
1476 *lost += consumer_lost;
1477 }
1478 ret = 0;
1479 DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1480 *lost, session_id);
1481
1482 end:
1483 rcu_read_unlock();
1484 return ret;
1485 }
This page took 0.060439 seconds and 4 git commands to generate.