Cleanup: ust-app.c: Dead assignment
[lttng-tools.git] / src / bin / lttng-sessiond / consumer.c
1 /*
2 * Copyright (C) 2012 David Goulet <dgoulet@efficios.com>
3 * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 *
5 * SPDX-License-Identifier: GPL-2.0-only
6 *
7 */
8
9 #define _LGPL_SOURCE
10 #include <assert.h>
11 #include <stdio.h>
12 #include <stdlib.h>
13 #include <string.h>
14 #include <sys/stat.h>
15 #include <sys/types.h>
16 #include <unistd.h>
17 #include <inttypes.h>
18
19 #include <common/common.h>
20 #include <common/defaults.h>
21 #include <common/uri.h>
22 #include <common/relayd/relayd.h>
23 #include <common/string-utils/format.h>
24
25 #include "consumer.h"
26 #include "health-sessiond.h"
27 #include "ust-app.h"
28 #include "utils.h"
29 #include "lttng-sessiond.h"
30
31 /*
32 * Return allocated full pathname of the session using the consumer trace path
33 * and subdir if available.
34 *
35 * The caller can safely free(3) the returned value. On error, NULL is
36 * returned.
37 */
38 char *setup_channel_trace_path(struct consumer_output *consumer,
39 const char *session_path, size_t *consumer_path_offset)
40 {
41 int ret;
42 char *pathname;
43
44 assert(consumer);
45 assert(session_path);
46
47 health_code_update();
48
49 /*
50 * Allocate the string ourself to make sure we never exceed
51 * LTTNG_PATH_MAX.
52 */
53 pathname = zmalloc(LTTNG_PATH_MAX);
54 if (!pathname) {
55 goto error;
56 }
57
58 /* Get correct path name destination */
59 if (consumer->type == CONSUMER_DST_NET &&
60 consumer->relay_major_version == 2 &&
61 consumer->relay_minor_version < 11) {
62 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s%s/%s/%s",
63 consumer->dst.net.base_dir,
64 consumer->chunk_path, consumer->domain_subdir,
65 session_path);
66 *consumer_path_offset = 0;
67 } else {
68 ret = snprintf(pathname, LTTNG_PATH_MAX, "%s/%s",
69 consumer->domain_subdir, session_path);
70 *consumer_path_offset = strlen(consumer->domain_subdir) + 1;
71 }
72 DBG3("Consumer trace path relative to current trace chunk: \"%s\"",
73 pathname);
74 if (ret < 0) {
75 PERROR("Failed to format channel path");
76 goto error;
77 } else if (ret >= LTTNG_PATH_MAX) {
78 ERR("Truncation occurred while formatting channel path");
79 goto error;
80 }
81
82 return pathname;
83 error:
84 free(pathname);
85 return NULL;
86 }
87
88 /*
89 * Send a data payload using a given consumer socket of size len.
90 *
91 * The consumer socket lock MUST be acquired before calling this since this
92 * function can change the fd value.
93 *
94 * Return 0 on success else a negative value on error.
95 */
96 int consumer_socket_send(
97 struct consumer_socket *socket, const void *msg, size_t len)
98 {
99 int fd;
100 ssize_t size;
101
102 assert(socket);
103 assert(socket->fd_ptr);
104 assert(msg);
105
106 /* Consumer socket is invalid. Stopping. */
107 fd = *socket->fd_ptr;
108 if (fd < 0) {
109 goto error;
110 }
111
112 size = lttcomm_send_unix_sock(fd, msg, len);
113 if (size < 0) {
114 /* The above call will print a PERROR on error. */
115 DBG("Error when sending data to consumer on sock %d", fd);
116 /*
117 * At this point, the socket is not usable anymore thus closing it and
118 * setting the file descriptor to -1 so it is not reused.
119 */
120
121 /* This call will PERROR on error. */
122 (void) lttcomm_close_unix_sock(fd);
123 *socket->fd_ptr = -1;
124 goto error;
125 }
126
127 return 0;
128
129 error:
130 return -1;
131 }
132
133 /*
134 * Receive a data payload using a given consumer socket of size len.
135 *
136 * The consumer socket lock MUST be acquired before calling this since this
137 * function can change the fd value.
138 *
139 * Return 0 on success else a negative value on error.
140 */
141 int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
142 {
143 int fd;
144 ssize_t size;
145
146 assert(socket);
147 assert(socket->fd_ptr);
148 assert(msg);
149
150 /* Consumer socket is invalid. Stopping. */
151 fd = *socket->fd_ptr;
152 if (fd < 0) {
153 goto error;
154 }
155
156 size = lttcomm_recv_unix_sock(fd, msg, len);
157 if (size <= 0) {
158 /* The above call will print a PERROR on error. */
159 DBG("Error when receiving data from the consumer socket %d", fd);
160 /*
161 * At this point, the socket is not usable anymore thus closing it and
162 * setting the file descriptor to -1 so it is not reused.
163 */
164
165 /* This call will PERROR on error. */
166 (void) lttcomm_close_unix_sock(fd);
167 *socket->fd_ptr = -1;
168 goto error;
169 }
170
171 return 0;
172
173 error:
174 return -1;
175 }
176
177 /*
178 * Receive a reply command status message from the consumer. Consumer socket
179 * lock MUST be acquired before calling this function.
180 *
181 * Return 0 on success, -1 on recv error or a negative lttng error code which
182 * was possibly returned by the consumer.
183 */
184 int consumer_recv_status_reply(struct consumer_socket *sock)
185 {
186 int ret;
187 struct lttcomm_consumer_status_msg reply;
188
189 assert(sock);
190
191 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
192 if (ret < 0) {
193 goto end;
194 }
195
196 if (reply.ret_code == LTTCOMM_CONSUMERD_SUCCESS) {
197 /* All good. */
198 ret = 0;
199 } else {
200 ret = -reply.ret_code;
201 DBG("Consumer ret code %d", ret);
202 }
203
204 end:
205 return ret;
206 }
207
208 /*
209 * Once the ASK_CHANNEL command is sent to the consumer, the channel
210 * information are sent back. This call receives that data and populates key
211 * and stream_count.
212 *
213 * On success return 0 and both key and stream_count are set. On error, a
214 * negative value is sent back and both parameters are untouched.
215 */
216 int consumer_recv_status_channel(struct consumer_socket *sock,
217 uint64_t *key, unsigned int *stream_count)
218 {
219 int ret;
220 struct lttcomm_consumer_status_channel reply;
221
222 assert(sock);
223 assert(stream_count);
224 assert(key);
225
226 ret = consumer_socket_recv(sock, &reply, sizeof(reply));
227 if (ret < 0) {
228 goto end;
229 }
230
231 /* An error is possible so don't touch the key and stream_count. */
232 if (reply.ret_code != LTTCOMM_CONSUMERD_SUCCESS) {
233 ret = -1;
234 goto end;
235 }
236
237 *key = reply.key;
238 *stream_count = reply.stream_count;
239 ret = 0;
240
241 end:
242 return ret;
243 }
244
245 /*
246 * Send destroy relayd command to consumer.
247 *
248 * On success return positive value. On error, negative value.
249 */
250 int consumer_send_destroy_relayd(struct consumer_socket *sock,
251 struct consumer_output *consumer)
252 {
253 int ret;
254 struct lttcomm_consumer_msg msg;
255
256 assert(consumer);
257 assert(sock);
258
259 DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
260
261 memset(&msg, 0, sizeof(msg));
262 msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
263 msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
264
265 pthread_mutex_lock(sock->lock);
266 ret = consumer_socket_send(sock, &msg, sizeof(msg));
267 if (ret < 0) {
268 goto error;
269 }
270
271 /* Don't check the return value. The caller will do it. */
272 ret = consumer_recv_status_reply(sock);
273
274 DBG2("Consumer send destroy relayd command done");
275
276 error:
277 pthread_mutex_unlock(sock->lock);
278 return ret;
279 }
280
281 /*
282 * For each consumer socket in the consumer output object, send a destroy
283 * relayd command.
284 */
285 void consumer_output_send_destroy_relayd(struct consumer_output *consumer)
286 {
287 struct lttng_ht_iter iter;
288 struct consumer_socket *socket;
289
290 assert(consumer);
291
292 /* Destroy any relayd connection */
293 if (consumer->type == CONSUMER_DST_NET) {
294 rcu_read_lock();
295 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
296 node.node) {
297 int ret;
298
299 /* Send destroy relayd command */
300 ret = consumer_send_destroy_relayd(socket, consumer);
301 if (ret < 0) {
302 DBG("Unable to send destroy relayd command to consumer");
303 /* Continue since we MUST delete everything at this point. */
304 }
305 }
306 rcu_read_unlock();
307 }
308 }
309
310 /*
311 * From a consumer_data structure, allocate and add a consumer socket to the
312 * consumer output.
313 *
314 * Return 0 on success, else negative value on error
315 */
316 int consumer_create_socket(struct consumer_data *data,
317 struct consumer_output *output)
318 {
319 int ret = 0;
320 struct consumer_socket *socket;
321
322 assert(data);
323
324 if (output == NULL || data->cmd_sock < 0) {
325 /*
326 * Not an error. Possible there is simply not spawned consumer or it's
327 * disabled for the tracing session asking the socket.
328 */
329 goto error;
330 }
331
332 rcu_read_lock();
333 socket = consumer_find_socket(data->cmd_sock, output);
334 rcu_read_unlock();
335 if (socket == NULL) {
336 socket = consumer_allocate_socket(&data->cmd_sock);
337 if (socket == NULL) {
338 ret = -1;
339 goto error;
340 }
341
342 socket->registered = 0;
343 socket->lock = &data->lock;
344 rcu_read_lock();
345 consumer_add_socket(socket, output);
346 rcu_read_unlock();
347 }
348
349 socket->type = data->type;
350
351 DBG3("Consumer socket created (fd: %d) and added to output",
352 data->cmd_sock);
353
354 error:
355 return ret;
356 }
357
358 /*
359 * Return the consumer socket from the given consumer output with the right
360 * bitness. On error, returns NULL.
361 *
362 * The caller MUST acquire a rcu read side lock and keep it until the socket
363 * object reference is not needed anymore.
364 */
365 struct consumer_socket *consumer_find_socket_by_bitness(int bits,
366 const struct consumer_output *consumer)
367 {
368 int consumer_fd;
369 struct consumer_socket *socket = NULL;
370
371 switch (bits) {
372 case 64:
373 consumer_fd = uatomic_read(&the_ust_consumerd64_fd);
374 break;
375 case 32:
376 consumer_fd = uatomic_read(&the_ust_consumerd32_fd);
377 break;
378 default:
379 assert(0);
380 goto end;
381 }
382
383 socket = consumer_find_socket(consumer_fd, consumer);
384 if (!socket) {
385 ERR("Consumer socket fd %d not found in consumer obj %p",
386 consumer_fd, consumer);
387 }
388
389 end:
390 return socket;
391 }
392
393 /*
394 * Find a consumer_socket in a consumer_output hashtable. Read side lock must
395 * be acquired before calling this function and across use of the
396 * returned consumer_socket.
397 */
398 struct consumer_socket *consumer_find_socket(int key,
399 const struct consumer_output *consumer)
400 {
401 struct lttng_ht_iter iter;
402 struct lttng_ht_node_ulong *node;
403 struct consumer_socket *socket = NULL;
404
405 /* Negative keys are lookup failures */
406 if (key < 0 || consumer == NULL) {
407 return NULL;
408 }
409
410 lttng_ht_lookup(consumer->socks, (void *)((unsigned long) key),
411 &iter);
412 node = lttng_ht_iter_get_node_ulong(&iter);
413 if (node != NULL) {
414 socket = caa_container_of(node, struct consumer_socket, node);
415 }
416
417 return socket;
418 }
419
420 /*
421 * Allocate a new consumer_socket and return the pointer.
422 */
423 struct consumer_socket *consumer_allocate_socket(int *fd)
424 {
425 struct consumer_socket *socket = NULL;
426
427 assert(fd);
428
429 socket = zmalloc(sizeof(struct consumer_socket));
430 if (socket == NULL) {
431 PERROR("zmalloc consumer socket");
432 goto error;
433 }
434
435 socket->fd_ptr = fd;
436 lttng_ht_node_init_ulong(&socket->node, *fd);
437
438 error:
439 return socket;
440 }
441
442 /*
443 * Add consumer socket to consumer output object. Read side lock must be
444 * acquired before calling this function.
445 */
446 void consumer_add_socket(struct consumer_socket *sock,
447 struct consumer_output *consumer)
448 {
449 assert(sock);
450 assert(consumer);
451
452 lttng_ht_add_unique_ulong(consumer->socks, &sock->node);
453 }
454
455 /*
456 * Delete consumer socket to consumer output object. Read side lock must be
457 * acquired before calling this function.
458 */
459 void consumer_del_socket(struct consumer_socket *sock,
460 struct consumer_output *consumer)
461 {
462 int ret;
463 struct lttng_ht_iter iter;
464
465 assert(sock);
466 assert(consumer);
467
468 iter.iter.node = &sock->node.node;
469 ret = lttng_ht_del(consumer->socks, &iter);
470 assert(!ret);
471 }
472
473 /*
474 * RCU destroy call function.
475 */
476 static void destroy_socket_rcu(struct rcu_head *head)
477 {
478 struct lttng_ht_node_ulong *node =
479 caa_container_of(head, struct lttng_ht_node_ulong, head);
480 struct consumer_socket *socket =
481 caa_container_of(node, struct consumer_socket, node);
482
483 free(socket);
484 }
485
486 /*
487 * Destroy and free socket pointer in a call RCU. Read side lock must be
488 * acquired before calling this function.
489 */
490 void consumer_destroy_socket(struct consumer_socket *sock)
491 {
492 assert(sock);
493
494 /*
495 * We DO NOT close the file descriptor here since it is global to the
496 * session daemon and is closed only if the consumer dies or a custom
497 * consumer was registered,
498 */
499 if (sock->registered) {
500 DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
501 lttcomm_close_unix_sock(*sock->fd_ptr);
502 }
503
504 call_rcu(&sock->node.head, destroy_socket_rcu);
505 }
506
507 /*
508 * Allocate and assign data to a consumer_output object.
509 *
510 * Return pointer to structure.
511 */
512 struct consumer_output *consumer_create_output(enum consumer_dst_type type)
513 {
514 struct consumer_output *output = NULL;
515
516 output = zmalloc(sizeof(struct consumer_output));
517 if (output == NULL) {
518 PERROR("zmalloc consumer_output");
519 goto error;
520 }
521
522 /* By default, consumer output is enabled */
523 output->enabled = 1;
524 output->type = type;
525 output->net_seq_index = (uint64_t) -1ULL;
526 urcu_ref_init(&output->ref);
527
528 output->socks = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
529
530 error:
531 return output;
532 }
533
534 /*
535 * Iterate over the consumer output socket hash table and destroy them. The
536 * socket file descriptor are only closed if the consumer output was
537 * registered meaning it's an external consumer.
538 */
539 void consumer_destroy_output_sockets(struct consumer_output *obj)
540 {
541 struct lttng_ht_iter iter;
542 struct consumer_socket *socket;
543
544 if (!obj->socks) {
545 return;
546 }
547
548 rcu_read_lock();
549 cds_lfht_for_each_entry(obj->socks->ht, &iter.iter, socket, node.node) {
550 consumer_del_socket(socket, obj);
551 consumer_destroy_socket(socket);
552 }
553 rcu_read_unlock();
554 }
555
556 /*
557 * Delete the consumer_output object from the list and free the ptr.
558 *
559 * Should *NOT* be called with RCU read-side lock held.
560 */
561 static void consumer_release_output(struct urcu_ref *ref)
562 {
563 struct consumer_output *obj =
564 caa_container_of(ref, struct consumer_output, ref);
565
566 consumer_destroy_output_sockets(obj);
567
568 if (obj->socks) {
569 /* Finally destroy HT */
570 ht_cleanup_push(obj->socks);
571 }
572
573 free(obj);
574 }
575
576 /*
577 * Get the consumer_output object.
578 */
579 void consumer_output_get(struct consumer_output *obj)
580 {
581 urcu_ref_get(&obj->ref);
582 }
583
584 /*
585 * Put the consumer_output object.
586 *
587 * Should *NOT* be called with RCU read-side lock held.
588 */
589 void consumer_output_put(struct consumer_output *obj)
590 {
591 if (!obj) {
592 return;
593 }
594 urcu_ref_put(&obj->ref, consumer_release_output);
595 }
596
597 /*
598 * Copy consumer output and returned the newly allocated copy.
599 *
600 * Should *NOT* be called with RCU read-side lock held.
601 */
602 struct consumer_output *consumer_copy_output(struct consumer_output *src)
603 {
604 int ret;
605 struct consumer_output *output;
606
607 assert(src);
608
609 output = consumer_create_output(src->type);
610 if (output == NULL) {
611 goto end;
612 }
613 output->enabled = src->enabled;
614 output->net_seq_index = src->net_seq_index;
615 memcpy(output->domain_subdir, src->domain_subdir,
616 sizeof(output->domain_subdir));
617 output->snapshot = src->snapshot;
618 output->relay_major_version = src->relay_major_version;
619 output->relay_minor_version = src->relay_minor_version;
620 output->relay_allows_clear = src->relay_allows_clear;
621 memcpy(&output->dst, &src->dst, sizeof(output->dst));
622 ret = consumer_copy_sockets(output, src);
623 if (ret < 0) {
624 goto error_put;
625 }
626 end:
627 return output;
628
629 error_put:
630 consumer_output_put(output);
631 return NULL;
632 }
633
634 /*
635 * Copy consumer sockets from src to dst.
636 *
637 * Return 0 on success or else a negative value.
638 */
639 int consumer_copy_sockets(struct consumer_output *dst,
640 struct consumer_output *src)
641 {
642 int ret = 0;
643 struct lttng_ht_iter iter;
644 struct consumer_socket *socket, *copy_sock;
645
646 assert(dst);
647 assert(src);
648
649 rcu_read_lock();
650 cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
651 /* Ignore socket that are already there. */
652 copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
653 if (copy_sock) {
654 continue;
655 }
656
657 /* Create new socket object. */
658 copy_sock = consumer_allocate_socket(socket->fd_ptr);
659 if (copy_sock == NULL) {
660 rcu_read_unlock();
661 ret = -ENOMEM;
662 goto error;
663 }
664
665 copy_sock->registered = socket->registered;
666 /*
667 * This is valid because this lock is shared accross all consumer
668 * object being the global lock of the consumer data structure of the
669 * session daemon.
670 */
671 copy_sock->lock = socket->lock;
672 consumer_add_socket(copy_sock, dst);
673 }
674 rcu_read_unlock();
675
676 error:
677 return ret;
678 }
679
680 /*
681 * Set network URI to the consumer output.
682 *
683 * Return 0 on success. Return 1 if the URI were equal. Else, negative value on
684 * error.
685 */
686 int consumer_set_network_uri(const struct ltt_session *session,
687 struct consumer_output *output,
688 struct lttng_uri *uri)
689 {
690 int ret;
691 struct lttng_uri *dst_uri = NULL;
692
693 /* Code flow error safety net. */
694 assert(output);
695 assert(uri);
696
697 switch (uri->stype) {
698 case LTTNG_STREAM_CONTROL:
699 dst_uri = &output->dst.net.control;
700 output->dst.net.control_isset = 1;
701 if (uri->port == 0) {
702 /* Assign default port. */
703 uri->port = DEFAULT_NETWORK_CONTROL_PORT;
704 } else {
705 if (output->dst.net.data_isset && uri->port ==
706 output->dst.net.data.port) {
707 ret = -LTTNG_ERR_INVALID;
708 goto error;
709 }
710 }
711 DBG3("Consumer control URI set with port %d", uri->port);
712 break;
713 case LTTNG_STREAM_DATA:
714 dst_uri = &output->dst.net.data;
715 output->dst.net.data_isset = 1;
716 if (uri->port == 0) {
717 /* Assign default port. */
718 uri->port = DEFAULT_NETWORK_DATA_PORT;
719 } else {
720 if (output->dst.net.control_isset && uri->port ==
721 output->dst.net.control.port) {
722 ret = -LTTNG_ERR_INVALID;
723 goto error;
724 }
725 }
726 DBG3("Consumer data URI set with port %d", uri->port);
727 break;
728 default:
729 ERR("Set network uri type unknown %d", uri->stype);
730 ret = -LTTNG_ERR_INVALID;
731 goto error;
732 }
733
734 ret = uri_compare(dst_uri, uri);
735 if (!ret) {
736 /* Same URI, don't touch it and return success. */
737 DBG3("URI network compare are the same");
738 goto equal;
739 }
740
741 /* URIs were not equal, replacing it. */
742 memcpy(dst_uri, uri, sizeof(struct lttng_uri));
743 output->type = CONSUMER_DST_NET;
744 if (dst_uri->stype != LTTNG_STREAM_CONTROL) {
745 /* Only the control uri needs to contain the path. */
746 goto end;
747 }
748
749 /*
750 * If the user has specified a subdir as part of the control
751 * URL, the session's base output directory is:
752 * /RELAYD_OUTPUT_PATH/HOSTNAME/USER_SPECIFIED_DIR
753 *
754 * Hence, the "base_dir" from which all stream files and
755 * session rotation chunks are created takes the form
756 * /HOSTNAME/USER_SPECIFIED_DIR
757 *
758 * If the user has not specified an output directory as part of
759 * the control URL, the base output directory has the form:
760 * /RELAYD_OUTPUT_PATH/HOSTNAME/SESSION_NAME-CREATION_TIME
761 *
762 * Hence, the "base_dir" from which all stream files and
763 * session rotation chunks are created takes the form
764 * /HOSTNAME/SESSION_NAME-CREATION_TIME
765 *
766 * Note that automatically generated session names already
767 * contain the session's creation time. In that case, the
768 * creation time is omitted to prevent it from being duplicated
769 * in the final directory hierarchy.
770 */
771 if (*uri->subdir) {
772 if (strstr(uri->subdir, "../")) {
773 ERR("Network URI subdirs are not allowed to walk up the path hierarchy");
774 ret = -LTTNG_ERR_INVALID;
775 goto error;
776 }
777 ret = snprintf(output->dst.net.base_dir,
778 sizeof(output->dst.net.base_dir),
779 "/%s/%s/", session->hostname, uri->subdir);
780 } else {
781 if (session->has_auto_generated_name) {
782 ret = snprintf(output->dst.net.base_dir,
783 sizeof(output->dst.net.base_dir),
784 "/%s/%s/", session->hostname,
785 session->name);
786 } else {
787 char session_creation_datetime[16];
788 size_t strftime_ret;
789 struct tm *timeinfo;
790
791 timeinfo = localtime(&session->creation_time);
792 if (!timeinfo) {
793 ret = -LTTNG_ERR_FATAL;
794 goto error;
795 }
796 strftime_ret = strftime(session_creation_datetime,
797 sizeof(session_creation_datetime),
798 "%Y%m%d-%H%M%S", timeinfo);
799 if (strftime_ret == 0) {
800 ERR("Failed to format session creation timestamp while setting network URI");
801 ret = -LTTNG_ERR_FATAL;
802 goto error;
803 }
804 ret = snprintf(output->dst.net.base_dir,
805 sizeof(output->dst.net.base_dir),
806 "/%s/%s-%s/", session->hostname,
807 session->name,
808 session_creation_datetime);
809 }
810 }
811 if (ret >= sizeof(output->dst.net.base_dir)) {
812 ret = -LTTNG_ERR_INVALID;
813 ERR("Truncation occurred while setting network output base directory");
814 goto error;
815 } else if (ret == -1) {
816 ret = -LTTNG_ERR_INVALID;
817 PERROR("Error occurred while setting network output base directory");
818 goto error;
819 }
820
821 DBG3("Consumer set network uri base_dir path %s",
822 output->dst.net.base_dir);
823
824 end:
825 return 0;
826 equal:
827 return 1;
828 error:
829 return ret;
830 }
831
832 /*
833 * Send file descriptor to consumer via sock.
834 *
835 * The consumer socket lock must be held by the caller.
836 */
837 int consumer_send_fds(struct consumer_socket *sock, const int *fds,
838 size_t nb_fd)
839 {
840 int ret;
841
842 assert(fds);
843 assert(sock);
844 assert(nb_fd > 0);
845 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
846
847 ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
848 if (ret < 0) {
849 /* The above call will print a PERROR on error. */
850 DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
851 goto error;
852 }
853
854 ret = consumer_recv_status_reply(sock);
855 error:
856 return ret;
857 }
858
859 /*
860 * Consumer send communication message structure to consumer.
861 *
862 * The consumer socket lock must be held by the caller.
863 */
864 int consumer_send_msg(struct consumer_socket *sock,
865 const struct lttcomm_consumer_msg *msg)
866 {
867 int ret;
868
869 assert(msg);
870 assert(sock);
871 assert(pthread_mutex_trylock(sock->lock) == EBUSY);
872
873 ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
874 if (ret < 0) {
875 goto error;
876 }
877
878 ret = consumer_recv_status_reply(sock);
879
880 error:
881 return ret;
882 }
883
884 /*
885 * Consumer send channel communication message structure to consumer.
886 *
887 * The consumer socket lock must be held by the caller.
888 */
889 int consumer_send_channel(struct consumer_socket *sock,
890 struct lttcomm_consumer_msg *msg)
891 {
892 int ret;
893
894 assert(msg);
895 assert(sock);
896
897 ret = consumer_send_msg(sock, msg);
898 if (ret < 0) {
899 goto error;
900 }
901
902 error:
903 return ret;
904 }
905
906 /*
907 * Populate the given consumer msg structure with the ask_channel command
908 * information.
909 */
910 void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg,
911 uint64_t subbuf_size,
912 uint64_t num_subbuf,
913 int overwrite,
914 unsigned int switch_timer_interval,
915 unsigned int read_timer_interval,
916 unsigned int live_timer_interval,
917 bool is_in_live_session,
918 unsigned int monitor_timer_interval,
919 int output,
920 int type,
921 uint64_t session_id,
922 const char *pathname,
923 const char *name,
924 uint64_t relayd_id,
925 uint64_t key,
926 unsigned char *uuid,
927 uint32_t chan_id,
928 uint64_t tracefile_size,
929 uint64_t tracefile_count,
930 uint64_t session_id_per_pid,
931 unsigned int monitor,
932 uint32_t ust_app_uid,
933 int64_t blocking_timeout,
934 const char *root_shm_path,
935 const char *shm_path,
936 struct lttng_trace_chunk *trace_chunk,
937 const struct lttng_credentials *buffer_credentials)
938 {
939 assert(msg);
940
941 /* Zeroed structure */
942 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
943 msg->u.ask_channel.buffer_credentials.uid = UINT32_MAX;
944 msg->u.ask_channel.buffer_credentials.gid = UINT32_MAX;
945
946 if (trace_chunk) {
947 uint64_t chunk_id;
948 enum lttng_trace_chunk_status chunk_status;
949
950 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
951 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
952 LTTNG_OPTIONAL_SET(&msg->u.ask_channel.chunk_id, chunk_id);
953 }
954 msg->u.ask_channel.buffer_credentials.uid =
955 lttng_credentials_get_uid(buffer_credentials);
956 msg->u.ask_channel.buffer_credentials.gid =
957 lttng_credentials_get_gid(buffer_credentials);
958
959 msg->cmd_type = LTTNG_CONSUMER_ASK_CHANNEL_CREATION;
960 msg->u.ask_channel.subbuf_size = subbuf_size;
961 msg->u.ask_channel.num_subbuf = num_subbuf ;
962 msg->u.ask_channel.overwrite = overwrite;
963 msg->u.ask_channel.switch_timer_interval = switch_timer_interval;
964 msg->u.ask_channel.read_timer_interval = read_timer_interval;
965 msg->u.ask_channel.live_timer_interval = live_timer_interval;
966 msg->u.ask_channel.is_live = is_in_live_session;
967 msg->u.ask_channel.monitor_timer_interval = monitor_timer_interval;
968 msg->u.ask_channel.output = output;
969 msg->u.ask_channel.type = type;
970 msg->u.ask_channel.session_id = session_id;
971 msg->u.ask_channel.session_id_per_pid = session_id_per_pid;
972 msg->u.ask_channel.relayd_id = relayd_id;
973 msg->u.ask_channel.key = key;
974 msg->u.ask_channel.chan_id = chan_id;
975 msg->u.ask_channel.tracefile_size = tracefile_size;
976 msg->u.ask_channel.tracefile_count = tracefile_count;
977 msg->u.ask_channel.monitor = monitor;
978 msg->u.ask_channel.ust_app_uid = ust_app_uid;
979 msg->u.ask_channel.blocking_timeout = blocking_timeout;
980
981 memcpy(msg->u.ask_channel.uuid, uuid, sizeof(msg->u.ask_channel.uuid));
982
983 if (pathname) {
984 strncpy(msg->u.ask_channel.pathname, pathname,
985 sizeof(msg->u.ask_channel.pathname));
986 msg->u.ask_channel.pathname[sizeof(msg->u.ask_channel.pathname)-1] = '\0';
987 }
988
989 strncpy(msg->u.ask_channel.name, name, sizeof(msg->u.ask_channel.name));
990 msg->u.ask_channel.name[sizeof(msg->u.ask_channel.name) - 1] = '\0';
991
992 if (root_shm_path) {
993 strncpy(msg->u.ask_channel.root_shm_path, root_shm_path,
994 sizeof(msg->u.ask_channel.root_shm_path));
995 msg->u.ask_channel.root_shm_path[sizeof(msg->u.ask_channel.root_shm_path) - 1] = '\0';
996 }
997 if (shm_path) {
998 strncpy(msg->u.ask_channel.shm_path, shm_path,
999 sizeof(msg->u.ask_channel.shm_path));
1000 msg->u.ask_channel.shm_path[sizeof(msg->u.ask_channel.shm_path) - 1] = '\0';
1001 }
1002 }
1003
1004 /*
1005 * Init channel communication message structure.
1006 */
1007 void consumer_init_add_channel_comm_msg(struct lttcomm_consumer_msg *msg,
1008 uint64_t channel_key,
1009 uint64_t session_id,
1010 const char *pathname,
1011 uid_t uid,
1012 gid_t gid,
1013 uint64_t relayd_id,
1014 const char *name,
1015 unsigned int nb_init_streams,
1016 enum lttng_event_output output,
1017 int type,
1018 uint64_t tracefile_size,
1019 uint64_t tracefile_count,
1020 unsigned int monitor,
1021 unsigned int live_timer_interval,
1022 bool is_in_live_session,
1023 unsigned int monitor_timer_interval,
1024 struct lttng_trace_chunk *trace_chunk)
1025 {
1026 assert(msg);
1027
1028 /* Zeroed structure */
1029 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1030
1031 if (trace_chunk) {
1032 uint64_t chunk_id;
1033 enum lttng_trace_chunk_status chunk_status;
1034
1035 chunk_status = lttng_trace_chunk_get_id(trace_chunk, &chunk_id);
1036 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
1037 LTTNG_OPTIONAL_SET(&msg->u.channel.chunk_id, chunk_id);
1038 }
1039
1040 /* Send channel */
1041 msg->cmd_type = LTTNG_CONSUMER_ADD_CHANNEL;
1042 msg->u.channel.channel_key = channel_key;
1043 msg->u.channel.session_id = session_id;
1044 msg->u.channel.relayd_id = relayd_id;
1045 msg->u.channel.nb_init_streams = nb_init_streams;
1046 msg->u.channel.output = output;
1047 msg->u.channel.type = type;
1048 msg->u.channel.tracefile_size = tracefile_size;
1049 msg->u.channel.tracefile_count = tracefile_count;
1050 msg->u.channel.monitor = monitor;
1051 msg->u.channel.live_timer_interval = live_timer_interval;
1052 msg->u.channel.is_live = is_in_live_session;
1053 msg->u.channel.monitor_timer_interval = monitor_timer_interval;
1054
1055 strncpy(msg->u.channel.pathname, pathname,
1056 sizeof(msg->u.channel.pathname));
1057 msg->u.channel.pathname[sizeof(msg->u.channel.pathname) - 1] = '\0';
1058
1059 strncpy(msg->u.channel.name, name, sizeof(msg->u.channel.name));
1060 msg->u.channel.name[sizeof(msg->u.channel.name) - 1] = '\0';
1061 }
1062
1063 /*
1064 * Init stream communication message structure.
1065 */
1066 void consumer_init_add_stream_comm_msg(struct lttcomm_consumer_msg *msg,
1067 uint64_t channel_key,
1068 uint64_t stream_key,
1069 int32_t cpu)
1070 {
1071 assert(msg);
1072
1073 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1074
1075 msg->cmd_type = LTTNG_CONSUMER_ADD_STREAM;
1076 msg->u.stream.channel_key = channel_key;
1077 msg->u.stream.stream_key = stream_key;
1078 msg->u.stream.cpu = cpu;
1079 }
1080
1081 void consumer_init_streams_sent_comm_msg(struct lttcomm_consumer_msg *msg,
1082 enum lttng_consumer_command cmd,
1083 uint64_t channel_key, uint64_t net_seq_idx)
1084 {
1085 assert(msg);
1086
1087 memset(msg, 0, sizeof(struct lttcomm_consumer_msg));
1088
1089 msg->cmd_type = cmd;
1090 msg->u.sent_streams.channel_key = channel_key;
1091 msg->u.sent_streams.net_seq_idx = net_seq_idx;
1092 }
1093
1094 /*
1095 * Send stream communication structure to the consumer.
1096 */
1097 int consumer_send_stream(struct consumer_socket *sock,
1098 struct consumer_output *dst, struct lttcomm_consumer_msg *msg,
1099 const int *fds, size_t nb_fd)
1100 {
1101 int ret;
1102
1103 assert(msg);
1104 assert(dst);
1105 assert(sock);
1106 assert(fds);
1107
1108 ret = consumer_send_msg(sock, msg);
1109 if (ret < 0) {
1110 goto error;
1111 }
1112
1113 ret = consumer_send_fds(sock, fds, nb_fd);
1114 if (ret < 0) {
1115 goto error;
1116 }
1117
1118 error:
1119 return ret;
1120 }
1121
1122 /*
1123 * Send relayd socket to consumer associated with a session name.
1124 *
1125 * The consumer socket lock must be held by the caller.
1126 *
1127 * On success return positive value. On error, negative value.
1128 */
1129 int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
1130 struct lttcomm_relayd_sock *rsock, struct consumer_output *consumer,
1131 enum lttng_stream_type type, uint64_t session_id,
1132 const char *session_name, const char *hostname,
1133 const char *base_path, int session_live_timer,
1134 const uint64_t *current_chunk_id, time_t session_creation_time,
1135 bool session_name_contains_creation_time)
1136 {
1137 int ret;
1138 struct lttcomm_consumer_msg msg;
1139
1140 /* Code flow error. Safety net. */
1141 assert(rsock);
1142 assert(consumer);
1143 assert(consumer_sock);
1144
1145 memset(&msg, 0, sizeof(msg));
1146 /* Bail out if consumer is disabled */
1147 if (!consumer->enabled) {
1148 ret = LTTNG_OK;
1149 goto error;
1150 }
1151
1152 if (type == LTTNG_STREAM_CONTROL) {
1153 char output_path[LTTNG_PATH_MAX] = {};
1154 uint64_t relayd_session_id;
1155
1156 ret = relayd_create_session(rsock, &relayd_session_id,
1157 session_name, hostname, base_path,
1158 session_live_timer, consumer->snapshot,
1159 session_id, the_sessiond_uuid, current_chunk_id,
1160 session_creation_time,
1161 session_name_contains_creation_time,
1162 output_path);
1163 if (ret < 0) {
1164 /* Close the control socket. */
1165 (void) relayd_close(rsock);
1166 goto error;
1167 }
1168 msg.u.relayd_sock.relayd_session_id = relayd_session_id;
1169 DBG("Created session on relay, output path reply: %s",
1170 output_path);
1171 }
1172
1173 msg.cmd_type = LTTNG_CONSUMER_ADD_RELAYD_SOCKET;
1174 /*
1175 * Assign network consumer output index using the temporary consumer since
1176 * this call should only be made from within a set_consumer_uri() function
1177 * call in the session daemon.
1178 */
1179 msg.u.relayd_sock.net_index = consumer->net_seq_index;
1180 msg.u.relayd_sock.type = type;
1181 msg.u.relayd_sock.session_id = session_id;
1182 memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
1183
1184 DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
1185 ret = consumer_send_msg(consumer_sock, &msg);
1186 if (ret < 0) {
1187 goto error;
1188 }
1189
1190 DBG3("Sending relayd socket file descriptor to consumer");
1191 ret = consumer_send_fds(consumer_sock, ALIGNED_CONST_PTR(rsock->sock.fd), 1);
1192 if (ret < 0) {
1193 goto error;
1194 }
1195
1196 DBG2("Consumer relayd socket sent");
1197
1198 error:
1199 return ret;
1200 }
1201
1202 static
1203 int consumer_send_pipe(struct consumer_socket *consumer_sock,
1204 enum lttng_consumer_command cmd, int pipe)
1205 {
1206 int ret;
1207 struct lttcomm_consumer_msg msg;
1208 const char *pipe_name;
1209 const char *command_name;
1210
1211 switch (cmd) {
1212 case LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE:
1213 pipe_name = "channel monitor";
1214 command_name = "SET_CHANNEL_MONITOR_PIPE";
1215 break;
1216 default:
1217 ERR("Unexpected command received in %s (cmd = %d)", __func__,
1218 (int) cmd);
1219 abort();
1220 }
1221
1222 /* Code flow error. Safety net. */
1223
1224 memset(&msg, 0, sizeof(msg));
1225 msg.cmd_type = cmd;
1226
1227 pthread_mutex_lock(consumer_sock->lock);
1228 DBG3("Sending %s command to consumer", command_name);
1229 ret = consumer_send_msg(consumer_sock, &msg);
1230 if (ret < 0) {
1231 goto error;
1232 }
1233
1234 DBG3("Sending %s pipe %d to consumer on socket %d",
1235 pipe_name,
1236 pipe, *consumer_sock->fd_ptr);
1237 ret = consumer_send_fds(consumer_sock, &pipe, 1);
1238 if (ret < 0) {
1239 goto error;
1240 }
1241
1242 DBG2("%s pipe successfully sent", pipe_name);
1243 error:
1244 pthread_mutex_unlock(consumer_sock->lock);
1245 return ret;
1246 }
1247
1248 int consumer_send_channel_monitor_pipe(struct consumer_socket *consumer_sock,
1249 int pipe)
1250 {
1251 return consumer_send_pipe(consumer_sock,
1252 LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, pipe);
1253 }
1254
1255 /*
1256 * Ask the consumer if the data is pending for the specific session id.
1257 * Returns 1 if data is pending, 0 otherwise, or < 0 on error.
1258 */
1259 int consumer_is_data_pending(uint64_t session_id,
1260 struct consumer_output *consumer)
1261 {
1262 int ret;
1263 int32_t ret_code = 0; /* Default is that the data is NOT pending */
1264 struct consumer_socket *socket;
1265 struct lttng_ht_iter iter;
1266 struct lttcomm_consumer_msg msg;
1267
1268 assert(consumer);
1269
1270 DBG3("Consumer data pending for id %" PRIu64, session_id);
1271
1272 memset(&msg, 0, sizeof(msg));
1273 msg.cmd_type = LTTNG_CONSUMER_DATA_PENDING;
1274 msg.u.data_pending.session_id = session_id;
1275
1276 /* Send command for each consumer */
1277 rcu_read_lock();
1278 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1279 node.node) {
1280 pthread_mutex_lock(socket->lock);
1281 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1282 if (ret < 0) {
1283 pthread_mutex_unlock(socket->lock);
1284 goto error_unlock;
1285 }
1286
1287 /*
1288 * No need for a recv reply status because the answer to the command is
1289 * the reply status message.
1290 */
1291
1292 ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
1293 if (ret < 0) {
1294 pthread_mutex_unlock(socket->lock);
1295 goto error_unlock;
1296 }
1297 pthread_mutex_unlock(socket->lock);
1298
1299 if (ret_code == 1) {
1300 break;
1301 }
1302 }
1303 rcu_read_unlock();
1304
1305 DBG("Consumer data is %s pending for session id %" PRIu64,
1306 ret_code == 1 ? "" : "NOT", session_id);
1307 return ret_code;
1308
1309 error_unlock:
1310 rcu_read_unlock();
1311 return -1;
1312 }
1313
1314 /*
1315 * Send a flush command to consumer using the given channel key.
1316 *
1317 * Return 0 on success else a negative value.
1318 */
1319 int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
1320 {
1321 int ret;
1322 struct lttcomm_consumer_msg msg;
1323
1324 assert(socket);
1325
1326 DBG2("Consumer flush channel key %" PRIu64, key);
1327
1328 memset(&msg, 0, sizeof(msg));
1329 msg.cmd_type = LTTNG_CONSUMER_FLUSH_CHANNEL;
1330 msg.u.flush_channel.key = key;
1331
1332 pthread_mutex_lock(socket->lock);
1333 health_code_update();
1334
1335 ret = consumer_send_msg(socket, &msg);
1336 if (ret < 0) {
1337 goto end;
1338 }
1339
1340 end:
1341 health_code_update();
1342 pthread_mutex_unlock(socket->lock);
1343 return ret;
1344 }
1345
1346 /*
1347 * Send a clear quiescent command to consumer using the given channel key.
1348 *
1349 * Return 0 on success else a negative value.
1350 */
1351 int consumer_clear_quiescent_channel(struct consumer_socket *socket, uint64_t key)
1352 {
1353 int ret;
1354 struct lttcomm_consumer_msg msg;
1355
1356 assert(socket);
1357
1358 DBG2("Consumer clear quiescent channel key %" PRIu64, key);
1359
1360 memset(&msg, 0, sizeof(msg));
1361 msg.cmd_type = LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL;
1362 msg.u.clear_quiescent_channel.key = key;
1363
1364 pthread_mutex_lock(socket->lock);
1365 health_code_update();
1366
1367 ret = consumer_send_msg(socket, &msg);
1368 if (ret < 0) {
1369 goto end;
1370 }
1371
1372 end:
1373 health_code_update();
1374 pthread_mutex_unlock(socket->lock);
1375 return ret;
1376 }
1377
1378 /*
1379 * Send a close metadata command to consumer using the given channel key.
1380 * Called with registry lock held.
1381 *
1382 * Return 0 on success else a negative value.
1383 */
1384 int consumer_close_metadata(struct consumer_socket *socket,
1385 uint64_t metadata_key)
1386 {
1387 int ret;
1388 struct lttcomm_consumer_msg msg;
1389
1390 assert(socket);
1391
1392 DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
1393
1394 memset(&msg, 0, sizeof(msg));
1395 msg.cmd_type = LTTNG_CONSUMER_CLOSE_METADATA;
1396 msg.u.close_metadata.key = metadata_key;
1397
1398 pthread_mutex_lock(socket->lock);
1399 health_code_update();
1400
1401 ret = consumer_send_msg(socket, &msg);
1402 if (ret < 0) {
1403 goto end;
1404 }
1405
1406 end:
1407 health_code_update();
1408 pthread_mutex_unlock(socket->lock);
1409 return ret;
1410 }
1411
1412 /*
1413 * Send a setup metdata command to consumer using the given channel key.
1414 *
1415 * Return 0 on success else a negative value.
1416 */
1417 int consumer_setup_metadata(struct consumer_socket *socket,
1418 uint64_t metadata_key)
1419 {
1420 int ret;
1421 struct lttcomm_consumer_msg msg;
1422
1423 assert(socket);
1424
1425 DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
1426
1427 memset(&msg, 0, sizeof(msg));
1428 msg.cmd_type = LTTNG_CONSUMER_SETUP_METADATA;
1429 msg.u.setup_metadata.key = metadata_key;
1430
1431 pthread_mutex_lock(socket->lock);
1432 health_code_update();
1433
1434 ret = consumer_send_msg(socket, &msg);
1435 if (ret < 0) {
1436 goto end;
1437 }
1438
1439 end:
1440 health_code_update();
1441 pthread_mutex_unlock(socket->lock);
1442 return ret;
1443 }
1444
1445 /*
1446 * Send metadata string to consumer.
1447 * RCU read-side lock must be held to guarantee existence of socket.
1448 *
1449 * Return 0 on success else a negative value.
1450 */
1451 int consumer_push_metadata(struct consumer_socket *socket,
1452 uint64_t metadata_key, char *metadata_str, size_t len,
1453 size_t target_offset, uint64_t version)
1454 {
1455 int ret;
1456 struct lttcomm_consumer_msg msg;
1457
1458 assert(socket);
1459
1460 DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
1461
1462 pthread_mutex_lock(socket->lock);
1463
1464 memset(&msg, 0, sizeof(msg));
1465 msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
1466 msg.u.push_metadata.key = metadata_key;
1467 msg.u.push_metadata.target_offset = target_offset;
1468 msg.u.push_metadata.len = len;
1469 msg.u.push_metadata.version = version;
1470
1471 health_code_update();
1472 ret = consumer_send_msg(socket, &msg);
1473 if (ret < 0 || len == 0) {
1474 goto end;
1475 }
1476
1477 DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
1478 len);
1479
1480 ret = consumer_socket_send(socket, metadata_str, len);
1481 if (ret < 0) {
1482 goto end;
1483 }
1484
1485 health_code_update();
1486 ret = consumer_recv_status_reply(socket);
1487 if (ret < 0) {
1488 goto end;
1489 }
1490
1491 end:
1492 pthread_mutex_unlock(socket->lock);
1493 health_code_update();
1494 return ret;
1495 }
1496
1497 /*
1498 * Ask the consumer to snapshot a specific channel using the key.
1499 *
1500 * Returns LTTNG_OK on success or else an LTTng error code.
1501 */
1502 enum lttng_error_code consumer_snapshot_channel(struct consumer_socket *socket,
1503 uint64_t key, const struct consumer_output *output, int metadata,
1504 uid_t uid, gid_t gid, const char *channel_path, int wait,
1505 uint64_t nb_packets_per_stream)
1506 {
1507 int ret;
1508 enum lttng_error_code status = LTTNG_OK;
1509 struct lttcomm_consumer_msg msg;
1510
1511 assert(socket);
1512 assert(output);
1513
1514 DBG("Consumer snapshot channel key %" PRIu64, key);
1515
1516 memset(&msg, 0, sizeof(msg));
1517 msg.cmd_type = LTTNG_CONSUMER_SNAPSHOT_CHANNEL;
1518 msg.u.snapshot_channel.key = key;
1519 msg.u.snapshot_channel.nb_packets_per_stream = nb_packets_per_stream;
1520 msg.u.snapshot_channel.metadata = metadata;
1521
1522 if (output->type == CONSUMER_DST_NET) {
1523 msg.u.snapshot_channel.relayd_id =
1524 output->net_seq_index;
1525 msg.u.snapshot_channel.use_relayd = 1;
1526 } else {
1527 msg.u.snapshot_channel.relayd_id = (uint64_t) -1ULL;
1528 }
1529 ret = lttng_strncpy(msg.u.snapshot_channel.pathname,
1530 channel_path,
1531 sizeof(msg.u.snapshot_channel.pathname));
1532 if (ret < 0) {
1533 ERR("Snapshot path exceeds the maximal allowed length of %zu bytes (%zu bytes required) with path \"%s\"",
1534 sizeof(msg.u.snapshot_channel.pathname),
1535 strlen(channel_path),
1536 channel_path);
1537 status = LTTNG_ERR_SNAPSHOT_FAIL;
1538 goto error;
1539 }
1540
1541 health_code_update();
1542 pthread_mutex_lock(socket->lock);
1543 ret = consumer_send_msg(socket, &msg);
1544 pthread_mutex_unlock(socket->lock);
1545 if (ret < 0) {
1546 switch (-ret) {
1547 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1548 status = LTTNG_ERR_CHAN_NOT_FOUND;
1549 break;
1550 default:
1551 status = LTTNG_ERR_SNAPSHOT_FAIL;
1552 break;
1553 }
1554 goto error;
1555 }
1556
1557 error:
1558 health_code_update();
1559 return status;
1560 }
1561
1562 /*
1563 * Ask the consumer the number of discarded events for a channel.
1564 */
1565 int consumer_get_discarded_events(uint64_t session_id, uint64_t channel_key,
1566 struct consumer_output *consumer, uint64_t *discarded)
1567 {
1568 int ret;
1569 struct consumer_socket *socket;
1570 struct lttng_ht_iter iter;
1571 struct lttcomm_consumer_msg msg;
1572
1573 assert(consumer);
1574
1575 DBG3("Consumer discarded events id %" PRIu64, session_id);
1576
1577 memset(&msg, 0, sizeof(msg));
1578 msg.cmd_type = LTTNG_CONSUMER_DISCARDED_EVENTS;
1579 msg.u.discarded_events.session_id = session_id;
1580 msg.u.discarded_events.channel_key = channel_key;
1581
1582 *discarded = 0;
1583
1584 /* Send command for each consumer */
1585 rcu_read_lock();
1586 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1587 node.node) {
1588 uint64_t consumer_discarded = 0;
1589 pthread_mutex_lock(socket->lock);
1590 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1591 if (ret < 0) {
1592 pthread_mutex_unlock(socket->lock);
1593 goto end;
1594 }
1595
1596 /*
1597 * No need for a recv reply status because the answer to the
1598 * command is the reply status message.
1599 */
1600 ret = consumer_socket_recv(socket, &consumer_discarded,
1601 sizeof(consumer_discarded));
1602 if (ret < 0) {
1603 ERR("get discarded events");
1604 pthread_mutex_unlock(socket->lock);
1605 goto end;
1606 }
1607 pthread_mutex_unlock(socket->lock);
1608 *discarded += consumer_discarded;
1609 }
1610 ret = 0;
1611 DBG("Consumer discarded %" PRIu64 " events in session id %" PRIu64,
1612 *discarded, session_id);
1613
1614 end:
1615 rcu_read_unlock();
1616 return ret;
1617 }
1618
1619 /*
1620 * Ask the consumer the number of lost packets for a channel.
1621 */
1622 int consumer_get_lost_packets(uint64_t session_id, uint64_t channel_key,
1623 struct consumer_output *consumer, uint64_t *lost)
1624 {
1625 int ret;
1626 struct consumer_socket *socket;
1627 struct lttng_ht_iter iter;
1628 struct lttcomm_consumer_msg msg;
1629
1630 assert(consumer);
1631
1632 DBG3("Consumer lost packets id %" PRIu64, session_id);
1633
1634 memset(&msg, 0, sizeof(msg));
1635 msg.cmd_type = LTTNG_CONSUMER_LOST_PACKETS;
1636 msg.u.lost_packets.session_id = session_id;
1637 msg.u.lost_packets.channel_key = channel_key;
1638
1639 *lost = 0;
1640
1641 /* Send command for each consumer */
1642 rcu_read_lock();
1643 cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
1644 node.node) {
1645 uint64_t consumer_lost = 0;
1646 pthread_mutex_lock(socket->lock);
1647 ret = consumer_socket_send(socket, &msg, sizeof(msg));
1648 if (ret < 0) {
1649 pthread_mutex_unlock(socket->lock);
1650 goto end;
1651 }
1652
1653 /*
1654 * No need for a recv reply status because the answer to the
1655 * command is the reply status message.
1656 */
1657 ret = consumer_socket_recv(socket, &consumer_lost,
1658 sizeof(consumer_lost));
1659 if (ret < 0) {
1660 ERR("get lost packets");
1661 pthread_mutex_unlock(socket->lock);
1662 goto end;
1663 }
1664 pthread_mutex_unlock(socket->lock);
1665 *lost += consumer_lost;
1666 }
1667 ret = 0;
1668 DBG("Consumer lost %" PRIu64 " packets in session id %" PRIu64,
1669 *lost, session_id);
1670
1671 end:
1672 rcu_read_unlock();
1673 return ret;
1674 }
1675
1676 /*
1677 * Ask the consumer to rotate a channel.
1678 *
1679 * The new_chunk_id is the session->rotate_count that has been incremented
1680 * when the rotation started. On the relay, this allows to keep track in which
1681 * chunk each stream is currently writing to (for the rotate_pending operation).
1682 */
1683 int consumer_rotate_channel(struct consumer_socket *socket, uint64_t key,
1684 uid_t uid, gid_t gid, struct consumer_output *output,
1685 bool is_metadata_channel)
1686 {
1687 int ret;
1688 struct lttcomm_consumer_msg msg;
1689
1690 assert(socket);
1691
1692 DBG("Consumer rotate channel key %" PRIu64, key);
1693
1694 pthread_mutex_lock(socket->lock);
1695 memset(&msg, 0, sizeof(msg));
1696 msg.cmd_type = LTTNG_CONSUMER_ROTATE_CHANNEL;
1697 msg.u.rotate_channel.key = key;
1698 msg.u.rotate_channel.metadata = !!is_metadata_channel;
1699
1700 if (output->type == CONSUMER_DST_NET) {
1701 msg.u.rotate_channel.relayd_id = output->net_seq_index;
1702 } else {
1703 msg.u.rotate_channel.relayd_id = (uint64_t) -1ULL;
1704 }
1705
1706 health_code_update();
1707 ret = consumer_send_msg(socket, &msg);
1708 if (ret < 0) {
1709 switch (-ret) {
1710 case LTTCOMM_CONSUMERD_CHAN_NOT_FOUND:
1711 ret = -LTTNG_ERR_CHAN_NOT_FOUND;
1712 break;
1713 default:
1714 ret = -LTTNG_ERR_ROTATION_FAIL_CONSUMER;
1715 break;
1716 }
1717 goto error;
1718 }
1719 error:
1720 pthread_mutex_unlock(socket->lock);
1721 health_code_update();
1722 return ret;
1723 }
1724
1725 int consumer_open_channel_packets(struct consumer_socket *socket, uint64_t key)
1726 {
1727 int ret;
1728 const struct lttcomm_consumer_msg msg = {
1729 .cmd_type = LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS,
1730 .u.open_channel_packets.key = key,
1731 };
1732
1733 assert(socket);
1734
1735 DBG("Consumer open channel packets: channel key = %" PRIu64, key);
1736
1737 health_code_update();
1738
1739 pthread_mutex_lock(socket->lock);
1740 ret = consumer_send_msg(socket, &msg);
1741 pthread_mutex_unlock(socket->lock);
1742 if (ret < 0) {
1743 goto error_socket;
1744 }
1745
1746 error_socket:
1747 health_code_update();
1748 return ret;
1749 }
1750
1751 int consumer_clear_channel(struct consumer_socket *socket, uint64_t key)
1752 {
1753 int ret;
1754 struct lttcomm_consumer_msg msg;
1755
1756 assert(socket);
1757
1758 DBG("Consumer clear channel %" PRIu64, key);
1759
1760 memset(&msg, 0, sizeof(msg));
1761 msg.cmd_type = LTTNG_CONSUMER_CLEAR_CHANNEL;
1762 msg.u.clear_channel.key = key;
1763
1764 health_code_update();
1765
1766 pthread_mutex_lock(socket->lock);
1767 ret = consumer_send_msg(socket, &msg);
1768 if (ret < 0) {
1769 goto error_socket;
1770 }
1771
1772 error_socket:
1773 pthread_mutex_unlock(socket->lock);
1774
1775 health_code_update();
1776 return ret;
1777 }
1778
1779 int consumer_init(struct consumer_socket *socket,
1780 const lttng_uuid sessiond_uuid)
1781 {
1782 int ret;
1783 struct lttcomm_consumer_msg msg = {
1784 .cmd_type = LTTNG_CONSUMER_INIT,
1785 };
1786
1787 assert(socket);
1788
1789 DBG("Sending consumer initialization command");
1790 lttng_uuid_copy(msg.u.init.sessiond_uuid, sessiond_uuid);
1791
1792 health_code_update();
1793 ret = consumer_send_msg(socket, &msg);
1794 if (ret < 0) {
1795 goto error;
1796 }
1797
1798 error:
1799 health_code_update();
1800 return ret;
1801 }
1802
1803 /*
1804 * Ask the consumer to create a new chunk for a given session.
1805 *
1806 * Called with the consumer socket lock held.
1807 */
1808 int consumer_create_trace_chunk(struct consumer_socket *socket,
1809 uint64_t relayd_id, uint64_t session_id,
1810 struct lttng_trace_chunk *chunk,
1811 const char *domain_subdir)
1812 {
1813 int ret;
1814 enum lttng_trace_chunk_status chunk_status;
1815 struct lttng_credentials chunk_credentials;
1816 const struct lttng_directory_handle *chunk_directory_handle = NULL;
1817 struct lttng_directory_handle *domain_handle = NULL;
1818 int domain_dirfd;
1819 const char *chunk_name;
1820 bool chunk_name_overridden;
1821 uint64_t chunk_id;
1822 time_t creation_timestamp;
1823 char creation_timestamp_buffer[ISO8601_STR_LEN];
1824 const char *creation_timestamp_str = "(none)";
1825 const bool chunk_has_local_output = relayd_id == -1ULL;
1826 enum lttng_trace_chunk_status tc_status;
1827 struct lttcomm_consumer_msg msg = {
1828 .cmd_type = LTTNG_CONSUMER_CREATE_TRACE_CHUNK,
1829 .u.create_trace_chunk.session_id = session_id,
1830 };
1831
1832 assert(socket);
1833 assert(chunk);
1834
1835 if (relayd_id != -1ULL) {
1836 LTTNG_OPTIONAL_SET(&msg.u.create_trace_chunk.relayd_id,
1837 relayd_id);
1838 }
1839
1840 chunk_status = lttng_trace_chunk_get_name(chunk, &chunk_name,
1841 &chunk_name_overridden);
1842 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK &&
1843 chunk_status != LTTNG_TRACE_CHUNK_STATUS_NONE) {
1844 ERR("Failed to get name of trace chunk");
1845 ret = -LTTNG_ERR_FATAL;
1846 goto error;
1847 }
1848 if (chunk_name_overridden) {
1849 ret = lttng_strncpy(msg.u.create_trace_chunk.override_name,
1850 chunk_name,
1851 sizeof(msg.u.create_trace_chunk.override_name));
1852 if (ret) {
1853 ERR("Trace chunk name \"%s\" exceeds the maximal length allowed by the consumer protocol",
1854 chunk_name);
1855 ret = -LTTNG_ERR_FATAL;
1856 goto error;
1857 }
1858 }
1859
1860 chunk_status = lttng_trace_chunk_get_creation_timestamp(chunk,
1861 &creation_timestamp);
1862 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1863 ret = -LTTNG_ERR_FATAL;
1864 goto error;
1865 }
1866 msg.u.create_trace_chunk.creation_timestamp =
1867 (uint64_t) creation_timestamp;
1868 /* Only used for logging purposes. */
1869 ret = time_to_iso8601_str(creation_timestamp,
1870 creation_timestamp_buffer,
1871 sizeof(creation_timestamp_buffer));
1872 creation_timestamp_str = !ret ? creation_timestamp_buffer :
1873 "(formatting error)";
1874
1875 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
1876 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1877 /*
1878 * Anonymous trace chunks should never be transmitted
1879 * to remote peers (consumerd and relayd). They are used
1880 * internally for backward-compatibility purposes.
1881 */
1882 ret = -LTTNG_ERR_FATAL;
1883 goto error;
1884 }
1885 msg.u.create_trace_chunk.chunk_id = chunk_id;
1886
1887 if (chunk_has_local_output) {
1888 chunk_status = lttng_trace_chunk_borrow_chunk_directory_handle(
1889 chunk, &chunk_directory_handle);
1890 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1891 ret = -LTTNG_ERR_FATAL;
1892 goto error;
1893 }
1894 chunk_status = lttng_trace_chunk_get_credentials(
1895 chunk, &chunk_credentials);
1896 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1897 /*
1898 * Not associating credentials to a sessiond chunk is a
1899 * fatal internal error.
1900 */
1901 ret = -LTTNG_ERR_FATAL;
1902 goto error;
1903 }
1904 tc_status = lttng_trace_chunk_create_subdirectory(
1905 chunk, domain_subdir);
1906 if (tc_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
1907 PERROR("Failed to create chunk domain output directory \"%s\"",
1908 domain_subdir);
1909 ret = -LTTNG_ERR_FATAL;
1910 goto error;
1911 }
1912 domain_handle = lttng_directory_handle_create_from_handle(
1913 domain_subdir,
1914 chunk_directory_handle);
1915 if (!domain_handle) {
1916 ret = -LTTNG_ERR_FATAL;
1917 goto error;
1918 }
1919
1920 /*
1921 * This will only compile on platforms that support
1922 * dirfd (POSIX.2008). This is fine as the session daemon
1923 * is only built for such platforms.
1924 *
1925 * The ownership of the chunk directory handle's is maintained
1926 * by the trace chunk.
1927 */
1928 domain_dirfd = lttng_directory_handle_get_dirfd(
1929 domain_handle);
1930 assert(domain_dirfd >= 0);
1931
1932 msg.u.create_trace_chunk.credentials.value.uid =
1933 lttng_credentials_get_uid(&chunk_credentials);
1934 msg.u.create_trace_chunk.credentials.value.gid =
1935 lttng_credentials_get_gid(&chunk_credentials);
1936 msg.u.create_trace_chunk.credentials.is_set = 1;
1937 }
1938
1939 DBG("Sending consumer create trace chunk command: relayd_id = %" PRId64
1940 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
1941 ", creation_timestamp = %s",
1942 relayd_id, session_id, chunk_id,
1943 creation_timestamp_str);
1944 health_code_update();
1945 ret = consumer_send_msg(socket, &msg);
1946 health_code_update();
1947 if (ret < 0) {
1948 ERR("Trace chunk creation error on consumer");
1949 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1950 goto error;
1951 }
1952
1953 if (chunk_has_local_output) {
1954 DBG("Sending trace chunk domain directory fd to consumer");
1955 health_code_update();
1956 ret = consumer_send_fds(socket, &domain_dirfd, 1);
1957 health_code_update();
1958 if (ret < 0) {
1959 ERR("Trace chunk creation error on consumer");
1960 ret = -LTTNG_ERR_CREATE_TRACE_CHUNK_FAIL_CONSUMER;
1961 goto error;
1962 }
1963 }
1964 error:
1965 lttng_directory_handle_put(domain_handle);
1966 return ret;
1967 }
1968
1969 /*
1970 * Ask the consumer to close a trace chunk for a given session.
1971 *
1972 * Called with the consumer socket lock held.
1973 */
1974 int consumer_close_trace_chunk(struct consumer_socket *socket,
1975 uint64_t relayd_id, uint64_t session_id,
1976 struct lttng_trace_chunk *chunk,
1977 char *closed_trace_chunk_path)
1978 {
1979 int ret;
1980 enum lttng_trace_chunk_status chunk_status;
1981 struct lttcomm_consumer_msg msg = {
1982 .cmd_type = LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
1983 .u.close_trace_chunk.session_id = session_id,
1984 };
1985 struct lttcomm_consumer_close_trace_chunk_reply reply;
1986 uint64_t chunk_id;
1987 time_t close_timestamp;
1988 enum lttng_trace_chunk_command_type close_command;
1989 const char *close_command_name = "none";
1990 struct lttng_dynamic_buffer path_reception_buffer;
1991
1992 assert(socket);
1993 lttng_dynamic_buffer_init(&path_reception_buffer);
1994
1995 if (relayd_id != -1ULL) {
1996 LTTNG_OPTIONAL_SET(
1997 &msg.u.close_trace_chunk.relayd_id, relayd_id);
1998 }
1999
2000 chunk_status = lttng_trace_chunk_get_close_command(
2001 chunk, &close_command);
2002 switch (chunk_status) {
2003 case LTTNG_TRACE_CHUNK_STATUS_OK:
2004 LTTNG_OPTIONAL_SET(&msg.u.close_trace_chunk.close_command,
2005 (uint32_t) close_command);
2006 break;
2007 case LTTNG_TRACE_CHUNK_STATUS_NONE:
2008 break;
2009 default:
2010 ERR("Failed to get trace chunk close command");
2011 ret = -1;
2012 goto error;
2013 }
2014
2015 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2016 /*
2017 * Anonymous trace chunks should never be transmitted to remote peers
2018 * (consumerd and relayd). They are used internally for
2019 * backward-compatibility purposes.
2020 */
2021 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2022 msg.u.close_trace_chunk.chunk_id = chunk_id;
2023
2024 chunk_status = lttng_trace_chunk_get_close_timestamp(chunk,
2025 &close_timestamp);
2026 /*
2027 * A trace chunk should be closed locally before being closed remotely.
2028 * Otherwise, the close timestamp would never be transmitted to the
2029 * peers.
2030 */
2031 assert(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
2032 msg.u.close_trace_chunk.close_timestamp = (uint64_t) close_timestamp;
2033
2034 if (msg.u.close_trace_chunk.close_command.is_set) {
2035 close_command_name = lttng_trace_chunk_command_type_get_name(
2036 close_command);
2037 }
2038 DBG("Sending consumer close trace chunk command: relayd_id = %" PRId64
2039 ", session_id = %" PRIu64 ", chunk_id = %" PRIu64
2040 ", close command = \"%s\"",
2041 relayd_id, session_id, chunk_id, close_command_name);
2042
2043 health_code_update();
2044 ret = consumer_socket_send(socket, &msg, sizeof(struct lttcomm_consumer_msg));
2045 if (ret < 0) {
2046 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2047 goto error;
2048 }
2049 ret = consumer_socket_recv(socket, &reply, sizeof(reply));
2050 if (ret < 0) {
2051 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2052 goto error;
2053 }
2054 if (reply.path_length >= LTTNG_PATH_MAX) {
2055 ERR("Invalid path returned by relay daemon: %" PRIu32 "bytes exceeds maximal allowed length of %d bytes",
2056 reply.path_length, LTTNG_PATH_MAX);
2057 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2058 goto error;
2059 }
2060 ret = lttng_dynamic_buffer_set_size(&path_reception_buffer,
2061 reply.path_length);
2062 if (ret) {
2063 ERR("Failed to allocate reception buffer of path returned by the \"close trace chunk\" command");
2064 ret = -LTTNG_ERR_NOMEM;
2065 goto error;
2066 }
2067 ret = consumer_socket_recv(socket, path_reception_buffer.data,
2068 path_reception_buffer.size);
2069 if (ret < 0) {
2070 ERR("Communication error while receiving path of closed trace chunk");
2071 ret = -LTTNG_ERR_CLOSE_TRACE_CHUNK_FAIL_CONSUMER;
2072 goto error;
2073 }
2074 if (path_reception_buffer.data[path_reception_buffer.size - 1] != '\0') {
2075 ERR("Invalid path returned by relay daemon: not null-terminated");
2076 ret = -LTTNG_ERR_INVALID_PROTOCOL;
2077 goto error;
2078 }
2079 if (closed_trace_chunk_path) {
2080 /*
2081 * closed_trace_chunk_path is assumed to have a length >=
2082 * LTTNG_PATH_MAX
2083 */
2084 memcpy(closed_trace_chunk_path, path_reception_buffer.data,
2085 path_reception_buffer.size);
2086 }
2087 error:
2088 lttng_dynamic_buffer_reset(&path_reception_buffer);
2089 health_code_update();
2090 return ret;
2091 }
2092
2093 /*
2094 * Ask the consumer if a trace chunk exists.
2095 *
2096 * Called with the consumer socket lock held.
2097 * Returns 0 on success, or a negative value on error.
2098 */
2099 int consumer_trace_chunk_exists(struct consumer_socket *socket,
2100 uint64_t relayd_id, uint64_t session_id,
2101 struct lttng_trace_chunk *chunk,
2102 enum consumer_trace_chunk_exists_status *result)
2103 {
2104 int ret;
2105 enum lttng_trace_chunk_status chunk_status;
2106 struct lttcomm_consumer_msg msg = {
2107 .cmd_type = LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,
2108 .u.trace_chunk_exists.session_id = session_id,
2109 };
2110 uint64_t chunk_id;
2111 const char *consumer_reply_str;
2112
2113 assert(socket);
2114
2115 if (relayd_id != -1ULL) {
2116 LTTNG_OPTIONAL_SET(&msg.u.trace_chunk_exists.relayd_id,
2117 relayd_id);
2118 }
2119
2120 chunk_status = lttng_trace_chunk_get_id(chunk, &chunk_id);
2121 if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
2122 /*
2123 * Anonymous trace chunks should never be transmitted
2124 * to remote peers (consumerd and relayd). They are used
2125 * internally for backward-compatibility purposes.
2126 */
2127 ret = -LTTNG_ERR_FATAL;
2128 goto error;
2129 }
2130 msg.u.trace_chunk_exists.chunk_id = chunk_id;
2131
2132 DBG("Sending consumer trace chunk exists command: relayd_id = %" PRId64
2133 ", session_id = %" PRIu64
2134 ", chunk_id = %" PRIu64, relayd_id, session_id, chunk_id);
2135
2136 health_code_update();
2137 ret = consumer_send_msg(socket, &msg);
2138 switch (-ret) {
2139 case LTTCOMM_CONSUMERD_UNKNOWN_TRACE_CHUNK:
2140 consumer_reply_str = "unknown trace chunk";
2141 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK;
2142 break;
2143 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_LOCAL:
2144 consumer_reply_str = "trace chunk exists locally";
2145 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_LOCAL;
2146 break;
2147 case LTTCOMM_CONSUMERD_TRACE_CHUNK_EXISTS_REMOTE:
2148 consumer_reply_str = "trace chunk exists on remote peer";
2149 *result = CONSUMER_TRACE_CHUNK_EXISTS_STATUS_EXISTS_REMOTE;
2150 break;
2151 default:
2152 ERR("Consumer returned an error from TRACE_CHUNK_EXISTS command");
2153 ret = -1;
2154 goto error;
2155 }
2156 DBG("Consumer reply to TRACE_CHUNK_EXISTS command: %s",
2157 consumer_reply_str);
2158 ret = 0;
2159 error:
2160 health_code_update();
2161 return ret;
2162 }
This page took 0.078607 seconds and 4 git commands to generate.