0cf43d2ca3da002ffb7d68aef558316c75b9006a
[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 _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 */
43 int 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
69 end:
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 */
81 int 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
111 end:
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 */
120 int 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
154 error_send:
155 pthread_mutex_unlock(sock->lock);
156 error:
157 return ret;
158 }
159
160 /*
161 * For each consumer socket in the consumer output object, send a destroy
162 * relayd command.
163 */
164 void 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 */
195 int 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
231 error:
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 */
242 struct 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
266 end:
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 */
275 struct 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 */
300 struct 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
313 error:
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 */
321 void 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 */
334 void 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 */
351 static 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 */
365 void 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 */
387 struct 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
404 error:
405 return output;
406 }
407
408 /*
409 * Delete the consumer_output object from the list and free the ptr.
410 */
411 void consumer_destroy_output(struct consumer_output *obj)
412 {
413 if (obj == NULL) {
414 return;
415 }
416
417 if (obj->socks) {
418 struct lttng_ht_iter iter;
419 struct consumer_socket *socket;
420
421 rcu_read_lock();
422 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
423 consumer_del_socket(socket, obj);
424 consumer_destroy_socket(socket);
425 }
426 rcu_read_unlock();
427
428 /* Finally destroy HT */
429 lttng_ht_destroy(obj->socks);
430 }
431
432 free(obj);
433 }
434
435 /*
436 * Copy consumer output and returned the newly allocated copy.
437 */
438 struct consumer_output *consumer_copy_output(struct consumer_output *obj)
439 {
440 struct lttng_ht *tmp_ht_ptr;
441 struct lttng_ht_iter iter;
442 struct consumer_socket *socket, *copy_sock;
443 struct consumer_output *output;
444
445 assert(obj);
446
447 output = consumer_create_output(obj->type);
448 if (output == NULL) {
449 goto error;
450 }
451 /* Avoid losing the HT reference after the memcpy() */
452 tmp_ht_ptr = output->socks;
453
454 memcpy(output, obj, sizeof(struct consumer_output));
455
456 /* Putting back the HT pointer and start copying socket(s). */
457 output->socks = tmp_ht_ptr;
458
459 rcu_read_lock();
460 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
461 /* Create new socket object. */
462 copy_sock = consumer_allocate_socket(socket->fd);
463 if (copy_sock == NULL) {
464 rcu_read_unlock();
465 goto malloc_error;
466 }
467
468 copy_sock->registered = socket->registered;
469 copy_sock->lock = socket->lock;
470 consumer_add_socket(copy_sock, output);
471 }
472 rcu_read_unlock();
473
474 error:
475 return output;
476
477 malloc_error:
478 consumer_destroy_output(output);
479 return NULL;
480 }
481
482 /*
483 * Set network URI to the consumer output object.
484 *
485 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
486 * error.
487 */
488 int consumer_set_network_uri(struct consumer_output *obj,
489 struct lttng_uri *uri)
490 {
491 int ret;
492 char tmp_path[PATH_MAX];
493 char hostname[HOST_NAME_MAX];
494 struct lttng_uri *dst_uri = NULL;
495
496 /* Code flow error safety net. */
497 assert(obj);
498 assert(uri);
499
500 switch (uri->stype) {
501 case LTTNG_STREAM_CONTROL:
502 dst_uri = &obj->dst.net.control;
503 obj->dst.net.control_isset = 1;
504 if (uri->port == 0) {
505 /* Assign default port. */
506 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
507 }
508 DBG3("Consumer control URI set with port %d", uri->port);
509 break;
510 case LTTNG_STREAM_DATA:
511 dst_uri = &obj->dst.net.data;
512 obj->dst.net.data_isset = 1;
513 if (uri->port == 0) {
514 /* Assign default port. */
515 uri->port = DEFAULT_NETWORK_DATA_PORT;
516 }
517 DBG3("Consumer data URI set with port %d", uri->port);
518 break;
519 default:
520 ERR("Set network uri type unknown %d", uri->stype);
521 goto error;
522 }
523
524 ret = uri_compare(dst_uri, uri);
525 if (!ret) {
526 /* Same URI, don't touch it and return success. */
527 DBG3("URI network compare are the same");
528 goto equal;
529 }
530
531 /* URIs were not equal, replacing it. */
532 memset(dst_uri, 0, sizeof(struct lttng_uri));
533 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
534 obj->type = CONSUMER_DST_NET;
535
536 /* Handle subdir and add hostname in front. */
537 if (dst_uri->stype == LTTNG_STREAM_CONTROL) {
538 /* Get hostname to append it in the pathname */
539 ret = gethostname(hostname, sizeof(hostname));
540 if (ret < 0) {
541 PERROR("gethostname. Fallback on default localhost");
542 strncpy(hostname, "localhost", sizeof(hostname));
543 }
544 hostname[sizeof(hostname) - 1] = '\0';
545
546 /* Setup consumer subdir if none present in the control URI */
547 if (strlen(dst_uri->subdir) == 0) {
548 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
549 hostname, obj->subdir);
550 } else {
551 ret = snprintf(tmp_path, sizeof(tmp_path), "%s/%s",
552 hostname, dst_uri->subdir);
553 }
554 if (ret < 0) {
555 PERROR("snprintf set consumer uri subdir");
556 goto error;
557 }
558
559 strncpy(obj->subdir, tmp_path, sizeof(obj->subdir));
560 DBG3("Consumer set network uri subdir path %s", tmp_path);
561 }
562
563 return 0;
564 equal:
565 return 1;
566 error:
567 return -1;
568 }
569
570 /*
571 * Send file descriptor to consumer via sock.
572 */
573 int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
574 {
575 int ret;
576
577 assert(fds);
578 assert(sock);
579 assert(nb_fd > 0);
580
581 ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
582 if (ret < 0) {
583 /* The above call will print a PERROR on error. */
584 DBG("Error when sending consumer fds on sock %d", sock->fd);
585 goto error;
586 }
587
588 ret = consumer_recv_status_reply(sock);
589
590 error:
591 return ret;
592 }
593
594 /*
595 * Consumer send communication message structure to consumer.
596 */
597 int consumer_send_msg(struct consumer_socket *sock,
598 struct lttcomm_consumer_msg *msg)
599 {
600 int ret;
601
602 assert(msg);
603 assert(sock);
604 assert(sock->fd >= 0);
605
606 ret = lttcomm_send_unix_sock(sock->fd, msg,
607 sizeof(struct lttcomm_consumer_msg));
608 if (ret < 0) {
609 /* The above call will print a PERROR on error. */
610 DBG("Error when sending consumer channel on sock %d", sock->fd);
611 goto error;
612 }
613
614 ret = consumer_recv_status_reply(sock);
615
616 error:
617 return ret;
618 }
619
620 /*
621 * Consumer send channel communication message structure to consumer.
622 */
623 int consumer_send_channel(struct consumer_socket *sock,
624 struct lttcomm_consumer_msg *msg)
625 {
626 int ret;
627
628 assert(msg);
629 assert(sock);
630 assert(sock->fd >= 0);
631
632 ret = lttcomm_send_unix_sock(sock->fd, msg,
633 sizeof(struct lttcomm_consumer_msg));
634 if (ret < 0) {
635 /* The above call will print a PERROR on error. */
636 DBG("Error when sending consumer channel on sock %d", sock->fd);
637 goto error;
638 }
639
640 ret = consumer_recv_status_reply(sock);
641
642 error:
643 return ret;
644 }
645
646 /*
647 * Populate the given consumer msg structure with the ask_channel command
648 * information.
649 */
650 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
651 uint64_t subbuf_size,
652 uint64_t num_subbuf,
653 int overwrite,
654 unsigned int switch_timer_interval,
655 unsigned int read_timer_interval,
656 int output,
657 int type,
658 uint64_t session_id,
659 const char *pathname,
660 const char *name,
661 uid_t uid,
662 gid_t gid,
663 uint64_t relayd_id,
664 uint64_t key,
665 unsigned char *uuid,
666 uint32_t chan_id,
667 uint64_t tracefile_size,
668 uint64_t tracefile_count)
669 {
670 assert(msg);
671
672 /* Zeroed structure */
673 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
674
675 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
676 msg->u.ask_channel.subbuf_size = subbuf_size;
677 msg->u.ask_channel.num_subbuf = num_subbuf ;
678 msg->u.ask_channel.overwrite = overwrite;
679 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
680 msg->u.ask_channel.read_timer_interval = read_timer_interval;
681 msg->u.ask_channel.output = output;
682 msg->u.ask_channel.type = type;
683 msg->u.ask_channel.session_id = session_id;
684 msg->u.ask_channel.uid = uid;
685 msg->u.ask_channel.gid = gid;
686 msg->u.ask_channel.relayd_id = relayd_id;
687 msg->u.ask_channel.key = key;
688 msg->u.ask_channel.chan_id = chan_id;
689 msg->u.ask_channel.tracefile_size = tracefile_size;
690 msg->u.ask_channel.tracefile_count = tracefile_count;
691
692 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
693
694 strncpy(msg->u.ask_channel.pathname, pathname,
695 sizeof(msg->u.ask_channel.pathname));
696 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
697
698 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
699 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
700 }
701
702 /*
703 * Init channel communication message structure.
704 */
705 void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
706 enum lttng_consumer_command cmd,
707 uint64_t channel_key,
708 uint64_t session_id,
709 const char *pathname,
710 uid_t uid,
711 gid_t gid,
712 uint64_t relayd_id,
713 const char *name,
714 unsigned int nb_init_streams,
715 enum lttng_event_output output,
716 int type,
717 uint64_t tracefile_size,
718 uint64_t tracefile_count)
719 {
720 assert(msg);
721
722 /* Zeroed structure */
723 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
724
725 /* Send channel */
726 msg->cmd_type = cmd;
727 msg->u.channel.channel_key = channel_key;
728 msg->u.channel.session_id = session_id;
729 msg->u.channel.uid = uid;
730 msg->u.channel.gid = gid;
731 msg->u.channel.relayd_id = relayd_id;
732 msg->u.channel.nb_init_streams = nb_init_streams;
733 msg->u.channel.output = output;
734 msg->u.channel.type = type;
735 msg->u.channel.tracefile_size = tracefile_size;
736 msg->u.channel.tracefile_count = tracefile_count;
737
738 strncpy(msg->u.channel.pathname, pathname,
739 sizeof(msg->u.channel.pathname));
740 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
741
742 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
743 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
744 }
745
746 /*
747 * Init stream communication message structure.
748 */
749 void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
750 enum lttng_consumer_command cmd,
751 uint64_t channel_key,
752 uint64_t stream_key,
753 int cpu)
754 {
755 assert(msg);
756
757 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
758
759 msg->cmd_type = cmd;
760 msg->u.stream.channel_key = channel_key;
761 msg->u.stream.stream_key = stream_key;
762 msg->u.stream.cpu = cpu;
763 }
764
765 /*
766 * Send stream communication structure to the consumer.
767 */
768 int consumer_send_stream(struct consumer_socket *sock,
769 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
770 int *fds, size_t nb_fd)
771 {
772 int ret;
773
774 assert(msg);
775 assert(dst);
776 assert(sock);
777 assert(fds);
778
779 /* Send on socket */
780 ret = lttcomm_send_unix_sock(sock->fd, msg,
781 sizeof(struct lttcomm_consumer_msg));
782 if (ret < 0) {
783 /* The above call will print a PERROR on error. */
784 DBG("Error when sending consumer stream on sock %d", sock->fd);
785 goto error;
786 }
787
788 ret = consumer_recv_status_reply(sock);
789 if (ret < 0) {
790 goto error;
791 }
792
793 ret = consumer_send_fds(sock, fds, nb_fd);
794 if (ret < 0) {
795 goto error;
796 }
797
798 error:
799 return ret;
800 }
801
802 /*
803 * Send relayd socket to consumer associated with a session name.
804 *
805 * On success return positive value. On error, negative value.
806 */
807 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
808 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
809 enum lttng_stream_type type, uint64_t session_id)
810 {
811 int ret;
812 struct lttcomm_consumer_msg msg;
813
814 /* Code flow error. Safety net. */
815 assert(rsock);
816 assert(consumer);
817 assert(consumer_sock);
818
819 /* Bail out if consumer is disabled */
820 if (!consumer->enabled) {
821 ret = LTTNG_OK;
822 goto error;
823 }
824
825 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
826 /*
827 * Assign network consumer output index using the temporary consumer since
828 * this call should only be made from within a set_consumer_uri() function
829 * call in the session daemon.
830 */
831 msg.u.relayd_sock.net_index = consumer->net_seq_index;
832 msg.u.relayd_sock.type = type;
833 msg.u.relayd_sock.session_id = session_id;
834 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
835
836 DBG3("Sending relayd sock info to consumer on %d", consumer_sock->fd);
837 ret = lttcomm_send_unix_sock(consumer_sock->fd, &msg, sizeof(msg));
838 if (ret < 0) {
839 /* The above call will print a PERROR on error. */
840 DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
841 goto error;
842 }
843
844 ret = consumer_recv_status_reply(consumer_sock);
845 if (ret < 0) {
846 goto error;
847 }
848
849 DBG3("Sending relayd socket file descriptor to consumer");
850 ret = consumer_send_fds(consumer_sock, &rsock->sock.fd, 1);
851 if (ret < 0) {
852 goto error;
853 }
854
855 DBG2("Consumer relayd socket sent");
856
857 error:
858 return ret;
859 }
860
861 /*
862 * Set consumer subdirectory using the session name and a generated datetime if
863 * needed. This is appended to the current subdirectory.
864 */
865 int consumer_set_subdir(struct consumer_output *consumer,
866 const char *session_name)
867 {
868 int ret = 0;
869 unsigned int have_default_name = 0;
870 char datetime[16], tmp_path[PATH_MAX];
871 time_t rawtime;
872 struct tm *timeinfo;
873
874 assert(consumer);
875 assert(session_name);
876
877 memset(tmp_path, 0, sizeof(tmp_path));
878
879 /* Flag if we have a default session. */
880 if (strncmp(session_name, DEFAULT_SESSION_NAME "-",
881 strlen(DEFAULT_SESSION_NAME) + 1) == 0) {
882 have_default_name = 1;
883 } else {
884 /* Get date and time for session path */
885 time(&rawtime);
886 timeinfo = localtime(&rawtime);
887 strftime(datetime, sizeof(datetime), "%Y%m%d-%H%M%S", timeinfo);
888 }
889
890 if (have_default_name) {
891 ret = snprintf(tmp_path, sizeof(tmp_path),
892 "%s/%s", consumer->subdir, session_name);
893 } else {
894 ret = snprintf(tmp_path, sizeof(tmp_path),
895 "%s/%s-%s/", consumer->subdir, session_name, datetime);
896 }
897 if (ret < 0) {
898 PERROR("snprintf session name date");
899 goto error;
900 }
901
902 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
903 DBG2("Consumer subdir set to %s", consumer->subdir);
904
905 error:
906 return ret;
907 }
908
909 /*
910 * Ask the consumer if the data is ready to read (NOT pending) for the specific
911 * session id.
912 *
913 * This function has a different behavior with the consumer i.e. that it waits
914 * for a reply from the consumer if yes or no the data is pending.
915 */
916 int consumer_is_data_pending(uint64_t session_id,
917 struct consumer_output *consumer)
918 {
919 int ret;
920 int32_t ret_code = 0; /* Default is that the data is NOT pending */
921 struct consumer_socket *socket;
922 struct lttng_ht_iter iter;
923 struct lttcomm_consumer_msg msg;
924
925 assert(consumer);
926
927 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
928
929 msg.u.data_pending.session_id = session_id;
930
931 DBG3("Consumer data pending for id %" PRIu64, session_id);
932
933 /* Send command for each consumer */
934 rcu_read_lock();
935 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
936 node.node) {
937 /* Code flow error */
938 assert(socket->fd >= 0);
939
940 pthread_mutex_lock(socket->lock);
941
942 ret = lttcomm_send_unix_sock(socket->fd, &msg, sizeof(msg));
943 if (ret < 0) {
944 /* The above call will print a PERROR on error. */
945 DBG("Error on consumer is data pending on sock %d", socket->fd);
946 pthread_mutex_unlock(socket->lock);
947 goto error_unlock;
948 }
949
950 /*
951 * No need for a recv reply status because the answer to the command is
952 * the reply status message.
953 */
954
955 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
956 if (ret <= 0) {
957 if (ret == 0) {
958 /* Orderly shutdown. Don't return 0 which means success. */
959 ret = -1;
960 }
961 /* The above call will print a PERROR on error. */
962 DBG("Error on recv consumer is data pending on sock %d", socket->fd);
963 pthread_mutex_unlock(socket->lock);
964 goto error_unlock;
965 }
966
967 pthread_mutex_unlock(socket->lock);
968
969 if (ret_code == 1) {
970 break;
971 }
972 }
973 rcu_read_unlock();
974
975 DBG("Consumer data is %s pending for session id %" PRIu64,
976 ret_code == 1 ? "" : "NOT", session_id);
977 return ret_code;
978
979 error_unlock:
980 rcu_read_unlock();
981 return -1;
982 }
983
984 /*
985 * Send a flush command to consumer using the given channel key.
986 *
987 * Return 0 on success else a negative value.
988 */
989 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
990 {
991 int ret;
992 struct lttcomm_consumer_msg msg;
993
994 assert(socket);
995 assert(socket->fd >= 0);
996
997 DBG2("Consumer flush channel key %" PRIu64, key);
998
999 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1000 msg.u.flush_channel.key = key;
1001
1002 pthread_mutex_lock(socket->lock);
1003 health_code_update();
1004
1005 ret = consumer_send_msg(socket, &msg);
1006 if (ret < 0) {
1007 goto end;
1008 }
1009
1010 end:
1011 health_code_update();
1012 pthread_mutex_unlock(socket->lock);
1013 return ret;
1014 }
1015
1016 /*
1017 * Send a close metdata command to consumer using the given channel key.
1018 *
1019 * Return 0 on success else a negative value.
1020 */
1021 int consumer_close_metadata(struct consumer_socket *socket,
1022 uint64_t metadata_key)
1023 {
1024 int ret;
1025 struct lttcomm_consumer_msg msg;
1026
1027 assert(socket);
1028 assert(socket->fd >= 0);
1029
1030 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1031
1032 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1033 msg.u.close_metadata.key = metadata_key;
1034
1035 pthread_mutex_lock(socket->lock);
1036 health_code_update();
1037
1038 ret = consumer_send_msg(socket, &msg);
1039 if (ret < 0) {
1040 goto end;
1041 }
1042
1043 end:
1044 health_code_update();
1045 pthread_mutex_unlock(socket->lock);
1046 return ret;
1047 }
1048
1049 /*
1050 * Send a setup metdata command to consumer using the given channel key.
1051 *
1052 * Return 0 on success else a negative value.
1053 */
1054 int consumer_setup_metadata(struct consumer_socket *socket,
1055 uint64_t metadata_key)
1056 {
1057 int ret;
1058 struct lttcomm_consumer_msg msg;
1059
1060 assert(socket);
1061 assert(socket->fd >= 0);
1062
1063 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1064
1065 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1066 msg.u.setup_metadata.key = metadata_key;
1067
1068 pthread_mutex_lock(socket->lock);
1069 health_code_update();
1070
1071 ret = consumer_send_msg(socket, &msg);
1072 if (ret < 0) {
1073 goto end;
1074 }
1075
1076 end:
1077 health_code_update();
1078 pthread_mutex_unlock(socket->lock);
1079 return ret;
1080 }
1081
1082 /*
1083 * Send metadata string to consumer. Socket lock MUST be acquired.
1084 *
1085 * Return 0 on success else a negative value.
1086 */
1087 int consumer_push_metadata(struct consumer_socket *socket,
1088 uint64_t metadata_key, char *metadata_str, size_t len,
1089 size_t target_offset)
1090 {
1091 int ret;
1092 struct lttcomm_consumer_msg msg;
1093
1094 assert(socket);
1095 assert(socket->fd >= 0);
1096
1097 DBG2("Consumer push metadata to consumer socket %d", socket->fd);
1098
1099 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1100 msg.u.push_metadata.key = metadata_key;
1101 msg.u.push_metadata.target_offset = target_offset;
1102 msg.u.push_metadata.len = len;
1103
1104 health_code_update();
1105 ret = consumer_send_msg(socket, &msg);
1106 if (ret < 0 || len == 0) {
1107 goto end;
1108 }
1109
1110 DBG3("Consumer pushing metadata on sock %d of len %lu", socket->fd, len);
1111
1112 ret = lttcomm_send_unix_sock(socket->fd, metadata_str, len);
1113 if (ret < 0) {
1114 goto end;
1115 }
1116
1117 health_code_update();
1118 ret = consumer_recv_status_reply(socket);
1119 if (ret < 0) {
1120 goto end;
1121 }
1122
1123 end:
1124 health_code_update();
1125 return ret;
1126 }
This page took 0.052857 seconds and 4 git commands to generate.