X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Frotation-thread.c;h=e6b9b6a1e72790702a7d96e1412dd3bcadc43662;hp=1704777a09ece70aef872745fd0ffc0c451ba939;hb=d73ee93f00ad2bf67dc79f2e3e09f74e2fb3d01f;hpb=d086f507d02078aed618ab291a0bc4a634958fa3

diff --git a/src/bin/lttng-sessiond/rotation-thread.c b/src/bin/lttng-sessiond/rotation-thread.c
index 1704777a0..e6b9b6a1e 100644
--- a/src/bin/lttng-sessiond/rotation-thread.c
+++ b/src/bin/lttng-sessiond/rotation-thread.c
@@ -33,6 +33,7 @@
 
 #include <common/kernel-ctl/kernel-ctl.h>
 #include <lttng/notification/channel-internal.h>
+#include <lttng/rotate-internal.h>
 
 #include "rotation-thread.h"
 #include "lttng-sessiond.h"
@@ -52,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;
 };
@@ -142,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;
 
@@ -183,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;
@@ -256,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
@@ -281,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;
 }
@@ -327,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) {
@@ -355,7 +384,7 @@ int handle_channel_rotation_pipe(int fd, uint32_t revents,
 		time_t now = time(NULL);
 
 		if (now == (time_t) -1) {
-			session->rotation_status = LTTNG_ROTATION_STATUS_ERROR;
+			session->rotation_state = LTTNG_ROTATION_STATE_ERROR;
 			ret = LTTNG_ERR_UNK;
 			goto end_unlock_session;
 		}
@@ -366,8 +395,19 @@ int handle_channel_rotation_pipe(int fd, uint32_t revents,
 			goto end_unlock_session;
 		}
 		session->rotate_pending = false;
-		session->rotation_status = LTTNG_ROTATION_STATUS_COMPLETED;
 		session->last_chunk_start_ts = session->current_chunk_start_ts;
+		if (session->rotate_pending_relay) {
+			ret = sessiond_timer_rotate_pending_start(
+					session,
+					DEFAULT_ROTATE_PENDING_RELAY_TIMER);
+			if (ret) {
+				ERR("Failed to enable rotate pending timer");
+				ret = -1;
+				goto end_unlock_session;
+			}
+		} else {
+			session->rotation_state = LTTNG_ROTATION_STATE_COMPLETED;
+		}
 		DBG("Rotation completed for session %s", session->name);
 	}
 
@@ -383,6 +423,323 @@ end:
 	return ret;
 }
 
