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 <lttng/notification/channel-internal.h>
37 #include "rotation-thread.h"
38 #include "lttng-sessiond.h"
39 #include "health-sessiond.h"
43 #include "sessiond-timer.h"
46 #include <urcu/list.h>
47 #include <urcu/rculfhash.h>
50 * Store a struct rotation_channel_info for each channel that is currently
51 * being rotated by the consumer.
53 struct cds_lfht
*channel_pending_rotate_ht
;
55 struct rotation_thread_state
{
56 struct lttng_poll_event events
;
60 void channel_rotation_info_destroy(struct rotation_channel_info
*channel_info
)
67 int match_channel_info(struct cds_lfht_node
*node
, const void *key
)
69 struct rotation_channel_key
*channel_key
= (struct rotation_channel_key
*) key
;
70 struct rotation_channel_info
*channel_info
;
72 channel_info
= caa_container_of(node
, struct rotation_channel_info
,
73 rotate_channels_ht_node
);
75 return !!((channel_key
->key
== channel_info
->channel_key
.key
) &&
76 (channel_key
->domain
== channel_info
->channel_key
.domain
));
80 struct rotation_channel_info
*lookup_channel_pending(uint64_t key
,
81 enum lttng_domain_type domain
)
83 struct cds_lfht_iter iter
;
84 struct cds_lfht_node
*node
;
85 struct rotation_channel_info
*channel_info
= NULL
;
86 struct rotation_channel_key channel_key
= { .key
= key
,
89 cds_lfht_lookup(channel_pending_rotate_ht
,
90 hash_channel_key(&channel_key
),
93 node
= cds_lfht_iter_get_node(&iter
);
98 channel_info
= caa_container_of(node
, struct rotation_channel_info
,
99 rotate_channels_ht_node
);
100 cds_lfht_del(channel_pending_rotate_ht
, node
);
106 * Destroy the thread data previously created by the init function.
108 void rotation_thread_handle_destroy(
109 struct rotation_thread_handle
*handle
)
117 if (handle
->ust32_consumer
>= 0) {
118 ret
= close(handle
->ust32_consumer
);
120 PERROR("close 32-bit consumer channel rotation pipe");
123 if (handle
->ust64_consumer
>= 0) {
124 ret
= close(handle
->ust64_consumer
);
126 PERROR("close 64-bit consumer channel rotation pipe");
129 if (handle
->kernel_consumer
>= 0) {
130 ret
= close(handle
->kernel_consumer
);
132 PERROR("close kernel consumer channel rotation pipe");
140 struct rotation_thread_handle
*rotation_thread_handle_create(
141 struct lttng_pipe
*ust32_channel_rotate_pipe
,
142 struct lttng_pipe
*ust64_channel_rotate_pipe
,
143 struct lttng_pipe
*kernel_channel_rotate_pipe
,
144 int thread_quit_pipe
,
145 struct rotation_thread_timer_queue
*rotation_timer_queue
)
147 struct rotation_thread_handle
*handle
;
149 handle
= zmalloc(sizeof(*handle
));
154 if (ust32_channel_rotate_pipe
) {
155 handle
->ust32_consumer
=
156 lttng_pipe_release_readfd(
157 ust32_channel_rotate_pipe
);
158 if (handle
->ust32_consumer
< 0) {
162 handle
->ust32_consumer
= -1;
164 if (ust64_channel_rotate_pipe
) {
165 handle
->ust64_consumer
=
166 lttng_pipe_release_readfd(
167 ust64_channel_rotate_pipe
);
168 if (handle
->ust64_consumer
< 0) {
172 handle
->ust64_consumer
= -1;
174 if (kernel_channel_rotate_pipe
) {
175 handle
->kernel_consumer
=
176 lttng_pipe_release_readfd(
177 kernel_channel_rotate_pipe
);
178 if (handle
->kernel_consumer
< 0) {
182 handle
->kernel_consumer
= -1;
184 handle
->thread_quit_pipe
= thread_quit_pipe
;
185 handle
->rotation_timer_queue
= rotation_timer_queue
;
190 rotation_thread_handle_destroy(handle
);
195 int init_poll_set(struct lttng_poll_event
*poll_set
,
196 struct rotation_thread_handle
*handle
)
201 * Create pollset with size 5:
202 * - sessiond quit pipe
203 * - sessiond timer pipe,
204 * - consumerd (32-bit user space) channel rotate pipe,
205 * - consumerd (64-bit user space) channel rotate pipe,
206 * - consumerd (kernel) channel rotate pipe,
208 ret
= lttng_poll_create(poll_set
, 5, LTTNG_CLOEXEC
);
213 ret
= lttng_poll_add(poll_set
, handle
->thread_quit_pipe
,
216 ERR("[rotation-thread] Failed to add thread_quit_pipe fd to pollset");
219 ret
= lttng_poll_add(poll_set
,
220 lttng_pipe_get_readfd(handle
->rotation_timer_queue
->event_pipe
),
223 ERR("[rotation-thread] Failed to add rotate_pending fd to pollset");
226 ret
= lttng_poll_add(poll_set
, handle
->ust32_consumer
,
229 ERR("[rotation-thread] Failed to add ust-32 channel rotation pipe fd to pollset");
232 ret
= lttng_poll_add(poll_set
, handle
->ust64_consumer
,
235 ERR("[rotation-thread] Failed to add ust-64 channel rotation pipe fd to pollset");
238 if (handle
->kernel_consumer
>= 0) {
239 ret
= lttng_poll_add(poll_set
, handle
->kernel_consumer
,
242 ERR("[rotation-thread] Failed to add kernel channel rotation pipe fd to pollset");
250 lttng_poll_clean(poll_set
);
255 void fini_thread_state(struct rotation_thread_state
*state
)
257 lttng_poll_clean(&state
->events
);
258 cds_lfht_destroy(channel_pending_rotate_ht
, NULL
);
262 int init_thread_state(struct rotation_thread_handle
*handle
,
263 struct rotation_thread_state
*state
)
267 memset(state
, 0, sizeof(*state
));
268 lttng_poll_init(&state
->events
);
270 ret
= init_poll_set(&state
->events
, handle
);
272 ERR("[rotation-thread] Failed to initialize rotation thread poll set");
276 channel_pending_rotate_ht
= cds_lfht_new(DEFAULT_HT_SIZE
,
277 1, 0, CDS_LFHT_AUTO_RESIZE
| CDS_LFHT_ACCOUNTING
, NULL
);
278 if (!channel_pending_rotate_ht
) {
279 ERR("[rotation-thread] Failed to create channel pending rotation hash table");
289 int handle_channel_rotation_pipe(int fd
, uint32_t revents
,
290 struct rotation_thread_handle
*handle
,
291 struct rotation_thread_state
*state
)
294 enum lttng_domain_type domain
;
295 struct rotation_channel_info
*channel_info
;
296 struct ltt_session
*session
= NULL
;
299 if (fd
== handle
->ust32_consumer
||
300 fd
== handle
->ust64_consumer
) {
301 domain
= LTTNG_DOMAIN_UST
;
302 } else if (fd
== handle
->kernel_consumer
) {
303 domain
= LTTNG_DOMAIN_KERNEL
;
305 ERR("[rotation-thread] Unknown channel rotation pipe fd %d",
310 if (revents
& (LPOLLERR
| LPOLLHUP
| LPOLLRDHUP
)) {
311 ret
= lttng_poll_del(&state
->events
, fd
);
313 ERR("[rotation-thread] Failed to remove consumer "
314 "rotation pipe from poll set");
320 ret
= read(fd
, &key
, sizeof(key
));
321 } while (ret
== -1 && errno
== EINTR
);
322 if (ret
!= sizeof(key
)) {
323 ERR("[rotation-thread] Failed to read from pipe (fd = %i)",
329 DBG("[rotation-thread] Received notification for chan %" PRIu64
330 ", domain %d\n", key
, domain
);
332 channel_info
= lookup_channel_pending(key
, domain
);
334 ERR("[rotation-thread] Failed to find channel_info (key = %"
341 session
= session_find_by_id(channel_info
->session_id
);
344 * The session may have been destroyed before we had a chance to
345 * perform this action, return gracefully.
347 DBG("[rotation-thread] Session %" PRIu64
" not found",
348 channel_info
->session_id
);
350 goto end_unlock_session_list
;
353 session_lock(session
);
354 if (--session
->nr_chan_rotate_pending
== 0) {
355 time_t now
= time(NULL
);
357 if (now
== (time_t) -1) {
358 session
->rotation_status
= LTTNG_ROTATION_STATUS_ERROR
;
360 goto end_unlock_session
;
363 ret
= rename_complete_chunk(session
, now
);
365 ERR("Failed to rename completed rotation chunk");
366 goto end_unlock_session
;
368 session
->rotate_pending
= false;
369 session
->rotation_status
= LTTNG_ROTATION_STATUS_COMPLETED
;
370 session
->last_chunk_start_ts
= session
->current_chunk_start_ts
;
371 DBG("Rotation completed for session %s", session
->name
);
377 channel_rotation_info_destroy(channel_info
);
378 session_unlock(session
);
379 end_unlock_session_list
:
380 session_unlock_list();
386 void *thread_rotation(void *data
)
389 struct rotation_thread_handle
*handle
= data
;
390 struct rotation_thread_state state
;
392 DBG("[rotation-thread] Started rotation thread");
395 ERR("[rotation-thread] Invalid thread context provided");
399 rcu_register_thread();
402 health_register(health_sessiond
, HEALTH_SESSIOND_TYPE_ROTATION
);
403 health_code_update();
405 ret
= init_thread_state(handle
, &state
);
410 /* Ready to handle client connections. */
411 sessiond_notify_ready();
417 DBG("[rotation-thread] Entering poll wait");
418 ret
= lttng_poll_wait(&state
.events
, -1);
419 DBG("[rotation-thread] Poll wait returned (%i)", ret
);
423 * Restart interrupted system call.
425 if (errno
== EINTR
) {
428 ERR("[rotation-thread] Error encountered during lttng_poll_wait (%i)", ret
);
433 for (i
= 0; i
< fd_count
; i
++) {
434 int fd
= LTTNG_POLL_GETFD(&state
.events
, i
);
435 uint32_t revents
= LTTNG_POLL_GETEV(&state
.events
, i
);
437 DBG("[rotation-thread] Handling fd (%i) activity (%u)",
440 if (fd
== handle
->thread_quit_pipe
) {
441 DBG("[rotation-thread] Quit pipe activity");
443 } else if (fd
== handle
->ust32_consumer
||
444 fd
== handle
->ust64_consumer
||
445 fd
== handle
->kernel_consumer
) {
446 ret
= handle_channel_rotation_pipe(fd
,
447 revents
, handle
, &state
);
449 ERR("[rotation-thread] Handle channel rotation pipe");
457 DBG("[rotation-thread] Exit");
458 fini_thread_state(&state
);
459 health_unregister(health_sessiond
);
460 rcu_thread_offline();
461 rcu_unregister_thread();