2 * Copyright (C) 2017 - Julien Desfossez <jdesfossez@efficios.com>
4 * This program is free software; you can redistribute it and/or modify it
5 * under the terms of the GNU General Public License, version 2 only, as
6 * published by the Free Software Foundation.
8 * This program is distributed in the hope that it will be useful, but WITHOUT
9 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
13 * You should have received a copy of the GNU General Public License along with
14 * this program; if not, write to the Free Software Foundation, Inc., 51
15 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
19 #include <lttng/trigger/trigger.h>
20 #include <common/error.h>
21 #include <common/config/session-config.h>
22 #include <common/defaults.h>
23 #include <common/utils.h>
24 #include <common/futex.h>
25 #include <common/align.h>
26 #include <common/time.h>
27 #include <common/hashtable/utils.h>
28 #include <sys/eventfd.h>
34 #include <common/kernel-ctl/kernel-ctl.h>
35 #include "rotation-thread.h"
36 #include "lttng-sessiond.h"
37 #include "health-sessiond.h"
41 #include "sessiond-timer.h"
44 #include <urcu/list.h>
45 #include <urcu/rculfhash.h>
47 struct cds_lfht
*channel_pending_rotate_ht
;
50 void channel_rotation_info_destroy(struct rotation_channel_info
*channel_info
)
57 int match_channel_info(struct cds_lfht_node
*node
, const void *key
)
59 struct rotation_channel_key
*channel_key
= (struct rotation_channel_key
*) key
;
60 struct rotation_channel_info
*channel_info
;
62 channel_info
= caa_container_of(node
, struct rotation_channel_info
,
63 rotate_channels_ht_node
);
65 return !!((channel_key
->key
== channel_info
->channel_key
.key
) &&
66 (channel_key
->domain
== channel_info
->channel_key
.domain
));
70 struct rotation_channel_info
*lookup_channel_pending(uint64_t key
,
71 enum lttng_domain_type domain
)
73 struct cds_lfht_iter iter
;
74 struct cds_lfht_node
*node
;
75 struct rotation_channel_info
*channel_info
= NULL
;
76 struct rotation_channel_key channel_key
= { .key
= key
,
79 cds_lfht_lookup(channel_pending_rotate_ht
,
80 hash_channel_key(&channel_key
),
83 node
= cds_lfht_iter_get_node(&iter
);
88 channel_info
= caa_container_of(node
, struct rotation_channel_info
,
89 rotate_channels_ht_node
);
90 cds_lfht_del(channel_pending_rotate_ht
, node
);
96 * Destroy the thread data previously created by the init function.
98 void rotation_thread_handle_destroy(
99 struct rotation_thread_handle
*handle
)
107 if (handle
->ust32_consumer
>= 0) {
108 ret
= close(handle
->ust32_consumer
);
110 PERROR("close 32-bit consumer channel rotation pipe");
113 if (handle
->ust64_consumer
>= 0) {
114 ret
= close(handle
->ust64_consumer
);
116 PERROR("close 64-bit consumer channel rotation pipe");
119 if (handle
->kernel_consumer
>= 0) {
120 ret
= close(handle
->kernel_consumer
);
122 PERROR("close kernel consumer channel rotation pipe");
130 struct rotation_thread_handle
*rotation_thread_handle_create(
131 struct lttng_pipe
*ust32_channel_rotate_pipe
,
132 struct lttng_pipe
*ust64_channel_rotate_pipe
,
133 struct lttng_pipe
*kernel_channel_rotate_pipe
,
134 int thread_quit_pipe
, int rotate_timer_pipe
)
136 struct rotation_thread_handle
*handle
;
138 handle
= zmalloc(sizeof(*handle
));
143 if (ust32_channel_rotate_pipe
) {
144 handle
->ust32_consumer
=
145 lttng_pipe_release_readfd(
146 ust32_channel_rotate_pipe
);
147 if (handle
->ust32_consumer
< 0) {
151 handle
->ust32_consumer
= -1;
153 if (ust64_channel_rotate_pipe
) {
154 handle
->ust64_consumer
=
155 lttng_pipe_release_readfd(
156 ust64_channel_rotate_pipe
);
157 if (handle
->ust64_consumer
< 0) {
161 handle
->ust64_consumer
= -1;
163 if (kernel_channel_rotate_pipe
) {
164 handle
->kernel_consumer
=
165 lttng_pipe_release_readfd(
166 kernel_channel_rotate_pipe
);
167 if (handle
->kernel_consumer
< 0) {
171 handle
->kernel_consumer
= -1;
173 handle
->thread_quit_pipe
= thread_quit_pipe
;
174 handle
->rotate_timer_pipe
= rotate_timer_pipe
;
179 rotation_thread_handle_destroy(handle
);
184 int init_poll_set(struct lttng_poll_event
*poll_set
,
185 struct rotation_thread_handle
*handle
)
190 * Create pollset with size 5:
191 * - sessiond quit pipe
192 * - consumerd (32-bit user space) channel rotate pipe,
193 * - consumerd (64-bit user space) channel rotate pipe,
194 * - consumerd (kernel) channel rotate pipe.
195 * - sessiond rotate pending pipe
197 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
202 ret
= lttng_poll_add(poll_set
, handle
->thread_quit_pipe
,
205 ERR("[rotation-thread] Failed to add thread_quit_pipe fd to pollset");
208 ret
= lttng_poll_add(poll_set
, handle
->rotate_timer_pipe
,
211 ERR("[rotation-thread] Failed to add rotate_pending fd to pollset");
214 ret
= lttng_poll_add(poll_set
, handle
->ust32_consumer
,
217 ERR("[rotation-thread] Failed to add ust-32 channel rotation pipe fd to pollset");
220 ret
= lttng_poll_add(poll_set
, handle
->ust64_consumer
,
223 ERR("[rotation-thread] Failed to add ust-64 channel rotation pipe fd to pollset");
226 if (handle
->kernel_consumer
< 0) {
229 ret
= lttng_poll_add(poll_set
, handle
->kernel_consumer
,
232 ERR("[rotation-thread] Failed to add kernel channel rotation pipe fd to pollset");
239 lttng_poll_clean(poll_set
);
244 void fini_thread_state(struct rotation_thread_state
*state
)
246 lttng_poll_clean(&state
->events
);
250 int init_thread_state(struct rotation_thread_handle
*handle
,
251 struct rotation_thread_state
*state
)
255 memset(state
, 0, sizeof(*state
));
256 lttng_poll_init(&state
->events
);
258 ret
= init_poll_set(&state
->events
, handle
);
263 channel_pending_rotate_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
264 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
265 if (!channel_pending_rotate_ht
) {
274 int handle_channel_rotation_pipe(int fd
, uint32_t revents
,
275 struct rotation_thread_handle
*handle
,
276 struct rotation_thread_state
*state
)
279 enum lttng_domain_type domain
;
280 struct rotation_channel_info
*channel_info
;
281 struct ltt_session
*session
= NULL
;
284 if (fd
== handle
->ust32_consumer
||
285 fd
== handle
->ust64_consumer
) {
286 domain
= LTTNG_DOMAIN_UST
;
287 } else if (fd
== handle
->kernel_consumer
) {
288 domain
= LTTNG_DOMAIN_KERNEL
;
293 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
294 ret
= lttng_poll_del(&state
->events
, fd
);
296 ERR("[rotation-thread] Failed to remove consumer "
297 "rotation pipe from poll set");
303 ret
= read(fd
, &key
, sizeof(key
));
304 } while (ret
== -1 && errno
== EINTR
);
305 if (ret
!= sizeof(key
)) {
306 ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
312 DBG("[rotation-thread] Received notification for chan %" PRIu64
313 ", domain %d\n", key
, domain
);
315 channel_info
= lookup_channel_pending(key
, domain
);
317 ERR("[rotation-thread] Failed to find channel_info (key = %"
324 session
= session_find_by_id(channel_info
->session_id
);
326 ERR("[rotation-thread] Session %" PRIu64
" not found",
327 channel_info
->session_id
);
332 if (--session
->nr_chan_rotate_pending
== 0) {
333 time_t now
= time(NULL
);
335 if (now
== (time_t) -1) {
336 session
->rotate_status
= LTTNG_ROTATE_ERROR
;
337 ret
= LTTNG_ERR_ROTATE_NOT_AVAILABLE
;
340 session_lock(session
);
342 ret
= rename_complete_chunk(session
, now
);
344 ERR("Failed to rename completed rotation chunk");
345 session_unlock(session
);
348 session
->rotate_pending
= false;
349 if (session
->rotate_pending_relay
) {
350 ret
= sessiond_timer_rotate_pending_start(
352 DEFAULT_ROTATE_PENDING_RELAY_TIMER
);
354 ERR("Enabling rotate pending timer");
356 session_unlock(session
);
360 session_unlock(session
);
363 channel_rotation_info_destroy(channel_info
);
368 session_unlock_list();
375 int rotate_pending_relay_timer(struct ltt_session
*session
)
379 DBG("[rotation-thread] Check rotate pending on session %" PRIu64
,
381 ret
= relay_rotate_pending(session
, session
->rotate_count
- 1);
383 ERR("[rotation-thread] Check relay rotate pending");
387 DBG("[rotation-thread] Rotation completed on the relay for "
388 "session %" PRIu64
, session
->id
);
390 * Stop the timer and clear the queue, the timers are currently
391 * ignored because of the rotate_pending_relay_check_in_progress
394 sessiond_timer_rotate_pending_stop(session
);
396 * Now we can clear the pending flag in the session. New
397 * rotations can start now.
399 session
->rotate_pending_relay
= false;
400 } else if (ret
== 1) {
401 DBG("[rotation-thread] Rotation still pending on the relay for "
402 "session %" PRIu64
, session
->id
);
405 * Allow the timer thread to send other notifications when needed.
407 session
->rotate_pending_relay_check_in_progress
= false;
416 int rotate_timer(struct ltt_session
*session
)
420 DBG("[rotation-thread] Rotate timer on session %" PRIu64
, session
->id
);
423 * If the session is stopped, we need to cancel this timer.
425 session_lock(session
);
426 if (!session
->active
&& session
->rotate_timer_enabled
) {
427 sessiond_rotate_timer_stop(session
);
430 ret
= cmd_rotate_session(session
, NULL
);
431 session_unlock(session
);
432 if (ret
== -LTTNG_ERR_ROTATE_PENDING
) {
435 } else if (ret
!= LTTNG_OK
) {
448 int handle_rotate_timer_pipe(int fd
, uint32_t revents
,
449 struct rotation_thread_handle
*handle
,
450 struct rotation_thread_state
*state
)
453 struct ltt_session
*session
;
454 struct sessiond_rotation_timer timer_data
;
456 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
457 ret
= lttng_poll_del(&state
->events
, fd
);
459 ERR("[rotation-thread] Failed to remove consumer "
460 "rotate pending pipe from poll set");
465 memset(&timer_data
, 0, sizeof(struct sessiond_rotation_timer
));
468 ret
= read(fd
, &timer_data
, sizeof(timer_data
));
469 } while (ret
== -1 && errno
== EINTR
);
470 if (ret
!= sizeof(timer_data
)) {
471 ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
479 session
= session_find_by_id(timer_data
.session_id
);
481 ERR("[rotation-thread] Session %" PRIu64
" not found",
482 timer_data
.session_id
);
484 * This is a non-fatal error, and we cannot report it to the
485 * user (timer), so just print the error and continue the
492 if (timer_data
.signal
== LTTNG_SESSIOND_SIG_ROTATE_PENDING
) {
493 ret
= rotate_pending_relay_timer(session
);
494 } else if (timer_data
.signal
== LTTNG_SESSIOND_SIG_ROTATE_TIMER
) {
495 ret
= rotate_timer(session
);
497 ERR("Unknown signal in rotate timer");
502 session_unlock_list();
508 void *thread_rotation(void *data
)
511 struct rotation_thread_handle
*handle
= data
;
512 struct rotation_thread_state state
;
514 DBG("[rotation-thread] Started rotation thread");
517 ERR("[rotation-thread] Invalid thread context provided");
521 rcu_register_thread();
524 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
525 health_code_update();
527 ret
= init_thread_state(handle
, &state
);
532 /* Ready to handle client connections. */
533 sessiond_notify_ready();
539 DBG("[rotation-thread] Entering poll wait");
540 ret
= lttng_poll_wait(&state
.events
, -1);
541 DBG("[rotation-thread] Poll wait returned (%i)", ret
);
545 * Restart interrupted system call.
547 if (errno
== EINTR
) {
550 ERR("[rotation-thread] Error encountered during lttng_poll_wait (%i)", ret
);
555 for (i
= 0; i
< fd_count
; i
++) {
556 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
557 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
559 DBG("[rotation-thread] Handling fd (%i) activity (%u)",
562 if (fd
== handle
->thread_quit_pipe
) {
563 DBG("[rotation-thread] Quit pipe activity");
565 } else if (fd
== handle
->rotate_timer_pipe
) {
566 ret
= handle_rotate_timer_pipe(fd
, revents
,
569 ERR("[rotation-thread] Rotate timer");
572 } else if (fd
== handle
->ust32_consumer
||
573 fd
== handle
->ust64_consumer
||
574 fd
== handle
->kernel_consumer
) {
575 ret
= handle_channel_rotation_pipe(fd
,
576 revents
, handle
, &state
);
578 ERR("[rotation-thread] Exit main loop");
586 fini_thread_state(&state
);
587 health_unregister(health_sessiond
);
588 rcu_thread_offline();
589 rcu_unregister_thread();