+/*
+ * Process the rotate_pending check, called with session lock held.
+ */
+static
+int rotate_pending_relay_timer(struct ltt_session *session)
+{
+	int ret;
+
+	DBG("[rotation-thread] Check rotate pending on session %" PRIu64,
+			session->id);
+	ret = relay_rotate_pending(session, session->rotate_count - 1);
+	if (ret < 0) {
+		ERR("[rotation-thread] Check relay rotate pending");
+		goto end;
+	}
+	if (ret == 0) {
+		DBG("[rotation-thread] Rotation completed on the relay for "
+				"session %" PRIu64, session->id);
+		/*
+		 * Now we can clear the pending flag in the session. New
+		 * rotations can start now.
+		 */
+		session->rotate_pending_relay = false;
+		session->rotation_state = LTTNG_ROTATION_STATE_COMPLETED;
+	} else if (ret == 1) {
+		DBG("[rotation-thread] Rotation still pending on the relay for "
+				"session %" PRIu64, session->id);
+		ret = sessiond_timer_rotate_pending_start(session,
+				DEFAULT_ROTATE_PENDING_RELAY_TIMER);
+		if (ret) {
+			ERR("Re-enabling rotate pending timer");
+			ret = -1;
+			goto end;
+		}
+	}
+
+	ret = 0;
+
+end:
+	return ret;
+}
+
+/*
+ * Process the rotate_timer, called with session lock held.
+ */
+static
+int rotate_timer(struct ltt_session *session)
+{
+	int ret;
+
+	/*
+	 * Complete _at most_ one scheduled rotation on a stopped session.
+	 */
+	if (!session->active && session->rotate_timer_enabled &&
+			session->rotated_after_last_stop) {
+		ret = 0;
+		goto end;
+	}
+
+	/* Ignore this timer if a rotation is already in progress. */
+	if (session->rotate_pending || session->rotate_pending_relay) {
+		ret = 0;
+		goto end;
+	}
+
+	DBG("[rotation-thread] Rotate timer on session %s", session->name);
+
+	ret = cmd_rotate_session(session, NULL);
+	if (ret == -LTTNG_ERR_ROTATION_PENDING) {
+		DBG("Scheduled rotation aborted since a rotation is already in progress");
+		ret = 0;
+		goto end;
+	} else if (ret != LTTNG_OK) {
+		ERR("[rotation-thread] Automatic time-triggered rotation failed with error code %i", ret);
+		ret = -1;
+		goto end;
+	}
+
+	ret = 0;
+
+end:
+	return ret;
+}
+
+static
+int handle_rotate_timer_pipe(uint32_t revents,
+		struct rotation_thread_handle *handle,
+		struct rotation_thread_state *state,
+		struct rotation_thread_timer_queue *queue)
+{
+	int ret = 0;
+	int fd = lttng_pipe_get_readfd(queue->event_pipe);
+	struct ltt_session *session;
+	char buf[1];
+
+	if (revents & (LPOLLERR | LPOLLHUP | LPOLLRDHUP)) {
+		ret = lttng_poll_del(&state->events, fd);
+		if (ret) {
+			ERR("[rotation-thread] Failed to remove consumer "
+					"rotate pending pipe from poll set");
+		}
+		goto end;
+	}
+
+	ret = lttng_read(fd, buf, 1);
+	if (ret != 1) {
+		ERR("[rotation-thread] Failed to read from wakeup pipe (fd = %i)", fd);
+		ret = -1;
+		goto end;
+	}
+
+	for (;;) {
+		struct sessiond_rotation_timer *timer_data;
+
+		/*
+		 * Take the queue lock only to pop elements from the list.
+		 */
+		pthread_mutex_lock(&queue->lock);
+		if (cds_list_empty(&queue->list)) {
+			pthread_mutex_unlock(&queue->lock);
+			break;
+		}
+		timer_data = cds_list_first_entry(&queue->list,
+				struct sessiond_rotation_timer, head);
+		cds_list_del(&timer_data->head);
+		pthread_mutex_unlock(&queue->lock);
+
+		/*
+		 * session lock to lookup the session ID.
+		 */
+		session_lock_list();
+		session = session_find_by_id(timer_data->session_id);
+		if (!session) {
+			DBG("[rotation-thread] Session %" PRIu64 " not found",
+					timer_data->session_id);
+			/*
+			 * This is a non-fatal error, and we cannot report it to the
+			 * user (timer), so just print the error and continue the
+			 * processing.
+			 */
+			session_unlock_list();
+			free(timer_data);
+			continue;
+		}
+
+		/*
+		 * Take the session lock and release the session_list lock.
+		 */
+		session_lock(session);
+		session_unlock_list();
+
+		if (timer_data->signal == LTTNG_SESSIOND_SIG_ROTATE_PENDING) {
+			ret = rotate_pending_relay_timer(session);
+		} else if (timer_data->signal == LTTNG_SESSIOND_SIG_ROTATE_TIMER) {
+			ret = rotate_timer(session);
+		} else {
+			ERR("Unknown signal in rotate timer %d", timer_data->signal);
+			ret = -1;
+		}
+		session_unlock(session);
+		free(timer_data);
+		if (ret) {
+			ERR("Error processing timer");
+			goto end;
+		}
+	}
+
+	ret = 0;
+
+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;
+	bool notification_pending;
+	struct lttng_notification *notification = NULL;
+	enum lttng_notification_channel_status status;
+	const struct lttng_evaluation *notification_evaluation;
+	const struct lttng_condition *notification_condition;
+
+	status = lttng_notification_channel_has_pending_notification(
+			rotate_notification_channel, &notification_pending);
+	if (status != LTTNG_NOTIFICATION_CHANNEL_STATUS_OK) {
+		ERR("[rotation-thread ]Error occured while checking for pending notification");
+		ret = -1;
+		goto end;
+	}
+
+	if (!notification_pending) {
+		ret = 0;
+		goto end;
+	}
+
+	/* Receive the next notification. */
+	status = lttng_notification_channel_get_next_notification(
+			rotate_notification_channel,
+			&notification);
+
+	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;
@@ -440,13 +797,27 @@ void *thread_rotation(void *data)
 			if (fd == handle->thread_quit_pipe) {
 				DBG("[rotation-thread] Quit pipe activity");
 				goto exit;
+			} else if (fd == lttng_pipe_get_readfd(handle->rotation_timer_queue->event_pipe)) {
+				ret = handle_rotate_timer_pipe(revents,
+						handle, &state, handle->rotation_timer_queue);
+				if (ret) {
+					ERR("[rotation-thread] Failed to handle rotation timer pipe event");
+					goto error;
+				}
 			} else if (fd == handle->ust32_consumer ||
 					fd == handle->ust64_consumer ||
 					fd == handle->kernel_consumer) {
 				ret = handle_channel_rotation_pipe(fd,
 						revents, handle, &state);
 				if (ret) {
-					ERR("[rotation-thread] Handle channel rotation pipe");
+					ERR("[rotation-thread] Failed to 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;
 				}
 			}