Fix: sessiond: abort called on undefined client command
[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,
139a8d25 490 session->id,
7fdbed1c
JG
491 session->last_archived_chunk_id.value,
492 location);
d3740619 493 lttng_trace_archive_location_put(location);
7fdbed1c 494 if (ret != LTTNG_OK) {
bd0514a5 495 ERR("Failed to notify notification thread of completed rotation for session %s",
7fdbed1c
JG
496 session->name);
497 }
92816cc3
JG
498 }
499
500 ret = 0;
6ae1bf46 501check_ongoing_rotation:
92816cc3 502 if (session->rotation_state == LTTNG_ROTATION_STATE_ONGOING) {
d2956687
JG
503 chunk_status = lttng_trace_chunk_get_id(
504 session->chunk_being_archived,
505 &chunk_being_archived_id);
a0377dfe 506 LTTNG_ASSERT(chunk_status == LTTNG_TRACE_CHUNK_STATUS_OK);
d2956687 507
bd0514a5 508 DBG("Rotation of trace archive %" PRIu64 " is still pending for session %s",
d2956687 509 chunk_being_archived_id, session->name);
92816cc3
JG
510 ret = timer_session_rotation_pending_check_start(session,
511 DEFAULT_ROTATE_PENDING_TIMER);
512 if (ret) {
d2956687 513 ERR("Failed to re-enable rotation pending timer");
92816cc3
JG
514 ret = -1;
515 goto end;
516 }
517 }
518
6ae1bf46 519end:
d88744a4
JD
520 return ret;
521}
522
ed1e52a3 523/* Call with the session and session_list locks held. */
259c2674 524static
92816cc3 525int launch_session_rotation(struct ltt_session *session)
259c2674
JD
526{
527 int ret;
92816cc3 528 struct lttng_rotate_session_return rotation_return;
259c2674 529
bd0514a5 530 DBG("Launching scheduled time-based rotation on session \"%s\"",
92816cc3 531 session->name);
259c2674 532
343defc2
MD
533 ret = cmd_rotate_session(session, &rotation_return, false,
534 LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
92816cc3 535 if (ret == LTTNG_OK) {
bd0514a5 536 DBG("Scheduled time-based rotation successfully launched on session \"%s\"",
92816cc3
JG
537 session->name);
538 } else {
539 /* Don't consider errors as fatal. */
bd0514a5 540 DBG("Scheduled time-based rotation aborted for session %s: %s",
92816cc3 541 session->name, lttng_strerror(ret));
259c2674 542 }
92816cc3
JG
543 return 0;
544}
259c2674 545
92816cc3
JG
546static
547int run_job(struct rotation_thread_job *job, struct ltt_session *session,
548 struct notification_thread_handle *notification_thread_handle)
549{
550 int ret;
259c2674 551
92816cc3
JG
552 switch (job->type) {
553 case ROTATION_THREAD_JOB_TYPE_SCHEDULED_ROTATION:
16100d7a 554 ret = launch_session_rotation(session);
92816cc3
JG
555 break;
556 case ROTATION_THREAD_JOB_TYPE_CHECK_PENDING_ROTATION:
557 ret = check_session_rotation_pending(session,
558 notification_thread_handle);
559 break;
560 default:
561 abort();
259c2674 562 }
259c2674
JD
563 return ret;
564}
565
d88744a4 566static
92816cc3 567int handle_job_queue(struct rotation_thread_handle *handle,
f46376a1 568 struct rotation_thread *state __attribute__((unused)),
d88744a4
JD
569 struct rotation_thread_timer_queue *queue)
570{
571 int ret = 0;
d88744a4
JD
572
573 for (;;) {
e32d7f27 574 struct ltt_session *session;
92816cc3 575 struct rotation_thread_job *job;
d88744a4 576
92816cc3 577 /* Take the queue lock only to pop an element from the list. */
d88744a4
JD
578 pthread_mutex_lock(&queue->lock);
579 if (cds_list_empty(&queue->list)) {
580 pthread_mutex_unlock(&queue->lock);
581 break;
582 }
92816cc3
JG
583 job = cds_list_first_entry(&queue->list,
584 typeof(*job), head);
585 cds_list_del(&job->head);
d88744a4
JD
586 pthread_mutex_unlock(&queue->lock);
587
d88744a4 588 session_lock_list();
c7031a2c 589 session = job->session;
d88744a4 590 if (!session) {
bd0514a5 591 DBG("Session \"%s\" not found",
739b8e17 592 session->name != NULL ? session->name : "");
d88744a4 593 /*
92816cc3
JG
594 * This is a non-fatal error, and we cannot report it to
595 * the user (timer), so just print the error and
596 * continue the processing.
597 *
598 * While the timer thread will purge pending signals for
599 * a session on the session's destruction, it is
600 * possible for a job targeting that session to have
601 * already been queued before it was destroyed.
d88744a4 602 */
92816cc3 603 free(job);
e32d7f27 604 session_put(session);
5b8139c6 605 session_unlock_list();
d88744a4
JD
606 continue;
607 }
608
d88744a4 609 session_lock(session);
16100d7a 610 ret = run_job(job, session, handle->notification_thread_handle);
d88744a4 611 session_unlock(session);
faf1bdcf 612 /* Release reference held by the job. */
e32d7f27 613 session_put(session);
ed1e52a3 614 session_unlock_list();
92816cc3 615 free(job);
d88744a4 616 if (ret) {
d88744a4
JD
617 goto end;
618 }
619 }
620
621 ret = 0;
622
623end:
624 return ret;
625}
626
92816cc3 627static
c08136a3 628int handle_condition(const struct lttng_notification *notification,
90936dcf
JD
629 struct notification_thread_handle *notification_thread_handle)
630{
631 int ret = 0;
632 const char *condition_session_name = NULL;
633 enum lttng_condition_type condition_type;
634 enum lttng_condition_status condition_status;
635 enum lttng_evaluation_status evaluation_status;
636 uint64_t consumed;
637 struct ltt_session *session;
c08136a3
JG
638 const struct lttng_condition *condition =
639 lttng_notification_get_const_condition(notification);
640 const struct lttng_evaluation *evaluation =
641 lttng_notification_get_const_evaluation(notification);
90936dcf
JD
642
643 condition_type = lttng_condition_get_type(condition);
644
645 if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) {
646 ret = -1;
bd0514a5 647 ERR("Condition type and session usage type are not the same");
90936dcf
JD
648 goto end;
649 }
650
651 /* Fetch info to test */
652 condition_status = lttng_condition_session_consumed_size_get_session_name(
653 condition, &condition_session_name);
654 if (condition_status != LTTNG_CONDITION_STATUS_OK) {
bd0514a5 655 ERR("Session name could not be fetched");
90936dcf
JD
656 ret = -1;
657 goto end;
658 }
659 evaluation_status = lttng_evaluation_session_consumed_size_get_consumed_size(evaluation,
660 &consumed);
661 if (evaluation_status != LTTNG_EVALUATION_STATUS_OK) {
bd0514a5 662 ERR("Failed to get evaluation");
90936dcf
JD
663 ret = -1;
664 goto end;
665 }
666
667 session_lock_list();
668 session = session_find_by_name(condition_session_name);
669 if (!session) {
eb2827a4
JG
670 DBG("Failed to find session while handling notification: notification type = %s, session name = `%s`",
671 lttng_condition_type_str(condition_type),
90936dcf 672 condition_session_name);
eb2827a4
JG
673 /*
674 * Not a fatal error: a session can be destroyed before we get
675 * the chance to handle the notification.
676 */
677 ret = 0;
678 session_unlock_list();
90936dcf
JD
679 goto end;
680 }
681 session_lock(session);
90936dcf 682
c08136a3
JG
683 if (!lttng_trigger_is_equal(session->rotate_trigger,
684 lttng_notification_get_const_trigger(notification))) {
685 /* Notification does not originate from our rotation trigger. */
686 ret = 0;
687 goto end_unlock;
688 }
689
90936dcf
JD
690 ret = unsubscribe_session_consumed_size_rotation(session,
691 notification_thread_handle);
692 if (ret) {
490b3229 693 goto end_unlock;
90936dcf
JD
694 }
695
2545db87
JG
696 ret = cmd_rotate_session(
697 session, NULL, false, LTTNG_TRACE_CHUNK_COMMAND_TYPE_MOVE_TO_COMPLETED);
698 switch (ret) {
699 case LTTNG_OK:
700 break;
701 case -LTTNG_ERR_ROTATION_PENDING:
90936dcf 702 DBG("Rotate already pending, subscribe to the next threshold value");
2545db87
JG
703 break;
704 case -LTTNG_ERR_ROTATION_MULTIPLE_AFTER_STOP:
705 DBG("Rotation already happened since last stop, subscribe to the next threshold value");
706 break;
707 case -LTTNG_ERR_ROTATION_AFTER_STOP_CLEAR:
708 DBG("Rotation already happened since last stop and clear, subscribe to the next threshold value");
709 break;
710 default:
711 ERR("Failed to rotate on size notification with error: %s", lttng_strerror(ret));
90936dcf
JD
712 ret = -1;
713 goto end_unlock;
714 }
2545db87
JG
715
716 ret = subscribe_session_consumed_size_rotation(
717 session, consumed + session->rotate_size, notification_thread_handle);
90936dcf 718 if (ret) {
bd0514a5 719 ERR("Failed to subscribe to session consumed size condition");
90936dcf
JD
720 goto end_unlock;
721 }
722 ret = 0;
723
724end_unlock:
725 session_unlock(session);
e32d7f27 726 session_put(session);
5b8139c6 727 session_unlock_list();
90936dcf
JD
728end:
729 return ret;
730}
731
732static
f46376a1 733int handle_notification_channel(int fd __attribute__((unused)),
90936dcf 734 struct rotation_thread_handle *handle,
f46376a1 735 struct rotation_thread *state __attribute__((unused)))
90936dcf
JD
736{
737 int ret;
d73ee93f
JG
738 bool notification_pending;
739 struct lttng_notification *notification = NULL;
90936dcf 740 enum lttng_notification_channel_status status;
90936dcf 741
d73ee93f
JG
742 status = lttng_notification_channel_has_pending_notification(
743 rotate_notification_channel, &notification_pending);
744 if (status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
bd0514a5 745 ERR("Error occurred while checking for pending notification");
d73ee93f
JG
746 ret = -1;
747 goto end;
748 }
749
750 if (!notification_pending) {
751 ret = 0;
752 goto end;
753 }
754
90936dcf
JD
755 /* Receive the next notification. */
756 status = lttng_notification_channel_get_next_notification(
757 rotate_notification_channel,
758 &notification);
759
760 switch (status) {
761 case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK:
762 break;
763 case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED:
764 /* Not an error, we will wait for the next one */
765 ret = 0;
766 goto end;;
767 case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED:
768 ERR("Notification channel was closed");
769 ret = -1;
770 goto end;
771 default:
772 /* Unhandled conditions / errors. */
773 ERR("Unknown notification channel status");
774 ret = -1;
775 goto end;
776 }
777
c08136a3 778 ret = handle_condition(notification,
90936dcf
JD
779 handle->notification_thread_handle);
780
781end:
782 lttng_notification_destroy(notification);
90936dcf
JD
783 return ret;
784}
785
2c0c9bbc 786static
db66e574
JD
787void *thread_rotation(void *data)
788{
789 int ret;
7966af57 790 struct rotation_thread_handle *handle = (rotation_thread_handle *) data;
92816cc3 791 struct rotation_thread thread;
87380d40 792 int queue_pipe_fd;
db66e574 793
bd0514a5 794 DBG("Started rotation thread");
f620cc28
JG
795 rcu_register_thread();
796 rcu_thread_online();
412d7227 797 health_register(the_health_sessiond, HEALTH_SESSIOND_TYPE_ROTATION);
f620cc28 798 health_code_update();
db66e574
JD
799
800 if (!handle) {
bd0514a5 801 ERR("Invalid thread context provided");
db66e574
JD
802 goto end;
803 }
804
87380d40
JR
805 queue_pipe_fd = lttng_pipe_get_readfd(
806 handle->rotation_timer_queue->event_pipe);
807
db66e574 808
92816cc3 809 ret = init_thread_state(handle, &thread);
db66e574 810 if (ret) {
f5f8e5cd 811 goto error;
db66e574
JD
812 }
813
db66e574
JD
814 while (true) {
815 int fd_count, i;
816
817 health_poll_entry();
bd0514a5 818 DBG("Entering poll wait");
92816cc3 819 ret = lttng_poll_wait(&thread.events, -1);
bd0514a5 820 DBG("Poll wait returned (%i)", ret);
db66e574
JD
821 health_poll_exit();
822 if (ret < 0) {
823 /*
824 * Restart interrupted system call.
825 */
826 if (errno == EINTR) {
827 continue;
828 }
bd0514a5 829 ERR("Error encountered during lttng_poll_wait (%i)", ret);
db66e574
JD
830 goto error;
831 }
832
833 fd_count = ret;
834 for (i = 0; i < fd_count; i++) {
92816cc3
JG
835 int fd = LTTNG_POLL_GETFD(&thread.events, i);
836 uint32_t revents = LTTNG_POLL_GETEV(&thread.events, i);
db66e574 837
bd0514a5 838 DBG("Handling fd (%i) activity (%u)",
db66e574
JD
839 fd, revents);
840
92816cc3 841 if (revents & LPOLLERR) {
bd0514a5 842 ERR("Polling returned an error on fd %i", fd);
92816cc3
JG
843 goto error;
844 }
845
85e17b27
JG
846 if (fd == rotate_notification_channel->socket) {
847 ret = handle_notification_channel(fd, handle,
848 &thread);
849 if (ret) {
bd0514a5 850 ERR("Error occurred while handling activity on notification channel socket");
85e17b27
JG
851 goto error;
852 }
853 } else {
854 /* Job queue or quit pipe activity. */
85e17b27
JG
855
856 /*
857 * The job queue is serviced if there is
858 * activity on the quit pipe to ensure it is
859 * flushed and all references held in the queue
860 * are released.
861 */
92816cc3
JG
862 ret = handle_job_queue(handle, &thread,
863 handle->rotation_timer_queue);
d88744a4 864 if (ret) {
bd0514a5 865 ERR("Failed to handle rotation timer pipe event");
d88744a4
JD
866 goto error;
867 }
85e17b27 868
64d9b072
JG
869 if (fd == queue_pipe_fd) {
870 char buf;
871
872 ret = lttng_read(fd, &buf, 1);
873 if (ret != 1) {
bd0514a5 874 ERR("Failed to read from wakeup pipe (fd = %i)", fd);
64d9b072
JG
875 goto error;
876 }
877 } else {
bd0514a5 878 DBG("Quit pipe activity");
85e17b27 879 goto exit;
90936dcf 880 }
db66e574
JD
881 }
882 }
883 }
884exit:
885error:
bd0514a5 886 DBG("Thread exit");
92816cc3 887 fini_thread_state(&thread);
f620cc28 888end:
412d7227 889 health_unregister(the_health_sessiond);
03732c32
JG
890 rcu_thread_offline();
891 rcu_unregister_thread();
db66e574
JD
892 return NULL;
893}
64d9b072
JG
894
895static
896bool shutdown_rotation_thread(void *thread_data)
897{
7966af57 898 struct rotation_thread_handle *handle = (rotation_thread_handle *) thread_data;
64d9b072
JG
899 const int write_fd = lttng_pipe_get_writefd(handle->quit_pipe);
900
901 return notify_thread_pipe(write_fd) == 1;
902}
903
904bool launch_rotation_thread(struct rotation_thread_handle *handle)
905{
906 struct lttng_thread *thread;
907
908 thread = lttng_thread_create("Rotation",
909 thread_rotation,
910 shutdown_rotation_thread,
911 NULL,
912 handle);
913 if (!thread) {
914 goto error;
915 }
916 lttng_thread_put(thread);
917 return true;
918error:
919 return false;
920}
This page took 0.154072 seconds and 4 git commands to generate.