+struct channel_info *channel_info_create(const char *channel_name,
+ struct channel_key *channel_key, uint64_t channel_capacity,
+ struct session_info *session_info)
+{
+ struct channel_info *channel_info = zmalloc(sizeof(*channel_info));
+
+ if (!channel_info) {
+ goto end;
+ }
+
+ cds_lfht_node_init(&channel_info->channels_ht_node);
+ cds_lfht_node_init(&channel_info->session_info_channels_ht_node);
+ memcpy(&channel_info->key, channel_key, sizeof(*channel_key));
+ channel_info->capacity = channel_capacity;
+
+ channel_info->name = strdup(channel_name);
+ if (!channel_info->name) {
+ goto error;
+ }
+
+ /*
+ * Set the references between session and channel infos:
+ * - channel_info holds a strong reference to session_info
+ * - session_info holds a weak reference to channel_info
+ */
+ session_info_get(session_info);
+ session_info_add_channel(session_info, channel_info);
+ channel_info->session_info = session_info;
+end:
+ return channel_info;
+error:
+ channel_info_destroy(channel_info);
+ return NULL;
+}
+
+LTTNG_HIDDEN
+bool notification_client_list_get(struct notification_client_list *list)
+{
+ return urcu_ref_get_unless_zero(&list->ref);
+}
+
+static
+void free_notification_client_list_rcu(struct rcu_head *node)
+{
+ free(caa_container_of(node, struct notification_client_list,
+ rcu_node));
+}
+
+static
+void notification_client_list_release(struct urcu_ref *list_ref)
+{
+ struct notification_client_list *list =
+ container_of(list_ref, typeof(*list), ref);
+ struct notification_client_list_element *client_list_element, *tmp;
+
+ lttng_condition_put(list->condition);
+
+ if (list->notification_trigger_clients_ht) {
+ rcu_read_lock();
+
+ cds_lfht_del(list->notification_trigger_clients_ht,
+ &list->notification_trigger_clients_ht_node);
+ rcu_read_unlock();
+ list->notification_trigger_clients_ht = NULL;
+ }
+ cds_list_for_each_entry_safe(client_list_element, tmp,
+ &list->clients_list, node) {
+ free(client_list_element);
+ }
+
+ assert(cds_list_empty(&list->triggers_list));
+
+ pthread_mutex_destroy(&list->lock);
+ call_rcu(&list->rcu_node, free_notification_client_list_rcu);
+}
+
+static
+bool condition_applies_to_client(const struct lttng_condition *condition,
+ struct notification_client *client)
+{
+ bool applies = false;
+ struct lttng_condition_list_element *condition_list_element;
+
+ cds_list_for_each_entry(condition_list_element, &client->condition_list,
+ node) {
+ applies = lttng_condition_is_equal(
+ condition_list_element->condition,
+ condition);
+ if (applies) {
+ break;
+ }
+ }
+
+ return applies;
+}
+
+static
+struct notification_client_list *notification_client_list_create(
+ struct notification_thread_state *state,
+ const struct lttng_condition *condition)
+{
+ struct notification_client *client;
+ struct cds_lfht_iter iter;
+ struct notification_client_list *client_list;
+
+ client_list = zmalloc(sizeof(*client_list));
+ if (!client_list) {
+ PERROR("Failed to allocate notification client list");
+ goto end;
+ }
+
+ pthread_mutex_init(&client_list->lock, NULL);
+ /*
+ * The trigger that owns the condition has the first reference to this
+ * client list.
+ */
+ urcu_ref_init(&client_list->ref);
+ cds_lfht_node_init(&client_list->notification_trigger_clients_ht_node);
+ CDS_INIT_LIST_HEAD(&client_list->clients_list);
+ CDS_INIT_LIST_HEAD(&client_list->triggers_list);
+
+ /*
+ * Create a copy of the condition so that it's independent of any
+ * trigger. The client list may outlive the trigger object (which owns
+ * the condition) that is used to create it.
+ */
+ client_list->condition = lttng_condition_copy(condition);
+
+ /* Build a list of clients to which this new condition applies. */
+ cds_lfht_for_each_entry (state->client_socket_ht, &iter, client,
+ client_socket_ht_node) {
+ struct notification_client_list_element *client_list_element;
+
+ if (!condition_applies_to_client(condition, client)) {
+ continue;
+ }
+
+ client_list_element = zmalloc(sizeof(*client_list_element));
+ if (!client_list_element) {
+ goto error_put_client_list;
+ }
+
+ CDS_INIT_LIST_HEAD(&client_list_element->node);
+ client_list_element->client = client;
+ cds_list_add(&client_list_element->node, &client_list->clients_list);
+ }
+
+ client_list->notification_trigger_clients_ht =
+ state->notification_trigger_clients_ht;
+
+ rcu_read_lock();
+ /*
+ * Add the client list to the global list of client list.
+ */
+ cds_lfht_add_unique(state->notification_trigger_clients_ht,
+ lttng_condition_hash(client_list->condition),
+ match_client_list_condition,
+ client_list->condition,
+ &client_list->notification_trigger_clients_ht_node);
+ rcu_read_unlock();
+ goto end;
+
+error_put_client_list:
+ notification_client_list_put(client_list);
+ client_list = NULL;
+
+end:
+ return client_list;
+}
+
+void notification_client_list_put(struct notification_client_list *list)
+{
+ if (!list) {
+ return;
+ }
+ return urcu_ref_put(&list->ref, notification_client_list_release);
+}
+
+/* Provides a reference to the returned list. */
+static
+struct notification_client_list *get_client_list_from_condition(
+ struct notification_thread_state *state,
+ const struct lttng_condition *condition)
+{
+ struct cds_lfht_node *node;
+ struct cds_lfht_iter iter;
+ struct notification_client_list *list = NULL;
+
+ rcu_read_lock();
+ cds_lfht_lookup(state->notification_trigger_clients_ht,
+ lttng_condition_hash(condition),
+ match_client_list_condition,
+ condition,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (node) {
+ list = container_of(node, struct notification_client_list,
+ notification_trigger_clients_ht_node);
+ list = notification_client_list_get(list) ? list : NULL;
+ }
+
+ rcu_read_unlock();
+ return list;
+}
+
+static
+int evaluate_channel_condition_for_client(
+ const struct lttng_condition *condition,
+ struct notification_thread_state *state,
+ struct lttng_evaluation **evaluation,
+ uid_t *session_uid, gid_t *session_gid)
+{
+ int ret;
+ struct cds_lfht_iter iter;
+ struct cds_lfht_node *node;
+ struct channel_info *channel_info = NULL;
+ struct channel_key *channel_key = NULL;
+ struct channel_state_sample *last_sample = NULL;
+ struct lttng_channel_trigger_list *channel_trigger_list = NULL;
+
+ rcu_read_lock();
+
+ /* Find the channel associated with the condition. */
+ cds_lfht_for_each_entry(state->channel_triggers_ht, &iter,
+ channel_trigger_list, channel_triggers_ht_node) {
+ struct lttng_trigger_list_element *element;
+
+ cds_list_for_each_entry(element, &channel_trigger_list->list, node) {
+ const struct lttng_condition *current_condition =
+ lttng_trigger_get_const_condition(
+ element->trigger);
+
+ assert(current_condition);
+ if (!lttng_condition_is_equal(condition,
+ current_condition)) {
+ continue;
+ }
+
+ /* Found the trigger, save the channel key. */
+ channel_key = &channel_trigger_list->channel_key;
+ break;
+ }
+ if (channel_key) {
+ /* The channel key was found stop iteration. */
+ break;
+ }
+ }
+
+ if (!channel_key){
+ /* No channel found; normal exit. */
+ DBG("[notification-thread] No known channel associated with newly subscribed-to condition");
+ ret = 0;
+ goto end;
+ }
+
+ /* Fetch channel info for the matching channel. */
+ cds_lfht_lookup(state->channels_ht,
+ hash_channel_key(channel_key),
+ match_channel_info,
+ channel_key,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ assert(node);
+ channel_info = caa_container_of(node, struct channel_info,
+ channels_ht_node);
+
+ /* Retrieve the channel's last sample, if it exists. */
+ cds_lfht_lookup(state->channel_state_ht,
+ hash_channel_key(channel_key),
+ match_channel_state_sample,
+ channel_key,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (node) {
+ last_sample = caa_container_of(node,
+ struct channel_state_sample,
+ channel_state_ht_node);
+ } else {
+ /* Nothing to evaluate, no sample was ever taken. Normal exit */
+ DBG("[notification-thread] No channel sample associated with newly subscribed-to condition");
+ ret = 0;
+ goto end;
+ }
+
+ ret = evaluate_buffer_condition(condition, evaluation, state,
+ NULL, last_sample,
+ 0, channel_info->session_info->consumed_data_size,
+ channel_info);
+ if (ret) {
+ WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition");
+ goto end;
+ }
+
+ *session_uid = channel_info->session_info->uid;
+ *session_gid = channel_info->session_info->gid;
+end:
+ rcu_read_unlock();
+ return ret;
+}
+
+static
+const char *get_condition_session_name(const struct lttng_condition *condition)
+{
+ const char *session_name = NULL;
+ enum lttng_condition_status status;
+
+ switch (lttng_condition_get_type(condition)) {
+ case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
+ case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
+ status = lttng_condition_buffer_usage_get_session_name(
+ condition, &session_name);
+ break;
+ case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
+ status = lttng_condition_session_consumed_size_get_session_name(
+ condition, &session_name);
+ break;
+ case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
+ case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
+ status = lttng_condition_session_rotation_get_session_name(
+ condition, &session_name);
+ break;
+ default:
+ abort();
+ }
+ if (status != LTTNG_CONDITION_STATUS_OK) {
+ ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
+ goto end;
+ }
+end:
+ return session_name;
+}
+
+static
+int evaluate_session_condition_for_client(
+ const struct lttng_condition *condition,
+ struct notification_thread_state *state,
+ struct lttng_evaluation **evaluation,
+ uid_t *session_uid, gid_t *session_gid)
+{
+ int ret;
+ struct cds_lfht_iter iter;
+ struct cds_lfht_node *node;
+ const char *session_name;
+ struct session_info *session_info = NULL;
+
+ rcu_read_lock();
+ session_name = get_condition_session_name(condition);
+
+ /* Find the session associated with the trigger. */
+ cds_lfht_lookup(state->sessions_ht,
+ hash_key_str(session_name, lttng_ht_seed),
+ match_session,
+ session_name,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (!node) {
+ DBG("[notification-thread] No known session matching name \"%s\"",
+ session_name);
+ ret = 0;
+ goto end;
+ }
+
+ session_info = caa_container_of(node, struct session_info,
+ sessions_ht_node);
+ session_info_get(session_info);
+
+ /*
+ * Evaluation is performed in-line here since only one type of
+ * session-bound condition is handled for the moment.
+ */
+ switch (lttng_condition_get_type(condition)) {
+ case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
+ if (!session_info->rotation.ongoing) {
+ ret = 0;
+ goto end_session_put;
+ }
+
+ *evaluation = lttng_evaluation_session_rotation_ongoing_create(
+ session_info->rotation.id);
+ if (!*evaluation) {
+ /* Fatal error. */
+ ERR("[notification-thread] Failed to create session rotation ongoing evaluation for session \"%s\"",
+ session_info->name);
+ ret = -1;
+ goto end_session_put;
+ }
+ ret = 0;
+ break;
+ default:
+ ret = 0;
+ goto end_session_put;
+ }
+
+ *session_uid = session_info->uid;
+ *session_gid = session_info->gid;
+
+end_session_put:
+ session_info_put(session_info);
+end:
+ rcu_read_unlock();
+ return ret;
+}
+
+static
+int evaluate_condition_for_client(const struct lttng_trigger *trigger,
+ const struct lttng_condition *condition,
+ struct notification_client *client,
+ struct notification_thread_state *state)
+{
+ int ret;
+ struct lttng_evaluation *evaluation = NULL;
+ struct notification_client_list client_list = {
+ .lock = PTHREAD_MUTEX_INITIALIZER,
+ };
+ struct notification_client_list_element client_list_element = { 0 };
+ uid_t object_uid = 0;
+ gid_t object_gid = 0;
+
+ assert(trigger);
+ assert(condition);
+ assert(client);
+ assert(state);
+
+ switch (get_condition_binding_object(condition)) {
+ case LTTNG_OBJECT_TYPE_SESSION:
+ ret = evaluate_session_condition_for_client(condition, state,
+ &evaluation, &object_uid, &object_gid);
+ break;
+ case LTTNG_OBJECT_TYPE_CHANNEL:
+ ret = evaluate_channel_condition_for_client(condition, state,
+ &evaluation, &object_uid, &object_gid);
+ break;
+ case LTTNG_OBJECT_TYPE_NONE:
+ DBG("[notification-thread] Newly subscribed-to condition not bound to object, nothing to evaluate");
+ ret = 0;
+ goto end;
+ case LTTNG_OBJECT_TYPE_UNKNOWN:
+ default:
+ ret = -1;
+ goto end;
+ }
+ if (ret) {
+ /* Fatal error. */
+ goto end;
+ }
+ if (!evaluation) {
+ /* Evaluation yielded nothing. Normal exit. */
+ DBG("[notification-thread] Newly subscribed-to condition evaluated to false, nothing to report to client");
+ ret = 0;
+ goto end;
+ }
+
+ /*
+ * Create a temporary client list with the client currently
+ * subscribing.
+ */
+ cds_lfht_node_init(&client_list.notification_trigger_clients_ht_node);
+ CDS_INIT_LIST_HEAD(&client_list.clients_list);
+
+ CDS_INIT_LIST_HEAD(&client_list_element.node);
+ client_list_element.client = client;
+ cds_list_add(&client_list_element.node, &client_list.clients_list);
+
+ /* Send evaluation result to the newly-subscribed client. */
+ DBG("[notification-thread] Newly subscribed-to condition evaluated to true, notifying client");
+ ret = send_evaluation_to_clients(trigger, evaluation, &client_list,
+ state, object_uid, object_gid);
+
+end:
+ return ret;
+}
+
+static
+int notification_thread_client_subscribe(struct notification_client *client,
+ struct lttng_condition *condition,
+ struct notification_thread_state *state,
+ enum lttng_notification_channel_status *_status)
+{
+ int ret = 0;
+ struct notification_client_list *client_list = NULL;
+ struct lttng_condition_list_element *condition_list_element = NULL;
+ struct notification_client_list_element *client_list_element = NULL;
+ enum lttng_notification_channel_status status =
+ LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
+
+ /*
+ * Ensure that the client has not already subscribed to this condition
+ * before.
+ */
+ cds_list_for_each_entry(condition_list_element, &client->condition_list, node) {
+ if (lttng_condition_is_equal(condition_list_element->condition,
+ condition)) {
+ status = LTTNG_NOTIFICATION_CHANNEL_STATUS_ALREADY_SUBSCRIBED;
+ goto end;
+ }
+ }
+
+ condition_list_element = zmalloc(sizeof(*condition_list_element));
+ if (!condition_list_element) {
+ ret = -1;
+ goto error;
+ }
+ client_list_element = zmalloc(sizeof(*client_list_element));
+ if (!client_list_element) {
+ ret = -1;
+ goto error;
+ }
+
+ /*
+ * Add the newly-subscribed condition to the client's subscription list.
+ */
+ CDS_INIT_LIST_HEAD(&condition_list_element->node);
+ condition_list_element->condition = condition;
+ cds_list_add(&condition_list_element->node, &client->condition_list);
+
+ client_list = get_client_list_from_condition(state, condition);
+ if (!client_list) {
+ /*
+ * No notification-emiting trigger registered with this
+ * condition. We don't evaluate the condition right away
+ * since this trigger is not registered yet.
+ */
+ free(client_list_element);
+ goto end;
+ }
+
+ /*
+ * The condition to which the client just subscribed is evaluated
+ * at this point so that conditions that are already TRUE result
+ * in a notification being sent out.
+ *
+ * The client_list's trigger is used without locking the list itself.
+ * This is correct since the list doesn't own the trigger and the
+ * object is immutable.
+ */
+ struct lttng_trigger_ht_element *trigger_ht_element;
+ pthread_mutex_lock(&client_list->lock);
+ cds_list_for_each_entry(trigger_ht_element,
+ &client_list->triggers_list, client_list_trigger_node) {
+ if (evaluate_condition_for_client(trigger_ht_element->trigger, condition,
+ client, state)) {
+ WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
+ ret = -1;
+ free(client_list_element);
+ goto end;
+ }
+ }
+ pthread_mutex_unlock(&client_list->lock);
+
+ /*
+ * Add the client to the list of clients interested in a given trigger
+ * if a "notification" trigger with a corresponding condition was
+ * added prior.
+ */
+ client_list_element->client = client;
+ CDS_INIT_LIST_HEAD(&client_list_element->node);
+
+ pthread_mutex_lock(&client_list->lock);
+ cds_list_add(&client_list_element->node, &client_list->clients_list);
+ pthread_mutex_unlock(&client_list->lock);
+end:
+ if (_status) {
+ *_status = status;
+ }
+ if (client_list) {
+ notification_client_list_put(client_list);
+ }
+ return ret;
+error:
+ free(condition_list_element);
+ free(client_list_element);
+ return ret;
+}
+
+static
+int notification_thread_client_unsubscribe(
+ struct notification_client *client,
+ struct lttng_condition *condition,
+ struct notification_thread_state *state,
+ enum lttng_notification_channel_status *_status)
+{
+ struct notification_client_list *client_list;
+ struct lttng_condition_list_element *condition_list_element,
+ *condition_tmp;
+ struct notification_client_list_element *client_list_element,
+ *client_tmp;
+ bool condition_found = false;
+ enum lttng_notification_channel_status status =
+ LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
+
+ /* Remove the condition from the client's condition list. */
+ cds_list_for_each_entry_safe(condition_list_element, condition_tmp,
+ &client->condition_list, node) {
+ if (!lttng_condition_is_equal(condition_list_element->condition,
+ condition)) {
+ continue;
+ }
+
+ cds_list_del(&condition_list_element->node);
+ /*
+ * The caller may be iterating on the client's conditions to
+ * tear down a client's connection. In this case, the condition
+ * will be destroyed at the end.
+ */
+ if (condition != condition_list_element->condition) {
+ lttng_condition_destroy(
+ condition_list_element->condition);
+ }
+ free(condition_list_element);
+ condition_found = true;
+ break;
+ }
+
+ if (!condition_found) {
+ status = LTTNG_NOTIFICATION_CHANNEL_STATUS_UNKNOWN_CONDITION;
+ goto end;
+ }
+
+ /*
+ * Remove the client from the list of clients interested the trigger
+ * matching the condition.
+ */
+ client_list = get_client_list_from_condition(state, condition);
+ if (!client_list) {
+ goto end;
+ }
+
+ pthread_mutex_lock(&client_list->lock);
+ cds_list_for_each_entry_safe(client_list_element, client_tmp,
+ &client_list->clients_list, node) {
+ if (client_list_element->client->id != client->id) {
+ continue;
+ }
+ cds_list_del(&client_list_element->node);
+ free(client_list_element);
+ break;
+ }
+ pthread_mutex_unlock(&client_list->lock);
+ notification_client_list_put(client_list);
+ client_list = NULL;
+end:
+ lttng_condition_destroy(condition);
+ if (_status) {
+ *_status = status;
+ }
+ return 0;
+}
+
+static
+void free_notification_client_rcu(struct rcu_head *node)
+{
+ free(caa_container_of(node, struct notification_client, rcu_node));
+}
+
+static
+void notification_client_destroy(struct notification_client *client,
+ struct notification_thread_state *state)
+{
+ if (!client) {
+ return;
+ }
+
+ /*
+ * The client object is not reachable by other threads, no need to lock
+ * the client here.
+ */
+ if (client->socket >= 0) {
+ (void) lttcomm_close_unix_sock(client->socket);
+ client->socket = -1;
+ }
+ client->communication.active = false;
+ lttng_payload_reset(&client->communication.inbound.payload);
+ lttng_payload_reset(&client->communication.outbound.payload);
+ pthread_mutex_destroy(&client->lock);
+ call_rcu(&client->rcu_node, free_notification_client_rcu);
+}
+
+/*
+ * Call with rcu_read_lock held (and hold for the lifetime of the returned
+ * client pointer).
+ */
+static
+struct notification_client *get_client_from_socket(int socket,
+ struct notification_thread_state *state)
+{
+ struct cds_lfht_iter iter;
+ struct cds_lfht_node *node;
+ struct notification_client *client = NULL;
+
+ cds_lfht_lookup(state->client_socket_ht,
+ hash_client_socket(socket),
+ match_client_socket,
+ (void *) (unsigned long) socket,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (!node) {
+ goto end;
+ }
+
+ client = caa_container_of(node, struct notification_client,
+ client_socket_ht_node);
+end:
+ return client;
+}
+
+/*
+ * Call with rcu_read_lock held (and hold for the lifetime of the returned
+ * client pointer).
+ */
+static
+struct notification_client *get_client_from_id(notification_client_id id,
+ struct notification_thread_state *state)
+{
+ struct cds_lfht_iter iter;
+ struct cds_lfht_node *node;
+ struct notification_client *client = NULL;
+
+ cds_lfht_lookup(state->client_id_ht,
+ hash_client_id(id),
+ match_client_id,
+ &id,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (!node) {
+ goto end;
+ }
+
+ client = caa_container_of(node, struct notification_client,
+ client_id_ht_node);
+end:
+ return client;
+}
+
+static
+bool buffer_usage_condition_applies_to_channel(
+ const struct lttng_condition *condition,
+ const struct channel_info *channel_info)
+{
+ enum lttng_condition_status status;
+ enum lttng_domain_type condition_domain;
+ const char *condition_session_name = NULL;
+ const char *condition_channel_name = NULL;
+
+ status = lttng_condition_buffer_usage_get_domain_type(condition,
+ &condition_domain);
+ assert(status == LTTNG_CONDITION_STATUS_OK);
+ if (channel_info->key.domain != condition_domain) {
+ goto fail;
+ }
+
+ status = lttng_condition_buffer_usage_get_session_name(
+ condition, &condition_session_name);
+ assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
+
+ status = lttng_condition_buffer_usage_get_channel_name(
+ condition, &condition_channel_name);
+ assert((status == LTTNG_CONDITION_STATUS_OK) && condition_channel_name);
+
+ if (strcmp(channel_info->session_info->name, condition_session_name)) {
+ goto fail;
+ }
+ if (strcmp(channel_info->name, condition_channel_name)) {
+ goto fail;
+ }
+
+ return true;
+fail:
+ return false;
+}
+
+static
+bool session_consumed_size_condition_applies_to_channel(
+ const struct lttng_condition *condition,
+ const struct channel_info *channel_info)
+{
+ enum lttng_condition_status status;
+ const char *condition_session_name = NULL;
+
+ status = lttng_condition_session_consumed_size_get_session_name(
+ condition, &condition_session_name);
+ assert((status == LTTNG_CONDITION_STATUS_OK) && condition_session_name);
+
+ if (strcmp(channel_info->session_info->name, condition_session_name)) {
+ goto fail;
+ }
+
+ return true;
+fail:
+ return false;
+}
+
+static
+bool trigger_applies_to_channel(const struct lttng_trigger *trigger,
+ const struct channel_info *channel_info)
+{
+ const struct lttng_condition *condition;
+ bool trigger_applies;
+
+ condition = lttng_trigger_get_const_condition(trigger);
+ if (!condition) {
+ goto fail;
+ }
+
+ switch (lttng_condition_get_type(condition)) {
+ case LTTNG_CONDITION_TYPE_BUFFER_USAGE_LOW:
+ case LTTNG_CONDITION_TYPE_BUFFER_USAGE_HIGH:
+ trigger_applies = buffer_usage_condition_applies_to_channel(
+ condition, channel_info);
+ break;
+ case LTTNG_CONDITION_TYPE_SESSION_CONSUMED_SIZE:
+ trigger_applies = session_consumed_size_condition_applies_to_channel(
+ condition, channel_info);
+ break;
+ default:
+ goto fail;
+ }
+
+ return trigger_applies;
+fail:
+ return false;
+}
+
+/* Must be called with RCU read lock held. */
+static
+struct lttng_session_trigger_list *get_session_trigger_list(
+ struct notification_thread_state *state,
+ const char *session_name)
+{
+ struct lttng_session_trigger_list *list = NULL;
+ struct cds_lfht_node *node;
+ struct cds_lfht_iter iter;
+
+ cds_lfht_lookup(state->session_triggers_ht,
+ hash_key_str(session_name, lttng_ht_seed),
+ match_session_trigger_list,
+ session_name,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (!node) {
+ /*
+ * Not an error, the list of triggers applying to that session
+ * will be initialized when the session is created.
+ */
+ DBG("[notification-thread] No trigger list found for session \"%s\" as it is not yet known to the notification system",
+ session_name);
+ goto end;
+ }
+
+ list = caa_container_of(node,
+ struct lttng_session_trigger_list,
+ session_triggers_ht_node);
+end:
+ return list;
+}
+
+/*
+ * Allocate an empty lttng_session_trigger_list for the session named
+ * 'session_name'.
+ *
+ * No ownership of 'session_name' is assumed by the session trigger list.
+ * It is the caller's responsability to ensure the session name is alive
+ * for as long as this list is.
+ */
+static
+struct lttng_session_trigger_list *lttng_session_trigger_list_create(
+ const char *session_name,
+ struct cds_lfht *session_triggers_ht)
+{
+ struct lttng_session_trigger_list *list;
+
+ list = zmalloc(sizeof(*list));
+ if (!list) {
+ goto end;
+ }
+ list->session_name = session_name;
+ CDS_INIT_LIST_HEAD(&list->list);
+ cds_lfht_node_init(&list->session_triggers_ht_node);
+ list->session_triggers_ht = session_triggers_ht;
+
+ rcu_read_lock();
+ /* Publish the list through the session_triggers_ht. */
+ cds_lfht_add(session_triggers_ht,
+ hash_key_str(session_name, lttng_ht_seed),
+ &list->session_triggers_ht_node);
+ rcu_read_unlock();
+end:
+ return list;
+}
+
+static
+void free_session_trigger_list_rcu(struct rcu_head *node)
+{
+ free(caa_container_of(node, struct lttng_session_trigger_list,
+ rcu_node));
+}
+
+static
+void lttng_session_trigger_list_destroy(struct lttng_session_trigger_list *list)
+{
+ struct lttng_trigger_list_element *trigger_list_element, *tmp;
+
+ /* Empty the list element by element, and then free the list itself. */
+ cds_list_for_each_entry_safe(trigger_list_element, tmp,
+ &list->list, node) {
+ cds_list_del(&trigger_list_element->node);
+ free(trigger_list_element);
+ }
+ rcu_read_lock();
+ /* Unpublish the list from the session_triggers_ht. */
+ cds_lfht_del(list->session_triggers_ht,
+ &list->session_triggers_ht_node);
+ rcu_read_unlock();
+ call_rcu(&list->rcu_node, free_session_trigger_list_rcu);
+}
+
+static
+int lttng_session_trigger_list_add(struct lttng_session_trigger_list *list,
+ struct lttng_trigger *trigger)
+{
+ int ret = 0;
+ struct lttng_trigger_list_element *new_element =
+ zmalloc(sizeof(*new_element));
+
+ if (!new_element) {
+ ret = -1;
+ goto end;
+ }
+ CDS_INIT_LIST_HEAD(&new_element->node);
+ new_element->trigger = trigger;
+ cds_list_add(&new_element->node, &list->list);
+end:
+ return ret;
+}
+
+static
+bool trigger_applies_to_session(const struct lttng_trigger *trigger,
+ const char *session_name)
+{
+ bool applies = false;
+ const struct lttng_condition *condition;
+
+ condition = lttng_trigger_get_const_condition(trigger);
+ switch (lttng_condition_get_type(condition)) {
+ case LTTNG_CONDITION_TYPE_SESSION_ROTATION_ONGOING:
+ case LTTNG_CONDITION_TYPE_SESSION_ROTATION_COMPLETED:
+ {
+ enum lttng_condition_status condition_status;
+ const char *condition_session_name;
+
+ condition_status = lttng_condition_session_rotation_get_session_name(
+ condition, &condition_session_name);
+ if (condition_status != LTTNG_CONDITION_STATUS_OK) {
+ ERR("[notification-thread] Failed to retrieve session rotation condition's session name");
+ goto end;
+ }
+
+ assert(condition_session_name);
+ applies = !strcmp(condition_session_name, session_name);
+ break;
+ }
+ default:
+ goto end;
+ }
+end:
+ return applies;
+}
+
+/*
+ * Allocate and initialize an lttng_session_trigger_list which contains
+ * all triggers that apply to the session named 'session_name'.
+ *
+ * No ownership of 'session_name' is assumed by the session trigger list.
+ * It is the caller's responsability to ensure the session name is alive
+ * for as long as this list is.
+ */
+static
+struct lttng_session_trigger_list *lttng_session_trigger_list_build(
+ const struct notification_thread_state *state,
+ const char *session_name)
+{
+ int trigger_count = 0;
+ struct lttng_session_trigger_list *session_trigger_list = NULL;
+ struct lttng_trigger_ht_element *trigger_ht_element = NULL;
+ struct cds_lfht_iter iter;
+
+ session_trigger_list = lttng_session_trigger_list_create(session_name,
+ state->session_triggers_ht);
+
+ /* Add all triggers applying to the session named 'session_name'. */
+ cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
+ node) {
+ int ret;
+
+ if (!trigger_applies_to_session(trigger_ht_element->trigger,
+ session_name)) {
+ continue;
+ }
+
+ ret = lttng_session_trigger_list_add(session_trigger_list,
+ trigger_ht_element->trigger);
+ if (ret) {
+ goto error;
+ }
+
+ trigger_count++;
+ }
+
+ DBG("[notification-thread] Found %i triggers that apply to newly created session",
+ trigger_count);
+ return session_trigger_list;
+error:
+ lttng_session_trigger_list_destroy(session_trigger_list);
+ return NULL;
+}
+
+static
+struct session_info *find_or_create_session_info(
+ struct notification_thread_state *state,
+ const char *name, uid_t uid, gid_t gid)
+{
+ struct session_info *session = NULL;
+ struct cds_lfht_node *node;
+ struct cds_lfht_iter iter;
+ struct lttng_session_trigger_list *trigger_list;
+
+ rcu_read_lock();
+ cds_lfht_lookup(state->sessions_ht,
+ hash_key_str(name, lttng_ht_seed),
+ match_session,
+ name,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+ if (node) {
+ DBG("[notification-thread] Found session info of session \"%s\" (uid = %i, gid = %i)",
+ name, uid, gid);
+ session = caa_container_of(node, struct session_info,
+ sessions_ht_node);
+ assert(session->uid == uid);
+ assert(session->gid == gid);
+ session_info_get(session);
+ goto end;
+ }
+
+ trigger_list = lttng_session_trigger_list_build(state, name);
+ if (!trigger_list) {
+ goto error;
+ }
+
+ session = session_info_create(name, uid, gid, trigger_list,
+ state->sessions_ht);
+ if (!session) {
+ ERR("[notification-thread] Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
+ name, uid, gid);
+ lttng_session_trigger_list_destroy(trigger_list);
+ goto error;
+ }
+ trigger_list = NULL;
+
+ cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
+ &session->sessions_ht_node);
+end:
+ rcu_read_unlock();
+ return session;
+error:
+ rcu_read_unlock();
+ session_info_put(session);
+ return NULL;
+}
+
+static
+int handle_notification_thread_command_add_channel(
+ struct notification_thread_state *state,
+ const char *session_name, uid_t session_uid, gid_t session_gid,
+ const char *channel_name, enum lttng_domain_type channel_domain,
+ uint64_t channel_key_int, uint64_t channel_capacity,
+ enum lttng_error_code *cmd_result)
+{
+ struct cds_list_head trigger_list;
+ struct channel_info *new_channel_info = NULL;
+ struct channel_key channel_key = {
+ .key = channel_key_int,
+ .domain = channel_domain,
+ };
+ struct lttng_channel_trigger_list *channel_trigger_list = NULL;
+ struct lttng_trigger_ht_element *trigger_ht_element = NULL;
+ int trigger_count = 0;
+ struct cds_lfht_iter iter;
+ struct session_info *session_info = NULL;
+
+ DBG("[notification-thread] Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
+ channel_name, session_name, channel_key_int,
+ lttng_domain_type_str(channel_domain));
+
+ CDS_INIT_LIST_HEAD(&trigger_list);
+
+ session_info = find_or_create_session_info(state, session_name,
+ session_uid, session_gid);
+ if (!session_info) {
+ /* Allocation error or an internal error occurred. */
+ goto error;
+ }
+
+ new_channel_info = channel_info_create(channel_name, &channel_key,
+ channel_capacity, session_info);
+ if (!new_channel_info) {
+ goto error;
+ }
+
+ rcu_read_lock();
+ /* Build a list of all triggers applying to the new channel. */
+ cds_lfht_for_each_entry(state->triggers_ht, &iter, trigger_ht_element,
+ node) {
+ struct lttng_trigger_list_element *new_element;
+
+ if (!trigger_applies_to_channel(trigger_ht_element->trigger,
+ new_channel_info)) {
+ continue;
+ }
+
+ new_element = zmalloc(sizeof(*new_element));
+ if (!new_element) {
+ rcu_read_unlock();
+ goto error;
+ }
+ CDS_INIT_LIST_HEAD(&new_element->node);
+ new_element->trigger = trigger_ht_element->trigger;
+ cds_list_add(&new_element->node, &trigger_list);
+ trigger_count++;
+ }
+ rcu_read_unlock();
+
+ DBG("[notification-thread] Found %i triggers that apply to newly added channel",
+ trigger_count);
+ channel_trigger_list = zmalloc(sizeof(*channel_trigger_list));
+ if (!channel_trigger_list) {
+ goto error;
+ }
+ channel_trigger_list->channel_key = new_channel_info->key;
+ CDS_INIT_LIST_HEAD(&channel_trigger_list->list);
+ cds_lfht_node_init(&channel_trigger_list->channel_triggers_ht_node);
+ cds_list_splice(&trigger_list, &channel_trigger_list->list);
+
+ rcu_read_lock();
+ /* Add channel to the channel_ht which owns the channel_infos. */
+ cds_lfht_add(state->channels_ht,
+ hash_channel_key(&new_channel_info->key),
+ &new_channel_info->channels_ht_node);
+ /*
+ * Add the list of triggers associated with this channel to the
+ * channel_triggers_ht.
+ */
+ cds_lfht_add(state->channel_triggers_ht,
+ hash_channel_key(&new_channel_info->key),
+ &channel_trigger_list->channel_triggers_ht_node);
+ rcu_read_unlock();
+ session_info_put(session_info);
+ *cmd_result = LTTNG_OK;
+ return 0;
+error:
+ channel_info_destroy(new_channel_info);
+ session_info_put(session_info);
+ return 1;
+}
+
+static
+void free_channel_trigger_list_rcu(struct rcu_head *node)
+{
+ free(caa_container_of(node, struct lttng_channel_trigger_list,
+ rcu_node));
+}
+
+static
+void free_channel_state_sample_rcu(struct rcu_head *node)
+{
+ free(caa_container_of(node, struct channel_state_sample,
+ rcu_node));
+}
+
+static
+int handle_notification_thread_command_remove_channel(
+ struct notification_thread_state *state,
+ uint64_t channel_key, enum lttng_domain_type domain,
+ enum lttng_error_code *cmd_result)