X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Frotation-thread.c;h=9f15ed06edacbae130cef7ac9e4c0c7793ecee87;hp=ecccc7bb01ab00bbbba7dd71675b88a7f4b08e3d;hb=90936dcf0968343f20b2f6fd365b9c015cdb9717;hpb=1d757b1cd3b4669b52e2d9ceafb03eafd42490ff diff --git a/src/bin/lttng-sessiond/rotation-thread.c b/src/bin/lttng-sessiond/rotation-thread.c index ecccc7bb0..9f15ed06e 100644 --- a/src/bin/lttng-sessiond/rotation-thread.c +++ b/src/bin/lttng-sessiond/rotation-thread.c @@ -53,6 +53,8 @@ */ struct cds_lfht *channel_pending_rotate_ht; +struct lttng_notification_channel *rotate_notification_channel = NULL; + struct rotation_thread_state { struct lttng_poll_event events; }; @@ -143,7 +145,9 @@ struct rotation_thread_handle *rotation_thread_handle_create( struct lttng_pipe *ust64_channel_rotate_pipe, struct lttng_pipe *kernel_channel_rotate_pipe, int thread_quit_pipe, - struct rotation_thread_timer_queue *rotation_timer_queue) + struct rotation_thread_timer_queue *rotation_timer_queue, + struct notification_thread_handle *notification_thread_handle, + sem_t *notification_thread_ready) { struct rotation_thread_handle *handle; @@ -184,6 +188,8 @@ struct rotation_thread_handle *rotation_thread_handle_create( } handle->thread_quit_pipe = thread_quit_pipe; handle->rotation_timer_queue = rotation_timer_queue; + handle->notification_thread_handle = notification_thread_handle; + handle->notification_thread_ready = notification_thread_ready; end: return handle; @@ -257,6 +263,9 @@ void fini_thread_state(struct rotation_thread_state *state) { lttng_poll_clean(&state->events); cds_lfht_destroy(channel_pending_rotate_ht, NULL); + if (rotate_notification_channel) { + lttng_notification_channel_destroy(rotate_notification_channel); + } } static @@ -282,6 +291,25 @@ int init_thread_state(struct rotation_thread_handle *handle, goto end; } + /* + * We wait until the notification thread is ready to create the + * notification channel and add it to the poll_set. + */ + sem_wait(handle->notification_thread_ready); + rotate_notification_channel = lttng_notification_channel_create( + lttng_session_daemon_notification_endpoint); + if (!rotate_notification_channel) { + ERR("[rotation-thread] Could not create notification channel"); + ret = -1; + goto end; + } + ret = lttng_poll_add(&state->events, rotate_notification_channel->socket, + LPOLLIN | LPOLLERR); + if (ret < 0) { + ERR("[rotation-thread] Failed to add notification fd to pollset"); + goto end; + } + end: return ret; } @@ -328,7 +356,7 @@ int handle_channel_rotation_pipe(int fd, uint32_t revents, } DBG("[rotation-thread] Received notification for chan %" PRIu64 - ", domain %d\n", key, domain); + ", domain %d", key, domain); channel_info = lookup_channel_pending(key, domain); if (!channel_info) { @@ -566,6 +594,136 @@ end: return ret; } +int handle_condition( + const struct lttng_condition *condition, + const struct lttng_evaluation *evaluation, + struct notification_thread_handle *notification_thread_handle) +{ + int ret = 0; + const char *condition_session_name = NULL; + enum lttng_condition_type condition_type; + enum lttng_condition_status condition_status; + enum lttng_evaluation_status evaluation_status; + uint64_t consumed; + struct ltt_session *session; + + condition_type = lttng_condition_get_type(condition); + + if (condition_type != LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE) { + ret = -1; + ERR("[rotation-thread] Condition type and session usage type are not the same"); + goto end; + } + + /* Fetch info to test */ + condition_status = lttng_condition_session_consumed_size_get_session_name( + condition, &condition_session_name); + if (condition_status != LTTNG_CONDITION_STATUS_OK) { + ERR("[rotation-thread] Session name could not be fetched"); + ret = -1; + goto end; + } + evaluation_status = lttng_evaluation_session_consumed_size_get_consumed_size(evaluation, + &consumed); + if (evaluation_status != LTTNG_EVALUATION_STATUS_OK) { + ERR("[rotation-thread] Failed to get evaluation"); + ret = -1; + goto end; + } + + session_lock_list(); + session = session_find_by_name(condition_session_name); + if (!session) { + ret = -1; + session_unlock_list(); + ERR("[rotation-thread] Session \"%s\" not found", + condition_session_name); + goto end; + } + session_lock(session); + session_unlock_list(); + + ret = unsubscribe_session_consumed_size_rotation(session, + notification_thread_handle); + if (ret) { + goto end; + } + + ret = cmd_rotate_session(session, NULL); + if (ret == -LTTNG_ERR_ROTATION_PENDING) { + DBG("Rotate already pending, subscribe to the next threshold value"); + ret = 0; + } else if (ret != LTTNG_OK) { + ERR("[rotation-thread] Failed to rotate on size notification with error: %s", + lttng_strerror(ret)); + ret = -1; + goto end_unlock; + } + ret = subscribe_session_consumed_size_rotation(session, + consumed + session->rotate_size, + notification_thread_handle); + if (ret) { + ERR("[rotation-thread] Failed to subscribe to session consumed size condition"); + goto end_unlock; + } + ret = 0; + +end_unlock: + session_unlock(session); +end: + return ret; +} + +static +int handle_notification_channel(int fd, uint32_t revents, + struct rotation_thread_handle *handle, + struct rotation_thread_state *state) +{ + int ret; + struct lttng_notification *notification; + enum lttng_notification_channel_status status; + const struct lttng_evaluation *notification_evaluation; + const struct lttng_condition *notification_condition; + + /* Receive the next notification. */ + status = lttng_notification_channel_get_next_notification( + rotate_notification_channel, + ¬ification); + + switch (status) { + case LTTNG_NOTIFICATION_CHANNEL_STATUS_OK: + break; + case LTTNG_NOTIFICATION_CHANNEL_STATUS_NOTIFICATIONS_DROPPED: + /* Not an error, we will wait for the next one */ + ret = 0; + goto end;; + case LTTNG_NOTIFICATION_CHANNEL_STATUS_CLOSED: + ERR("Notification channel was closed"); + ret = -1; + goto end; + default: + /* Unhandled conditions / errors. */ + ERR("Unknown notification channel status"); + ret = -1; + goto end; + } + + notification_condition = lttng_notification_get_condition(notification); + notification_evaluation = lttng_notification_get_evaluation(notification); + + ret = handle_condition(notification_condition, notification_evaluation, + handle->notification_thread_handle); + +end: + lttng_notification_destroy(notification); + if (ret != 0) { + goto end; + } + + + return ret; +} + void *thread_rotation(void *data) { int ret; @@ -639,6 +797,13 @@ void *thread_rotation(void *data) ERR("[rotation-thread] Handle channel rotation pipe"); goto error; } + } else if (fd == rotate_notification_channel->socket) { + ret = handle_notification_channel(fd, revents, + handle, &state); + if (ret) { + ERR("[rotation-thread] Error occured while handling activity on notification channel socket"); + goto error; + } } } }