projects / lttng-tools.git / blob
? search:


abefba8659d7cc429aa39588186e580de084d51b
[lttng-tools.git] / src / bin / lttng-relayd / main.c
1 /*
2 * Copyright (C) 2012 - Julien Desfossez <jdesfossez@efficios.com>
3 * David Goulet <dgoulet@efficios.com>
4 *
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License, version 2 only,
7 * as published by the Free Software Foundation.
8 *
9 * This program is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
12 * more details.
13 *
14 * You should have received a copy of the GNU General Public License along
15 * with this program; if not, write to the Free Software Foundation, Inc.,
16 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
17 */
18
19 #define _GNU_SOURCE
20 #include <getopt.h>
21 #include <grp.h>
22 #include <limits.h>
23 #include <pthread.h>
24 #include <signal.h>
25 #include <stdio.h>
26 #include <stdlib.h>
27 #include <string.h>
28 #include <sys/mman.h>
29 #include <sys/mount.h>
30 #include <sys/resource.h>
31 #include <sys/socket.h>
32 #include <sys/stat.h>
33 #include <sys/types.h>
34 #include <sys/wait.h>
35 #include <urcu/futex.h>
36 #include <urcu/uatomic.h>
37 #include <unistd.h>
38 #include <fcntl.h>
39 #include <config.h>
40
41 #include <lttng/lttng.h>
42 #include <common/common.h>
43 #include <common/compat/poll.h>
44 #include <common/compat/socket.h>
45 #include <common/defaults.h>
46 #include <common/futex.h>
47 #include <common/sessiond-comm/sessiond-comm.h>
48 #include <common/sessiond-comm/inet.h>
49 #include <common/hashtable/hashtable.h>
50 #include <common/sessiond-comm/relayd.h>
51 #include <common/uri.h>
52 #include <common/utils.h>
53
54 #include "lttng-relayd.h"
55
56 /* command line options */
57 static int opt_daemon;
58 static char *opt_output_path;
59 static struct lttng_uri *control_uri;
60 static struct lttng_uri *data_uri;
61
62 const char *progname;
63 static int is_root; /* Set to 1 if the daemon is running as root */
64
65 /*
66 * Quit pipe for all threads. This permits a single cancellation point
67 * for all threads when receiving an event on the pipe.
68 */
69 static int thread_quit_pipe[2] = { -1, -1 };
70
71 /*
72 * This pipe is used to inform the worker thread that a command is queued and
73 * ready to be processed.
74 */
75 static int relay_cmd_pipe[2] = { -1, -1 };
76
77 /* Shared between threads */
78 static int dispatch_thread_exit;
79
80 static pthread_t listener_thread;
81 static pthread_t dispatcher_thread;
82 static pthread_t worker_thread;
83
84 static uint64_t last_relay_stream_id;
85 static uint64_t last_relay_session_id;
86
87 /*
88 * Relay command queue.
89 *
90 * The relay_thread_listener and relay_thread_dispatcher communicate with this
91 * queue.
92 */
93 static struct relay_cmd_queue relay_cmd_queue;
94
95 /* buffer allocated at startup, used to store the trace data */
96 static char *data_buffer;
97 static unsigned int data_buffer_size;
98
99 /*
100 * usage function on stderr
101 */
102 static
103 void usage(void)
104 {
105 fprintf(stderr, "Usage: %s OPTIONS\n\nOptions:\n", progname);
106 fprintf(stderr, " -h, --help Display this usage.\n");
107 fprintf(stderr, " -d, --daemonize Start as a daemon.\n");
108 fprintf(stderr, " -C, --control-port Control port listening (URI)\n");
109 fprintf(stderr, " -D, --data-port Data port listening (URI)\n");
110 fprintf(stderr, " -o, --output Output path for traces (PATH)\n");
111 fprintf(stderr, " -v, --verbose Verbose mode. Activate DBG() macro.\n");
112 }
113
114 static
115 int parse_args(int argc, char **argv)
116 {
117 int c;
118 int ret = 0;
119 char *default_address;
120
121 static struct option long_options[] = {
122 { "control-port", 1, 0, 'C', },
123 { "data-port", 1, 0, 'D', },
124 { "daemonize", 0, 0, 'd', },
125 { "help", 0, 0, 'h', },
126 { "output", 1, 0, 'o', },
127 { "verbose", 0, 0, 'v', },
128 { NULL, 0, 0, 0, },
129 };
130
131 while (1) {
132 int option_index = 0;
133 c = getopt_long(argc, argv, "dhv" "C:D:o:",
134 long_options, &option_index);
135 if (c == -1) {
136 break;
137 }
138
139 switch (c) {
140 case 0:
141 fprintf(stderr, "option %s", long_options[option_index].name);
142 if (optarg) {
143 fprintf(stderr, " with arg %s\n", optarg);
144 }
145 break;
146 case 'C':
147 ret = uri_parse(optarg, &control_uri);
148 if (ret < 0) {
149 ERR("Invalid control URI specified");
150 goto exit;
151 }
152 if (control_uri->port == 0) {
153 control_uri->port = DEFAULT_NETWORK_CONTROL_PORT;
154 }
155 break;
156 case 'D':
157 ret = uri_parse(optarg, &data_uri);
158 if (ret < 0) {
159 ERR("Invalid data URI specified");
160 goto exit;
161 }
162 if (data_uri->port == 0) {
163 data_uri->port = DEFAULT_NETWORK_DATA_PORT;
164 }
165 break;
166 case 'd':
167 opt_daemon = 1;
168 break;
169 case 'h':
170 usage();
171 exit(EXIT_FAILURE);
172 case 'o':
173 ret = asprintf(&opt_output_path, "%s", optarg);
174 if (ret < 0) {
175 PERROR("asprintf opt_output_path");
176 goto exit;
177 }
178 break;
179 case 'v':
180 /* Verbose level can increase using multiple -v */
181 lttng_opt_verbose += 1;
182 break;
183 default:
184 /* Unknown option or other error.
185 * Error is printed by getopt, just return */
186 ret = -1;
187 goto exit;
188 }
189 }
190
191 /* assign default values */
192 if (control_uri == NULL) {
193 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
194 DEFAULT_NETWORK_CONTROL_PORT);
195 if (ret < 0) {
196 PERROR("asprintf default data address");
197 goto exit;
198 }
199
200 ret = uri_parse(default_address, &control_uri);
201 free(default_address);
202 if (ret < 0) {
203 ERR("Invalid control URI specified");
204 goto exit;
205 }
206 }
207 if (data_uri == NULL) {
208 ret = asprintf(&default_address, "tcp://0.0.0.0:%d",
209 DEFAULT_NETWORK_DATA_PORT);
210 if (ret < 0) {
211 PERROR("asprintf default data address");
212 goto exit;
213 }
214
215 ret = uri_parse(default_address, &data_uri);
216 free(default_address);
217 if (ret < 0) {
218 ERR("Invalid data URI specified");
219 goto exit;
220 }
221 }
222
223 exit:
224 return ret;
225 }
226
227 /*
228 * Cleanup the daemon
229 */
230 static
231 void cleanup(void)
232 {
233 DBG("Cleaning up");
234
235 /* free the dynamically allocated opt_output_path */
236 free(opt_output_path);
237
238 /* Close thread quit pipes */
239 utils_close_pipe(thread_quit_pipe);
240
241 /* Close relay cmd pipes */
242 utils_close_pipe(relay_cmd_pipe);
243 }
244
245 /*
246 * Write to writable pipe used to notify a thread.
247 */
248 static
249 int notify_thread_pipe(int wpipe)
250 {
251 int ret;
252
253 do {
254 ret = write(wpipe, "!", 1);
255 } while (ret < 0 && errno == EINTR);
256 if (ret < 0) {
257 PERROR("write poll pipe");
258 }
259
260 return ret;
261 }
262
263 /*
264 * Stop all threads by closing the thread quit pipe.
265 */
266 static
267 void stop_threads(void)
268 {
269 int ret;
270
271 /* Stopping all threads */
272 DBG("Terminating all threads");
273 ret = notify_thread_pipe(thread_quit_pipe[1]);
274 if (ret < 0) {
275 ERR("write error on thread quit pipe");
276 }
277
278 /* Dispatch thread */
279 CMM_STORE_SHARED(dispatch_thread_exit, 1);
280 futex_nto1_wake(&relay_cmd_queue.futex);
281 }
282
283 /*
284 * Signal handler for the daemon
285 *
286 * Simply stop all worker threads, leaving main() return gracefully after
287 * joining all threads and calling cleanup().
288 */
289 static
290 void sighandler(int sig)
291 {
292 switch (sig) {
293 case SIGPIPE:
294 DBG("SIGPIPE caught");
295 return;
296 case SIGINT:
297 DBG("SIGINT caught");
298 stop_threads();
299 break;
300 case SIGTERM:
301 DBG("SIGTERM caught");
302 stop_threads();
303 break;
304 default:
305 break;
306 }
307 }
308
309 /*
310 * Setup signal handler for :
311 * SIGINT, SIGTERM, SIGPIPE
312 */
313 static
314 int set_signal_handler(void)
315 {
316 int ret = 0;
317 struct sigaction sa;
318 sigset_t sigset;
319
320 if ((ret = sigemptyset(&sigset)) < 0) {
321 PERROR("sigemptyset");
322 return ret;
323 }
324
325 sa.sa_handler = sighandler;
326 sa.sa_mask = sigset;
327 sa.sa_flags = 0;
328 if ((ret = sigaction(SIGTERM, &sa, NULL)) < 0) {
329 PERROR("sigaction");
330 return ret;
331 }
332
333 if ((ret = sigaction(SIGINT, &sa, NULL)) < 0) {
334 PERROR("sigaction");
335 return ret;
336 }
337
338 if ((ret = sigaction(SIGPIPE, &sa, NULL)) < 0) {
339 PERROR("sigaction");
340 return ret;
341 }
342
343 DBG("Signal handler set for SIGTERM, SIGPIPE and SIGINT");
344
345 return ret;
346 }
347
348 /*
349 * Init thread quit pipe.
350 *
351 * Return -1 on error or 0 if all pipes are created.
352 */
353 static
354 int init_thread_quit_pipe(void)
355 {
356 int ret;
357
358 ret = utils_create_pipe_cloexec(thread_quit_pipe);
359
360 return ret;
361 }
362
363 /*
364 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
365 */
366 static
367 int create_thread_poll_set(struct lttng_poll_event *events, int size)
368 {
369 int ret;
370
371 if (events == NULL || size == 0) {
372 ret = -1;
373 goto error;
374 }
375
376 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
377 if (ret < 0) {
378 goto error;
379 }
380
381 /* Add quit pipe */
382 ret = lttng_poll_add(events, thread_quit_pipe[0], LPOLLIN);
383 if (ret < 0) {
384 goto error;
385 }
386
387 return 0;
388
389 error:
390 return ret;
391 }
392
393 /*
394 * Check if the thread quit pipe was triggered.
395 *
396 * Return 1 if it was triggered else 0;
397 */
398 static
399 int check_thread_quit_pipe(int fd, uint32_t events)
400 {
401 if (fd == thread_quit_pipe[0] && (events & LPOLLIN)) {
402 return 1;
403 }
404
405 return 0;
406 }
407
408 /*
409 * Create and init socket from uri.
410 */
411 static
412 struct lttcomm_sock *relay_init_sock(struct lttng_uri *uri)
413 {
414 int ret;
415 struct lttcomm_sock *sock = NULL;
416
417 sock = lttcomm_alloc_sock_from_uri(uri);
418 if (sock == NULL) {
419 ERR("Allocating socket");
420 goto error;
421 }
422
423 ret = lttcomm_create_sock(sock);
424 if (ret < 0) {
425 goto error;
426 }
427 DBG("Listening on sock %d", sock->fd);
428
429 ret = sock->ops->bind(sock);
430 if (ret < 0) {
431 goto error;
432 }
433
434 ret = sock->ops->listen(sock, -1);
435 if (ret < 0) {
436 goto error;
437
438 }
439
440 return sock;
441
442 error:
443 if (sock) {
444 lttcomm_destroy_sock(sock);
445 }
446 return NULL;
447 }
448
449 /*
450 * This thread manages the listening for new connections on the network
451 */
452 static
453 void *relay_thread_listener(void *data)
454 {
455 int i, ret, pollfd, err = -1;
456 int val = 1;
457 uint32_t revents, nb_fd;
458 struct lttng_poll_event events;
459 struct lttcomm_sock *control_sock, *data_sock;
460
461 /*
462 * Get allocated in this thread, enqueued to a global queue, dequeued and
463 * freed in the worker thread.
464 */
465 struct relay_command *relay_cmd = NULL;
466
467 DBG("[thread] Relay listener started");
468
469 control_sock = relay_init_sock(control_uri);
470 if (!control_sock) {
471 goto error_sock_control;
472 }
473
474 data_sock = relay_init_sock(data_uri);
475 if (!data_sock) {
476 goto error_sock_relay;
477 }
478
479 /*
480 * Pass 3 as size here for the thread quit pipe, control and data socket.
481 */
482 ret = create_thread_poll_set(&events, 3);
483 if (ret < 0) {
484 goto error_create_poll;
485 }
486
487 /* Add the control socket */
488 ret = lttng_poll_add(&events, control_sock->fd, LPOLLIN | LPOLLRDHUP);
489 if (ret < 0) {
490 goto error_poll_add;
491 }
492
493 /* Add the data socket */
494 ret = lttng_poll_add(&events, data_sock->fd, LPOLLIN | LPOLLRDHUP);
495 if (ret < 0) {
496 goto error_poll_add;
497 }
498
499 while (1) {
500 DBG("Listener accepting connections");
501
502 nb_fd = LTTNG_POLL_GETNB(&events);
503
504 restart:
505 ret = lttng_poll_wait(&events, -1);
506 if (ret < 0) {
507 /*
508 * Restart interrupted system call.
509 */
510 if (errno == EINTR) {
511 goto restart;
512 }
513 goto error;
514 }
515
516 DBG("Relay new connection received");
517 for (i = 0; i < nb_fd; i++) {
518 /* Fetch once the poll data */
519 revents = LTTNG_POLL_GETEV(&events, i);
520 pollfd = LTTNG_POLL_GETFD(&events, i);
521
522 /* Thread quit pipe has been closed. Killing thread. */
523 ret = check_thread_quit_pipe(pollfd, revents);
524 if (ret) {
525 err = 0;
526 goto exit;
527 }
528
529 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
530 ERR("socket poll error");
531 goto error;
532 } else if (revents & LPOLLIN) {
533 struct lttcomm_sock *newsock = NULL;
534
535 relay_cmd = zmalloc(sizeof(struct relay_command));
536 if (relay_cmd == NULL) {
537 PERROR("relay command zmalloc");
538 goto error;
539 }
540
541 if (pollfd == data_sock->fd) {
542 newsock = data_sock->ops->accept(data_sock);
543 if (newsock < 0) {
544 PERROR("accepting data sock");
545 goto error;
546 }
547 relay_cmd->type = RELAY_DATA;
548 DBG("Relay data connection accepted, socket %d", newsock->fd);
549 } else if (pollfd == control_sock->fd) {
550 newsock = control_sock->ops->accept(control_sock);
551 if (newsock < 0) {
552 PERROR("accepting control sock");
553 goto error;
554 }
555 relay_cmd->type = RELAY_CONTROL;
556 DBG("Relay control connection accepted, socket %d", newsock->fd);
557 }
558 ret = setsockopt(newsock->fd, SOL_SOCKET, SO_REUSEADDR,
559 &val, sizeof(int));
560 if (ret < 0) {
561 PERROR("setsockopt inet");
562 goto error;
563 }
564 relay_cmd->sock = newsock;
565 /*
566 * Lock free enqueue the request.
567 */
568 cds_wfq_enqueue(&relay_cmd_queue.queue, &relay_cmd->node);
569
570 /*
571 * Wake the dispatch queue futex. Implicit memory
572 * barrier with the exchange in cds_wfq_enqueue.
573 */
574 futex_nto1_wake(&relay_cmd_queue.futex);
575 }
576 }
577 }
578
579 exit:
580 error:
581 error_poll_add:
582 lttng_poll_clean(&events);
583 error_create_poll:
584 if (data_sock->fd >= 0) {
585 ret = data_sock->ops->close(data_sock);
586 if (ret) {
587 PERROR("close");
588 }
589 }
590 lttcomm_destroy_sock(data_sock);
591 error_sock_relay:
592 if (control_sock->fd >= 0) {
593 ret = control_sock->ops->close(control_sock);
594 if (ret) {
595 PERROR("close");
596 }
597 }
598 lttcomm_destroy_sock(control_sock);
599 error_sock_control:
600 if (err) {
601 DBG("Thread exited with error");
602 }
603 DBG("Relay listener thread cleanup complete");
604 stop_threads();
605 return NULL;
606 }
607
608 /*
609 * This thread manages the dispatching of the requests to worker threads
610 */
611 static
612 void *relay_thread_dispatcher(void *data)
613 {
614 int ret;
615 struct cds_wfq_node *node;
616 struct relay_command *relay_cmd = NULL;
617
618 DBG("[thread] Relay dispatcher started");
619
620 while (!CMM_LOAD_SHARED(dispatch_thread_exit)) {
621 /* Atomically prepare the queue futex */
622 futex_nto1_prepare(&relay_cmd_queue.futex);
623
624 do {
625 /* Dequeue commands */
626 node = cds_wfq_dequeue_blocking(&relay_cmd_queue.queue);
627 if (node == NULL) {
628 DBG("Woken up but nothing in the relay command queue");
629 /* Continue thread execution */
630 break;
631 }
632
633 relay_cmd = caa_container_of(node, struct relay_command, node);
634 DBG("Dispatching request waiting on sock %d", relay_cmd->sock->fd);
635
636 /*
637 * Inform worker thread of the new request. This
638 * call is blocking so we can be assured that the data will be read
639 * at some point in time or wait to the end of the world :)
640 */
641 do {
642 ret = write(relay_cmd_pipe[1], relay_cmd,
643 sizeof(struct relay_command));
644 } while (ret < 0 && errno == EINTR);
645 free(relay_cmd);
646 if (ret < 0) {
647 PERROR("write cmd pipe");
648 goto error;
649 }
650 } while (node != NULL);
651
652 /* Futex wait on queue. Blocking call on futex() */
653 futex_nto1_wait(&relay_cmd_queue.futex);
654 }
655
656 error:
657 DBG("Dispatch thread dying");
658 stop_threads();
659 return NULL;
660 }
661
662 /*
663 * Return the realpath(3) of the path even if the last directory token does not
664 * exist. For example, with /tmp/test1/test2, if test2/ does not exist but the
665 * /tmp/test1 does, the real path is returned. In normal time, realpath(3)
666 * fails if the end point directory does not exist.
667 */
668 static
669 char *expand_full_path(const char *path)
670 {
671 const char *end_path = path;
672 char *next, *cut_path, *expanded_path, *respath;
673
674 /* Find last token delimited by '/' */
675 while ((next = strpbrk(end_path + 1, "/"))) {
676 end_path = next;
677 }
678
679 /* Cut last token from original path */
680 cut_path = strndup(path, end_path - path);
681
682 expanded_path = malloc(PATH_MAX);
683 if (expanded_path == NULL) {
684 respath = NULL;
685 goto end;
686 }
687
688 respath = realpath(cut_path, expanded_path);
689 if (respath == NULL) {
690 switch (errno) {
691 case ENOENT:
692 ERR("%s: No such file or directory", cut_path);
693 break;
694 default:
695 PERROR("realpath");
696 break;
697 }
698 free(expanded_path);
699 } else {
700 /* Add end part to expanded path */
701 strcat(respath, end_path);
702 }
703 end:
704 free(cut_path);
705 return respath;
706 }
707
708
709 /*
710 * config_get_default_path
711 *
712 * Returns the HOME directory path. Caller MUST NOT free(3) the return pointer.
713 */
714 static
715 char *config_get_default_path(void)
716 {
717 return getenv("HOME");
718 }
719
720 /*
721 * Create recursively directory using the FULL path.
722 */
723 static
724 int mkdir_recursive(char *path, mode_t mode)
725 {
726 char *p, tmp[PATH_MAX];
727 struct stat statbuf;
728 size_t len;
729 int ret;
730
731 ret = snprintf(tmp, sizeof(tmp), "%s", path);
732 if (ret < 0) {
733 PERROR("snprintf mkdir");
734 goto error;
735 }
736
737 len = ret;
738 if (tmp[len - 1] == '/') {
739 tmp[len - 1] = 0;
740 }
741
742 for (p = tmp + 1; *p; p++) {
743 if (*p == '/') {
744 *p = 0;
745 if (tmp[strlen(tmp) - 1] == '.' &&
746 tmp[strlen(tmp) - 2] == '.' &&
747 tmp[strlen(tmp) - 3] == '/') {
748 ERR("Using '/../' is not permitted in the trace path (%s)",
749 tmp);
750 ret = -1;
751 goto error;
752 }
753 ret = stat(tmp, &statbuf);
754 if (ret < 0) {
755 ret = mkdir(tmp, mode);
756 if (ret < 0) {
757 if (errno != EEXIST) {
758 PERROR("mkdir recursive");
759 ret = -errno;
760 goto error;
761 }
762 }
763 }
764 *p = '/';
765 }
766 }
767
768 ret = mkdir(tmp, mode);
769 if (ret < 0) {
770 if (errno != EEXIST) {
771 PERROR("mkdir recursive last piece");
772 ret = -errno;
773 } else {
774 ret = 0;
775 }
776 }
777
778 error:
779 return ret;
780 }
781
782 static
783 char *create_output_path_auto(char *path_name)
784 {
785 int ret;
786 char *traces_path = NULL;
787 char *alloc_path = NULL;
788 char *default_path;
789
790 default_path = config_get_default_path();
791 if (default_path == NULL) {
792 ERR("Home path not found.\n \
793 Please specify an output path using -o, --output PATH");
794 goto exit;
795 }
796 alloc_path = strdup(default_path);
797 if (alloc_path == NULL) {
798 PERROR("Path allocation");
799 goto exit;
800 }
801 ret = asprintf(&traces_path, "%s/" DEFAULT_TRACE_DIR_NAME
802 "/%s", alloc_path, path_name);
803 if (ret < 0) {
804 PERROR("asprintf trace dir name");
805 goto exit;
806 }
807 exit:
808 free(alloc_path);
809 return traces_path;
810 }
811
812 static
813 char *create_output_path_noauto(char *path_name)
814 {
815 int ret;
816 char *traces_path = NULL;
817 char *full_path;
818
819 full_path = expand_full_path(opt_output_path);
820 ret = asprintf(&traces_path, "%s/%s", full_path, path_name);
821 if (ret < 0) {
822 PERROR("asprintf trace dir name");
823 goto exit;
824 }
825 exit:
826 free(full_path);
827 return traces_path;
828 }
829
830 /*
831 * create_output_path: create the output trace directory
832 */
833 static
834 char *create_output_path(char *path_name)
835 {
836 if (opt_output_path == NULL) {
837 return create_output_path_auto(path_name);
838 } else {
839 return create_output_path_noauto(path_name);
840 }
841 }
842
843 static
844 void deferred_free_stream(struct rcu_head *head)
845 {
846 struct relay_stream *stream =
847 caa_container_of(head, struct relay_stream, rcu_node);
848 free(stream);
849 }
850
851 /*
852 * relay_delete_session: Free all memory associated with a session and
853 * close all the FDs
854 */
855 static
856 void relay_delete_session(struct relay_command *cmd, struct lttng_ht *streams_ht)
857 {
858 struct lttng_ht_iter iter;
859 struct lttng_ht_node_ulong *node;
860 struct relay_stream *stream;
861 int ret;
862
863 if (!cmd->session) {
864 return;
865 }
866
867 DBG("Relay deleting session %lu", cmd->session->id);
868 free(cmd->session->sock);
869
870 rcu_read_lock();
871 cds_lfht_for_each_entry(streams_ht->ht, &iter.iter, node, node) {
872 node = lttng_ht_iter_get_node_ulong(&iter);
873 if (node) {
874 stream = caa_container_of(node,
875 struct relay_stream, stream_n);
876 if (stream->session == cmd->session) {
877 close(stream->fd);
878 ret = lttng_ht_del(streams_ht, &iter);
879 assert(!ret);
880 call_rcu(&stream->rcu_node,
881 deferred_free_stream);
882 }
883 }
884 }
885 rcu_read_unlock();
886 }
887
888 /*
889 * relay_add_stream: allocate a new stream for a session
890 */
891 static
892 int relay_add_stream(struct lttcomm_relayd_hdr *recv_hdr,
893 struct relay_command *cmd, struct lttng_ht *streams_ht)
894 {
895 struct relay_session *session = cmd->session;
896 struct lttcomm_relayd_add_stream stream_info;
897 struct relay_stream *stream = NULL;
898 struct lttcomm_relayd_status_stream reply;
899 char *path = NULL, *root_path = NULL;
900 int ret, send_ret;
901
902 if (!session || session->version_check_done == 0) {
903 ERR("Trying to add a stream before version check");
904 ret = -1;
905 goto end_no_session;
906 }
907
908 /* FIXME : use data_size for something ? */
909 ret = cmd->sock->ops->recvmsg(cmd->sock, &stream_info,
910 sizeof(struct lttcomm_relayd_add_stream), MSG_WAITALL);
911 if (ret < sizeof(struct lttcomm_relayd_add_stream)) {
912 ERR("Relay didn't receive valid add_stream struct size : %d", ret);
913 ret = -1;
914 goto end_no_session;
915 }
916 stream = zmalloc(sizeof(struct relay_stream));
917 if (stream == NULL) {
918 PERROR("relay stream zmalloc");
919 ret = -1;
920 goto end_no_session;
921 }
922
923 rcu_read_lock();
924 stream->stream_handle = ++last_relay_stream_id;
925 stream->seq = 0;
926 stream->session = session;
927
928 root_path = create_output_path(stream_info.pathname);
929 if (!root_path) {
930 ret = -1;
931 goto end;
932 }
933 ret = mkdir_recursive(root_path, S_IRWXU | S_IRWXG);
934 if (ret < 0) {
935 ERR("relay creating output directory");
936 goto end;
937 }
938
939 ret = asprintf(&path, "%s/%s", root_path, stream_info.channel_name);
940 if (ret < 0) {
941 PERROR("asprintf stream path");
942 goto end;
943 }
944
945 ret = open(path, O_WRONLY|O_CREAT|O_TRUNC, S_IRWXU|S_IRWXG|S_IRWXO);
946 if (ret < 0) {
947 PERROR("Relay creating trace file");
948 goto end;
949 }
950
951 stream->fd = ret;
952 DBG("Tracefile %s created", path);
953
954 lttng_ht_node_init_ulong(&stream->stream_n,
955 (unsigned long) stream->stream_handle);
956 lttng_ht_add_unique_ulong(streams_ht,
957 &stream->stream_n);
958
959 DBG("Relay new stream added %s", stream_info.channel_name);
960
961 end:
962 free(path);
963 free(root_path);
964 /* send the session id to the client or a negative return code on error */
965 if (ret < 0) {
966 reply.ret_code = htobe32(LTTCOMM_ERR);
967 } else {
968 reply.ret_code = htobe32(LTTCOMM_OK);
969 }
970 reply.handle = htobe64(stream->stream_handle);
971 send_ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
972 sizeof(struct lttcomm_relayd_status_stream), 0);
973 if (send_ret < 0) {
974 ERR("Relay sending stream id");
975 }
976 rcu_read_unlock();
977
978 end_no_session:
979 return ret;
980 }
981
982 /*
983 * relay_unknown_command: send -1 if received unknown command
984 */
985 static
986 void relay_unknown_command(struct relay_command *cmd)
987 {
988 struct lttcomm_relayd_generic_reply reply;
989 int ret;
990
991 reply.ret_code = htobe32(LTTCOMM_ERR);
992 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
993 sizeof(struct lttcomm_relayd_generic_reply), 0);
994 if (ret < 0) {
995 ERR("Relay sending unknown command");
996 }
997 }
998
999 /*
1000 * relay_start: send an acknowledgment to the client to tell if we are
1001 * ready to receive data. We are ready if a session is established.
1002 */
1003 static
1004 int relay_start(struct lttcomm_relayd_hdr *recv_hdr,
1005 struct relay_command *cmd)
1006 {
1007 int ret = htobe32(LTTCOMM_OK);
1008 struct lttcomm_relayd_generic_reply reply;
1009 struct relay_session *session = cmd->session;
1010
1011 if (!session) {
1012 DBG("Trying to start the streaming without a session established");
1013 ret = htobe32(LTTCOMM_ERR);
1014 }
1015
1016 reply.ret_code = ret;
1017 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1018 sizeof(struct lttcomm_relayd_generic_reply), 0);
1019 if (ret < 0) {
1020 ERR("Relay sending start ack");
1021 }
1022
1023 return ret;
1024 }
1025
1026 /*
1027 * Get stream from stream id.
1028 * Need to be called with RCU read-side lock held.
1029 */
1030 static
1031 struct relay_stream *relay_stream_from_stream_id(uint64_t stream_id,
1032 struct lttng_ht *streams_ht)
1033 {
1034 struct lttng_ht_node_ulong *node;
1035 struct lttng_ht_iter iter;
1036 struct relay_stream *ret;
1037
1038 lttng_ht_lookup(streams_ht,
1039 (void *)((unsigned long) stream_id),
1040 &iter);
1041 node = lttng_ht_iter_get_node_ulong(&iter);
1042 if (node == NULL) {
1043 DBG("Relay stream %lu not found", stream_id);
1044 ret = NULL;
1045 goto end;
1046 }
1047
1048 ret = caa_container_of(node, struct relay_stream, stream_n);
1049
1050 end:
1051 return ret;
1052 }
1053
1054 /*
1055 * relay_recv_metadata: receive the metada for the session.
1056 */
1057 static
1058 int relay_recv_metadata(struct lttcomm_relayd_hdr *recv_hdr,
1059 struct relay_command *cmd, struct lttng_ht *streams_ht)
1060 {
1061 int ret = htobe32(LTTCOMM_OK);
1062 struct relay_session *session = cmd->session;
1063 struct lttcomm_relayd_metadata_payload *metadata_struct;
1064 struct relay_stream *metadata_stream;
1065 uint64_t data_size, payload_size;
1066
1067 if (!session) {
1068 ERR("Metadata sent before version check");
1069 ret = -1;
1070 goto end;
1071 }
1072
1073 data_size = be64toh(recv_hdr->data_size);
1074 payload_size = data_size - sizeof(uint64_t);
1075 if (data_buffer_size < data_size) {
1076 data_buffer = realloc(data_buffer, data_size);
1077 if (!data_buffer) {
1078 ERR("Allocating data buffer");
1079 ret = -1;
1080 goto end;
1081 }
1082 data_buffer_size = data_size;
1083 }
1084 memset(data_buffer, 0, data_size);
1085 DBG2("Relay receiving metadata, waiting for %lu bytes", data_size);
1086 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1087 if (ret < 0 || ret != data_size) {
1088 ret = -1;
1089 ERR("Relay didn't receive the whole metadata");
1090 goto end;
1091 }
1092 metadata_struct = (struct lttcomm_relayd_metadata_payload *) data_buffer;
1093
1094 rcu_read_lock();
1095 metadata_stream = relay_stream_from_stream_id(
1096 be64toh(metadata_struct->stream_id), streams_ht);
1097 if (!metadata_stream) {
1098 ret = -1;
1099 goto end_unlock;
1100 }
1101
1102 do {
1103 ret = write(metadata_stream->fd, metadata_struct->payload,
1104 payload_size);
1105 } while (ret < 0 && errno == EINTR);
1106 if (ret < (payload_size)) {
1107 ERR("Relay error writing metadata on file");
1108 ret = -1;
1109 goto end_unlock;
1110 }
1111 DBG2("Relay metadata written");
1112
1113 end_unlock:
1114 rcu_read_unlock();
1115 end:
1116 return ret;
1117 }
1118
1119 /*
1120 * relay_send_version: send relayd version number
1121 */
1122 static
1123 int relay_send_version(struct lttcomm_relayd_hdr *recv_hdr,
1124 struct relay_command *cmd)
1125 {
1126 int ret = htobe32(LTTCOMM_OK);
1127 struct lttcomm_relayd_version reply;
1128 struct relay_session *session = NULL;
1129
1130 if (cmd->session == NULL) {
1131 session = zmalloc(sizeof(struct relay_session));
1132 if (session == NULL) {
1133 PERROR("relay session zmalloc");
1134 ret = -1;
1135 goto end;
1136 }
1137 session->id = ++last_relay_session_id;
1138 DBG("Created session %lu", session->id);
1139 cmd->session = session;
1140 }
1141 session->version_check_done = 1;
1142
1143 sscanf(VERSION, "%u.%u", &reply.major, &reply.minor);
1144 reply.major = htobe32(reply.major);
1145 reply.minor = htobe32(reply.minor);
1146 ret = cmd->sock->ops->sendmsg(cmd->sock, &reply,
1147 sizeof(struct lttcomm_relayd_version), 0);
1148 if (ret < 0) {
1149 ERR("Relay sending version");
1150 }
1151 DBG("Version check done");
1152
1153 end:
1154 return ret;
1155 }
1156
1157 /*
1158 * relay_process_control: Process the commands received on the control socket
1159 */
1160 static
1161 int relay_process_control(struct lttcomm_relayd_hdr *recv_hdr,
1162 struct relay_command *cmd, struct lttng_ht *streams_ht)
1163 {
1164 int ret = 0;
1165
1166 switch (be32toh(recv_hdr->cmd)) {
1167 /*
1168 case RELAYD_CREATE_SESSION:
1169 ret = relay_create_session(recv_hdr, cmd);
1170 break;
1171 */
1172 case RELAYD_ADD_STREAM:
1173 ret = relay_add_stream(recv_hdr, cmd, streams_ht);
1174 break;
1175 case RELAYD_START_DATA:
1176 ret = relay_start(recv_hdr, cmd);
1177 break;
1178 case RELAYD_SEND_METADATA:
1179 ret = relay_recv_metadata(recv_hdr, cmd, streams_ht);
1180 break;
1181 case RELAYD_VERSION:
1182 ret = relay_send_version(recv_hdr, cmd);
1183 break;
1184 case RELAYD_UPDATE_SYNC_INFO:
1185 default:
1186 ERR("Received unknown command (%u)", be32toh(recv_hdr->cmd));
1187 relay_unknown_command(cmd);
1188 ret = -1;
1189 goto end;
1190 }
1191
1192 end:
1193 return ret;
1194 }
1195
1196 /*
1197 * relay_process_data: Process the data received on the data socket
1198 */
1199 static
1200 int relay_process_data(struct relay_command *cmd, struct lttng_ht *streams_ht)
1201 {
1202 int ret = 0;
1203 struct relay_stream *stream;
1204 struct lttcomm_relayd_data_hdr data_hdr;
1205 uint64_t stream_id;
1206 uint32_t data_size;
1207
1208 ret = cmd->sock->ops->recvmsg(cmd->sock, &data_hdr,
1209 sizeof(struct lttcomm_relayd_data_hdr), MSG_WAITALL);
1210 if (ret <= 0) {
1211 ERR("Connections seems to be closed");
1212 ret = -1;
1213 goto end;
1214 }
1215
1216 stream_id = be64toh(data_hdr.stream_id);
1217
1218 rcu_read_lock();
1219 stream = relay_stream_from_stream_id(stream_id, streams_ht);
1220 if (!stream) {
1221 ret = -1;
1222 goto end_unlock;
1223 }
1224
1225 data_size = be32toh(data_hdr.data_size);
1226 if (data_buffer_size < data_size) {
1227 data_buffer = realloc(data_buffer, data_size);
1228 if (!data_buffer) {
1229 ERR("Allocating data buffer");
1230 ret = -1;
1231 goto end_unlock;
1232 }
1233 data_buffer_size = data_size;
1234 }
1235 memset(data_buffer, 0, data_size);
1236
1237 ret = cmd->sock->ops->recvmsg(cmd->sock, data_buffer, data_size, MSG_WAITALL);
1238 if (ret <= 0) {
1239 ret = -1;
1240 goto end_unlock;
1241 }
1242
1243 do {
1244 ret = write(stream->fd, data_buffer, data_size);
1245 } while (ret < 0 && errno == EINTR);
1246 if (ret < data_size) {
1247 ERR("Relay error writing data to file");
1248 ret = -1;
1249 goto end_unlock;
1250 }
1251 DBG2("Relay wrote %d bytes to tracefile for stream id %lu", ret, stream->stream_handle);
1252
1253 end_unlock:
1254 rcu_read_unlock();
1255 end:
1256 return ret;
1257 }
1258
1259 static
1260 void relay_cleanup_poll_connection(struct lttng_poll_event *events, int pollfd)
1261 {
1262 int ret;
1263
1264 lttng_poll_del(events, pollfd);
1265
1266 ret = close(pollfd);
1267 if (ret < 0) {
1268 ERR("Closing pollfd %d", pollfd);
1269 }
1270 }
1271
1272 static
1273 int relay_add_connection(int fd, struct lttng_poll_event *events,
1274 struct lttng_ht *relay_connections_ht)
1275 {
1276 struct relay_command *relay_connection;
1277 int ret;
1278
1279 relay_connection = zmalloc(sizeof(struct relay_command));
1280 if (relay_connection == NULL) {
1281 PERROR("Relay command zmalloc");
1282 goto error;
1283 }
1284 ret = read(fd, relay_connection, sizeof(struct relay_command));
1285 if (ret < 0 || ret < sizeof(relay_connection)) {
1286 PERROR("read relay cmd pipe");
1287 goto error_read;
1288 }
1289
1290 lttng_ht_node_init_ulong(&relay_connection->sock_n,
1291 (unsigned long) relay_connection->sock->fd);
1292 rcu_read_lock();
1293 lttng_ht_add_unique_ulong(relay_connections_ht,
1294 &relay_connection->sock_n);
1295 rcu_read_unlock();
1296 return lttng_poll_add(events,
1297 relay_connection->sock->fd,
1298 LPOLLIN | LPOLLRDHUP);
1299
1300 error_read:
1301 free(relay_connection);
1302 error:
1303 return -1;
1304 }
1305
1306 static
1307 void deferred_free_connection(struct rcu_head *head)
1308 {
1309 struct relay_command *relay_connection =
1310 caa_container_of(head, struct relay_command, rcu_node);
1311 free(relay_connection);
1312 }
1313
1314 static
1315 void relay_del_connection(struct lttng_ht *relay_connections_ht,
1316 struct lttng_ht *streams_ht, struct lttng_ht_iter *iter,
1317 struct relay_command *relay_connection)
1318 {
1319 int ret;
1320
1321 ret = lttng_ht_del(relay_connections_ht, iter);
1322 assert(!ret);
1323 if (relay_connection->type == RELAY_CONTROL) {
1324 relay_delete_session(relay_connection, streams_ht);
1325 }
1326 call_rcu(&relay_connection->rcu_node,
1327 deferred_free_connection);
1328 }
1329
1330 /*
1331 * This thread does the actual work
1332 */
1333 static
1334 void *relay_thread_worker(void *data)
1335 {
1336 int i, ret, pollfd, err = -1;
1337 uint32_t revents, nb_fd;
1338 struct relay_command *relay_connection;
1339 struct lttng_poll_event events;
1340 struct lttng_ht *relay_connections_ht;
1341 struct lttng_ht_node_ulong *node;
1342 struct lttng_ht_iter iter;
1343 struct lttng_ht *streams_ht;
1344 struct lttcomm_relayd_hdr recv_hdr;
1345
1346 DBG("[thread] Relay worker started");
1347
1348 rcu_register_thread();
1349
1350 /* table of connections indexed on socket */
1351 relay_connections_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1352 if (!relay_connections_ht) {
1353 goto relay_connections_ht_error;
1354 }
1355
1356 /* tables of streams indexed by stream ID */
1357 streams_ht = lttng_ht_new(0, LTTNG_HT_TYPE_ULONG);
1358 if (!streams_ht) {
1359 goto streams_ht_error;
1360 }
1361
1362 ret = create_thread_poll_set(&events, 2);
1363 if (ret < 0) {
1364 goto error_poll_create;
1365 }
1366
1367 ret = lttng_poll_add(&events, relay_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1368 if (ret < 0) {
1369 goto error;
1370 }
1371
1372 while (1) {
1373 /* Zeroed the events structure */
1374 lttng_poll_reset(&events);
1375
1376 nb_fd = LTTNG_POLL_GETNB(&events);
1377
1378 /* Infinite blocking call, waiting for transmission */
1379 restart:
1380 ret = lttng_poll_wait(&events, -1);
1381 if (ret < 0) {
1382 /*
1383 * Restart interrupted system call.
1384 */
1385 if (errno == EINTR) {
1386 goto restart;
1387 }
1388 goto error;
1389 }
1390
1391 for (i = 0; i < nb_fd; i++) {
1392 /* Fetch once the poll data */
1393 revents = LTTNG_POLL_GETEV(&events, i);
1394 pollfd = LTTNG_POLL_GETFD(&events, i);
1395
1396 /* Thread quit pipe has been closed. Killing thread. */
1397 ret = check_thread_quit_pipe(pollfd, revents);
1398 if (ret) {
1399 err = 0;
1400 goto exit;
1401 }
1402
1403 /* Inspect the relay cmd pipe for new connection */
1404 if (pollfd == relay_cmd_pipe[0]) {
1405 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1406 ERR("Relay pipe error");
1407 goto error;
1408 } else if (revents & LPOLLIN) {
1409 DBG("Relay command received");
1410 ret = relay_add_connection(relay_cmd_pipe[0],
1411 &events, relay_connections_ht);
1412 if (ret < 0) {
1413 goto error;
1414 }
1415 }
1416 } else if (revents > 0) {
1417 rcu_read_lock();
1418 lttng_ht_lookup(relay_connections_ht,
1419 (void *)((unsigned long) pollfd),
1420 &iter);
1421 node = lttng_ht_iter_get_node_ulong(&iter);
1422 if (node == NULL) {
1423 DBG2("Relay sock %d not found", pollfd);
1424 rcu_read_unlock();
1425 goto error;
1426 }
1427 relay_connection = caa_container_of(node,
1428 struct relay_command, sock_n);
1429
1430 if (revents & (LPOLLERR)) {
1431 ERR("POLL ERROR");
1432 relay_cleanup_poll_connection(&events, pollfd);
1433 relay_del_connection(relay_connections_ht,
1434 streams_ht, &iter,
1435 relay_connection);
1436 } else if (revents & (LPOLLHUP | LPOLLRDHUP)) {
1437 DBG("Socket %d hung up", pollfd);
1438 relay_cleanup_poll_connection(&events, pollfd);
1439 relay_del_connection(relay_connections_ht,
1440 streams_ht, &iter,
1441 relay_connection);
1442 } else if (revents & LPOLLIN) {
1443 /* control socket */
1444 if (relay_connection->type == RELAY_CONTROL) {
1445 ret = relay_connection->sock->ops->recvmsg(
1446 relay_connection->sock, &recv_hdr,
1447 sizeof(struct lttcomm_relayd_hdr), MSG_WAITALL);
1448 /* connection closed */
1449 if (ret <= 0) {
1450 relay_cleanup_poll_connection(&events, pollfd);
1451 relay_del_connection(relay_connections_ht,
1452 streams_ht, &iter,
1453 relay_connection);
1454 DBG("Control connection closed with %d", pollfd);
1455 } else {
1456 if (relay_connection->session) {
1457 DBG2("Relay worker receiving data for session : %lu",
1458 relay_connection->session->id);
1459 }
1460 ret = relay_process_control(&recv_hdr,
1461 relay_connection,
1462 streams_ht);
1463 /*
1464 * there was an error in processing a control
1465 * command: clear the session
1466 * */
1467 if (ret < 0) {
1468 relay_cleanup_poll_connection(&events, pollfd);
1469 relay_del_connection(relay_connections_ht,
1470 streams_ht, &iter,
1471 relay_connection);
1472 DBG("Connection closed with %d", pollfd);
1473 }
1474 }
1475 /* data socket */
1476 } else if (relay_connection->type == RELAY_DATA) {
1477 ret = relay_process_data(relay_connection, streams_ht);
1478 /* connection closed */
1479 if (ret < 0) {
1480 relay_cleanup_poll_connection(&events, pollfd);
1481 relay_del_connection(relay_connections_ht,
1482 streams_ht, &iter,
1483 relay_connection);
1484 DBG("Data connection closed with %d", pollfd);
1485 }
1486 }
1487 }
1488 rcu_read_unlock();
1489 }
1490 }
1491 }
1492
1493 exit:
1494 error:
1495 lttng_poll_clean(&events);
1496
1497 /* empty the hash table and free the memory */
1498 rcu_read_lock();
1499 cds_lfht_for_each_entry(relay_connections_ht->ht, &iter.iter, node, node) {
1500 node = lttng_ht_iter_get_node_ulong(&iter);
1501 if (node) {
1502 relay_connection = caa_container_of(node,
1503 struct relay_command, sock_n);
1504 relay_del_connection(relay_connections_ht,
1505 streams_ht, &iter,
1506 relay_connection);
1507 }
1508 }
1509 rcu_read_unlock();
1510 error_poll_create:
1511 lttng_ht_destroy(streams_ht);
1512 streams_ht_error:
1513 lttng_ht_destroy(relay_connections_ht);
1514 relay_connections_ht_error:
1515 if (err) {
1516 DBG("Thread exited with error");
1517 }
1518 DBG("Worker thread cleanup complete");
1519 free(data_buffer);
1520 stop_threads();
1521 rcu_unregister_thread();
1522 return NULL;
1523 }
1524
1525 /*
1526 * Create the relay command pipe to wake thread_manage_apps.
1527 * Closed in cleanup().
1528 */
1529 static int create_relay_cmd_pipe(void)
1530 {
1531 int ret;
1532
1533 ret = utils_create_pipe_cloexec(relay_cmd_pipe);
1534
1535 return ret;
1536 }
1537
1538 /*
1539 * main
1540 */
1541 int main(int argc, char **argv)
1542 {
1543 int ret = 0;
1544 void *status;
1545
1546 /* Create thread quit pipe */
1547 if ((ret = init_thread_quit_pipe()) < 0) {
1548 goto error;
1549 }
1550
1551 /* Parse arguments */
1552 progname = argv[0];
1553 if ((ret = parse_args(argc, argv) < 0)) {
1554 goto exit;
1555 }
1556
1557 if ((ret = set_signal_handler()) < 0) {
1558 goto exit;
1559 }
1560
1561 /* Daemonize */
1562 if (opt_daemon) {
1563 ret = daemon(0, 0);
1564 if (ret < 0) {
1565 PERROR("daemon");
1566 goto exit;
1567 }
1568 }
1569
1570 /* Check if daemon is UID = 0 */
1571 is_root = !getuid();
1572
1573 if (!is_root) {
1574 if (control_uri->port < 1024 || data_uri->port < 1024) {
1575 ERR("Need to be root to use ports < 1024");
1576 ret = -1;
1577 goto exit;
1578 }
1579 }
1580
1581 /* Setup the thread apps communication pipe. */
1582 if ((ret = create_relay_cmd_pipe()) < 0) {
1583 goto exit;
1584 }
1585
1586 /* Init relay command queue. */
1587 cds_wfq_init(&relay_cmd_queue.queue);
1588
1589 /* Set up max poll set size */
1590 lttng_poll_set_max_size();
1591
1592 /* Setup the dispatcher thread */
1593 ret = pthread_create(&dispatcher_thread, NULL,
1594 relay_thread_dispatcher, (void *) NULL);
1595 if (ret != 0) {
1596 PERROR("pthread_create dispatcher");
1597 goto exit_dispatcher;
1598 }
1599
1600 /* Setup the worker thread */
1601 ret = pthread_create(&worker_thread, NULL,
1602 relay_thread_worker, (void *) NULL);
1603 if (ret != 0) {
1604 PERROR("pthread_create worker");
1605 goto exit_worker;
1606 }
1607
1608 /* Setup the listener thread */
1609 ret = pthread_create(&listener_thread, NULL,
1610 relay_thread_listener, (void *) NULL);
1611 if (ret != 0) {
1612 PERROR("pthread_create listener");
1613 goto exit_listener;
1614 }
1615
1616 exit_listener:
1617 ret = pthread_join(listener_thread, &status);
1618 if (ret != 0) {
1619 PERROR("pthread_join");
1620 goto error; /* join error, exit without cleanup */
1621 }
1622
1623 exit_worker:
1624 ret = pthread_join(worker_thread, &status);
1625 if (ret != 0) {
1626 PERROR("pthread_join");
1627 goto error; /* join error, exit without cleanup */
1628 }
1629
1630 exit_dispatcher:
1631 ret = pthread_join(dispatcher_thread, &status);
1632 if (ret != 0) {
1633 PERROR("pthread_join");
1634 goto error; /* join error, exit without cleanup */
1635 }
1636
1637 exit:
1638 cleanup();
1639 if (!ret) {
1640 exit(EXIT_SUCCESS);
1641 }
1642
1643 error:
1644 exit(EXIT_FAILURE);
1645 }
LTTng 2.0 tools and control repository
This page took 0.094923 seconds and 4 git commands to generate.