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