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