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