+ 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.list);
+ client_list.trigger = trigger;
+
+ CDS_INIT_LIST_HEAD(&client_list_element.node);
+ client_list_element.client = client;
+ cds_list_add(&client_list_element.node, &client_list.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.
+ */
+ if (evaluate_condition_for_client(client_list->trigger, condition,
+ client, state)) {
+ WARN("[notification-thread] Evaluation of a condition on client subscription failed, aborting.");
+ ret = -1;
+ free(client_list_element);
+ goto end;
+ }
+
+ /*
+ * 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->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->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;
+}
+
+static
+bool trigger_applies_to_client(struct lttng_trigger *trigger,
+ 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,
+ lttng_trigger_get_condition(trigger));
+ if (applies) {
+ break;
+ }
+ }
+ return applies;
+}
+
+/* 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;
+ }