2 * Copyright (C) 2011 EfficiOS Inc.
3 * Copyright (C) 2011 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
4 * Copyright (C) 2013 Jérémie Galarneau <jeremie.galarneau@efficios.com>
6 * SPDX-License-Identifier: GPL-2.0-only
12 #include <common/pipe.h>
13 #include <common/utils.h>
15 #include "manage-consumer.h"
16 #include "testpoint.h"
17 #include "health-sessiond.h"
20 #include "ust-consumer.h"
22 struct thread_notifiers
{
23 struct lttng_pipe
*quit_pipe
;
24 struct consumer_data
*consumer_data
;
26 int initialization_result
;
29 static void mark_thread_as_ready(struct thread_notifiers
*notifiers
)
31 DBG("Marking consumer management thread as ready");
32 notifiers
->initialization_result
= 0;
33 sem_post(¬ifiers
->ready
);
36 static void mark_thread_intialization_as_failed(
37 struct thread_notifiers
*notifiers
)
39 ERR("Consumer management thread entering error state");
40 notifiers
->initialization_result
= -1;
41 sem_post(¬ifiers
->ready
);
44 static void wait_until_thread_is_ready(struct thread_notifiers
*notifiers
)
46 DBG("Waiting for consumer management thread to be ready");
47 sem_wait(¬ifiers
->ready
);
48 DBG("Consumer management thread is ready");
52 * This thread manage the consumer error sent back to the session daemon.
54 static void *thread_consumer_management(void *data
)
56 int sock
= -1, i
, ret
, pollfd
, err
= -1, should_quit
= 0;
57 uint32_t revents
, nb_fd
;
58 enum lttcomm_return_code code
;
59 struct lttng_poll_event events
;
60 struct thread_notifiers
*notifiers
= data
;
61 struct consumer_data
*consumer_data
= notifiers
->consumer_data
;
62 const int quit_pipe_read_fd
= lttng_pipe_get_readfd(notifiers
->quit_pipe
);
63 struct consumer_socket
*cmd_socket_wrapper
= NULL
;
65 DBG("[thread] Manage consumer started");
67 rcu_register_thread();
70 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_CONSUMER
);
75 * Pass 3 as size here for the thread quit pipe, consumerd_err_sock and the
76 * metadata_sock. Nothing more will be added to this poll set.
78 ret
= lttng_poll_create(&events
, 3, LTTNG_CLOEXEC
);
80 mark_thread_intialization_as_failed(notifiers
);
84 ret
= lttng_poll_add(&events
, quit_pipe_read_fd
, LPOLLIN
| LPOLLERR
);
86 mark_thread_intialization_as_failed(notifiers
);
91 * The error socket here is already in a listening state which was done
92 * just before spawning this thread to avoid a race between the consumer
93 * daemon exec trying to connect and the listen() call.
95 ret
= lttng_poll_add(&events
, consumer_data
->err_sock
, LPOLLIN
| LPOLLRDHUP
);
97 mark_thread_intialization_as_failed(notifiers
);
101 health_code_update();
103 /* Infinite blocking call, waiting for transmission */
106 if (testpoint(sessiond_thread_manage_consumer
)) {
107 mark_thread_intialization_as_failed(notifiers
);
111 ret
= lttng_poll_wait(&events
, -1);
114 mark_thread_intialization_as_failed(notifiers
);
120 for (i
= 0; i
< nb_fd
; i
++) {
121 /* Fetch once the poll data */
122 revents
= LTTNG_POLL_GETEV(&events
, i
);
123 pollfd
= LTTNG_POLL_GETFD(&events
, i
);
125 health_code_update();
127 /* Thread quit pipe has been closed. Killing thread. */
128 if (pollfd
== quit_pipe_read_fd
) {
130 mark_thread_intialization_as_failed(notifiers
);
132 } else if (pollfd
== consumer_data
->err_sock
) {
133 /* Event on the registration socket */
134 if (revents
& LPOLLIN
) {
136 } else if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
137 ERR("consumer err socket poll error");
138 mark_thread_intialization_as_failed(notifiers
);
141 ERR("Unexpected poll events %u for sock %d", revents
, pollfd
);
142 mark_thread_intialization_as_failed(notifiers
);
148 sock
= lttcomm_accept_unix_sock(consumer_data
->err_sock
);
150 mark_thread_intialization_as_failed(notifiers
);
155 * Set the CLOEXEC flag. Return code is useless because either way, the
158 (void) utils_set_fd_cloexec(sock
);
160 health_code_update();
162 DBG2("Receiving code from consumer err_sock");
164 /* Getting status code from consumerd */
166 int32_t comm_code
= 0;
168 ret
= lttcomm_recv_unix_sock(sock
, &comm_code
, sizeof(comm_code
));
169 code
= (typeof(code
)) comm_code
;
172 mark_thread_intialization_as_failed(notifiers
);
176 health_code_update();
177 if (code
!= LTTCOMM_CONSUMERD_COMMAND_SOCK_READY
) {
178 ERR("consumer error when waiting for SOCK_READY : %s",
179 lttcomm_get_readable_code(-code
));
180 mark_thread_intialization_as_failed(notifiers
);
184 /* Connect both command and metadata sockets. */
185 consumer_data
->cmd_sock
=
186 lttcomm_connect_unix_sock(
187 consumer_data
->cmd_unix_sock_path
);
188 consumer_data
->metadata_fd
=
189 lttcomm_connect_unix_sock(
190 consumer_data
->cmd_unix_sock_path
);
191 if (consumer_data
->cmd_sock
< 0 || consumer_data
->metadata_fd
< 0) {
192 PERROR("consumer connect cmd socket");
193 mark_thread_intialization_as_failed(notifiers
);
197 consumer_data
->metadata_sock
.fd_ptr
= &consumer_data
->metadata_fd
;
199 /* Create metadata socket lock. */
200 consumer_data
->metadata_sock
.lock
= zmalloc(sizeof(pthread_mutex_t
));
201 if (consumer_data
->metadata_sock
.lock
== NULL
) {
202 PERROR("zmalloc pthread mutex");
203 mark_thread_intialization_as_failed(notifiers
);
206 pthread_mutex_init(consumer_data
->metadata_sock
.lock
, NULL
);
208 DBG("Consumer command socket ready (fd: %d)", consumer_data
->cmd_sock
);
209 DBG("Consumer metadata socket ready (fd: %d)",
210 consumer_data
->metadata_fd
);
213 * Remove the consumerd error sock since we've established a connection.
215 ret
= lttng_poll_del(&events
, consumer_data
->err_sock
);
217 mark_thread_intialization_as_failed(notifiers
);
221 /* Add new accepted error socket. */
222 ret
= lttng_poll_add(&events
, sock
, LPOLLIN
| LPOLLRDHUP
);
224 mark_thread_intialization_as_failed(notifiers
);
228 /* Add metadata socket that is successfully connected. */
229 ret
= lttng_poll_add(&events
, consumer_data
->metadata_fd
,
230 LPOLLIN
| LPOLLRDHUP
);
232 mark_thread_intialization_as_failed(notifiers
);
236 health_code_update();
239 * Transfer the write-end of the channel monitoring pipe to the consumer
240 * by issuing a SET_CHANNEL_MONITOR_PIPE command.
242 cmd_socket_wrapper
= consumer_allocate_socket(&consumer_data
->cmd_sock
);
243 if (!cmd_socket_wrapper
) {
244 mark_thread_intialization_as_failed(notifiers
);
247 cmd_socket_wrapper
->lock
= &consumer_data
->lock
;
249 pthread_mutex_lock(cmd_socket_wrapper
->lock
);
250 ret
= consumer_init(cmd_socket_wrapper
, sessiond_uuid
);
252 ERR("Failed to send sessiond uuid to consumer daemon");
253 mark_thread_intialization_as_failed(notifiers
);
254 pthread_mutex_unlock(cmd_socket_wrapper
->lock
);
257 pthread_mutex_unlock(cmd_socket_wrapper
->lock
);
259 ret
= consumer_send_channel_monitor_pipe(cmd_socket_wrapper
,
260 consumer_data
->channel_monitor_pipe
);
262 mark_thread_intialization_as_failed(notifiers
);
266 /* Discard the socket wrapper as it is no longer needed. */
267 consumer_destroy_socket(cmd_socket_wrapper
);
268 cmd_socket_wrapper
= NULL
;
270 /* The thread is completely initialized, signal that it is ready. */
271 mark_thread_as_ready(notifiers
);
273 /* Infinite blocking call, waiting for transmission */
275 health_code_update();
277 /* Exit the thread because the thread quit pipe has been triggered. */
279 /* Not a health error. */
285 ret
= lttng_poll_wait(&events
, -1);
293 for (i
= 0; i
< nb_fd
; i
++) {
294 /* Fetch once the poll data */
295 revents
= LTTNG_POLL_GETEV(&events
, i
);
296 pollfd
= LTTNG_POLL_GETFD(&events
, i
);
298 health_code_update();
301 * Thread quit pipe has been triggered, flag that we should stop
302 * but continue the current loop to handle potential data from
305 if (pollfd
== quit_pipe_read_fd
) {
307 } else if (pollfd
== sock
) {
308 /* Event on the consumerd socket */
309 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)
310 && !(revents
& LPOLLIN
)) {
311 ERR("consumer err socket second poll error");
314 health_code_update();
315 /* Wait for any consumerd error */
317 int32_t comm_code
= 0;
319 ret
= lttcomm_recv_unix_sock(
320 sock
, &comm_code
, sizeof(comm_code
));
321 code
= (typeof(code
)) comm_code
;
324 ERR("consumer closed the command socket");
328 ERR("consumer return code : %s",
329 lttcomm_get_readable_code(-code
));
332 } else if (pollfd
== consumer_data
->metadata_fd
) {
333 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)
334 && !(revents
& LPOLLIN
)) {
335 ERR("consumer err metadata socket second poll error");
338 /* UST metadata requests */
339 ret
= ust_consumer_metadata_request(
340 &consumer_data
->metadata_sock
);
342 ERR("Handling metadata request");
346 /* No need for an else branch all FDs are tested prior. */
348 health_code_update();
354 * We lock here because we are about to close the sockets and some other
355 * thread might be using them so get exclusive access which will abort all
356 * other consumer command by other threads.
358 pthread_mutex_lock(&consumer_data
->lock
);
360 /* Immediately set the consumerd state to stopped */
361 if (consumer_data
->type
== LTTNG_CONSUMER_KERNEL
) {
362 uatomic_set(&kernel_consumerd_state
, CONSUMER_ERROR
);
363 } else if (consumer_data
->type
== LTTNG_CONSUMER64_UST
||
364 consumer_data
->type
== LTTNG_CONSUMER32_UST
) {
365 uatomic_set(&ust_consumerd_state
, CONSUMER_ERROR
);
367 /* Code flow error... */
371 if (consumer_data
->err_sock
>= 0) {
372 ret
= close(consumer_data
->err_sock
);
376 consumer_data
->err_sock
= -1;
378 if (consumer_data
->cmd_sock
>= 0) {
379 ret
= close(consumer_data
->cmd_sock
);
383 consumer_data
->cmd_sock
= -1;
385 if (consumer_data
->metadata_sock
.fd_ptr
&&
386 *consumer_data
->metadata_sock
.fd_ptr
>= 0) {
387 ret
= close(*consumer_data
->metadata_sock
.fd_ptr
);
399 unlink(consumer_data
->err_unix_sock_path
);
400 unlink(consumer_data
->cmd_unix_sock_path
);
401 pthread_mutex_unlock(&consumer_data
->lock
);
403 /* Cleanup metadata socket mutex. */
404 if (consumer_data
->metadata_sock
.lock
) {
405 pthread_mutex_destroy(consumer_data
->metadata_sock
.lock
);
406 free(consumer_data
->metadata_sock
.lock
);
408 lttng_poll_clean(&events
);
410 if (cmd_socket_wrapper
) {
411 consumer_destroy_socket(cmd_socket_wrapper
);
416 ERR("Health error occurred in %s", __func__
);
418 health_unregister(health_sessiond
);
419 DBG("consumer thread cleanup completed");
421 rcu_thread_offline();
422 rcu_unregister_thread();
427 static bool shutdown_consumer_management_thread(void *data
)
429 struct thread_notifiers
*notifiers
= data
;
430 const int write_fd
= lttng_pipe_get_writefd(notifiers
->quit_pipe
);
432 return notify_thread_pipe(write_fd
) == 1;
435 static void cleanup_consumer_management_thread(void *data
)
437 struct thread_notifiers
*notifiers
= data
;
439 lttng_pipe_destroy(notifiers
->quit_pipe
);
443 bool launch_consumer_management_thread(struct consumer_data
*consumer_data
)
445 struct lttng_pipe
*quit_pipe
;
446 struct thread_notifiers
*notifiers
= NULL
;
447 struct lttng_thread
*thread
;
449 notifiers
= zmalloc(sizeof(*notifiers
));
454 quit_pipe
= lttng_pipe_open(FD_CLOEXEC
);
458 notifiers
->quit_pipe
= quit_pipe
;
459 notifiers
->consumer_data
= consumer_data
;
460 sem_init(¬ifiers
->ready
, 0, 0);
462 thread
= lttng_thread_create("Consumer management",
463 thread_consumer_management
,
464 shutdown_consumer_management_thread
,
465 cleanup_consumer_management_thread
,
470 wait_until_thread_is_ready(notifiers
);
471 lttng_thread_put(thread
);
472 if (notifiers
->initialization_result
) {
477 cleanup_consumer_management_thread(notifiers
);