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