Fix: thread exit vs futex wait/wakeup race
[lttng-tools.git] / src / bin / lttng-sessiond / main.c
1 /*
2 * Copyright (C) 2011 - David Goulet <david.goulet@polymtl.ca>
3 * Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * 2013 - Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 *
6 * This program is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License, version 2 only,
8 * as published by the Free Software Foundation.
9 *
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
14 *
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
18 */
19
20 #define _LGPL_SOURCE
21 #include <getopt.h>
22 #include <grp.h>
23 #include <limits.h>
24 #include <paths.h>
25 #include <pthread.h>
26 #include <signal.h>
27 #include <stdio.h>
28 #include <stdlib.h>
29 #include <string.h>
30 #include <inttypes.h>
31 #include <sys/mman.h>
32 #include <sys/mount.h>
33 #include <sys/resource.h>
34 #include <sys/socket.h>
35 #include <sys/stat.h>
36 #include <sys/types.h>
37 #include <sys/wait.h>
38 #include <urcu/uatomic.h>
39 #include <unistd.h>
40 #include <ctype.h>
41
42 #include <common/common.h>
43 #include <common/compat/socket.h>
44 #include <common/compat/getenv.h>
45 #include <common/defaults.h>
46 #include <common/kernel-consumer/kernel-consumer.h>
47 #include <common/futex.h>
48 #include <common/relayd/relayd.h>
49 #include <common/utils.h>
50 #include <common/daemonize.h>
51 #include <common/config/session-config.h>
52
53 #include "lttng-sessiond.h"
54 #include "buffer-registry.h"
55 #include "channel.h"
56 #include "cmd.h"
57 #include "consumer.h"
58 #include "context.h"
59 #include "event.h"
60 #include "kernel.h"
61 #include "kernel-consumer.h"
62 #include "modprobe.h"
63 #include "shm.h"
64 #include "ust-ctl.h"
65 #include "ust-consumer.h"
66 #include "utils.h"
67 #include "fd-limit.h"
68 #include "health-sessiond.h"
69 #include "testpoint.h"
70 #include "ust-thread.h"
71 #include "agent-thread.h"
72 #include "save.h"
73 #include "load-session-thread.h"
74 #include "notification-thread.h"
75 #include "notification-thread-commands.h"
76 #include "syscall.h"
77 #include "agent.h"
78 #include "ht-cleanup.h"
79
80 #define CONSUMERD_FILE "lttng-consumerd"
81
82 static const char *help_msg =
83 #ifdef LTTNG_EMBED_HELP
84 #include <lttng-sessiond.8.h>
85 #else
86 NULL
87 #endif
88 ;
89
90 const char *progname;
91 const char *tracing_group_name = DEFAULT_TRACING_GROUP;
92 static int tracing_group_name_override;
93 static char *opt_pidfile;
94 static int opt_sig_parent;
95 static int opt_verbose_consumer;
96 static int opt_daemon, opt_background;
97 static int opt_no_kernel;
98 static char *opt_load_session_path;
99 static pid_t ppid; /* Parent PID for --sig-parent option */
100 static pid_t child_ppid; /* Internal parent PID use with daemonize. */
101 static char *rundir;
102 static int lockfile_fd = -1;
103
104 /* Set to 1 when a SIGUSR1 signal is received. */
105 static int recv_child_signal;
106
107 /*
108 * Consumer daemon specific control data. Every value not initialized here is
109 * set to 0 by the static definition.
110 */
111 static struct consumer_data kconsumer_data = {
112 .type = LTTNG_CONSUMER_KERNEL,
113 .err_unix_sock_path = DEFAULT_KCONSUMERD_ERR_SOCK_PATH,
114 .cmd_unix_sock_path = DEFAULT_KCONSUMERD_CMD_SOCK_PATH,
115 .err_sock = -1,
116 .cmd_sock = -1,
117 .channel_monitor_pipe = -1,
118 .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
119 .lock = PTHREAD_MUTEX_INITIALIZER,
120 .cond = PTHREAD_COND_INITIALIZER,
121 .cond_mutex = PTHREAD_MUTEX_INITIALIZER,
122 };
123 static struct consumer_data ustconsumer64_data = {
124 .type = LTTNG_CONSUMER64_UST,
125 .err_unix_sock_path = DEFAULT_USTCONSUMERD64_ERR_SOCK_PATH,
126 .cmd_unix_sock_path = DEFAULT_USTCONSUMERD64_CMD_SOCK_PATH,
127 .err_sock = -1,
128 .cmd_sock = -1,
129 .channel_monitor_pipe = -1,
130 .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
131 .lock = PTHREAD_MUTEX_INITIALIZER,
132 .cond = PTHREAD_COND_INITIALIZER,
133 .cond_mutex = PTHREAD_MUTEX_INITIALIZER,
134 };
135 static struct consumer_data ustconsumer32_data = {
136 .type = LTTNG_CONSUMER32_UST,
137 .err_unix_sock_path = DEFAULT_USTCONSUMERD32_ERR_SOCK_PATH,
138 .cmd_unix_sock_path = DEFAULT_USTCONSUMERD32_CMD_SOCK_PATH,
139 .err_sock = -1,
140 .cmd_sock = -1,
141 .channel_monitor_pipe = -1,
142 .pid_mutex = PTHREAD_MUTEX_INITIALIZER,
143 .lock = PTHREAD_MUTEX_INITIALIZER,
144 .cond = PTHREAD_COND_INITIALIZER,
145 .cond_mutex = PTHREAD_MUTEX_INITIALIZER,
146 };
147
148 /* Command line options */
149 static const struct option long_options[] = {
150 { "client-sock", required_argument, 0, 'c' },
151 { "apps-sock", required_argument, 0, 'a' },
152 { "kconsumerd-cmd-sock", required_argument, 0, '\0' },
153 { "kconsumerd-err-sock", required_argument, 0, '\0' },
154 { "ustconsumerd32-cmd-sock", required_argument, 0, '\0' },
155 { "ustconsumerd32-err-sock", required_argument, 0, '\0' },
156 { "ustconsumerd64-cmd-sock", required_argument, 0, '\0' },
157 { "ustconsumerd64-err-sock", required_argument, 0, '\0' },
158 { "consumerd32-path", required_argument, 0, '\0' },
159 { "consumerd32-libdir", required_argument, 0, '\0' },
160 { "consumerd64-path", required_argument, 0, '\0' },
161 { "consumerd64-libdir", required_argument, 0, '\0' },
162 { "daemonize", no_argument, 0, 'd' },
163 { "background", no_argument, 0, 'b' },
164 { "sig-parent", no_argument, 0, 'S' },
165 { "help", no_argument, 0, 'h' },
166 { "group", required_argument, 0, 'g' },
167 { "version", no_argument, 0, 'V' },
168 { "quiet", no_argument, 0, 'q' },
169 { "verbose", no_argument, 0, 'v' },
170 { "verbose-consumer", no_argument, 0, '\0' },
171 { "no-kernel", no_argument, 0, '\0' },
172 { "pidfile", required_argument, 0, 'p' },
173 { "agent-tcp-port", required_argument, 0, '\0' },
174 { "config", required_argument, 0, 'f' },
175 { "load", required_argument, 0, 'l' },
176 { "kmod-probes", required_argument, 0, '\0' },
177 { "extra-kmod-probes", required_argument, 0, '\0' },
178 { NULL, 0, 0, 0 }
179 };
180
181 /* Command line options to ignore from configuration file */
182 static const char *config_ignore_options[] = { "help", "version", "config" };
183
184 /* Shared between threads */
185 static int dispatch_thread_exit;
186
187 /* Global application Unix socket path */
188 static char apps_unix_sock_path[PATH_MAX];
189 /* Global client Unix socket path */
190 static char client_unix_sock_path[PATH_MAX];
191 /* global wait shm path for UST */
192 static char wait_shm_path[PATH_MAX];
193 /* Global health check unix path */
194 static char health_unix_sock_path[PATH_MAX];
195
196 /* Sockets and FDs */
197 static int client_sock = -1;
198 static int apps_sock = -1;
199 int kernel_tracer_fd = -1;
200 static int kernel_poll_pipe[2] = { -1, -1 };
201
202 /*
203 * Quit pipe for all threads. This permits a single cancellation point
204 * for all threads when receiving an event on the pipe.
205 */
206 static int thread_quit_pipe[2] = { -1, -1 };
207
208 /*
209 * This pipe is used to inform the thread managing application communication
210 * that a command is queued and ready to be processed.
211 */
212 static int apps_cmd_pipe[2] = { -1, -1 };
213
214 int apps_cmd_notify_pipe[2] = { -1, -1 };
215
216 /* Pthread, Mutexes and Semaphores */
217 static pthread_t apps_thread;
218 static pthread_t apps_notify_thread;
219 static pthread_t reg_apps_thread;
220 static pthread_t client_thread;
221 static pthread_t kernel_thread;
222 static pthread_t dispatch_thread;
223 static pthread_t health_thread;
224 static pthread_t ht_cleanup_thread;
225 static pthread_t agent_reg_thread;
226 static pthread_t load_session_thread;
227 static pthread_t notification_thread;
228
229 /*
230 * UST registration command queue. This queue is tied with a futex and uses a N
231 * wakers / 1 waiter implemented and detailed in futex.c/.h
232 *
233 * The thread_registration_apps and thread_dispatch_ust_registration uses this
234 * queue along with the wait/wake scheme. The thread_manage_apps receives down
235 * the line new application socket and monitors it for any I/O error or clean
236 * close that triggers an unregistration of the application.
237 */
238 static struct ust_cmd_queue ust_cmd_queue;
239
240 /*
241 * Pointer initialized before thread creation.
242 *
243 * This points to the tracing session list containing the session count and a
244 * mutex lock. The lock MUST be taken if you iterate over the list. The lock
245 * MUST NOT be taken if you call a public function in session.c.
246 *
247 * The lock is nested inside the structure: session_list_ptr->lock. Please use
248 * session_lock_list and session_unlock_list for lock acquisition.
249 */
250 static struct ltt_session_list *session_list_ptr;
251
252 int ust_consumerd64_fd = -1;
253 int ust_consumerd32_fd = -1;
254
255 static const char *consumerd32_bin = CONFIG_CONSUMERD32_BIN;
256 static const char *consumerd64_bin = CONFIG_CONSUMERD64_BIN;
257 static const char *consumerd32_libdir = CONFIG_CONSUMERD32_LIBDIR;
258 static const char *consumerd64_libdir = CONFIG_CONSUMERD64_LIBDIR;
259 static int consumerd32_bin_override;
260 static int consumerd64_bin_override;
261 static int consumerd32_libdir_override;
262 static int consumerd64_libdir_override;
263
264 static const char *module_proc_lttng = "/proc/lttng";
265
266 /*
267 * Consumer daemon state which is changed when spawning it, killing it or in
268 * case of a fatal error.
269 */
270 enum consumerd_state {
271 CONSUMER_STARTED = 1,
272 CONSUMER_STOPPED = 2,
273 CONSUMER_ERROR = 3,
274 };
275
276 /*
277 * This consumer daemon state is used to validate if a client command will be
278 * able to reach the consumer. If not, the client is informed. For instance,
279 * doing a "lttng start" when the consumer state is set to ERROR will return an
280 * error to the client.
281 *
282 * The following example shows a possible race condition of this scheme:
283 *
284 * consumer thread error happens
285 * client cmd arrives
286 * client cmd checks state -> still OK
287 * consumer thread exit, sets error
288 * client cmd try to talk to consumer
289 * ...
290 *
291 * However, since the consumer is a different daemon, we have no way of making
292 * sure the command will reach it safely even with this state flag. This is why
293 * we consider that up to the state validation during command processing, the
294 * command is safe. After that, we can not guarantee the correctness of the
295 * client request vis-a-vis the consumer.
296 */
297 static enum consumerd_state ust_consumerd_state;
298 static enum consumerd_state kernel_consumerd_state;
299
300 /*
301 * Socket timeout for receiving and sending in seconds.
302 */
303 static int app_socket_timeout;
304
305 /* Set in main() with the current page size. */
306 long page_size;
307
308 /* Application health monitoring */
309 struct health_app *health_sessiond;
310
311 /* Agent TCP port for registration. Used by the agent thread. */
312 unsigned int agent_tcp_port = DEFAULT_AGENT_TCP_PORT;
313
314 /* Am I root or not. */
315 int is_root; /* Set to 1 if the daemon is running as root */
316
317 const char * const config_section_name = "sessiond";
318
319 /* Load session thread information to operate. */
320 struct load_session_thread_data *load_info;
321
322 /* Notification thread handle. */
323 struct notification_thread_handle *notification_thread_handle;
324
325 /* Global hash tables */
326 struct lttng_ht *agent_apps_ht_by_sock = NULL;
327
328 /*
329 * Whether sessiond is ready for commands/notification channel/health check
330 * requests.
331 * NR_LTTNG_SESSIOND_READY must match the number of calls to
332 * sessiond_notify_ready().
333 */
334 #define NR_LTTNG_SESSIOND_READY 4
335 int lttng_sessiond_ready = NR_LTTNG_SESSIOND_READY;
336
337 int sessiond_check_thread_quit_pipe(int fd, uint32_t events)
338 {
339 return (fd == thread_quit_pipe[0] && (events & LPOLLIN)) ? 1 : 0;
340 }
341
342 /* Notify parents that we are ready for cmd and health check */
343 LTTNG_HIDDEN
344 void sessiond_notify_ready(void)
345 {
346 if (uatomic_sub_return(&lttng_sessiond_ready, 1) == 0) {
347 /*
348 * Notify parent pid that we are ready to accept command
349 * for client side. This ppid is the one from the
350 * external process that spawned us.
351 */
352 if (opt_sig_parent) {
353 kill(ppid, SIGUSR1);
354 }
355
356 /*
357 * Notify the parent of the fork() process that we are
358 * ready.
359 */
360 if (opt_daemon || opt_background) {
361 kill(child_ppid, SIGUSR1);
362 }
363 }
364 }
365
366 static
367 void setup_consumerd_path(void)
368 {
369 const char *bin, *libdir;
370
371 /*
372 * Allow INSTALL_BIN_PATH to be used as a target path for the
373 * native architecture size consumer if CONFIG_CONSUMER*_PATH
374 * has not been defined.
375 */
376 #if (CAA_BITS_PER_LONG == 32)
377 if (!consumerd32_bin[0]) {
378 consumerd32_bin = INSTALL_BIN_PATH "/" CONSUMERD_FILE;
379 }
380 if (!consumerd32_libdir[0]) {
381 consumerd32_libdir = INSTALL_LIB_PATH;
382 }
383 #elif (CAA_BITS_PER_LONG == 64)
384 if (!consumerd64_bin[0]) {
385 consumerd64_bin = INSTALL_BIN_PATH "/" CONSUMERD_FILE;
386 }
387 if (!consumerd64_libdir[0]) {
388 consumerd64_libdir = INSTALL_LIB_PATH;
389 }
390 #else
391 #error "Unknown bitness"
392 #endif
393
394 /*
395 * runtime env. var. overrides the build default.
396 */
397 bin = lttng_secure_getenv("LTTNG_CONSUMERD32_BIN");
398 if (bin) {
399 consumerd32_bin = bin;
400 }
401 bin = lttng_secure_getenv("LTTNG_CONSUMERD64_BIN");
402 if (bin) {
403 consumerd64_bin = bin;
404 }
405 libdir = lttng_secure_getenv("LTTNG_CONSUMERD32_LIBDIR");
406 if (libdir) {
407 consumerd32_libdir = libdir;
408 }
409 libdir = lttng_secure_getenv("LTTNG_CONSUMERD64_LIBDIR");
410 if (libdir) {
411 consumerd64_libdir = libdir;
412 }
413 }
414
415 static
416 int __sessiond_set_thread_pollset(struct lttng_poll_event *events, size_t size,
417 int *a_pipe)
418 {
419 int ret;
420
421 assert(events);
422
423 ret = lttng_poll_create(events, size, LTTNG_CLOEXEC);
424 if (ret < 0) {
425 goto error;
426 }
427
428 /* Add quit pipe */
429 ret = lttng_poll_add(events, a_pipe[0], LPOLLIN | LPOLLERR);
430 if (ret < 0) {
431 goto error;
432 }
433
434 return 0;
435
436 error:
437 return ret;
438 }
439
440 /*
441 * Create a poll set with O_CLOEXEC and add the thread quit pipe to the set.
442 */
443 int sessiond_set_thread_pollset(struct lttng_poll_event *events, size_t size)
444 {
445 return __sessiond_set_thread_pollset(events, size, thread_quit_pipe);
446 }
447
448 /*
449 * Init thread quit pipe.
450 *
451 * Return -1 on error or 0 if all pipes are created.
452 */
453 static int __init_thread_quit_pipe(int *a_pipe)
454 {
455 int ret, i;
456
457 ret = pipe(a_pipe);
458 if (ret < 0) {
459 PERROR("thread quit pipe");
460 goto error;
461 }
462
463 for (i = 0; i < 2; i++) {
464 ret = fcntl(a_pipe[i], F_SETFD, FD_CLOEXEC);
465 if (ret < 0) {
466 PERROR("fcntl");
467 goto error;
468 }
469 }
470
471 error:
472 return ret;
473 }
474
475 static int init_thread_quit_pipe(void)
476 {
477 return __init_thread_quit_pipe(thread_quit_pipe);
478 }
479
480 /*
481 * Stop all threads by closing the thread quit pipe.
482 */
483 static void stop_threads(void)
484 {
485 int ret;
486
487 /* Stopping all threads */
488 DBG("Terminating all threads");
489 ret = notify_thread_pipe(thread_quit_pipe[1]);
490 if (ret < 0) {
491 ERR("write error on thread quit pipe");
492 }
493
494 /* Dispatch thread */
495 CMM_STORE_SHARED(dispatch_thread_exit, 1);
496 futex_nto1_wake(&ust_cmd_queue.futex);
497 }
498
499 /*
500 * Close every consumer sockets.
501 */
502 static void close_consumer_sockets(void)
503 {
504 int ret;
505
506 if (kconsumer_data.err_sock >= 0) {
507 ret = close(kconsumer_data.err_sock);
508 if (ret < 0) {
509 PERROR("kernel consumer err_sock close");
510 }
511 }
512 if (ustconsumer32_data.err_sock >= 0) {
513 ret = close(ustconsumer32_data.err_sock);
514 if (ret < 0) {
515 PERROR("UST consumerd32 err_sock close");
516 }
517 }
518 if (ustconsumer64_data.err_sock >= 0) {
519 ret = close(ustconsumer64_data.err_sock);
520 if (ret < 0) {
521 PERROR("UST consumerd64 err_sock close");
522 }
523 }
524 if (kconsumer_data.cmd_sock >= 0) {
525 ret = close(kconsumer_data.cmd_sock);
526 if (ret < 0) {
527 PERROR("kernel consumer cmd_sock close");
528 }
529 }
530 if (ustconsumer32_data.cmd_sock >= 0) {
531 ret = close(ustconsumer32_data.cmd_sock);
532 if (ret < 0) {
533 PERROR("UST consumerd32 cmd_sock close");
534 }
535 }
536 if (ustconsumer64_data.cmd_sock >= 0) {
537 ret = close(ustconsumer64_data.cmd_sock);
538 if (ret < 0) {
539 PERROR("UST consumerd64 cmd_sock close");
540 }
541 }
542 if (kconsumer_data.channel_monitor_pipe >= 0) {
543 ret = close(kconsumer_data.channel_monitor_pipe);
544 if (ret < 0) {
545 PERROR("kernel consumer channel monitor pipe close");
546 }
547 }
548 if (ustconsumer32_data.channel_monitor_pipe >= 0) {
549 ret = close(ustconsumer32_data.channel_monitor_pipe);
550 if (ret < 0) {
551 PERROR("UST consumerd32 channel monitor pipe close");
552 }
553 }
554 if (ustconsumer64_data.channel_monitor_pipe >= 0) {
555 ret = close(ustconsumer64_data.channel_monitor_pipe);
556 if (ret < 0) {
557 PERROR("UST consumerd64 channel monitor pipe close");
558 }
559 }
560 }
561
562 /*
563 * Generate the full lock file path using the rundir.
564 *
565 * Return the snprintf() return value thus a negative value is an error.
566 */
567 static int generate_lock_file_path(char *path, size_t len)
568 {
569 int ret;
570
571 assert(path);
572 assert(rundir);
573
574 /* Build lockfile path from rundir. */
575 ret = snprintf(path, len, "%s/" DEFAULT_LTTNG_SESSIOND_LOCKFILE, rundir);
576 if (ret < 0) {
577 PERROR("snprintf lockfile path");
578 }
579
580 return ret;
581 }
582
583 /*
584 * Wait on consumer process termination.
585 *
586 * Need to be called with the consumer data lock held or from a context
587 * ensuring no concurrent access to data (e.g: cleanup).
588 */
589 static void wait_consumer(struct consumer_data *consumer_data)
590 {
591 pid_t ret;
592 int status;
593
594 if (consumer_data->pid <= 0) {
595 return;
596 }
597
598 DBG("Waiting for complete teardown of consumerd (PID: %d)",
599 consumer_data->pid);
600 ret = waitpid(consumer_data->pid, &status, 0);
601 if (ret == -1) {
602 PERROR("consumerd waitpid pid: %d", consumer_data->pid)
603 } else if (!WIFEXITED(status)) {
604 ERR("consumerd termination with error: %d",
605 WEXITSTATUS(ret));
606 }
607 consumer_data->pid = 0;
608 }
609
610 /*
611 * Cleanup the session daemon's data structures.
612 */
613 static void sessiond_cleanup(void)
614 {
615 int ret;
616 struct ltt_session *sess, *stmp;
617 char path[PATH_MAX];
618
619 DBG("Cleanup sessiond");
620
621 /*
622 * Close the thread quit pipe. It has already done its job,
623 * since we are now called.
624 */
625 utils_close_pipe(thread_quit_pipe);
626
627 /*
628 * If opt_pidfile is undefined, the default file will be wiped when
629 * removing the rundir.
630 */
631 if (opt_pidfile) {
632 ret = remove(opt_pidfile);
633 if (ret < 0) {
634 PERROR("remove pidfile %s", opt_pidfile);
635 }
636 }
637
638 DBG("Removing sessiond and consumerd content of directory %s", rundir);
639
640 /* sessiond */
641 snprintf(path, PATH_MAX,
642 "%s/%s",
643 rundir, DEFAULT_LTTNG_SESSIOND_PIDFILE);
644 DBG("Removing %s", path);
645 (void) unlink(path);
646
647 snprintf(path, PATH_MAX, "%s/%s", rundir,
648 DEFAULT_LTTNG_SESSIOND_AGENTPORT_FILE);
649 DBG("Removing %s", path);
650 (void) unlink(path);
651
652 /* kconsumerd */
653 snprintf(path, PATH_MAX,
654 DEFAULT_KCONSUMERD_ERR_SOCK_PATH,
655 rundir);
656 DBG("Removing %s", path);
657 (void) unlink(path);
658
659 snprintf(path, PATH_MAX,
660 DEFAULT_KCONSUMERD_PATH,
661 rundir);
662 DBG("Removing directory %s", path);
663 (void) rmdir(path);
664
665 /* ust consumerd 32 */
666 snprintf(path, PATH_MAX,
667 DEFAULT_USTCONSUMERD32_ERR_SOCK_PATH,
668 rundir);
669 DBG("Removing %s", path);
670 (void) unlink(path);
671
672 snprintf(path, PATH_MAX,
673 DEFAULT_USTCONSUMERD32_PATH,
674 rundir);
675 DBG("Removing directory %s", path);
676 (void) rmdir(path);
677
678 /* ust consumerd 64 */
679 snprintf(path, PATH_MAX,
680 DEFAULT_USTCONSUMERD64_ERR_SOCK_PATH,
681 rundir);
682 DBG("Removing %s", path);
683 (void) unlink(path);
684
685 snprintf(path, PATH_MAX,
686 DEFAULT_USTCONSUMERD64_PATH,
687 rundir);
688 DBG("Removing directory %s", path);
689 (void) rmdir(path);
690
691 DBG("Cleaning up all sessions");
692
693 /* Destroy session list mutex */
694 if (session_list_ptr != NULL) {
695 pthread_mutex_destroy(&session_list_ptr->lock);
696
697 /* Cleanup ALL session */
698 cds_list_for_each_entry_safe(sess, stmp,
699 &session_list_ptr->head, list) {
700 cmd_destroy_session(sess, kernel_poll_pipe[1]);
701 }
702 }
703
704 wait_consumer(&kconsumer_data);
705 wait_consumer(&ustconsumer64_data);
706 wait_consumer(&ustconsumer32_data);
707
708 DBG("Cleaning up all agent apps");
709 agent_app_ht_clean();
710
711 DBG("Closing all UST sockets");
712 ust_app_clean_list();
713 buffer_reg_destroy_registries();
714
715 if (is_root && !opt_no_kernel) {
716 DBG2("Closing kernel fd");
717 if (kernel_tracer_fd >= 0) {
718 ret = close(kernel_tracer_fd);
719 if (ret) {
720 PERROR("close");
721 }
722 }
723 DBG("Unloading kernel modules");
724 modprobe_remove_lttng_all();
725 free(syscall_table);
726 }
727
728 close_consumer_sockets();
729
730 if (load_info) {
731 load_session_destroy_data(load_info);
732 free(load_info);
733 }
734
735 /*
736 * Cleanup lock file by deleting it and finaly closing it which will
737 * release the file system lock.
738 */
739 if (lockfile_fd >= 0) {
740 char lockfile_path[PATH_MAX];
741
742 ret = generate_lock_file_path(lockfile_path,
743 sizeof(lockfile_path));
744 if (ret > 0) {
745 ret = remove(lockfile_path);
746 if (ret < 0) {
747 PERROR("remove lock file");
748 }
749 ret = close(lockfile_fd);
750 if (ret < 0) {
751 PERROR("close lock file");
752 }
753 }
754 }
755
756 /*
757 * We do NOT rmdir rundir because there are other processes
758 * using it, for instance lttng-relayd, which can start in
759 * parallel with this teardown.
760 */
761
762 free(rundir);
763 }
764
765 /*
766 * Cleanup the daemon's option data structures.
767 */
768 static void sessiond_cleanup_options(void)
769 {
770 DBG("Cleaning up options");
771
772 /*
773 * If the override option is set, the pointer points to a *non* const
774 * thus freeing it even though the variable type is set to const.
775 */
776 if (tracing_group_name_override) {
777 free((void *) tracing_group_name);
778 }
779 if (consumerd32_bin_override) {
780 free((void *) consumerd32_bin);
781 }
782 if (consumerd64_bin_override) {
783 free((void *) consumerd64_bin);
784 }
785 if (consumerd32_libdir_override) {
786 free((void *) consumerd32_libdir);
787 }
788 if (consumerd64_libdir_override) {
789 free((void *) consumerd64_libdir);
790 }
791
792 free(opt_pidfile);
793 free(opt_load_session_path);
794 free(kmod_probes_list);
795 free(kmod_extra_probes_list);
796
797 run_as_destroy_worker();
798 }
799
800 /*
801 * Send data on a unix socket using the liblttsessiondcomm API.
802 *
803 * Return lttcomm error code.
804 */
805 static int send_unix_sock(int sock, void *buf, size_t len)
806 {
807 /* Check valid length */
808 if (len == 0) {
809 return -1;
810 }
811
812 return lttcomm_send_unix_sock(sock, buf, len);
813 }
814
815 /*
816 * Free memory of a command context structure.
817 */
818 static void clean_command_ctx(struct command_ctx **cmd_ctx)
819 {
820 DBG("Clean command context structure");
821 if (*cmd_ctx) {
822 if ((*cmd_ctx)->llm) {
823 free((*cmd_ctx)->llm);
824 }
825 if ((*cmd_ctx)->lsm) {
826 free((*cmd_ctx)->lsm);
827 }
828 free(*cmd_ctx);
829 *cmd_ctx = NULL;
830 }
831 }
832
833 /*
834 * Notify UST applications using the shm mmap futex.
835 */
836 static int notify_ust_apps(int active)
837 {
838 char *wait_shm_mmap;
839
840 DBG("Notifying applications of session daemon state: %d", active);
841
842 /* See shm.c for this call implying mmap, shm and futex calls */
843 wait_shm_mmap = shm_ust_get_mmap(wait_shm_path, is_root);
844 if (wait_shm_mmap == NULL) {
845 goto error;
846 }
847
848 /* Wake waiting process */
849 futex_wait_update((int32_t *) wait_shm_mmap, active);
850
851 /* Apps notified successfully */
852 return 0;
853
854 error:
855 return -1;
856 }
857
858 /*
859 * Setup the outgoing data buffer for the response (llm) by allocating the
860 * right amount of memory and copying the original information from the lsm
861 * structure.
862 *
863 * Return 0 on success, negative value on error.
864 */
865 static int setup_lttng_msg(struct command_ctx *cmd_ctx,
866 const void *payload_buf, size_t payload_len,
867 const void *cmd_header_buf, size_t cmd_header_len)
868 {
869 int ret = 0;
870 const size_t header_len = sizeof(struct lttcomm_lttng_msg);
871 const size_t cmd_header_offset = header_len;
872 const size_t payload_offset = cmd_header_offset + cmd_header_len;
873 const size_t total_msg_size = header_len + cmd_header_len + payload_len;
874
875 cmd_ctx->llm = zmalloc(total_msg_size);
876
877 if (cmd_ctx->llm == NULL) {
878 PERROR("zmalloc");
879 ret = -ENOMEM;
880 goto end;
881 }
882
883 /* Copy common data */
884 cmd_ctx->llm->cmd_type = cmd_ctx->lsm->cmd_type;
885 cmd_ctx->llm->pid = cmd_ctx->lsm->domain.attr.pid;
886 cmd_ctx->llm->cmd_header_size = cmd_header_len;
887 cmd_ctx->llm->data_size = payload_len;
888 cmd_ctx->lttng_msg_size = total_msg_size;
889
890 /* Copy command header */
891 if (cmd_header_len) {
892 memcpy(((uint8_t *) cmd_ctx->llm) + cmd_header_offset, cmd_header_buf,
893 cmd_header_len);
894 }
895
896 /* Copy payload */
897 if (payload_len) {
898 memcpy(((uint8_t *) cmd_ctx->llm) + payload_offset, payload_buf,
899 payload_len);
900 }
901
902 end:
903 return ret;
904 }
905
906 /*
907 * Version of setup_lttng_msg() without command header.
908 */
909 static int setup_lttng_msg_no_cmd_header(struct command_ctx *cmd_ctx,
910 void *payload_buf, size_t payload_len)
911 {
912 return setup_lttng_msg(cmd_ctx, payload_buf, payload_len, NULL, 0);
913 }
914 /*
915 * Update the kernel poll set of all channel fd available over all tracing
916 * session. Add the wakeup pipe at the end of the set.
917 */
918 static int update_kernel_poll(struct lttng_poll_event *events)
919 {
920 int ret;
921 struct ltt_session *session;
922 struct ltt_kernel_channel *channel;
923
924 DBG("Updating kernel poll set");
925
926 session_lock_list();
927 cds_list_for_each_entry(session, &session_list_ptr->head, list) {
928 session_lock(session);
929 if (session->kernel_session == NULL) {
930 session_unlock(session);
931 continue;
932 }
933
934 cds_list_for_each_entry(channel,
935 &session->kernel_session->channel_list.head, list) {
936 /* Add channel fd to the kernel poll set */
937 ret = lttng_poll_add(events, channel->fd, LPOLLIN | LPOLLRDNORM);
938 if (ret < 0) {
939 session_unlock(session);
940 goto error;
941 }
942 DBG("Channel fd %d added to kernel set", channel->fd);
943 }
944 session_unlock(session);
945 }
946 session_unlock_list();
947
948 return 0;
949
950 error:
951 session_unlock_list();
952 return -1;
953 }
954
955 /*
956 * Find the channel fd from 'fd' over all tracing session. When found, check
957 * for new channel stream and send those stream fds to the kernel consumer.
958 *
959 * Useful for CPU hotplug feature.
960 */
961 static int update_kernel_stream(struct consumer_data *consumer_data, int fd)
962 {
963 int ret = 0;
964 struct ltt_session *session;
965 struct ltt_kernel_session *ksess;
966 struct ltt_kernel_channel *channel;
967
968 DBG("Updating kernel streams for channel fd %d", fd);
969
970 session_lock_list();
971 cds_list_for_each_entry(session, &session_list_ptr->head, list) {
972 session_lock(session);
973 if (session->kernel_session == NULL) {
974 session_unlock(session);
975 continue;
976 }
977 ksess = session->kernel_session;
978
979 cds_list_for_each_entry(channel,
980 &ksess->channel_list.head, list) {
981 struct lttng_ht_iter iter;
982 struct consumer_socket *socket;
983
984 if (channel->fd != fd) {
985 continue;
986 }
987 DBG("Channel found, updating kernel streams");
988 ret = kernel_open_channel_stream(channel);
989 if (ret < 0) {
990 goto error;
991 }
992 /* Update the stream global counter */
993 ksess->stream_count_global += ret;
994
995 /*
996 * Have we already sent fds to the consumer? If yes, it
997 * means that tracing is started so it is safe to send
998 * our updated stream fds.
999 */
1000 if (ksess->consumer_fds_sent != 1
1001 || ksess->consumer == NULL) {
1002 ret = -1;
1003 goto error;
1004 }
1005
1006 rcu_read_lock();
1007 cds_lfht_for_each_entry(ksess->consumer->socks->ht,
1008 &iter.iter, socket, node.node) {
1009 pthread_mutex_lock(socket->lock);
1010 ret = kernel_consumer_send_channel_stream(socket,
1011 channel, ksess,
1012 session->output_traces ? 1 : 0);
1013 pthread_mutex_unlock(socket->lock);
1014 if (ret < 0) {
1015 rcu_read_unlock();
1016 goto error;
1017 }
1018 }
1019 rcu_read_unlock();
1020 }
1021 session_unlock(session);
1022 }
1023 session_unlock_list();
1024 return ret;
1025
1026 error:
1027 session_unlock(session);
1028 session_unlock_list();
1029 return ret;
1030 }
1031
1032 /*
1033 * For each tracing session, update newly registered apps. The session list
1034 * lock MUST be acquired before calling this.
1035 */
1036 static void update_ust_app(int app_sock)
1037 {
1038 struct ltt_session *sess, *stmp;
1039
1040 /* Consumer is in an ERROR state. Stop any application update. */
1041 if (uatomic_read(&ust_consumerd_state) == CONSUMER_ERROR) {
1042 /* Stop the update process since the consumer is dead. */
1043 return;
1044 }
1045
1046 /* For all tracing session(s) */
1047 cds_list_for_each_entry_safe(sess, stmp, &session_list_ptr->head, list) {
1048 struct ust_app *app;
1049
1050 session_lock(sess);
1051 if (!sess->ust_session) {
1052 goto unlock_session;
1053 }
1054
1055 rcu_read_lock();
1056 assert(app_sock >= 0);
1057 app = ust_app_find_by_sock(app_sock);
1058 if (app == NULL) {
1059 /*
1060 * Application can be unregistered before so
1061 * this is possible hence simply stopping the
1062 * update.
1063 */
1064 DBG3("UST app update failed to find app sock %d",
1065 app_sock);
1066 goto unlock_rcu;
1067 }
1068 ust_app_global_update(sess->ust_session, app);
1069 unlock_rcu:
1070 rcu_read_unlock();
1071 unlock_session:
1072 session_unlock(sess);
1073 }
1074 }
1075
1076 /*
1077 * This thread manage event coming from the kernel.
1078 *
1079 * Features supported in this thread:
1080 * -) CPU Hotplug
1081 */
1082 static void *thread_manage_kernel(void *data)
1083 {
1084 int ret, i, pollfd, update_poll_flag = 1, err = -1;
1085 uint32_t revents, nb_fd;
1086 char tmp;
1087 struct lttng_poll_event events;
1088
1089 DBG("[thread] Thread manage kernel started");
1090
1091 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_KERNEL);
1092
1093 /*
1094 * This first step of the while is to clean this structure which could free
1095 * non NULL pointers so initialize it before the loop.
1096 */
1097 lttng_poll_init(&events);
1098
1099 if (testpoint(sessiond_thread_manage_kernel)) {
1100 goto error_testpoint;
1101 }
1102
1103 health_code_update();
1104
1105 if (testpoint(sessiond_thread_manage_kernel_before_loop)) {
1106 goto error_testpoint;
1107 }
1108
1109 while (1) {
1110 health_code_update();
1111
1112 if (update_poll_flag == 1) {
1113 /* Clean events object. We are about to populate it again. */
1114 lttng_poll_clean(&events);
1115
1116 ret = sessiond_set_thread_pollset(&events, 2);
1117 if (ret < 0) {
1118 goto error_poll_create;
1119 }
1120
1121 ret = lttng_poll_add(&events, kernel_poll_pipe[0], LPOLLIN);
1122 if (ret < 0) {
1123 goto error;
1124 }
1125
1126 /* This will add the available kernel channel if any. */
1127 ret = update_kernel_poll(&events);
1128 if (ret < 0) {
1129 goto error;
1130 }
1131 update_poll_flag = 0;
1132 }
1133
1134 DBG("Thread kernel polling");
1135
1136 /* Poll infinite value of time */
1137 restart:
1138 health_poll_entry();
1139 ret = lttng_poll_wait(&events, -1);
1140 DBG("Thread kernel return from poll on %d fds",
1141 LTTNG_POLL_GETNB(&events));
1142 health_poll_exit();
1143 if (ret < 0) {
1144 /*
1145 * Restart interrupted system call.
1146 */
1147 if (errno == EINTR) {
1148 goto restart;
1149 }
1150 goto error;
1151 } else if (ret == 0) {
1152 /* Should not happen since timeout is infinite */
1153 ERR("Return value of poll is 0 with an infinite timeout.\n"
1154 "This should not have happened! Continuing...");
1155 continue;
1156 }
1157
1158 nb_fd = ret;
1159
1160 for (i = 0; i < nb_fd; i++) {
1161 /* Fetch once the poll data */
1162 revents = LTTNG_POLL_GETEV(&events, i);
1163 pollfd = LTTNG_POLL_GETFD(&events, i);
1164
1165 health_code_update();
1166
1167 if (!revents) {
1168 /* No activity for this FD (poll implementation). */
1169 continue;
1170 }
1171
1172 /* Thread quit pipe has been closed. Killing thread. */
1173 ret = sessiond_check_thread_quit_pipe(pollfd, revents);
1174 if (ret) {
1175 err = 0;
1176 goto exit;
1177 }
1178
1179 /* Check for data on kernel pipe */
1180 if (revents & LPOLLIN) {
1181 if (pollfd == kernel_poll_pipe[0]) {
1182 (void) lttng_read(kernel_poll_pipe[0],
1183 &tmp, 1);
1184 /*
1185 * Ret value is useless here, if this pipe gets any actions an
1186 * update is required anyway.
1187 */
1188 update_poll_flag = 1;
1189 continue;
1190 } else {
1191 /*
1192 * New CPU detected by the kernel. Adding kernel stream to
1193 * kernel session and updating the kernel consumer
1194 */
1195 ret = update_kernel_stream(&kconsumer_data, pollfd);
1196 if (ret < 0) {
1197 continue;
1198 }
1199 break;
1200 }
1201 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1202 update_poll_flag = 1;
1203 continue;
1204 } else {
1205 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1206 goto error;
1207 }
1208 }
1209 }
1210
1211 exit:
1212 error:
1213 lttng_poll_clean(&events);
1214 error_poll_create:
1215 error_testpoint:
1216 utils_close_pipe(kernel_poll_pipe);
1217 kernel_poll_pipe[0] = kernel_poll_pipe[1] = -1;
1218 if (err) {
1219 health_error();
1220 ERR("Health error occurred in %s", __func__);
1221 WARN("Kernel thread died unexpectedly. "
1222 "Kernel tracing can continue but CPU hotplug is disabled.");
1223 }
1224 health_unregister(health_sessiond);
1225 DBG("Kernel thread dying");
1226 return NULL;
1227 }
1228
1229 /*
1230 * Signal pthread condition of the consumer data that the thread.
1231 */
1232 static void signal_consumer_condition(struct consumer_data *data, int state)
1233 {
1234 pthread_mutex_lock(&data->cond_mutex);
1235
1236 /*
1237 * The state is set before signaling. It can be any value, it's the waiter
1238 * job to correctly interpret this condition variable associated to the
1239 * consumer pthread_cond.
1240 *
1241 * A value of 0 means that the corresponding thread of the consumer data
1242 * was not started. 1 indicates that the thread has started and is ready
1243 * for action. A negative value means that there was an error during the
1244 * thread bootstrap.
1245 */
1246 data->consumer_thread_is_ready = state;
1247 (void) pthread_cond_signal(&data->cond);
1248
1249 pthread_mutex_unlock(&data->cond_mutex);
1250 }
1251
1252 /*
1253 * This thread manage the consumer error sent back to the session daemon.
1254 */
1255 static void *thread_manage_consumer(void *data)
1256 {
1257 int sock = -1, i, ret, pollfd, err = -1, should_quit = 0;
1258 uint32_t revents, nb_fd;
1259 enum lttcomm_return_code code;
1260 struct lttng_poll_event events;
1261 struct consumer_data *consumer_data = data;
1262 struct consumer_socket *cmd_socket_wrapper = NULL;
1263
1264 DBG("[thread] Manage consumer started");
1265
1266 rcu_register_thread();
1267 rcu_thread_online();
1268
1269 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_CONSUMER);
1270
1271 health_code_update();
1272
1273 /*
1274 * Pass 3 as size here for the thread quit pipe, consumerd_err_sock and the
1275 * metadata_sock. Nothing more will be added to this poll set.
1276 */
1277 ret = sessiond_set_thread_pollset(&events, 3);
1278 if (ret < 0) {
1279 goto error_poll;
1280 }
1281
1282 /*
1283 * The error socket here is already in a listening state which was done
1284 * just before spawning this thread to avoid a race between the consumer
1285 * daemon exec trying to connect and the listen() call.
1286 */
1287 ret = lttng_poll_add(&events, consumer_data->err_sock, LPOLLIN | LPOLLRDHUP);
1288 if (ret < 0) {
1289 goto error;
1290 }
1291
1292 health_code_update();
1293
1294 /* Infinite blocking call, waiting for transmission */
1295 restart:
1296 health_poll_entry();
1297
1298 if (testpoint(sessiond_thread_manage_consumer)) {
1299 goto error;
1300 }
1301
1302 ret = lttng_poll_wait(&events, -1);
1303 health_poll_exit();
1304 if (ret < 0) {
1305 /*
1306 * Restart interrupted system call.
1307 */
1308 if (errno == EINTR) {
1309 goto restart;
1310 }
1311 goto error;
1312 }
1313
1314 nb_fd = ret;
1315
1316 for (i = 0; i < nb_fd; i++) {
1317 /* Fetch once the poll data */
1318 revents = LTTNG_POLL_GETEV(&events, i);
1319 pollfd = LTTNG_POLL_GETFD(&events, i);
1320
1321 health_code_update();
1322
1323 if (!revents) {
1324 /* No activity for this FD (poll implementation). */
1325 continue;
1326 }
1327
1328 /* Thread quit pipe has been closed. Killing thread. */
1329 ret = sessiond_check_thread_quit_pipe(pollfd, revents);
1330 if (ret) {
1331 err = 0;
1332 goto exit;
1333 }
1334
1335 /* Event on the registration socket */
1336 if (pollfd == consumer_data->err_sock) {
1337 if (revents & LPOLLIN) {
1338 continue;
1339 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1340 ERR("consumer err socket poll error");
1341 goto error;
1342 } else {
1343 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1344 goto error;
1345 }
1346 }
1347 }
1348
1349 sock = lttcomm_accept_unix_sock(consumer_data->err_sock);
1350 if (sock < 0) {
1351 goto error;
1352 }
1353
1354 /*
1355 * Set the CLOEXEC flag. Return code is useless because either way, the
1356 * show must go on.
1357 */
1358 (void) utils_set_fd_cloexec(sock);
1359
1360 health_code_update();
1361
1362 DBG2("Receiving code from consumer err_sock");
1363
1364 /* Getting status code from kconsumerd */
1365 ret = lttcomm_recv_unix_sock(sock, &code,
1366 sizeof(enum lttcomm_return_code));
1367 if (ret <= 0) {
1368 goto error;
1369 }
1370
1371 health_code_update();
1372 if (code != LTTCOMM_CONSUMERD_COMMAND_SOCK_READY) {
1373 ERR("consumer error when waiting for SOCK_READY : %s",
1374 lttcomm_get_readable_code(-code));
1375 goto error;
1376 }
1377
1378 /* Connect both command and metadata sockets. */
1379 consumer_data->cmd_sock =
1380 lttcomm_connect_unix_sock(
1381 consumer_data->cmd_unix_sock_path);
1382 consumer_data->metadata_fd =
1383 lttcomm_connect_unix_sock(
1384 consumer_data->cmd_unix_sock_path);
1385 if (consumer_data->cmd_sock < 0 || consumer_data->metadata_fd < 0) {
1386 PERROR("consumer connect cmd socket");
1387 /* On error, signal condition and quit. */
1388 signal_consumer_condition(consumer_data, -1);
1389 goto error;
1390 }
1391
1392 consumer_data->metadata_sock.fd_ptr = &consumer_data->metadata_fd;
1393
1394 /* Create metadata socket lock. */
1395 consumer_data->metadata_sock.lock = zmalloc(sizeof(pthread_mutex_t));
1396 if (consumer_data->metadata_sock.lock == NULL) {
1397 PERROR("zmalloc pthread mutex");
1398 goto error;
1399 }
1400 pthread_mutex_init(consumer_data->metadata_sock.lock, NULL);
1401
1402 DBG("Consumer command socket ready (fd: %d", consumer_data->cmd_sock);
1403 DBG("Consumer metadata socket ready (fd: %d)",
1404 consumer_data->metadata_fd);
1405
1406 /*
1407 * Remove the consumerd error sock since we've established a connection.
1408 */
1409 ret = lttng_poll_del(&events, consumer_data->err_sock);
1410 if (ret < 0) {
1411 goto error;
1412 }
1413
1414 /* Add new accepted error socket. */
1415 ret = lttng_poll_add(&events, sock, LPOLLIN | LPOLLRDHUP);
1416 if (ret < 0) {
1417 goto error;
1418 }
1419
1420 /* Add metadata socket that is successfully connected. */
1421 ret = lttng_poll_add(&events, consumer_data->metadata_fd,
1422 LPOLLIN | LPOLLRDHUP);
1423 if (ret < 0) {
1424 goto error;
1425 }
1426
1427 health_code_update();
1428
1429 /*
1430 * Transfer the write-end of the channel monitoring pipe to the
1431 * by issuing a SET_CHANNEL_MONITOR_PIPE command.
1432 */
1433 cmd_socket_wrapper = consumer_allocate_socket(&consumer_data->cmd_sock);
1434 if (!cmd_socket_wrapper) {
1435 goto error;
1436 }
1437
1438 ret = consumer_send_channel_monitor_pipe(cmd_socket_wrapper,
1439 consumer_data->channel_monitor_pipe);
1440 if (ret) {
1441 goto error;
1442 }
1443 /* Discard the socket wrapper as it is no longer needed. */
1444 consumer_destroy_socket(cmd_socket_wrapper);
1445 cmd_socket_wrapper = NULL;
1446
1447 /* The thread is completely initialized, signal that it is ready. */
1448 signal_consumer_condition(consumer_data, 1);
1449
1450 /* Infinite blocking call, waiting for transmission */
1451 restart_poll:
1452 while (1) {
1453 health_code_update();
1454
1455 /* Exit the thread because the thread quit pipe has been triggered. */
1456 if (should_quit) {
1457 /* Not a health error. */
1458 err = 0;
1459 goto exit;
1460 }
1461
1462 health_poll_entry();
1463 ret = lttng_poll_wait(&events, -1);
1464 health_poll_exit();
1465 if (ret < 0) {
1466 /*
1467 * Restart interrupted system call.
1468 */
1469 if (errno == EINTR) {
1470 goto restart_poll;
1471 }
1472 goto error;
1473 }
1474
1475 nb_fd = ret;
1476
1477 for (i = 0; i < nb_fd; i++) {
1478 /* Fetch once the poll data */
1479 revents = LTTNG_POLL_GETEV(&events, i);
1480 pollfd = LTTNG_POLL_GETFD(&events, i);
1481
1482 health_code_update();
1483
1484 if (!revents) {
1485 /* No activity for this FD (poll implementation). */
1486 continue;
1487 }
1488
1489 /*
1490 * Thread quit pipe has been triggered, flag that we should stop
1491 * but continue the current loop to handle potential data from
1492 * consumer.
1493 */
1494 should_quit = sessiond_check_thread_quit_pipe(pollfd, revents);
1495
1496 if (pollfd == sock) {
1497 /* Event on the consumerd socket */
1498 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
1499 && !(revents & LPOLLIN)) {
1500 ERR("consumer err socket second poll error");
1501 goto error;
1502 }
1503 health_code_update();
1504 /* Wait for any kconsumerd error */
1505 ret = lttcomm_recv_unix_sock(sock, &code,
1506 sizeof(enum lttcomm_return_code));
1507 if (ret <= 0) {
1508 ERR("consumer closed the command socket");
1509 goto error;
1510 }
1511
1512 ERR("consumer return code : %s",
1513 lttcomm_get_readable_code(-code));
1514
1515 goto exit;
1516 } else if (pollfd == consumer_data->metadata_fd) {
1517 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)
1518 && !(revents & LPOLLIN)) {
1519 ERR("consumer err metadata socket second poll error");
1520 goto error;
1521 }
1522 /* UST metadata requests */
1523 ret = ust_consumer_metadata_request(
1524 &consumer_data->metadata_sock);
1525 if (ret < 0) {
1526 ERR("Handling metadata request");
1527 goto error;
1528 }
1529 }
1530 /* No need for an else branch all FDs are tested prior. */
1531 }
1532 health_code_update();
1533 }
1534
1535 exit:
1536 error:
1537 /*
1538 * We lock here because we are about to close the sockets and some other
1539 * thread might be using them so get exclusive access which will abort all
1540 * other consumer command by other threads.
1541 */
1542 pthread_mutex_lock(&consumer_data->lock);
1543
1544 /* Immediately set the consumerd state to stopped */
1545 if (consumer_data->type == LTTNG_CONSUMER_KERNEL) {
1546 uatomic_set(&kernel_consumerd_state, CONSUMER_ERROR);
1547 } else if (consumer_data->type == LTTNG_CONSUMER64_UST ||
1548 consumer_data->type == LTTNG_CONSUMER32_UST) {
1549 uatomic_set(&ust_consumerd_state, CONSUMER_ERROR);
1550 } else {
1551 /* Code flow error... */
1552 assert(0);
1553 }
1554
1555 if (consumer_data->err_sock >= 0) {
1556 ret = close(consumer_data->err_sock);
1557 if (ret) {
1558 PERROR("close");
1559 }
1560 consumer_data->err_sock = -1;
1561 }
1562 if (consumer_data->cmd_sock >= 0) {
1563 ret = close(consumer_data->cmd_sock);
1564 if (ret) {
1565 PERROR("close");
1566 }
1567 consumer_data->cmd_sock = -1;
1568 }
1569 if (consumer_data->metadata_sock.fd_ptr &&
1570 *consumer_data->metadata_sock.fd_ptr >= 0) {
1571 ret = close(*consumer_data->metadata_sock.fd_ptr);
1572 if (ret) {
1573 PERROR("close");
1574 }
1575 }
1576 if (sock >= 0) {
1577 ret = close(sock);
1578 if (ret) {
1579 PERROR("close");
1580 }
1581 }
1582
1583 unlink(consumer_data->err_unix_sock_path);
1584 unlink(consumer_data->cmd_unix_sock_path);
1585 pthread_mutex_unlock(&consumer_data->lock);
1586
1587 /* Cleanup metadata socket mutex. */
1588 if (consumer_data->metadata_sock.lock) {
1589 pthread_mutex_destroy(consumer_data->metadata_sock.lock);
1590 free(consumer_data->metadata_sock.lock);
1591 }
1592 lttng_poll_clean(&events);
1593
1594 if (cmd_socket_wrapper) {
1595 consumer_destroy_socket(cmd_socket_wrapper);
1596 }
1597 error_poll:
1598 if (err) {
1599 health_error();
1600 ERR("Health error occurred in %s", __func__);
1601 }
1602 health_unregister(health_sessiond);
1603 DBG("consumer thread cleanup completed");
1604
1605 rcu_thread_offline();
1606 rcu_unregister_thread();
1607
1608 return NULL;
1609 }
1610
1611 /*
1612 * This thread manage application communication.
1613 */
1614 static void *thread_manage_apps(void *data)
1615 {
1616 int i, ret, pollfd, err = -1;
1617 ssize_t size_ret;
1618 uint32_t revents, nb_fd;
1619 struct lttng_poll_event events;
1620
1621 DBG("[thread] Manage application started");
1622
1623 rcu_register_thread();
1624 rcu_thread_online();
1625
1626 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_APP_MANAGE);
1627
1628 if (testpoint(sessiond_thread_manage_apps)) {
1629 goto error_testpoint;
1630 }
1631
1632 health_code_update();
1633
1634 ret = sessiond_set_thread_pollset(&events, 2);
1635 if (ret < 0) {
1636 goto error_poll_create;
1637 }
1638
1639 ret = lttng_poll_add(&events, apps_cmd_pipe[0], LPOLLIN | LPOLLRDHUP);
1640 if (ret < 0) {
1641 goto error;
1642 }
1643
1644 if (testpoint(sessiond_thread_manage_apps_before_loop)) {
1645 goto error;
1646 }
1647
1648 health_code_update();
1649
1650 while (1) {
1651 DBG("Apps thread polling");
1652
1653 /* Inifinite blocking call, waiting for transmission */
1654 restart:
1655 health_poll_entry();
1656 ret = lttng_poll_wait(&events, -1);
1657 DBG("Apps thread return from poll on %d fds",
1658 LTTNG_POLL_GETNB(&events));
1659 health_poll_exit();
1660 if (ret < 0) {
1661 /*
1662 * Restart interrupted system call.
1663 */
1664 if (errno == EINTR) {
1665 goto restart;
1666 }
1667 goto error;
1668 }
1669
1670 nb_fd = ret;
1671
1672 for (i = 0; i < nb_fd; i++) {
1673 /* Fetch once the poll data */
1674 revents = LTTNG_POLL_GETEV(&events, i);
1675 pollfd = LTTNG_POLL_GETFD(&events, i);
1676
1677 health_code_update();
1678
1679 if (!revents) {
1680 /* No activity for this FD (poll implementation). */
1681 continue;
1682 }
1683
1684 /* Thread quit pipe has been closed. Killing thread. */
1685 ret = sessiond_check_thread_quit_pipe(pollfd, revents);
1686 if (ret) {
1687 err = 0;
1688 goto exit;
1689 }
1690
1691 /* Inspect the apps cmd pipe */
1692 if (pollfd == apps_cmd_pipe[0]) {
1693 if (revents & LPOLLIN) {
1694 int sock;
1695
1696 /* Empty pipe */
1697 size_ret = lttng_read(apps_cmd_pipe[0], &sock, sizeof(sock));
1698 if (size_ret < sizeof(sock)) {
1699 PERROR("read apps cmd pipe");
1700 goto error;
1701 }
1702
1703 health_code_update();
1704
1705 /*
1706 * Since this is a command socket (write then read),
1707 * we only monitor the error events of the socket.
1708 */
1709 ret = lttng_poll_add(&events, sock,
1710 LPOLLERR | LPOLLHUP | LPOLLRDHUP);
1711 if (ret < 0) {
1712 goto error;
1713 }
1714
1715 DBG("Apps with sock %d added to poll set", sock);
1716 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1717 ERR("Apps command pipe error");
1718 goto error;
1719 } else {
1720 ERR("Unknown poll events %u for sock %d", revents, pollfd);
1721 goto error;
1722 }
1723 } else {
1724 /*
1725 * At this point, we know that a registered application made
1726 * the event at poll_wait.
1727 */
1728 if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
1729 /* Removing from the poll set */
1730 ret = lttng_poll_del(&events, pollfd);
1731 if (ret < 0) {
1732 goto error;
1733 }
1734
1735 /* Socket closed on remote end. */
1736 ust_app_unregister(pollfd);
1737 } else {
1738 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1739 goto error;
1740 }
1741 }
1742
1743 health_code_update();
1744 }
1745 }
1746
1747 exit:
1748 error:
1749 lttng_poll_clean(&events);
1750 error_poll_create:
1751 error_testpoint:
1752 utils_close_pipe(apps_cmd_pipe);
1753 apps_cmd_pipe[0] = apps_cmd_pipe[1] = -1;
1754
1755 /*
1756 * We don't clean the UST app hash table here since already registered
1757 * applications can still be controlled so let them be until the session
1758 * daemon dies or the applications stop.
1759 */
1760
1761 if (err) {
1762 health_error();
1763 ERR("Health error occurred in %s", __func__);
1764 }
1765 health_unregister(health_sessiond);
1766 DBG("Application communication apps thread cleanup complete");
1767 rcu_thread_offline();
1768 rcu_unregister_thread();
1769 return NULL;
1770 }
1771
1772 /*
1773 * Send a socket to a thread This is called from the dispatch UST registration
1774 * thread once all sockets are set for the application.
1775 *
1776 * The sock value can be invalid, we don't really care, the thread will handle
1777 * it and make the necessary cleanup if so.
1778 *
1779 * On success, return 0 else a negative value being the errno message of the
1780 * write().
1781 */
1782 static int send_socket_to_thread(int fd, int sock)
1783 {
1784 ssize_t ret;
1785
1786 /*
1787 * It's possible that the FD is set as invalid with -1 concurrently just
1788 * before calling this function being a shutdown state of the thread.
1789 */
1790 if (fd < 0) {
1791 ret = -EBADF;
1792 goto error;
1793 }
1794
1795 ret = lttng_write(fd, &sock, sizeof(sock));
1796 if (ret < sizeof(sock)) {
1797 PERROR("write apps pipe %d", fd);
1798 if (ret < 0) {
1799 ret = -errno;
1800 }
1801 goto error;
1802 }
1803
1804 /* All good. Don't send back the write positive ret value. */
1805 ret = 0;
1806 error:
1807 return (int) ret;
1808 }
1809
1810 /*
1811 * Sanitize the wait queue of the dispatch registration thread meaning removing
1812 * invalid nodes from it. This is to avoid memory leaks for the case the UST
1813 * notify socket is never received.
1814 */
1815 static void sanitize_wait_queue(struct ust_reg_wait_queue *wait_queue)
1816 {
1817 int ret, nb_fd = 0, i;
1818 unsigned int fd_added = 0;
1819 struct lttng_poll_event events;
1820 struct ust_reg_wait_node *wait_node = NULL, *tmp_wait_node;
1821
1822 assert(wait_queue);
1823
1824 lttng_poll_init(&events);
1825
1826 /* Just skip everything for an empty queue. */
1827 if (!wait_queue->count) {
1828 goto end;
1829 }
1830
1831 ret = lttng_poll_create(&events, wait_queue->count, LTTNG_CLOEXEC);
1832 if (ret < 0) {
1833 goto error_create;
1834 }
1835
1836 cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
1837 &wait_queue->head, head) {
1838 assert(wait_node->app);
1839 ret = lttng_poll_add(&events, wait_node->app->sock,
1840 LPOLLHUP | LPOLLERR);
1841 if (ret < 0) {
1842 goto error;
1843 }
1844
1845 fd_added = 1;
1846 }
1847
1848 if (!fd_added) {
1849 goto end;
1850 }
1851
1852 /*
1853 * Poll but don't block so we can quickly identify the faulty events and
1854 * clean them afterwards from the wait queue.
1855 */
1856 ret = lttng_poll_wait(&events, 0);
1857 if (ret < 0) {
1858 goto error;
1859 }
1860 nb_fd = ret;
1861
1862 for (i = 0; i < nb_fd; i++) {
1863 /* Get faulty FD. */
1864 uint32_t revents = LTTNG_POLL_GETEV(&events, i);
1865 int pollfd = LTTNG_POLL_GETFD(&events, i);
1866
1867 if (!revents) {
1868 /* No activity for this FD (poll implementation). */
1869 continue;
1870 }
1871
1872 cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
1873 &wait_queue->head, head) {
1874 if (pollfd == wait_node->app->sock &&
1875 (revents & (LPOLLHUP | LPOLLERR))) {
1876 cds_list_del(&wait_node->head);
1877 wait_queue->count--;
1878 ust_app_destroy(wait_node->app);
1879 free(wait_node);
1880 /*
1881 * Silence warning of use-after-free in
1882 * cds_list_for_each_entry_safe which uses
1883 * __typeof__(*wait_node).
1884 */
1885 wait_node = NULL;
1886 break;
1887 } else {
1888 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
1889 goto error;
1890 }
1891 }
1892 }
1893
1894 if (nb_fd > 0) {
1895 DBG("Wait queue sanitized, %d node were cleaned up", nb_fd);
1896 }
1897
1898 end:
1899 lttng_poll_clean(&events);
1900 return;
1901
1902 error:
1903 lttng_poll_clean(&events);
1904 error_create:
1905 ERR("Unable to sanitize wait queue");
1906 return;
1907 }
1908
1909 /*
1910 * Dispatch request from the registration threads to the application
1911 * communication thread.
1912 */
1913 static void *thread_dispatch_ust_registration(void *data)
1914 {
1915 int ret, err = -1;
1916 struct cds_wfcq_node *node;
1917 struct ust_command *ust_cmd = NULL;
1918 struct ust_reg_wait_node *wait_node = NULL, *tmp_wait_node;
1919 struct ust_reg_wait_queue wait_queue = {
1920 .count = 0,
1921 };
1922
1923 rcu_register_thread();
1924
1925 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_APP_REG_DISPATCH);
1926
1927 if (testpoint(sessiond_thread_app_reg_dispatch)) {
1928 goto error_testpoint;
1929 }
1930
1931 health_code_update();
1932
1933 CDS_INIT_LIST_HEAD(&wait_queue.head);
1934
1935 DBG("[thread] Dispatch UST command started");
1936
1937 for (;;) {
1938 health_code_update();
1939
1940 /* Atomically prepare the queue futex */
1941 futex_nto1_prepare(&ust_cmd_queue.futex);
1942
1943 if (CMM_LOAD_SHARED(dispatch_thread_exit)) {
1944 break;
1945 }
1946
1947 do {
1948 struct ust_app *app = NULL;
1949 ust_cmd = NULL;
1950
1951 /*
1952 * Make sure we don't have node(s) that have hung up before receiving
1953 * the notify socket. This is to clean the list in order to avoid
1954 * memory leaks from notify socket that are never seen.
1955 */
1956 sanitize_wait_queue(&wait_queue);
1957
1958 health_code_update();
1959 /* Dequeue command for registration */
1960 node = cds_wfcq_dequeue_blocking(&ust_cmd_queue.head, &ust_cmd_queue.tail);
1961 if (node == NULL) {
1962 DBG("Woken up but nothing in the UST command queue");
1963 /* Continue thread execution */
1964 break;
1965 }
1966
1967 ust_cmd = caa_container_of(node, struct ust_command, node);
1968
1969 DBG("Dispatching UST registration pid:%d ppid:%d uid:%d"
1970 " gid:%d sock:%d name:%s (version %d.%d)",
1971 ust_cmd->reg_msg.pid, ust_cmd->reg_msg.ppid,
1972 ust_cmd->reg_msg.uid, ust_cmd->reg_msg.gid,
1973 ust_cmd->sock, ust_cmd->reg_msg.name,
1974 ust_cmd->reg_msg.major, ust_cmd->reg_msg.minor);
1975
1976 if (ust_cmd->reg_msg.type == USTCTL_SOCKET_CMD) {
1977 wait_node = zmalloc(sizeof(*wait_node));
1978 if (!wait_node) {
1979 PERROR("zmalloc wait_node dispatch");
1980 ret = close(ust_cmd->sock);
1981 if (ret < 0) {
1982 PERROR("close ust sock dispatch %d", ust_cmd->sock);
1983 }
1984 lttng_fd_put(LTTNG_FD_APPS, 1);
1985 free(ust_cmd);
1986 goto error;
1987 }
1988 CDS_INIT_LIST_HEAD(&wait_node->head);
1989
1990 /* Create application object if socket is CMD. */
1991 wait_node->app = ust_app_create(&ust_cmd->reg_msg,
1992 ust_cmd->sock);
1993 if (!wait_node->app) {
1994 ret = close(ust_cmd->sock);
1995 if (ret < 0) {
1996 PERROR("close ust sock dispatch %d", ust_cmd->sock);
1997 }
1998 lttng_fd_put(LTTNG_FD_APPS, 1);
1999 free(wait_node);
2000 free(ust_cmd);
2001 continue;
2002 }
2003 /*
2004 * Add application to the wait queue so we can set the notify
2005 * socket before putting this object in the global ht.
2006 */
2007 cds_list_add(&wait_node->head, &wait_queue.head);
2008 wait_queue.count++;
2009
2010 free(ust_cmd);
2011 /*
2012 * We have to continue here since we don't have the notify
2013 * socket and the application MUST be added to the hash table
2014 * only at that moment.
2015 */
2016 continue;
2017 } else {
2018 /*
2019 * Look for the application in the local wait queue and set the
2020 * notify socket if found.
2021 */
2022 cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
2023 &wait_queue.head, head) {
2024 health_code_update();
2025 if (wait_node->app->pid == ust_cmd->reg_msg.pid) {
2026 wait_node->app->notify_sock = ust_cmd->sock;
2027 cds_list_del(&wait_node->head);
2028 wait_queue.count--;
2029 app = wait_node->app;
2030 free(wait_node);
2031 DBG3("UST app notify socket %d is set", ust_cmd->sock);
2032 break;
2033 }
2034 }
2035
2036 /*
2037 * With no application at this stage the received socket is
2038 * basically useless so close it before we free the cmd data
2039 * structure for good.
2040 */
2041 if (!app) {
2042 ret = close(ust_cmd->sock);
2043 if (ret < 0) {
2044 PERROR("close ust sock dispatch %d", ust_cmd->sock);
2045 }
2046 lttng_fd_put(LTTNG_FD_APPS, 1);
2047 }
2048 free(ust_cmd);
2049 }
2050
2051 if (app) {
2052 /*
2053 * @session_lock_list
2054 *
2055 * Lock the global session list so from the register up to the
2056 * registration done message, no thread can see the application
2057 * and change its state.
2058 */
2059 session_lock_list();
2060 rcu_read_lock();
2061
2062 /*
2063 * Add application to the global hash table. This needs to be
2064 * done before the update to the UST registry can locate the
2065 * application.
2066 */
2067 ust_app_add(app);
2068
2069 /* Set app version. This call will print an error if needed. */
2070 (void) ust_app_version(app);
2071
2072 /* Send notify socket through the notify pipe. */
2073 ret = send_socket_to_thread(apps_cmd_notify_pipe[1],
2074 app->notify_sock);
2075 if (ret < 0) {
2076 rcu_read_unlock();
2077 session_unlock_list();
2078 /*
2079 * No notify thread, stop the UST tracing. However, this is
2080 * not an internal error of the this thread thus setting
2081 * the health error code to a normal exit.
2082 */
2083 err = 0;
2084 goto error;
2085 }
2086
2087 /*
2088 * Update newly registered application with the tracing
2089 * registry info already enabled information.
2090 */
2091 update_ust_app(app->sock);
2092
2093 /*
2094 * Don't care about return value. Let the manage apps threads
2095 * handle app unregistration upon socket close.
2096 */
2097 (void) ust_app_register_done(app);
2098
2099 /*
2100 * Even if the application socket has been closed, send the app
2101 * to the thread and unregistration will take place at that
2102 * place.
2103 */
2104 ret = send_socket_to_thread(apps_cmd_pipe[1], app->sock);
2105 if (ret < 0) {
2106 rcu_read_unlock();
2107 session_unlock_list();
2108 /*
2109 * No apps. thread, stop the UST tracing. However, this is
2110 * not an internal error of the this thread thus setting
2111 * the health error code to a normal exit.
2112 */
2113 err = 0;
2114 goto error;
2115 }
2116
2117 rcu_read_unlock();
2118 session_unlock_list();
2119 }
2120 } while (node != NULL);
2121
2122 health_poll_entry();
2123 /* Futex wait on queue. Blocking call on futex() */
2124 futex_nto1_wait(&ust_cmd_queue.futex);
2125 health_poll_exit();
2126 }
2127 /* Normal exit, no error */
2128 err = 0;
2129
2130 error:
2131 /* Clean up wait queue. */
2132 cds_list_for_each_entry_safe(wait_node, tmp_wait_node,
2133 &wait_queue.head, head) {
2134 cds_list_del(&wait_node->head);
2135 wait_queue.count--;
2136 free(wait_node);
2137 }
2138
2139 /* Empty command queue. */
2140 for (;;) {
2141 /* Dequeue command for registration */
2142 node = cds_wfcq_dequeue_blocking(&ust_cmd_queue.head, &ust_cmd_queue.tail);
2143 if (node == NULL) {
2144 break;
2145 }
2146 ust_cmd = caa_container_of(node, struct ust_command, node);
2147 ret = close(ust_cmd->sock);
2148 if (ret < 0) {
2149 PERROR("close ust sock exit dispatch %d", ust_cmd->sock);
2150 }
2151 lttng_fd_put(LTTNG_FD_APPS, 1);
2152 free(ust_cmd);
2153 }
2154
2155 error_testpoint:
2156 DBG("Dispatch thread dying");
2157 if (err) {
2158 health_error();
2159 ERR("Health error occurred in %s", __func__);
2160 }
2161 health_unregister(health_sessiond);
2162 rcu_unregister_thread();
2163 return NULL;
2164 }
2165
2166 /*
2167 * This thread manage application registration.
2168 */
2169 static void *thread_registration_apps(void *data)
2170 {
2171 int sock = -1, i, ret, pollfd, err = -1;
2172 uint32_t revents, nb_fd;
2173 struct lttng_poll_event events;
2174 /*
2175 * Get allocated in this thread, enqueued to a global queue, dequeued and
2176 * freed in the manage apps thread.
2177 */
2178 struct ust_command *ust_cmd = NULL;
2179
2180 DBG("[thread] Manage application registration started");
2181
2182 health_register(health_sessiond, HEALTH_SESSIOND_TYPE_APP_REG);
2183
2184 if (testpoint(sessiond_thread_registration_apps)) {
2185 goto error_testpoint;
2186 }
2187
2188 ret = lttcomm_listen_unix_sock(apps_sock);
2189 if (ret < 0) {
2190 goto error_listen;
2191 }
2192
2193 /*
2194 * Pass 2 as size here for the thread quit pipe and apps socket. Nothing
2195 * more will be added to this poll set.
2196 */
2197 ret = sessiond_set_thread_pollset(&events, 2);
2198 if (ret < 0) {
2199 goto error_create_poll;
2200 }
2201
2202 /* Add the application registration socket */
2203 ret = lttng_poll_add(&events, apps_sock, LPOLLIN | LPOLLRDHUP);
2204 if (ret < 0) {
2205 goto error_poll_add;
2206 }
2207
2208 /* Notify all applications to register */
2209 ret = notify_ust_apps(1);
2210 if (ret < 0) {
2211 ERR("Failed to notify applications or create the wait shared memory.\n"
2212 "Execution continues but there might be problem for already\n"
2213 "running applications that wishes to register.");
2214 }
2215
2216 while (1) {
2217 DBG("Accepting application registration");
2218
2219 /* Inifinite blocking call, waiting for transmission */
2220 restart:
2221 health_poll_entry();
2222 ret = lttng_poll_wait(&events, -1);
2223 health_poll_exit();
2224 if (ret < 0) {
2225 /*
2226 * Restart interrupted system call.
2227 */
2228 if (errno == EINTR) {
2229 goto restart;
2230 }
2231 goto error;
2232 }
2233
2234 nb_fd = ret;
2235
2236 for (i = 0; i < nb_fd; i++) {
2237 health_code_update();
2238
2239 /* Fetch once the poll data */
2240 revents = LTTNG_POLL_GETEV(&events, i);
2241 pollfd = LTTNG_POLL_GETFD(&events, i);
2242
2243 if (!revents) {
2244 /* No activity for this FD (poll implementation). */
2245 continue;
2246 }
2247
2248 /* Thread quit pipe has been closed. Killing thread. */
2249 ret = sessiond_check_thread_quit_pipe(pollfd, revents);
2250 if (ret) {
2251 err = 0;
2252 goto exit;
2253 }
2254
2255 /* Event on the registration socket */
2256 if (pollfd == apps_sock) {
2257 if (revents & LPOLLIN) {
2258 sock = lttcomm_accept_unix_sock(apps_sock);
2259 if (sock < 0) {
2260 goto error;
2261 }
2262
2263 /*
2264 * Set socket timeout for both receiving and ending.
2265 * app_socket_timeout is in seconds, whereas
2266 * lttcomm_setsockopt_rcv_timeout and
2267 * lttcomm_setsockopt_snd_timeout expect msec as
2268 * parameter.
2269 */
2270 if (app_socket_timeout >= 0) {
2271 (void) lttcomm_setsockopt_rcv_timeout(sock,
2272 app_socket_timeout * 1000);
2273 (void) lttcomm_setsockopt_snd_timeout(sock,
2274 app_socket_timeout * 1000);
2275 }
2276
2277 /*
2278 * Set the CLOEXEC flag. Return code is useless because
2279 * either way, the show must go on.
2280 */
2281 (void) utils_set_fd_cloexec(sock);
2282
2283 /* Create UST registration command for enqueuing */
2284 ust_cmd = zmalloc(sizeof(struct ust_command));
2285 if (ust_cmd == NULL) {
2286 PERROR("ust command zmalloc");
2287 ret = close(sock);
2288 if (ret) {
2289 PERROR("close");
2290 }
2291 goto error;
2292 }
2293
2294 /*
2295 * Using message-based transmissions to ensure we don't
2296 * have to deal with partially received messages.
2297 */
2298 ret = lttng_fd_get(LTTNG_FD_APPS, 1);
2299 if (ret < 0) {
2300 ERR("Exhausted file descriptors allowed for applications.");
2301 free(ust_cmd);
2302 ret = close(sock);
2303 if (ret) {
2304 PERROR("close");
2305 }
2306 sock = -1;
2307 continue;
2308 }
2309
2310 health_code_update();
2311 ret = ust_app_recv_registration(sock, &ust_cmd->reg_msg);
2312 if (ret < 0) {
2313 free(ust_cmd);
2314 /* Close socket of the application. */
2315 ret = close(sock);
2316 if (ret) {
2317 PERROR("close");
2318 }
2319 lttng_fd_put(LTTNG_FD_APPS, 1);
2320 sock = -1;
2321 continue;
2322 }
2323 health_code_update();
2324
2325 ust_cmd->sock = sock;
2326 sock = -1;
2327
2328 DBG("UST registration received with pid:%d ppid:%d uid:%d"
2329 " gid:%d sock:%d name:%s (version %d.%d)",
2330 ust_cmd->reg_msg.pid, ust_cmd->reg_msg.ppid,
2331 ust_cmd->reg_msg.uid, ust_cmd->reg_msg.gid,
2332 ust_cmd->sock, ust_cmd->reg_msg.name,
2333 ust_cmd->reg_msg.major, ust_cmd->reg_msg.minor);
2334
2335 /*
2336 * Lock free enqueue the registration request. The red pill
2337 * has been taken! This apps will be part of the *system*.
2338 */
2339 cds_wfcq_enqueue(&ust_cmd_queue.head, &ust_cmd_queue.tail, &ust_cmd->node);
2340
2341 /*
2342 * Wake the registration queue futex. Implicit memory
2343 * barrier with the exchange in cds_wfcq_enqueue.
2344 */
2345 futex_nto1_wake(&ust_cmd_queue.futex);
2346 } else if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
2347 ERR("Register apps socket poll error");
2348 goto error;
2349 } else {
2350 ERR("Unexpected poll events %u for sock %d", revents, pollfd);
2351 goto error;
2352 }
2353 }
2354 }
2355 }
2356
2357 exit:
2358 error:
2359 /* Notify that the registration thread is gone */
2360 notify_ust_apps(0);
2361
2362 if (apps_sock >= 0) {
2363 ret = close(apps_sock);
2364 if (ret) {
2365 PERROR("close");
2366 }
2367 }
2368 if (sock >= 0) {
2369 ret = close(sock);
2370 if (ret) {
2371 PERROR("close");
2372 }
2373 lttng_fd_put(LTTNG_FD_APPS, 1);
2374 }
2375 unlink(apps_unix_sock_path);
2376
2377 error_poll_add:
2378 lttng_poll_clean(&events);
2379 error_listen:
2380 error_create_poll:
2381 error_testpoint:
2382 DBG("UST Registration thread cleanup complete");
2383 if (err) {
2384 health_error();
2385 ERR("Health error occurred in %s", __func__);
2386 }
2387 health_unregister(health_sessiond);
2388
2389 return NULL;
2390 }
2391
2392 /*
2393 * Start the thread_manage_consumer. This must be done after a lttng-consumerd
2394 * exec or it will fails.
2395 */
2396 static int spawn_consumer_thread(struct consumer_data *consumer_data)
2397 {
2398 int ret, clock_ret;
2399 struct timespec timeout;
2400
2401 /*
2402 * Make sure we set the readiness flag to 0 because we are NOT ready.
2403 * This access to consumer_thread_is_ready does not need to be
2404 * protected by consumer_data.cond_mutex (yet) since the consumer
2405 * management thread has not been started at this point.
2406 */
2407 consumer_data->consumer_thread_is_ready = 0;
2408
2409 /* Setup pthread condition */
2410 ret = pthread_condattr_init(&consumer_data->condattr);
2411 if (ret) {
2412 errno = ret;
2413 PERROR("pthread_condattr_init consumer data");
2414 goto error;
2415 }
2416
2417 /*
2418 * Set the monotonic clock in order to make sure we DO NOT jump in time
2419 * between the clock_gettime() call and the timedwait call. See bug #324
2420 * for a more details and how we noticed it.
2421 */
2422 ret = pthread_condattr_setclock(&consumer_data->condattr, CLOCK_MONOTONIC);
2423 if (ret) {
2424 errno = ret;
2425 PERROR("pthread_condattr_setclock consumer data");
2426 goto error;
2427 }
2428
2429 ret = pthread_cond_init(&consumer_data->cond, &consumer_data->condattr);
2430 if (ret) {
2431 errno = ret;
2432 PERROR("pthread_cond_init consumer data");
2433 goto error;
2434 }
2435
2436 ret = pthread_create(&consumer_data->thread, default_pthread_attr(),
2437 thread_manage_consumer, consumer_data);
2438 if (ret) {
2439 errno = ret;
2440 PERROR("pthread_create consumer");
2441 ret = -1;
2442 goto error;
2443 }
2444
2445 /* We are about to wait on a pthread condition */
2446 pthread_mutex_lock(&consumer_data->cond_mutex);
2447
2448 /* Get time for sem_timedwait absolute timeout */
2449 clock_ret = lttng_clock_gettime(CLOCK_MONOTONIC, &timeout);
2450 /*
2451 * Set the timeout for the condition timed wait even if the clock gettime
2452 * call fails since we might loop on that call and we want to avoid to
2453 * increment the timeout too many times.
2454 */
2455 timeout.tv_sec += DEFAULT_SEM_WAIT_TIMEOUT;
2456
2457 /*
2458 * The following loop COULD be skipped in some conditions so this is why we
2459 * set ret to 0 in order to make sure at least one round of the loop is
2460 * done.
2461 */
2462 ret = 0;
2463
2464 /*
2465 * Loop until the condition is reached or when a timeout is reached. Note
2466 * that the pthread_cond_timedwait(P) man page specifies that EINTR can NOT
2467 * be returned but the pthread_cond(3), from the glibc-doc, says that it is
2468 * possible. This loop does not take any chances and works with both of
2469 * them.
2470 */
2471 while (!consumer_data->consumer_thread_is_ready && ret != ETIMEDOUT) {
2472 if (clock_ret < 0) {
2473 PERROR("clock_gettime spawn consumer");
2474 /* Infinite wait for the consumerd thread to be ready */
2475 ret = pthread_cond_wait(&consumer_data->cond,
2476 &consumer_data->cond_mutex);
2477 } else {
2478 ret = pthread_cond_timedwait(&consumer_data->cond,
2479 &consumer_data->cond_mutex, &timeout);
2480 }
2481 }
2482
2483 /* Release the pthread condition */
2484 pthread_mutex_unlock(&consumer_data->cond_mutex);
2485
2486 if (ret != 0) {
2487 errno = ret;
2488 if (ret == ETIMEDOUT) {
2489 int pth_ret;
2490
2491 /*
2492 * Call has timed out so we kill the kconsumerd_thread and return
2493 * an error.
2494 */
2495 ERR("Condition timed out. The consumer thread was never ready."
2496 " Killing it");
2497 pth_ret = pthread_cancel(consumer_data->thread);
2498 if (pth_ret < 0) {
2499 PERROR("pthread_cancel consumer thread");
2500 }
2501 } else {
2502 PERROR("pthread_cond_wait failed consumer thread");
2503 }
2504 /* Caller is expecting a negative value on failure. */
2505 ret = -1;
2506 goto error;
2507 }
2508
2509 pthread_mutex_lock(&consumer_data->pid_mutex);
2510 if (consumer_data->pid == 0) {
2511 ERR("Consumerd did not start");
2512 pthread_mutex_unlock(&consumer_data->pid_mutex);
2513 goto error;
2514 }
2515 pthread_mutex_unlock(&consumer_data->pid_mutex);
2516
2517 return 0;
2518
2519 error:
2520 return ret;
2521 }
2522
2523 /*
2524 * Join consumer thread
2525 */
2526 static int join_consumer_thread(struct consumer_data *consumer_data)
2527 {
2528 void *status;
2529
2530 /* Consumer pid must be a real one. */
2531 if (consumer_data->pid > 0) {
2532 int ret;
2533 ret = kill(consumer_data->pid, SIGTERM);
2534 if (ret) {
2535 PERROR("Error killing consumer daemon");
2536 return ret;
2537 }
2538 return pthread_join(consumer_data->thread, &status);
2539 } else {
2540 return 0;
2541 }
2542 }
2543
2544 /*
2545 * Fork and exec a consumer daemon (consumerd).
2546 *
2547 * Return pid if successful else -1.
2548 */
2549 static pid_t spawn_consumerd(struct consumer_data *consumer_data)
2550 {
2551 int ret;
2552 pid_t pid;
2553 const char *consumer_to_use;
2554 const char *verbosity;
2555 struct stat st;
2556
2557 DBG("Spawning consumerd");
2558
2559 pid = fork();
2560 if (pid == 0) {
2561 /*
2562 * Exec consumerd.
2563 */
2564 if (opt_verbose_consumer) {
2565 verbosity = "--verbose";
2566 } else if (lttng_opt_quiet) {
2567 verbosity = "--quiet";
2568 } else {
2569 verbosity = "";
2570 }
2571
2572 switch (consumer_data->type) {
2573 case LTTNG_CONSUMER_KERNEL:
2574 /*
2575 * Find out which consumerd to execute. We will first try the
2576 * 64-bit path, then the sessiond's installation directory, and
2577 * fallback on the 32-bit one,
2578 */
2579 DBG3("Looking for a kernel consumer at these locations:");
2580 DBG3(" 1) %s", consumerd64_bin);
2581 DBG3(" 2) %s/%s", INSTALL_BIN_PATH, CONSUMERD_FILE);
2582 DBG3(" 3) %s", consumerd32_bin);
2583 if (stat(consumerd64_bin, &st) == 0) {
2584 DBG3("Found location #1");
2585 consumer_to_use = consumerd64_bin;
2586 } else if (stat(INSTALL_BIN_PATH "/" CONSUMERD_FILE, &st) == 0) {
2587 DBG3("Found location #2");
2588 consumer_to_use = INSTALL_BIN_PATH "/" CONSUMERD_FILE;
2589 } else if (stat(consumerd32_bin, &st) == 0) {
2590 DBG3("Found location #3");
2591 consumer_to_use = consumerd32_bin;
2592 } else {
2593 DBG("Could not find any valid consumerd executable");
2594 ret = -EINVAL;
2595 break;
2596 }
2597 DBG("Using kernel consumer at: %s", consumer_to_use);
2598 ret = execl(consumer_to_use,
2599 "lttng-consumerd", verbosity, "-k",
2600 "--consumerd-cmd-sock", consumer_data->cmd_unix_sock_path,
2601 "--consumerd-err-sock", consumer_data->err_unix_sock_path,
2602 "--group", tracing_group_name,
2603 NULL);
2604 break;
2605 case LTTNG_CONSUMER64_UST:
2606 {
2607 char *tmpnew = NULL;
2608
2609 if (consumerd64_libdir[0] != '\0') {
2610 char *tmp;
2611 size_t tmplen;
2612
2613 tmp = lttng_secure_getenv("LD_LIBRARY_PATH");
2614 if (!tmp) {
2615 tmp = "";
2616 }
2617 tmplen = strlen("LD_LIBRARY_PATH=")
2618 + strlen(consumerd64_libdir) + 1 /* : */ + strlen(tmp);
2619 tmpnew = zmalloc(tmplen + 1 /* \0 */);
2620 if (!tmpnew) {
2621 ret = -ENOMEM;
2622 goto error;
2623 }
2624 strcpy(tmpnew, "LD_LIBRARY_PATH=");
2625 strcat(tmpnew, consumerd64_libdir);
2626 if (tmp[0] != '\0') {
2627 strcat(tmpnew, ":");
2628 strcat(tmpnew, tmp);
2629 }
2630 ret = putenv(tmpnew);
2631 if (ret) {
2632 ret = -errno;
2633 free(tmpnew);
2634 goto error;
2635 }
2636 }
2637 DBG("Using 64-bit UST consumer at: %s", consumerd64_bin);
2638 ret = execl(consumerd64_bin, "lttng-consumerd", verbosity, "-u",
2639 "--consumerd-cmd-sock", consumer_data->cmd_unix_sock_path,
2640 "--consumerd-err-sock", consumer_data->err_unix_sock_path,
2641 "--group", tracing_group_name,
2642 NULL);
2643 if (consumerd64_libdir[0] != '\0') {
2644 free(tmpnew);
2645 }
2646 break;
2647 }
2648 case LTTNG_CONSUMER32_UST:
2649 {
2650 char *tmpnew = NULL;
2651
2652 if (consumerd32_libdir[0] != '\0') {
2653 char *tmp;
2654 size_t tmplen;
2655
2656 tmp = lttng_secure_getenv("LD_LIBRARY_PATH");
2657 if (!tmp) {
2658 tmp = "";
2659 }
2660 tmplen = strlen("LD_LIBRARY_PATH=")
2661 + strlen(consumerd32_libdir) + 1 /* : */ + strlen(tmp);
2662 tmpnew = zmalloc(tmplen + 1 /* \0 */);
2663 if (!tmpnew) {
2664 ret = -ENOMEM;
2665 goto error;
2666 }
2667 strcpy(tmpnew, "LD_LIBRARY_PATH=");
2668 strcat(tmpnew, consumerd32_libdir);
2669 if (tmp[0] != '\0') {
2670 strcat(tmpnew, ":");
2671 strcat(tmpnew, tmp);
2672 }
2673 ret = putenv(tmpnew);
2674 if (ret) {
2675 ret = -errno;
2676 free(tmpnew);
2677 goto error;
2678 }
2679 }
2680 DBG("Using 32-bit UST consumer at: %s", consumerd32_bin);
2681 ret = execl(consumerd32_bin, "lttng-consumerd", verbosity, "-u",
2682 "--consumerd-cmd-sock", consumer_data->cmd_unix_sock_path,
2683 "--consumerd-err-sock", consumer_data->err_unix_sock_path,
2684 "--group", tracing_group_name,
2685 NULL);
2686 if (consumerd32_libdir[0] != '\0') {
2687 free(tmpnew);
2688 }
2689 break;
2690 }
2691 default:
2692 PERROR("unknown consumer type");
2693 exit(EXIT_FAILURE);
2694 }
2695 if (errno != 0) {
2696 PERROR("Consumer execl()");
2697 }
2698 /* Reaching this point, we got a failure on our execl(). */
2699 exit(EXIT_FAILURE);
2700 } else if (pid > 0) {
2701 ret = pid;
2702 } else {
2703 PERROR("start consumer fork");
2704 ret = -errno;
2705 }
2706 error:
2707 return ret;
2708 }
2709
2710 /*
2711 * Spawn the consumerd daemon and session daemon thread.
2712 */
2713 static int start_consumerd(struct consumer_data *consumer_data)
2714 {
2715 int ret;
2716
2717 /*
2718 * Set the listen() state on the socket since there is a possible race
2719 * between the exec() of the consumer daemon and this call if place in the
2720 * consumer thread. See bug #366 for more details.
2721 */
2722 ret = lttcomm_listen_unix_sock(consumer_data->err_sock);
2723 if (ret < 0) {
2724 goto error;
2725 }
2726
2727 pthread_mutex_lock(&consumer_data->pid_mutex);
2728 if (consumer_data->pid != 0) {
2729 pthread_mutex_unlock(&consumer_data->pid_mutex);
2730 goto end;
2731 }
2732
2733 ret = spawn_consumerd(consumer_data);
2734 if (ret < 0) {
2735 ERR("Spawning consumerd failed");
2736 pthread_mutex_unlock(&consumer_data->pid_mutex);
2737 goto error;
2738 }
2739
2740 /* Setting up the consumer_data pid */
2741 consumer_data->pid = ret;
2742 DBG2("Consumer pid %d", consumer_data->pid);
2743 pthread_mutex_unlock(&consumer_data->pid_mutex);
2744
2745 DBG2("Spawning consumer control thread");
2746 ret = spawn_consumer_thread(consumer_data);
2747 if (ret < 0) {
2748 ERR("Fatal error spawning consumer control thread");
2749 goto error;
2750 }
2751
2752 end:
2753 return 0;
2754
2755 error:
2756 /* Cleanup already created sockets on error. */
2757 if (consumer_data->err_sock >= 0) {
2758 int err;
2759
2760 err = close(consumer_data->err_sock);
2761 if (err < 0) {
2762 PERROR("close consumer data error socket");
2763 }
2764 }
2765 return ret;
2766 }
2767
2768 /*
2769 * Setup necessary data for kernel tracer action.
2770 */
2771 static int init_kernel_tracer(void)
2772 {
2773 int ret;
2774
2775 /* Modprobe lttng kernel modules */
2776 ret = modprobe_lttng_control();
2777 if (ret < 0) {
2778 goto error;
2779 }
2780
2781 /* Open debugfs lttng */
2782 kernel_tracer_fd = open(module_proc_lttng, O_RDWR);
2783 if (kernel_tracer_fd < 0) {
2784 DBG("Failed to open %s", module_proc_lttng);
2785 ret = -1;
2786 goto error_open;
2787 }
2788
2789 /* Validate kernel version */
2790 ret = kernel_validate_version(kernel_tracer_fd);
2791 if (ret < 0) {
2792 goto error_version;
2793 }
2794
2795 ret = modprobe_lttng_data();
2796 if (ret < 0) {
2797 goto error_modules;
2798 }
2799
2800 DBG("Kernel tracer fd %d", kernel_tracer_fd);
2801 return 0;
2802
2803 error_version:
2804 modprobe_remove_lttng_control();
2805 ret = close(kernel_tracer_fd);
2806 if (ret) {
2807 PERROR("close");
2808 }
2809 kernel_tracer_fd = -1;
2810 return LTTNG_ERR_KERN_VERSION;
2811
2812 error_modules:
2813 ret = close(kernel_tracer_fd);
2814 if (ret) {
2815 PERROR("close");
2816 }
2817
2818 error_open:
2819 modprobe_remove_lttng_control();
2820
2821 error:
2822 WARN("No kernel tracer available");
2823 kernel_tracer_fd = -1;
2824 if (!is_root) {
2825 return LTTNG_ERR_NEED_ROOT_SESSIOND;
2826 } else {
2827 return LTTNG_ERR_KERN_NA;
2828 }
2829 }
2830
2831
2832 /*
2833 * Copy consumer output from the tracing session to the domain session. The
2834 * function also applies the right modification on a per domain basis for the
2835 * trace files destination directory.
2836 *
2837 * Should *NOT* be called with RCU read-side lock held.
2838 */
2839 static int copy_session_consumer(int domain, struct ltt_session *session)
2840 {
2841 int ret;
2842 const char *dir_name;
2843 struct consumer_output *consumer;
2844
2845 assert(session);
2846 assert(session->consumer);
2847
2848 switch (domain) {
2849 case LTTNG_DOMAIN_KERNEL:
2850 DBG3("Copying tracing session consumer output in kernel session");
2851 /*
2852 * XXX: We should audit the session creation and what this function
2853 * does "extra" in order to avoid a destroy since this function is used
2854 * in the domain session creation (kernel and ust) only. Same for UST
2855 * domain.
2856 */
2857 if (session->kernel_session->consumer) {
2858 consumer_output_put(session->kernel_session->consumer);
2859 }
2860 session->kernel_session->consumer =
2861 consumer_copy_output(session->consumer);
2862 /* Ease our life a bit for the next part */
2863 consumer = session->kernel_session->consumer;
2864 dir_name = DEFAULT_KERNEL_TRACE_DIR;
2865 break;
2866 case LTTNG_DOMAIN_JUL:
2867 case LTTNG_DOMAIN_LOG4J:
2868 case LTTNG_DOMAIN_PYTHON:
2869 case LTTNG_DOMAIN_UST:
2870 DBG3("Copying tracing session consumer output in UST session");
2871 if (session->ust_session->consumer) {
2872 consumer_output_put(session->ust_session->consumer);
2873 }
2874 session->ust_session->consumer =
2875 consumer_copy_output(session->consumer);
2876 /* Ease our life a bit for the next part */
2877 consumer = session->ust_session->consumer;
2878 dir_name = DEFAULT_UST_TRACE_DIR;
2879 break;
2880 default:
2881 ret = LTTNG_ERR_UNKNOWN_DOMAIN;
2882 goto error;
2883 }
2884
2885 /* Append correct directory to subdir */
2886 strncat(consumer->subdir, dir_name,
2887 sizeof(consumer->subdir) - strlen(consumer->subdir) - 1);
2888 DBG3("Copy session consumer subdir %s", consumer->subdir);
2889
2890 ret = LTTNG_OK;
2891
2892 error:
2893 return ret;
2894 }
2895
2896 /*
2897 * Create an UST session and add it to the session ust list.
2898 *
2899 * Should *NOT* be called with RCU read-side lock held.
2900 */
2901 static int create_ust_session(struct ltt_session *session,
2902 struct lttng_domain *domain)
2903 {
2904 int ret;
2905 struct ltt_ust_session *lus = NULL;
2906
2907 assert(session);
2908 assert(domain);
2909 assert(session->consumer);
2910
2911 switch (domain->type) {
2912 case LTTNG_DOMAIN_JUL:
2913 case LTTNG_DOMAIN_LOG4J:
2914 case LTTNG_DOMAIN_PYTHON:
2915 case LTTNG_DOMAIN_UST:
2916 break;
2917 default:
2918 ERR("Unknown UST domain on create session %d", domain->type);
2919 ret = LTTNG_ERR_UNKNOWN_DOMAIN;
2920 goto error;
2921 }
2922
2923 DBG("Creating UST session");
2924
2925 lus = trace_ust_create_session(session->id);
2926 if (lus == NULL) {
2927 ret = LTTNG_ERR_UST_SESS_FAIL;
2928 goto error;
2929 }
2930
2931 lus->uid = session->uid;
2932 lus->gid = session->gid;
2933 lus->output_traces = session->output_traces;
2934 lus->snapshot_mode = session->snapshot_mode;
2935 lus->live_timer_interval = session->live_timer;
2936 session->ust_session = lus;
2937 if (session->shm_path[0]) {
2938 strncpy(lus->root_shm_path, session->shm_path,
2939 sizeof(lus->root_shm_path));
2940 lus->root_shm_path[sizeof(lus->root_shm_path) - 1] = '\0';
2941 strncpy(lus->shm_path, session->shm_path,
2942 sizeof(lus->shm_path));
2943 lus->shm_path[sizeof(lus->shm_path) - 1] = '\0';
2944 strncat(lus->shm_path, "/ust",
2945 sizeof(lus->shm_path) - strlen(lus->shm_path) - 1);
2946 }
2947 /* Copy session output to the newly created UST session */
2948 ret = copy_session_consumer(domain->type, session);
2949 if (ret != LTTNG_OK) {
2950 goto error;
2951 }
2952
2953 return LTTNG_OK;
2954
2955 error:
2956 free(lus);
2957 session->ust_session = NULL;
2958 return ret;
2959 }
2960
2961 /*
2962 * Create a kernel tracer session then create the default channel.
2963 */
2964 static int create_kernel_session(struct ltt_session *session)
2965 {
2966 int ret;
2967
2968 DBG("Creating kernel session");
2969
2970 ret = kernel_create_session(session, kernel_tracer_fd);
2971 if (ret < 0) {
2972 ret = LTTNG_ERR_KERN_SESS_FAIL;
2973 goto error;
2974 }
2975
2976 /* Code flow safety */
2977 assert(session->kernel_session);
2978
2979 /* Copy session output to the newly created Kernel session */
2980 ret = copy_session_consumer(LTTNG_DOMAIN_KERNEL, session);
2981 if (ret != LTTNG_OK) {
2982 goto error;
2983 }
2984
2985 /* Create directory(ies) on local filesystem. */
2986 if (session->kernel_session->consumer->type == CONSUMER_DST_LOCAL &&
2987 strlen(session->kernel_session->consumer->dst.trace_path) > 0) {
2988 ret = run_as_mkdir_recursive(
2989 session->kernel_session->consumer->dst.trace_path,
2990 S_IRWXU | S_IRWXG, session->uid, session->gid);
2991 if (ret < 0) {
2992 if (errno != EEXIST) {
2993 ERR("Trace directory creation error");
2994 goto error;
2995 }
2996 }
2997 }
2998
2999 session->kernel_session->uid = session->uid;
3000 session->kernel_session->gid = session->gid;
3001 session->kernel_session->output_traces = session->output_traces;
3002 session->kernel_session->snapshot_mode = session->snapshot_mode;
3003
3004 return LTTNG_OK;
3005
3006 error:
3007 trace_kernel_destroy_session(session->kernel_session);
3008 session->kernel_session = NULL;
3009 return ret;
3010 }
3011
3012 /*
3013 * Count number of session permitted by uid/gid.
3014 */
3015 static unsigned int lttng_sessions_count(uid_t uid, gid_t gid)
3016 {
3017 unsigned int i = 0;
3018 struct ltt_session *session;
3019
3020 DBG("Counting number of available session for UID %d GID %d",
3021 uid, gid);
3022 cds_list_for_each_entry(session, &session_list_ptr->head, list) {
3023 /*
3024 * Only list the sessions the user can control.
3025 */
3026 if (!session_access_ok(session, uid, gid)) {
3027 continue;
3028 }
3029 i++;
3030 }
3031 return i;
3032 }
3033
3034 /*
3035 * Process the command requested by the lttng client within the command
3036 * context structure. This function make sure that the return structure (llm)
3037 * is set and ready for transmission before returning.
3038 *
3039 * Return any error encountered or 0 for success.
3040 *
3041 * "sock" is only used for special-case var. len data.
3042 *
3043 * Should *NOT* be called with RCU read-side lock held.
3044 */
3045 static int process_client_msg(struct command_ctx *cmd_ctx, int sock,
3046 int *sock_error)
3047 {
3048 int ret = LTTNG_OK;
3049 int need_tracing_session = 1;
3050 int need_domain;
3051
3052 DBG("Processing client command %d", cmd_ctx->lsm->cmd_type);
3053
3054 assert(!rcu_read_ongoing());
3055
3056 *sock_error = 0;
3057
3058 switch (cmd_ctx->lsm->cmd_type) {
3059 case LTTNG_CREATE_SESSION:
3060 case LTTNG_CREATE_SESSION_SNAPSHOT:
3061 case LTTNG_CREATE_SESSION_LIVE:
3062 case LTTNG_DESTROY_SESSION:
3063 case LTTNG_LIST_SESSIONS:
3064 case LTTNG_LIST_DOMAINS:
3065 case LTTNG_START_TRACE:
3066 case LTTNG_STOP_TRACE:
3067 case LTTNG_DATA_PENDING:
3068 case LTTNG_SNAPSHOT_ADD_OUTPUT:
3069 case LTTNG_SNAPSHOT_DEL_OUTPUT:
3070 case LTTNG_SNAPSHOT_LIST_OUTPUT:
3071 case LTTNG_SNAPSHOT_RECORD:
3072 case LTTNG_SAVE_SESSION:
3073 case LTTNG_SET_SESSION_SHM_PATH:
3074 case LTTNG_REGENERATE_METADATA:
3075 case LTTNG_REGENERATE_STATEDUMP:
3076 case LTTNG_REGISTER_TRIGGER:
3077 case LTTNG_UNREGISTER_TRIGGER:
3078 need_domain = 0;
3079 break;
3080 default:
3081 need_domain = 1;
3082 }
3083
3084 if (opt_no_kernel && need_domain
3085 && cmd_ctx->lsm->domain.type == LTTNG_DOMAIN_KERNEL) {
3086 if (!is_root) {
3087 ret = LTTNG_ERR_NEED_ROOT_SESSIOND;
3088 } else {
3089 ret = LTTNG_ERR_KERN_NA;
3090 }
3091 goto error;
3092 }
3093
3094 /* Deny register consumer if we already have a spawned consumer. */
3095 if (cmd_ctx->lsm->cmd_type == LTTNG_REGISTER_CONSUMER) {
3096 pthread_mutex_lock(&kconsumer_data.pid_mutex);
3097 if (kconsumer_data.pid > 0) {
3098 ret = LTTNG_ERR_KERN_CONSUMER_FAIL;
3099 pthread_mutex_unlock(&kconsumer_data.pid_mutex);
3100 goto error;
3101 }
3102 pthread_mutex_unlock(&kconsumer_data.pid_mutex);
3103 }
3104
3105 /*
3106 * Check for command that don't needs to allocate a returned payload. We do
3107 * this here so we don't have to make the call for no payload at each
3108 * command.
3109 */
3110 switch(cmd_ctx->lsm->cmd_type) {
3111 case LTTNG_LIST_SESSIONS:
3112 case LTTNG_LIST_TRACEPOINTS:
3113 case LTTNG_LIST_TRACEPOINT_FIELDS:
3114 case LTTNG_LIST_DOMAINS:
3115 case LTTNG_LIST_CHANNELS:
3116 case LTTNG_LIST_EVENTS:
3117 case LTTNG_LIST_SYSCALLS:
3118 case LTTNG_LIST_TRACKER_PIDS:
3119 case LTTNG_DATA_PENDING:
3120 break;
3121 default:
3122 /* Setup lttng message with no payload */
3123 ret = setup_lttng_msg_no_cmd_header(cmd_ctx, NULL, 0);
3124 if (ret < 0) {
3125 /* This label does not try to unlock the session */
3126 goto init_setup_error;
3127 }
3128 }
3129
3130 /* Commands that DO NOT need a session. */
3131 switch (cmd_ctx->lsm->cmd_type) {
3132 case LTTNG_CREATE_SESSION:
3133 case LTTNG_CREATE_SESSION_SNAPSHOT:
3134 case LTTNG_CREATE_SESSION_LIVE:
3135 case LTTNG_LIST_SESSIONS:
3136 case LTTNG_LIST_TRACEPOINTS:
3137 case LTTNG_LIST_SYSCALLS:
3138 case LTTNG_LIST_TRACEPOINT_FIELDS:
3139 case LTTNG_SAVE_SESSION:
3140 case LTTNG_REGISTER_TRIGGER:
3141 case LTTNG_UNREGISTER_TRIGGER:
3142 need_tracing_session = 0;
3143 break;
3144 default:
3145 DBG("Getting session %s by name", cmd_ctx->lsm->session.name);
3146 /*
3147 * We keep the session list lock across _all_ commands
3148 * for now, because the per-session lock does not
3149 * handle teardown properly.
3150 */
3151 session_lock_list();
3152 cmd_ctx->session = session_find_by_name(cmd_ctx->lsm->session.name);
3153 if (cmd_ctx->session == NULL) {
3154 ret = LTTNG_ERR_SESS_NOT_FOUND;
3155 goto error;
3156 } else {
3157 /* Acquire lock for the session */
3158 session_lock(cmd_ctx->session);
3159 }
3160 break;
3161 }
3162
3163 /*
3164 * Commands that need a valid session but should NOT create one if none
3165 * exists. Instead of creating one and destroying it when the command is
3166 * handled, process that right before so we save some round trip in useless
3167 * code path.
3168 */
3169 switch (cmd_ctx->lsm->cmd_type) {
3170 case LTTNG_DISABLE_CHANNEL:
3171 case LTTNG_DISABLE_EVENT:
3172 switch (cmd_ctx->lsm->domain.type) {
3173 case LTTNG_DOMAIN_KERNEL:
3174 if (!cmd_ctx->session->kernel_session) {
3175 ret = LTTNG_ERR_NO_CHANNEL;
3176 goto error;
3177 }
3178 break;
3179 case LTTNG_DOMAIN_JUL:
3180 case LTTNG_DOMAIN_LOG4J:
3181 case LTTNG_DOMAIN_PYTHON:
3182 case LTTNG_DOMAIN_UST:
3183 if (!cmd_ctx->session->ust_session) {
3184 ret = LTTNG_ERR_NO_CHANNEL;
3185 goto error;
3186 }
3187 break;
3188 default:
3189 ret = LTTNG_ERR_UNKNOWN_DOMAIN;
3190 goto error;
3191 }
3192 default:
3193 break;
3194 }
3195
3196 if (!need_domain) {
3197 goto skip_domain;
3198 }
3199
3200 /*
3201 * Check domain type for specific "pre-action".
3202 */
3203 switch (cmd_ctx->lsm->domain.type) {
3204 case LTTNG_DOMAIN_KERNEL:
3205 if (!is_root) {
3206 ret = LTTNG_ERR_NEED_ROOT_SESSIOND;
3207 goto error;
3208 }
3209
3210 /* Kernel tracer check */
3211 if (kernel_tracer_fd == -1) {
3212 /* Basically, load kernel tracer modules */
3213 ret = init_kernel_tracer();
3214 if (ret != 0) {
3215 goto error;
3216 }
3217 }
3218
3219 /* Consumer is in an ERROR state. Report back to client */
3220 if (uatomic_read(&kernel_consumerd_state) == CONSUMER_ERROR) {
3221 ret = LTTNG_ERR_NO_KERNCONSUMERD;
3222 goto error;
3223 }
3224
3225 /* Need a session for kernel command */
3226 if (need_tracing_session) {
3227 if (cmd_ctx->session->kernel_session == NULL) {
3228 ret = create_kernel_session(cmd_ctx->session);
3229 if (ret < 0) {
3230 ret = LTTNG_ERR_KERN_SESS_FAIL;
3231 goto error;
3232 }
3233 }
3234
3235 /* Start the kernel consumer daemon */
3236 pthread_mutex_lock(&kconsumer_data.pid_mutex);
3237 if (kconsumer_data.pid == 0 &&