Move ust channel registry inside session registry
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
CommitLineData
00e2e675
DG
1/*
2 * Copyright (C) 2012 - David Goulet <dgoulet@efficios.com>
3 *
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
7 *
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
11 * more details.
12 *
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
16 */
17
18#define _GNU_SOURCE
19#include <assert.h>
20#include <stdio.h>
21#include <stdlib.h>
22#include <string.h>
23#include <sys/stat.h>
24#include <sys/types.h>
25#include <unistd.h>
d88aee68 26#include <inttypes.h>
00e2e675
DG
27
28#include <common/common.h>
29#include <common/defaults.h>
30#include <common/uri.h>
31
32#include "consumer.h"
33
f50f23d9
DG
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 */
41int 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));
a6cd2b97
DG
49 if (ret <= 0) {
50 if (ret == 0) {
51 /* Orderly shutdown. Don't return 0 which means success. */
52 ret = -1;
53 }
3448e266
DG
54 /* The above call will print a PERROR on error. */
55 DBG("Fail to receive status reply on sock %d", sock->fd);
f50f23d9
DG
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;
ffe60014 64 DBG("Consumer ret code %d", ret);
f50f23d9
DG
65 }
66
67end:
68 return ret;
69}
70
ffe60014
DG
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 */
79int consumer_recv_status_channel(struct consumer_socket *sock,
d88aee68 80 uint64_t *key, unsigned int *stream_count)
ffe60014
DG
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
109end:
110 return ret;
111}
112
2f77fc4b
DG
113/*
114 * Send destroy relayd command to consumer.
115 *
116 * On success return positive value. On error, negative value.
117 */
118int 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
ffe60014 127 DBG2("Sending destroy relayd command to consumer sock %d", sock->fd);
2f77fc4b
DG
128
129 /* Bail out if consumer is disabled */
130 if (!consumer->enabled) {
f73fabfd 131 ret = LTTNG_OK;
2f77fc4b
DG
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));
2f77fc4b 141 if (ret < 0) {
c5c45efa
DG
142 /* Indicate that the consumer is probably closing at this point. */
143 DBG("send consumer destroy relayd command");
f50f23d9 144 goto error_send;
2f77fc4b
DG
145 }
146
f50f23d9
DG
147 /* Don't check the return value. The caller will do it. */
148 ret = consumer_recv_status_reply(sock);
149
2f77fc4b
DG
150 DBG2("Consumer send destroy relayd command done");
151
f50f23d9
DG
152error_send:
153 pthread_mutex_unlock(sock->lock);
2f77fc4b
DG
154error:
155 return ret;
156}
157
158/*
159 * For each consumer socket in the consumer output object, send a destroy
160 * relayd command.
161 */
162void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
163{
2f77fc4b
DG
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) {
c617c0c6
MD
174 int ret;
175
2f77fc4b
DG
176 /* Send destroy relayd command */
177 ret = consumer_send_destroy_relayd(socket, consumer);
178 if (ret < 0) {
c5c45efa 179 DBG("Unable to send destroy relayd command to consumer");
2f77fc4b
DG
180 /* Continue since we MUST delete everything at this point. */
181 }
182 }
183 rcu_read_unlock();
184 }
185}
186
a4b92340
DG
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 */
193int 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
2f77fc4b 219 socket->registered = 0;
a4b92340
DG
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
229error:
230 return ret;
231}
232
173af62f
DG
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 */
238struct 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 */
a4b92340 246 if (key < 0 || consumer == NULL) {
173af62f
DG
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 */
263struct 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
276error:
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 */
284void 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 */
297void 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 */
314static 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 */
328void 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
2f77fc4b
DG
334 * session daemon and is closed only if the consumer dies or a custom
335 * consumer was registered,
173af62f 336 */
2f77fc4b
DG
337 if (sock->registered) {
338 DBG3("Consumer socket was registered. Closing fd %d", sock->fd);
339 lttcomm_close_unix_sock(sock->fd);
340 }
173af62f
DG
341
342 call_rcu(&sock->node.head, destroy_socket_rcu);
343}
344
00e2e675
DG
345/*
346 * Allocate and assign data to a consumer_output object.
347 *
348 * Return pointer to structure.
349 */
350struct 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;
d88aee68 363 output->net_seq_index = (uint64_t) -1ULL;
173af62f
DG
364
365 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
00e2e675
DG
366
367error:
368 return output;
369}
370
371/*
372 * Delete the consumer_output object from the list and free the ptr.
373 */
374void consumer_destroy_output(struct consumer_output *obj)
375{
376 if (obj == NULL) {
377 return;
378 }
379
173af62f
DG
380 if (obj->socks) {
381 struct lttng_ht_iter iter;
382 struct consumer_socket *socket;
383
2f77fc4b 384 rcu_read_lock();
173af62f 385 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
2f77fc4b 386 consumer_del_socket(socket, obj);
173af62f
DG
387 consumer_destroy_socket(socket);
388 }
2f77fc4b
DG
389 rcu_read_unlock();
390
391 /* Finally destroy HT */
392 lttng_ht_destroy(obj->socks);
00e2e675 393 }
173af62f 394
00e2e675
DG
395 free(obj);
396}
397
398/*
399 * Copy consumer output and returned the newly allocated copy.
400 */
401struct consumer_output *consumer_copy_output(struct consumer_output *obj)
402{
09a90bcd 403 struct lttng_ht *tmp_ht_ptr;
173af62f
DG
404 struct lttng_ht_iter iter;
405 struct consumer_socket *socket, *copy_sock;
00e2e675
DG
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 }
09a90bcd
DG
414 /* Avoid losing the HT reference after the memcpy() */
415 tmp_ht_ptr = output->socks;
00e2e675
DG
416
417 memcpy(output, obj, sizeof(struct consumer_output));
418
09a90bcd
DG
419 /* Putting back the HT pointer and start copying socket(s). */
420 output->socks = tmp_ht_ptr;
173af62f 421
b82c5c4d 422 rcu_read_lock();
173af62f
DG
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) {
b82c5c4d 427 rcu_read_unlock();
173af62f
DG
428 goto malloc_error;
429 }
430
09a90bcd 431 copy_sock->registered = socket->registered;
173af62f
DG
432 copy_sock->lock = socket->lock;
433 consumer_add_socket(copy_sock, output);
434 }
b82c5c4d 435 rcu_read_unlock();
173af62f 436
00e2e675
DG
437error:
438 return output;
173af62f
DG
439
440malloc_error:
441 consumer_destroy_output(output);
442 return NULL;
00e2e675
DG
443}
444
445/*
446 * Set network URI to the consumer output object.
447 *
ad20f474
DG
448 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
449 * error.
00e2e675
DG
450 */
451int 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 }
ad20f474 471 DBG3("Consumer control URI set with port %d", uri->port);
00e2e675
DG
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 }
ad20f474 480 DBG3("Consumer data URI set with port %d", uri->port);
00e2e675
DG
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");
ad20f474 491 goto equal;
00e2e675
DG
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
00e2e675 526 return 0;
ad20f474
DG
527equal:
528 return 1;
00e2e675
DG
529error:
530 return -1;
531}
532
533/*
534 * Send file descriptor to consumer via sock.
535 */
f50f23d9 536int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
00e2e675
DG
537{
538 int ret;
539
540 assert(fds);
f50f23d9 541 assert(sock);
00e2e675
DG
542 assert(nb_fd > 0);
543
f50f23d9 544 ret = lttcomm_send_fds_unix_sock(sock->fd, fds, nb_fd);
00e2e675 545 if (ret < 0) {
3448e266
DG
546 /* The above call will print a PERROR on error. */
547 DBG("Error when sending consumer fds on sock %d", sock->fd);
00e2e675
DG
548 goto error;
549 }
550
f50f23d9
DG
551 ret = consumer_recv_status_reply(sock);
552
00e2e675
DG
553error:
554 return ret;
555}
556
ffe60014
DG
557/*
558 * Consumer send communication message structure to consumer.
559 */
560int 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
579error:
580 return ret;
581}
582
00e2e675
DG
583/*
584 * Consumer send channel communication message structure to consumer.
585 */
f50f23d9
DG
586int consumer_send_channel(struct consumer_socket *sock,
587 struct lttcomm_consumer_msg *msg)
00e2e675
DG
588{
589 int ret;
590
591 assert(msg);
f50f23d9
DG
592 assert(sock);
593 assert(sock->fd >= 0);
00e2e675 594
f50f23d9 595 ret = lttcomm_send_unix_sock(sock->fd, msg,
00e2e675
DG
596 sizeof(struct lttcomm_consumer_msg));
597 if (ret < 0) {
3448e266
DG
598 /* The above call will print a PERROR on error. */
599 DBG("Error when sending consumer channel on sock %d", sock->fd);
00e2e675
DG
600 goto error;
601 }
602
f50f23d9
DG
603 ret = consumer_recv_status_reply(sock);
604
00e2e675
DG
605error:
606 return ret;
607}
608
ffe60014
DG
609/*
610 * Populate the given consumer msg structure with the ask_channel command
611 * information.
612 */
613void 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,
d88aee68
DG
626 uint64_t relayd_id,
627 uint64_t key,
ffe60014
DG
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
00e2e675
DG
659/*
660 * Init channel communication message structure.
661 */
662void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg,
663 enum lttng_consumer_command cmd,
d88aee68 664 uint64_t channel_key,
ffe60014
DG
665 uint64_t session_id,
666 const char *pathname,
667 uid_t uid,
668 gid_t gid,
d88aee68 669 uint64_t relayd_id,
c30aaa51 670 const char *name,
ffe60014
DG
671 unsigned int nb_init_streams,
672 enum lttng_event_output output,
673 int type)
00e2e675
DG
674{
675 assert(msg);
676
00e2e675
DG
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;
ffe60014
DG
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;
c30aaa51 687 msg->u.channel.nb_init_streams = nb_init_streams;
ffe60014
DG
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';
00e2e675
DG
697}
698
699/*
700 * Init stream communication message structure.
701 */
702void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg,
703 enum lttng_consumer_command cmd,
d88aee68
DG
704 uint64_t channel_key,
705 uint64_t stream_key,
ffe60014 706 int cpu)
00e2e675
DG
707{
708 assert(msg);
709
710 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
711
00e2e675
DG
712 msg->cmd_type = cmd;
713 msg->u.stream.channel_key = channel_key;
714 msg->u.stream.stream_key = stream_key;
ffe60014 715 msg->u.stream.cpu = cpu;
00e2e675
DG
716}
717
718/*
719 * Send stream communication structure to the consumer.
720 */
f50f23d9
DG
721int consumer_send_stream(struct consumer_socket *sock,
722 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
723 int *fds, size_t nb_fd)
00e2e675
DG
724{
725 int ret;
726
727 assert(msg);
728 assert(dst);
f50f23d9 729 assert(sock);
ffe60014 730 assert(fds);
00e2e675
DG
731
732 /* Send on socket */
f50f23d9 733 ret = lttcomm_send_unix_sock(sock->fd, msg,
00e2e675
DG
734 sizeof(struct lttcomm_consumer_msg));
735 if (ret < 0) {
3448e266
DG
736 /* The above call will print a PERROR on error. */
737 DBG("Error when sending consumer stream on sock %d", sock->fd);
00e2e675
DG
738 goto error;
739 }
740
f50f23d9
DG
741 ret = consumer_recv_status_reply(sock);
742 if (ret < 0) {
743 goto error;
744 }
745
00e2e675
DG
746 ret = consumer_send_fds(sock, fds, nb_fd);
747 if (ret < 0) {
748 goto error;
749 }
750
751error:
752 return ret;
753}
37278a1e
DG
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 */
f50f23d9 760int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
37278a1e 761 struct lttcomm_sock *sock, struct consumer_output *consumer,
d88aee68 762 enum lttng_stream_type type, uint64_t session_id)
37278a1e
DG
763{
764 int ret;
765 struct lttcomm_consumer_msg msg;
766
767 /* Code flow error. Safety net. */
768 assert(sock);
769 assert(consumer);
f50f23d9 770 assert(consumer_sock);
37278a1e
DG
771
772 /* Bail out if consumer is disabled */
773 if (!consumer->enabled) {
f73fabfd 774 ret = LTTNG_OK;
37278a1e
DG
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;
46e6455f 786 msg.u.relayd_sock.session_id = session_id;
37278a1e
DG
787 memcpy(&msg.u.relayd_sock.sock, sock, sizeof(msg.u.relayd_sock.sock));
788
f50f23d9
DG
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));
37278a1e 791 if (ret < 0) {
3448e266
DG
792 /* The above call will print a PERROR on error. */
793 DBG("Error when sending relayd sockets on sock %d", sock->fd);
37278a1e
DG
794 goto error;
795 }
796
f50f23d9
DG
797 ret = consumer_recv_status_reply(consumer_sock);
798 if (ret < 0) {
799 goto error;
800 }
801
37278a1e
DG
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
810error:
811 return ret;
812}
173af62f
DG
813
814/*
2f77fc4b
DG
815 * Set consumer subdirectory using the session name and a generated datetime if
816 * needed. This is appended to the current subdirectory.
173af62f 817 */
2f77fc4b
DG
818int consumer_set_subdir(struct consumer_output *consumer,
819 const char *session_name)
173af62f 820{
2f77fc4b
DG
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;
173af62f
DG
826
827 assert(consumer);
2f77fc4b
DG
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);
173af62f
DG
841 }
842
2f77fc4b
DG
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 }
173af62f 850 if (ret < 0) {
2f77fc4b 851 PERROR("snprintf session name date");
173af62f
DG
852 goto error;
853 }
854
2f77fc4b
DG
855 strncpy(consumer->subdir, tmp_path, sizeof(consumer->subdir));
856 DBG2("Consumer subdir set to %s", consumer->subdir);
173af62f
DG
857
858error:
859 return ret;
860}
806e2684
DG
861
862/*
6d805429 863 * Ask the consumer if the data is ready to read (NOT pending) for the specific
806e2684
DG
864 * session id.
865 *
866 * This function has a different behavior with the consumer i.e. that it waits
6d805429 867 * for a reply from the consumer if yes or no the data is pending.
806e2684 868 */
d88aee68 869int consumer_is_data_pending(uint64_t session_id,
806e2684
DG
870 struct consumer_output *consumer)
871{
872 int ret;
6d805429 873 int32_t ret_code = 0; /* Default is that the data is NOT pending */
806e2684
DG
874 struct consumer_socket *socket;
875 struct lttng_ht_iter iter;
876 struct lttcomm_consumer_msg msg;
877
878 assert(consumer);
879
6d805429 880 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
806e2684 881
d88aee68 882 msg.u.data_pending.session_id = session_id;
806e2684 883
d88aee68 884 DBG3("Consumer data pending for id %" PRIu64, session_id);
806e2684 885
c8f59ee5 886 /* Send command for each consumer */
b82c5c4d 887 rcu_read_lock();
806e2684
DG
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) {
3448e266
DG
897 /* The above call will print a PERROR on error. */
898 DBG("Error on consumer is data pending on sock %d", socket->fd);
806e2684 899 pthread_mutex_unlock(socket->lock);
b82c5c4d 900 goto error_unlock;
806e2684
DG
901 }
902
f50f23d9
DG
903 /*
904 * No need for a recv reply status because the answer to the command is
905 * the reply status message.
906 */
907
806e2684 908 ret = lttcomm_recv_unix_sock(socket->fd, &ret_code, sizeof(ret_code));
a6cd2b97
DG
909 if (ret <= 0) {
910 if (ret == 0) {
911 /* Orderly shutdown. Don't return 0 which means success. */
912 ret = -1;
913 }
3448e266
DG
914 /* The above call will print a PERROR on error. */
915 DBG("Error on recv consumer is data pending on sock %d", socket->fd);
806e2684 916 pthread_mutex_unlock(socket->lock);
b82c5c4d 917 goto error_unlock;
806e2684
DG
918 }
919
920 pthread_mutex_unlock(socket->lock);
921
6d805429 922 if (ret_code == 1) {
806e2684
DG
923 break;
924 }
925 }
b82c5c4d 926 rcu_read_unlock();
806e2684 927
d88aee68
DG
928 DBG("Consumer data is %s pending for session id %" PRIu64,
929 ret_code == 1 ? "" : "NOT", session_id);
806e2684
DG
930 return ret_code;
931
b82c5c4d
DG
932error_unlock:
933 rcu_read_unlock();
806e2684
DG
934 return -1;
935}
This page took 0.109549 seconds and 4 git commands to generate.