Fix: sessiond: rotation trigger leak
[lttng-tools.git] / src / bin / lttng-sessiond / rotation-thread.cpp
CommitLineData
db66e574 1/*
ab5be9fa
MJ
2 * Copyright (C) 2017 Julien Desfossez <jdesfossez@efficios.com>
3 * Copyright (C) 2018 Jérémie Galarneau <jeremie.galarneau@efficios.com>
db66e574 4 *
ab5be9fa 5 * SPDX-License-Identifier: GPL-2.0-only
db66e574 6 *
db66e574
JD
7 */
8
9#define _LGPL_SOURCE
10#include <lttng/trigger/trigger.h>
c9e313bc
SM
11#include <common/error.hpp>
12#include <common/config/session-config.hpp>
13#include <common/defaults.hpp>
14#include <common/utils.hpp>
15#include <common/futex.hpp>
16#include <common/align.hpp>
17#include <common/time.hpp>
18#include <common/hashtable/utils.hpp>
db66e574
JD
19#include <sys/stat.h>
20#include <time.h>
21#include <signal.h>
22#include <inttypes.h>
23
c9e313bc
SM
24#include <common/kernel-ctl/kernel-ctl.hpp>
25#include <lttng/notification/channel-internal.hpp>
26#include <lttng/rotate-internal.hpp>
27#include <lttng/location-internal.hpp>
28#include <lttng/condition/condition-internal.hpp>
c08136a3 29#include <lttng/notification/notification-internal.hpp>
db66e574 30
c9e313bc
SM
31#include "rotation-thread.hpp"
32#include "lttng-sessiond.hpp"
33#include "health-sessiond.hpp"
34#include "rotate.hpp"
35#include "cmd.hpp"
36#include "session.hpp"
37#include "timer.hpp"
38#include "notification-thread-commands.hpp"
39#include "utils.hpp"
40#include "thread.hpp"
db66e574
JD
41
42#include <urcu.h>
43#include <urcu/list.h>
db66e574 44
90936dcf
JD
45struct lttng_notification_channel *rotate_notification_channel = NULL;
46
92816cc3 47struct rotation_thread {
db66e574
JD
48 struct lttng_poll_event events;
49};
50
92816cc3
JG
51/*
52 * The timer thread enqueues jobs and wakes up the rotation thread.
53 * When the rotation thread wakes up, it empties the queue.
54 */
55struct rotation_thread_timer_queue {
56 struct lttng_pipe *event_pipe;
57 struct cds_list_head list;
58 pthread_mutex_t lock;
59};
60
61struct rotation_thread_handle {
92816cc3
JG
62 struct rotation_thread_timer_queue *rotation_timer_queue;
63 /* Access to the notification thread cmd_queue */
64 struct notification_thread_handle *notification_thread_handle;
64d9b072
JG
65 /* Thread-specific quit pipe. */
66 struct lttng_pipe *quit_pipe;
92816cc3
JG
67};
68
f1494934
JG
69namespace {
70struct rotation_thread_job {
71 enum rotation_thread_job_type type;
72 struct ltt_session *session;
73 /* List member in struct rotation_thread_timer_queue. */
74 struct cds_list_head head;
75};
76} /* namespace */
77
db66e574 78static
92816cc3 79const char *get_job_type_str(enum rotation_thread_job_type job_type)
db66e574 80{
92816cc3
JG
81 switch (job_type) {
82 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
83 return "CHECK_PENDING_ROTATION";
84 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
85 return "SCHEDULED_ROTATION";
86 default:
87 abort();
88 }
db66e574
JD
89}
90
92816cc3 91struct rotation_thread_timer_queue *rotation_thread_timer_queue_create(void)
db66e574 92{
92816cc3 93 struct rotation_thread_timer_queue *queue = NULL;
db66e574 94
64803277 95 queue = zmalloc<rotation_thread_timer_queue>();
92816cc3
JG
96 if (!queue) {
97 PERROR("Failed to allocate timer rotate queue");
98 goto end;
99 }
db66e574 100
92816cc3
JG
101 queue->event_pipe = lttng_pipe_open(FD_CLOEXEC | O_NONBLOCK);
102 CDS_INIT_LIST_HEAD(&queue->list);
103 pthread_mutex_init(&queue->lock, NULL);
104end:
105 return queue;
db66e574
JD
106}
107
92816cc3
JG
108void rotation_thread_timer_queue_destroy(
109 struct rotation_thread_timer_queue *queue)
db66e574 110{
92816cc3
JG
111 if (!queue) {
112 return;
db66e574
JD
113 }
114
92816cc3
JG
115 lttng_pipe_destroy(queue->event_pipe);
116
117 pthread_mutex_lock(&queue->lock);
a0377dfe 118 LTTNG_ASSERT(cds_list_empty(&queue->list));
92816cc3
JG
119 pthread_mutex_unlock(&queue->lock);
120 pthread_mutex_destroy(&queue->lock);
121 free(queue);
122}
db66e574 123
92816cc3
JG
124/*
125 * Destroy the thread data previously created by the init function.
126 */
127void rotation_thread_handle_destroy(
128 struct rotation_thread_handle *handle)
129{
64d9b072 130 lttng_pipe_destroy(handle->quit_pipe);
db66e574
JD
131 free(handle);
132}
133
134struct rotation_thread_handle *rotation_thread_handle_create(
90936dcf 135 struct rotation_thread_timer_queue *rotation_timer_queue,
c8a9de5a 136 struct notification_thread_handle *notification_thread_handle)
db66e574
JD
137{
138 struct rotation_thread_handle *handle;
139
64803277 140 handle = zmalloc<rotation_thread_handle>();
db66e574
JD
141 if (!handle) {
142 goto end;
143 }
144
92816cc3
JG
145 handle->rotation_timer_queue = rotation_timer_queue;
146 handle->notification_thread_handle = notification_thread_handle;
64d9b072
JG
147 handle->quit_pipe = lttng_pipe_open(FD_CLOEXEC);
148 if (!handle->quit_pipe) {
149 goto error;
150 }
92816cc3
JG
151
152end:
153 return handle;
64d9b072
JG
154error:
155 rotation_thread_handle_destroy(handle);
156 return NULL;
92816cc3
JG
157}
158
159/*
160 * Called with the rotation_thread_timer_queue lock held.
161 * Return true if the same timer job already exists in the queue, false if not.
162 */
163static
164bool timer_job_exists(const struct rotation_thread_timer_queue *queue,
c7031a2c
JG
165 enum rotation_thread_job_type job_type,
166 struct ltt_session *session)
92816cc3
JG
167{
168 bool exists = false;
169 struct rotation_thread_job *job;
170
171 cds_list_for_each_entry(job, &queue->list, head) {
c7031a2c 172 if (job->session == session && job->type == job_type) {
92816cc3
JG
173 exists = true;
174 goto end;
db66e574 175 }
db66e574 176 }
92816cc3
JG
177end:
178 return exists;
179}
180
181void rotation_thread_enqueue_job(struct rotation_thread_timer_queue *queue,
c7031a2c
JG
182 enum rotation_thread_job_type job_type,
183 struct ltt_session *session)
92816cc3
JG
184{
185 int ret;
be2956e7 186 const char dummy = '!';
92816cc3
JG
187 struct rotation_thread_job *job = NULL;
188 const char *job_type_str = get_job_type_str(job_type);
189
190 pthread_mutex_lock(&queue->lock);
c7031a2c 191 if (timer_job_exists(queue, job_type, session)) {
92816cc3
JG
192 /*
193 * This timer job is already pending, we don't need to add
194 * it.
195 */
196 goto end;
db66e574 197 }
92816cc3 198
64803277 199 job = zmalloc<rotation_thread_job>();
92816cc3 200 if (!job) {
c7031a2c
JG
201 PERROR("Failed to allocate rotation thread job of type \"%s\" for session \"%s\"",
202 job_type_str, session->name);
92816cc3
JG
203 goto end;
204 }
c7031a2c
JG
205 /* No reason for this to fail as the caller must hold a reference. */
206 (void) session_get(session);
207
208 job->session = session;
92816cc3 209 job->type = job_type;
92816cc3
JG
210 cds_list_add_tail(&job->head, &queue->list);
211
be2956e7
JG
212 ret = lttng_write(lttng_pipe_get_writefd(queue->event_pipe), &dummy,
213 sizeof(dummy));
92816cc3
JG
214 if (ret < 0) {
215 /*
216 * We do not want to block in the timer handler, the job has
217 * been enqueued in the list, the wakeup pipe is probably full,
218 * the job will be processed when the rotation_thread catches
219 * up.
220 */
942003e5
MJ
221 DIAGNOSTIC_PUSH
222 DIAGNOSTIC_IGNORE_LOGICAL_OP
92816cc3 223 if (errno == EAGAIN || errno == EWOULDBLOCK) {
942003e5 224 DIAGNOSTIC_POP
92816cc3
JG
225 /*
226 * Not an error, but would be surprising and indicate
227 * that the rotation thread can't keep up with the
228 * current load.
229 */
230 DBG("Wake-up pipe of rotation thread job queue is full");
231 goto end;
db66e574 232 }
c7031a2c
JG
233 PERROR("Failed to wake-up the rotation thread after pushing a job of type \"%s\" for session \"%s\"",
234 job_type_str, session->name);
92816cc3 235 goto end;
db66e574 236 }
db66e574
JD
237
238end:
92816cc3 239 pthread_mutex_unlock(&queue->lock);
db66e574
JD
240}
241
242static
243int init_poll_set(struct lttng_poll_event *poll_set,
244 struct rotation_thread_handle *handle)
245{
246 int ret;
247
248 /*
64d9b072
JG
249 * Create pollset with size 3:
250 * - rotation thread quit pipe,
92816cc3 251 * - rotation thread timer queue pipe,
64d9b072 252 * - notification channel sock,
db66e574 253 */
64d9b072
JG
254 ret = lttng_poll_create(poll_set, 5, LTTNG_CLOEXEC);
255 if (ret < 0) {
db66e574
JD
256 goto error;
257 }
64d9b072
JG
258
259 ret = lttng_poll_add(poll_set,
260 lttng_pipe_get_readfd(handle->quit_pipe),
261 LPOLLIN | LPOLLERR);
262 if (ret < 0) {
bd0514a5 263 ERR("Failed to add quit pipe read fd to poll set");
64d9b072
JG
264 goto error;
265 }
266
d086f507
JD
267 ret = lttng_poll_add(poll_set,
268 lttng_pipe_get_readfd(handle->rotation_timer_queue->event_pipe),
269 LPOLLIN | LPOLLERR);
270 if (ret < 0) {
bd0514a5 271 ERR("Failed to add rotate_pending fd to poll set");
d086f507
JD
272 goto error;
273 }
db66e574 274
db66e574
JD
275 return ret;
276error:
277 lttng_poll_clean(poll_set);
278 return ret;
279}
280
281static
92816cc3 282void fini_thread_state(struct rotation_thread *state)
db66e574
JD
283{
284 lttng_poll_clean(&state->events);
90936dcf
JD
285 if (rotate_notification_channel) {
286 lttng_notification_channel_destroy(rotate_notification_channel);
287 }
db66e574
JD
288}
289
290static
291int init_thread_state(struct rotation_thread_handle *handle,
92816cc3 292 struct rotation_thread *state)
db66e574
JD
293{
294 int ret;
295
296 memset(state, 0, sizeof(*state));
297 lttng_poll_init(&state->events);
298
299 ret = init_poll_set(&state->events, handle);
300 if (ret) {
bd0514a5 301 ERR("Failed to initialize rotation thread poll set");
db66e574
JD
302 goto end;
303 }
304
90936dcf
JD
305 rotate_notification_channel = lttng_notification_channel_create(
306 lttng_session_daemon_notification_endpoint);
307 if (!rotate_notification_channel) {
bd0514a5 308 ERR("Could not create notification channel");
90936dcf
JD
309 ret = -1;
310 goto end;
311 }
312 ret = lttng_poll_add(&state->events, rotate_notification_channel->socket,
313 LPOLLIN | LPOLLERR);
314 if (ret < 0) {
bd0514a5 315 ERR("Failed to add notification fd to pollset");
90936dcf
JD
316 goto end;
317 }
318
db66e574
JD
319end:
320 return ret;
321}
322
323static
d2956687
JG
324void check_session_rotation_pending_on_consumers(struct ltt_session *session,
325 bool *_rotation_completed)
92816cc3 326{
db582e11 327 int ret = 0;
92816cc3
JG
328 struct consumer_socket *socket;
329 struct cds_lfht_iter iter;
d2956687
JG
330 enum consumer_trace_chunk_exists_status exists_status;
331 uint64_t relayd_id;
332 bool chunk_exists_on_peer = false;
333 enum lttng_trace_chunk_status chunk_status;
334
a0377dfe 335 LTTNG_ASSERT(session->chunk_being_archived);
92816cc3
JG
336
337 /*
338 * Check for a local pending rotation on all consumers (32-bit
339 * user space, 64-bit user space, and kernel).
340 */
92816cc3
JG
341 rcu_read_lock();
342 if (!session->ust_session) {
343 goto skip_ust;
344 }
345 cds_lfht_for_each_entry(session->ust_session->consumer->socks->ht,
346 &iter, socket, node.node) {
d2956687
JG
347 relayd_id = session->ust_session->consumer->type == CONSUMER_DST_LOCAL ?
348 -1ULL :
349 session->ust_session->consumer->net_seq_index;
350
351 pthread_mutex_lock(socket->lock);
352 ret = consumer_trace_chunk_exists(socket,
353 relayd_id,
354 session->id, session->chunk_being_archived,
355 &exists_status);
356 if (ret) {
357 pthread_mutex_unlock(socket->lock);
83ed9e90 358 ERR("Error occurred while checking rotation status on consumer daemon");
92816cc3 359 goto end;
db66e574 360 }
d2956687 361
16100d7a 362 if (exists_status != CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK) {
d2956687
JG
363 pthread_mutex_unlock(socket->lock);
364 chunk_exists_on_peer = true;
365 goto end;
16100d7a 366 }
d2956687 367 pthread_mutex_unlock(socket->lock);
16100d7a 368 }
db66e574 369
92816cc3
JG
370skip_ust:
371 if (!session->kernel_session) {
372 goto skip_kernel;
db66e574 373 }
92816cc3
JG
374 cds_lfht_for_each_entry(session->kernel_session->consumer->socks->ht,
375 &iter, socket, node.node) {
d2956687
JG
376 pthread_mutex_lock(socket->lock);
377 relayd_id = session->kernel_session->consumer->type == CONSUMER_DST_LOCAL ?
378 -1ULL :
379 session->kernel_session->consumer->net_seq_index;
380
381 ret = consumer_trace_chunk_exists(socket,
382 relayd_id,
383 session->id, session->chunk_being_archived,
384 &exists_status);
385 if (ret) {
386 pthread_mutex_unlock(socket->lock);
83ed9e90 387 ERR("Error occurred while checking rotation status on consumer daemon");
92816cc3
JG
388 goto end;
389 }
d2956687 390
16100d7a 391 if (exists_status != CONSUMER_TRACE_CHUNK_EXISTS_STATUS_UNKNOWN_CHUNK) {
d2956687
JG
392 pthread_mutex_unlock(socket->lock);
393 chunk_exists_on_peer = true;
394 goto end;
16100d7a 395 }
d2956687 396 pthread_mutex_unlock(socket->lock);
92816cc3
JG
397 }
398skip_kernel:
399end:
400 rcu_read_unlock();
db66e574 401
d2956687
JG
402 if (!chunk_exists_on_peer) {
403 uint64_t chunk_being_archived_id;
404
405 chunk_status = lttng_trace_chunk_get_id(
406 session->chunk_being_archived,
407 &chunk_being_archived_id);
a0377dfe 408 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
bd0514a5 409 DBG("Rotation of trace archive %" PRIu64 " of session \"%s\" is complete on all consumers",
d2956687 410 chunk_being_archived_id,
92816cc3 411 session->name);
db66e574 412 }
d2956687 413 *_rotation_completed = !chunk_exists_on_peer;
92816cc3 414 if (ret) {
2961f09e
JG
415 ret = session_reset_rotation_state(session,
416 LTTNG_ROTATION_STATE_ERROR);
417 if (ret) {
418 ERR("Failed to reset rotation state of session \"%s\"",
419 session->name);
420 }
db66e574 421 }
db66e574
JD
422}
423
d88744a4 424/*
92816cc3 425 * Check if the last rotation was completed, called with session lock held.
d2956687
JG
426 * Should only return non-zero in the event of a fatal error. Doing so will
427 * shutdown the thread.
d88744a4
JD
428 */
429static
92816cc3
JG
430int check_session_rotation_pending(struct ltt_session *session,
431 struct notification_thread_handle *notification_thread_handle)
d88744a4
JD
432{
433 int ret;
92816cc3 434 struct lttng_trace_archive_location *location;
d2956687
JG
435 enum lttng_trace_chunk_status chunk_status;
436 bool rotation_completed = false;
437 const char *archived_chunk_name;
438 uint64_t chunk_being_archived_id;
439
dc1d5967
FD
440 if (!session->chunk_being_archived) {
441 ret = 0;
442 goto end;
443 }
444
d2956687
JG
445 chunk_status = lttng_trace_chunk_get_id(session->chunk_being_archived,
446 &chunk_being_archived_id);
a0377dfe 447 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
d88744a4 448
bd0514a5 449 DBG("Checking for pending rotation on session \"%s\", trace archive %" PRIu64,
d2956687
JG
450 session->name, chunk_being_archived_id);
451
faf1bdcf
JG
452 /*
453 * The rotation-pending check timer of a session is launched in
454 * one-shot mode. If the rotation is incomplete, the rotation
455 * thread will re-enable the pending-check timer.
456 *
457 * The timer thread can't stop the timer itself since it is involved
458 * in the check for the timer's quiescence.
459 */
460 ret = timer_session_rotation_pending_check_stop(session);
461 if (ret) {
6ae1bf46 462 goto check_ongoing_rotation;
faf1bdcf
JG
463 }
464
d2956687
JG
465 check_session_rotation_pending_on_consumers(session,
466 &rotation_completed);
d2956687
JG
467 if (!rotation_completed ||
468 session->rotation_state == LTTNG_ROTATION_STATE_ERROR) {
6ae1bf46 469 goto check_ongoing_rotation;
92816cc3
JG
470 }
471
92816cc3
JG
472 /*
473 * Now we can clear the "ONGOING" state in the session. New
474 * rotations can start now.
475 */
d2956687
JG
476 chunk_status = lttng_trace_chunk_get_name(session->chunk_being_archived,
477 &archived_chunk_name, NULL);
a0377dfe 478 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
d2956687
JG
479 free(session->last_archived_chunk_name);
480 session->last_archived_chunk_name = strdup(archived_chunk_name);
481 if (!session->last_archived_chunk_name) {
482 PERROR("Failed to duplicate archived chunk name");
483 }
484 session_reset_rotation_state(session, LTTNG_ROTATION_STATE_COMPLETED);
92816cc3 485
7fdbed1c
JG
486 if (!session->quiet_rotation) {
487 location = session_get_trace_archive_location(session);
7fdbed1c
JG
488 ret = notification_thread_command_session_rotation_completed(
489 notification_thread_handle,
490 session->name,
491 session->uid,
492 session->gid,
493 session->last_archived_chunk_id.value,
494 location);
d3740619 495 lttng_trace_archive_location_put(location);
7fdbed1c 496 if (ret != LTTNG_OK) {
bd0514a5 497 ERR("Failed to notify notification thread of completed rotation for session %s",
7fdbed1c
JG
498 session->name);
499 }
92816cc3
JG
500 }
501
502 ret = 0;
6ae1bf46 503check_ongoing_rotation:
92816cc3 504 if (session->rotation_state == LTTNG_ROTATION_STATE_ONGOING) {
d2956687
JG
505 chunk_status = lttng_trace_chunk_get_id(
506 session->chunk_being_archived,
507 &chunk_being_archived_id);
a0377dfe 508 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
d2956687 509
bd0514a5 510 DBG("Rotation of trace archive %" PRIu64 " is still pending for session %s",
d2956687 511 chunk_being_archived_id, session->name);
92816cc3
JG
512 ret = timer_session_rotation_pending_check_start(session,
513 DEFAULT_ROTATE_PENDING_TIMER);
514 if (ret) {
d2956687 515 ERR("Failed to re-enable rotation pending timer");
92816cc3
JG
516 ret = -1;
517 goto end;
518 }
519 }
520
6ae1bf46 521end:
d88744a4
JD
522 return ret;
523}
524
ed1e52a3 525/* Call with the session and session_list locks held. */
259c2674 526static
92816cc3 527int launch_session_rotation(struct ltt_session *session)
259c2674
JD
528{
529 int ret;
92816cc3 530 struct lttng_rotate_session_return rotation_return;
259c2674 531
bd0514a5 532 DBG("Launching scheduled time-based rotation on session \"%s\"",
92816cc3 533 session->name);
259c2674 534
343defc2
MD
535 ret = cmd_rotate_session(session, &rotation_return, false,
536 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
92816cc3 537 if (ret == LTTNG_OK) {
bd0514a5 538 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
92816cc3
JG
539 session->name);
540 } else {
541 /* Don't consider errors as fatal. */
bd0514a5 542 DBG("Scheduled time-based rotation aborted for session %s: %s",
92816cc3 543 session->name, lttng_strerror(ret));
259c2674 544 }
92816cc3
JG
545 return 0;
546}
259c2674 547
92816cc3
JG
548static
549int run_job(struct rotation_thread_job *job, struct ltt_session *session,
550 struct notification_thread_handle *notification_thread_handle)
551{
552 int ret;
259c2674 553
92816cc3
JG
554 switch (job->type) {
555 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
16100d7a 556 ret = launch_session_rotation(session);
92816cc3
JG
557 break;
558 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
559 ret = check_session_rotation_pending(session,
560 notification_thread_handle);
561 break;
562 default:
563 abort();
259c2674 564 }
259c2674
JD
565 return ret;
566}
567
d88744a4 568static
92816cc3 569int handle_job_queue(struct rotation_thread_handle *handle,
f46376a1 570 struct rotation_thread *state __attribute__((unused)),
d88744a4
JD
571 struct rotation_thread_timer_queue *queue)
572{
573 int ret = 0;
d88744a4
JD
574
575 for (;;) {
e32d7f27 576 struct ltt_session *session;
92816cc3 577 struct rotation_thread_job *job;
d88744a4 578
92816cc3 579 /* Take the queue lock only to pop an element from the list. */
d88744a4
JD
580 pthread_mutex_lock(&queue->lock);
581 if (cds_list_empty(&queue->list)) {
582 pthread_mutex_unlock(&queue->lock);
583 break;
584 }
92816cc3
JG
585 job = cds_list_first_entry(&queue->list,
586 typeof(*job), head);
587 cds_list_del(&job->head);
d88744a4
JD
588 pthread_mutex_unlock(&queue->lock);
589
d88744a4 590 session_lock_list();
c7031a2c 591 session = job->session;
d88744a4 592 if (!session) {
bd0514a5 593 DBG("Session \"%s\" not found",
739b8e17 594 session->name != NULL ? session->name : "");
d88744a4 595 /*
92816cc3
JG
596 * This is a non-fatal error, and we cannot report it to
597 * the user (timer), so just print the error and
598 * continue the processing.
599 *
600 * While the timer thread will purge pending signals for
601 * a session on the session's destruction, it is
602 * possible for a job targeting that session to have
603 * already been queued before it was destroyed.
d88744a4 604 */
92816cc3 605 free(job);
e32d7f27 606 session_put(session);
5b8139c6 607 session_unlock_list();
d88744a4
JD
608 continue;
609 }
610
d88744a4 611 session_lock(session);
16100d7a 612 ret = run_job(job, session, handle->notification_thread_handle);
d88744a4 613 session_unlock(session);
faf1bdcf 614 /* Release reference held by the job. */
e32d7f27 615 session_put(session);
ed1e52a3 616 session_unlock_list();
92816cc3 617 free(job);
d88744a4 618 if (ret) {
d88744a4
JD
619 goto end;
620 }
621 }
622
623 ret = 0;
624
625end:
626 return ret;
627}
628
92816cc3 629static
c08136a3 630int handle_condition(const struct lttng_notification *notification,
90936dcf
JD
631 struct notification_thread_handle *notification_thread_handle)
632{
633 int ret = 0;
634 const char *condition_session_name = NULL;
635 enum lttng_condition_type condition_type;
636 enum lttng_condition_status condition_status;
637 enum lttng_evaluation_status evaluation_status;
638 uint64_t consumed;
639 struct ltt_session *session;
c08136a3
JG
640 const struct lttng_condition *condition =
641 lttng_notification_get_const_condition(notification);
642 const struct lttng_evaluation *evaluation =
643 lttng_notification_get_const_evaluation(notification);
90936dcf
JD
644
645 condition_type = lttng_condition_get_type(condition);
646
647 if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) {
648 ret = -1;
bd0514a5 649 ERR("Condition type and session usage type are not the same");
90936dcf
JD
650 goto end;
651 }
652
653 /* Fetch info to test */
654 condition_status = lttng_condition_session_consumed_size_get_session_name(
655 condition, &condition_session_name);
656 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
bd0514a5 657 ERR("Session name could not be fetched");
90936dcf
JD
658 ret = -1;
659 goto end;
660 }
661 evaluation_status = lttng_evaluation_session_consumed_size_get_consumed_size(evaluation,
662 &consumed);
663 if (evaluation_status != LTTNG_EVALUATION_STATUS_OK) {
bd0514a5 664 ERR("Failed to get evaluation");
90936dcf
JD
665 ret = -1;
666 goto end;
667 }
668
669 session_lock_list();
670 session = session_find_by_name(condition_session_name);
671 if (!session) {
eb2827a4
JG
672 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
673 lttng_condition_type_str(condition_type),
90936dcf 674 condition_session_name);
eb2827a4
JG
675 /*
676 * Not a fatal error: a session can be destroyed before we get
677 * the chance to handle the notification.
678 */
679 ret = 0;
680 session_unlock_list();
90936dcf
JD
681 goto end;
682 }
683 session_lock(session);
90936dcf 684
c08136a3
JG
685 if (!lttng_trigger_is_equal(session->rotate_trigger,
686 lttng_notification_get_const_trigger(notification))) {
687 /* Notification does not originate from our rotation trigger. */
688 ret = 0;
689 goto end_unlock;
690 }
691
90936dcf
JD
692 ret = unsubscribe_session_consumed_size_rotation(session,
693 notification_thread_handle);
694 if (ret) {
490b3229 695 goto end_unlock;
90936dcf
JD
696 }
697
343defc2
MD
698 ret = cmd_rotate_session(session, NULL, false,
699 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
90936dcf
JD
700 if (ret == -LTTNG_ERR_ROTATION_PENDING) {
701 DBG("Rotate already pending, subscribe to the next threshold value");
90936dcf 702 } else if (ret != LTTNG_OK) {
bd0514a5 703 ERR("Failed to rotate on size notification with error: %s",
90936dcf
JD
704 lttng_strerror(ret));
705 ret = -1;
706 goto end_unlock;
707 }
708 ret = subscribe_session_consumed_size_rotation(session,
709 consumed + session->rotate_size,
710 notification_thread_handle);
711 if (ret) {
bd0514a5 712 ERR("Failed to subscribe to session consumed size condition");
90936dcf
JD
713 goto end_unlock;
714 }
715 ret = 0;
716
717end_unlock:
718 session_unlock(session);
e32d7f27 719 session_put(session);
5b8139c6 720 session_unlock_list();
90936dcf
JD
721end:
722 return ret;
723}
724
725static
f46376a1 726int handle_notification_channel(int fd __attribute__((unused)),
90936dcf 727 struct rotation_thread_handle *handle,
f46376a1 728 struct rotation_thread *state __attribute__((unused)))
90936dcf
JD
729{
730 int ret;
d73ee93f
JG
731 bool notification_pending;
732 struct lttng_notification *notification = NULL;
90936dcf 733 enum lttng_notification_channel_status status;
90936dcf 734
d73ee93f
JG
735 status = lttng_notification_channel_has_pending_notification(
736 rotate_notification_channel, &notification_pending);
737 if (status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
bd0514a5 738 ERR("Error occurred while checking for pending notification");
d73ee93f
JG
739 ret = -1;
740 goto end;
741 }
742
743 if (!notification_pending) {
744 ret = 0;
745 goto end;
746 }
747
90936dcf
JD
748 /* Receive the next notification. */
749 status = lttng_notification_channel_get_next_notification(
750 rotate_notification_channel,
751 &notification);
752
753 switch (status) {
754 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK:
755 break;
756 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED:
757 /* Not an error, we will wait for the next one */
758 ret = 0;
759 goto end;;
760 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED:
761 ERR("Notification channel was closed");
762 ret = -1;
763 goto end;
764 default:
765 /* Unhandled conditions / errors. */
766 ERR("Unknown notification channel status");
767 ret = -1;
768 goto end;
769 }
770
c08136a3 771 ret = handle_condition(notification,
90936dcf
JD
772 handle->notification_thread_handle);
773
774end:
775 lttng_notification_destroy(notification);
90936dcf
JD
776 return ret;
777}
778
2c0c9bbc 779static
db66e574
JD
780void *thread_rotation(void *data)
781{
782 int ret;
7966af57 783 struct rotation_thread_handle *handle = (rotation_thread_handle *) data;
92816cc3 784 struct rotation_thread thread;
87380d40 785 int queue_pipe_fd;
db66e574 786
bd0514a5 787 DBG("Started rotation thread");
f620cc28
JG
788 rcu_register_thread();
789 rcu_thread_online();
412d7227 790 health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
f620cc28 791 health_code_update();
db66e574
JD
792
793 if (!handle) {
bd0514a5 794 ERR("Invalid thread context provided");
db66e574
JD
795 goto end;
796 }
797
87380d40
JR
798 queue_pipe_fd = lttng_pipe_get_readfd(
799 handle->rotation_timer_queue->event_pipe);
800
db66e574 801
92816cc3 802 ret = init_thread_state(handle, &thread);
db66e574 803 if (ret) {
f5f8e5cd 804 goto error;
db66e574
JD
805 }
806
db66e574
JD
807 while (true) {
808 int fd_count, i;
809
810 health_poll_entry();
bd0514a5 811 DBG("Entering poll wait");
92816cc3 812 ret = lttng_poll_wait(&thread.events, -1);
bd0514a5 813 DBG("Poll wait returned (%i)", ret);
db66e574
JD
814 health_poll_exit();
815 if (ret < 0) {
816 /*
817 * Restart interrupted system call.
818 */
819 if (errno == EINTR) {
820 continue;
821 }
bd0514a5 822 ERR("Error encountered during lttng_poll_wait (%i)", ret);
db66e574
JD
823 goto error;
824 }
825
826 fd_count = ret;
827 for (i = 0; i < fd_count; i++) {
92816cc3
JG
828 int fd = LTTNG_POLL_GETFD(&thread.events, i);
829 uint32_t revents = LTTNG_POLL_GETEV(&thread.events, i);
db66e574 830
bd0514a5 831 DBG("Handling fd (%i) activity (%u)",
db66e574
JD
832 fd, revents);
833
92816cc3 834 if (revents & LPOLLERR) {
bd0514a5 835 ERR("Polling returned an error on fd %i", fd);
92816cc3
JG
836 goto error;
837 }
838
85e17b27
JG
839 if (fd == rotate_notification_channel->socket) {
840 ret = handle_notification_channel(fd, handle,
841 &thread);
842 if (ret) {
bd0514a5 843 ERR("Error occurred while handling activity on notification channel socket");
85e17b27
JG
844 goto error;
845 }
846 } else {
847 /* Job queue or quit pipe activity. */
85e17b27
JG
848
849 /*
850 * The job queue is serviced if there is
851 * activity on the quit pipe to ensure it is
852 * flushed and all references held in the queue
853 * are released.
854 */
92816cc3
JG
855 ret = handle_job_queue(handle, &thread,
856 handle->rotation_timer_queue);
d88744a4 857 if (ret) {
bd0514a5 858 ERR("Failed to handle rotation timer pipe event");
d88744a4
JD
859 goto error;
860 }
85e17b27 861
64d9b072
JG
862 if (fd == queue_pipe_fd) {
863 char buf;
864
865 ret = lttng_read(fd, &buf, 1);
866 if (ret != 1) {
bd0514a5 867 ERR("Failed to read from wakeup pipe (fd = %i)", fd);
64d9b072
JG
868 goto error;
869 }
870 } else {
bd0514a5 871 DBG("Quit pipe activity");
85e17b27 872 goto exit;
90936dcf 873 }
db66e574
JD
874 }
875 }
876 }
877exit:
878error:
bd0514a5 879 DBG("Thread exit");
92816cc3 880 fini_thread_state(&thread);
f620cc28 881end:
412d7227 882 health_unregister(the_health_sessiond);
03732c32
JG
883 rcu_thread_offline();
884 rcu_unregister_thread();
db66e574
JD
885 return NULL;
886}
64d9b072
JG
887
888static
889bool shutdown_rotation_thread(void *thread_data)
890{
7966af57 891 struct rotation_thread_handle *handle = (rotation_thread_handle *) thread_data;
64d9b072
JG
892 const int write_fd = lttng_pipe_get_writefd(handle->quit_pipe);
893
894 return notify_thread_pipe(write_fd) == 1;
895}
896
897bool launch_rotation_thread(struct rotation_thread_handle *handle)
898{
899 struct lttng_thread *thread;
900
901 thread = lttng_thread_create("Rotation",
902 thread_rotation,
903 shutdown_rotation_thread,
904 NULL,
905 handle);
906 if (!thread) {
907 goto error;
908 }
909 lttng_thread_put(thread);
910 return true;
911error:
912 return false;
913}
This page took 0.138205 seconds and 4 git commands to generate.