X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fnotification-thread-events.c;h=4799c718641bfb0be35bf9433405abf71ab82ba4;hb=0fe4c3bf415eef3325b53383cbc3bc087db55d7f;hp=751e665bdc624936a8401fd307ae75ff93656978;hpb=8a9bd316322585e614fe4db6ef4bd2841c85b3f0;p=lttng-tools.git diff --git a/src/bin/lttng-sessiond/notification-thread-events.c b/src/bin/lttng-sessiond/notification-thread-events.c index 751e665bd..4799c7186 100644 --- a/src/bin/lttng-sessiond/notification-thread-events.c +++ b/src/bin/lttng-sessiond/notification-thread-events.c @@ -58,11 +58,15 @@ struct lttng_channel_trigger_list { struct channel_key channel_key; struct cds_list_head list; struct cds_lfht_node channel_triggers_ht_node; + /* call_rcu delayed reclaim. */ + struct rcu_head rcu_node; }; struct lttng_trigger_ht_element { struct lttng_trigger *trigger; struct cds_lfht_node node; + /* call_rcu delayed reclaim. */ + struct rcu_head rcu_node; }; struct lttng_condition_list_element { @@ -79,6 +83,8 @@ struct notification_client_list { struct lttng_trigger *trigger; struct cds_list_head list; struct cds_lfht_node notification_trigger_ht_node; + /* call_rcu delayed reclaim. */ + struct rcu_head rcu_node; }; struct notification_client { @@ -88,7 +94,7 @@ struct notification_client { uid_t uid; gid_t gid; /* - * Indicates if the credentials and versions of the client has been + * Indicates if the credentials and versions of the client have been * checked. */ bool validated; @@ -145,6 +151,8 @@ struct notification_client { struct lttng_dynamic_buffer buffer; } outbound; } communication; + /* call_rcu delayed reclaim. */ + struct rcu_head rcu_node; }; struct channel_state_sample { @@ -152,8 +160,24 @@ struct channel_state_sample { struct cds_lfht_node channel_state_ht_node; uint64_t highest_usage; uint64_t lowest_usage; + /* call_rcu delayed reclaim. */ + struct rcu_head rcu_node; }; +static unsigned long hash_channel_key(struct channel_key *key); +static int evaluate_condition(struct lttng_condition *condition, + struct lttng_evaluation **evaluation, + struct notification_thread_state *state, + struct channel_state_sample *previous_sample, + struct channel_state_sample *latest_sample, + uint64_t buffer_capacity); +static +int send_evaluation_to_clients(struct lttng_trigger *trigger, + struct lttng_evaluation *evaluation, + struct notification_client_list *client_list, + struct notification_thread_state *state, + uid_t channel_uid, gid_t channel_gid); + static int match_client(struct cds_lfht_node *node, const void *key) { @@ -281,7 +305,7 @@ unsigned long lttng_condition_buffer_usage_hash( val = condition->threshold_ratio.value * (double) UINT32_MAX; hash ^= hash_key_u64(&val, lttng_ht_seed); - } else if (condition->threshold_ratio.set) { + } else if (condition->threshold_bytes.set) { uint64_t val; val = condition->threshold_bytes.value; @@ -308,6 +332,22 @@ unsigned long lttng_condition_hash(struct lttng_condition *condition) } } +static +unsigned long hash_channel_key(struct channel_key *key) +{ + unsigned long key_hash = hash_key_u64(&key->key, lttng_ht_seed); + unsigned long domain_hash = hash_key_ulong( + (void *) (unsigned long) key->domain, lttng_ht_seed); + + return key_hash ^ domain_hash; +} + +static +void free_channel_info_rcu(struct rcu_head *node) +{ + free(caa_container_of(node, struct channel_info, rcu_node)); +} + static void channel_info_destroy(struct channel_info *channel_info) { @@ -321,7 +361,7 @@ void channel_info_destroy(struct channel_info *channel_info) if (channel_info->channel_name) { free(channel_info->channel_name); } - free(channel_info); + call_rcu(&channel_info->rcu_node, free_channel_info_rcu); } static @@ -357,6 +397,126 @@ error: return NULL; } +/* This function must be called with the RCU read lock held. */ +static +int evaluate_condition_for_client(struct lttng_trigger *trigger, + struct lttng_condition *condition, + struct notification_client *client, + struct notification_thread_state *state) +{ + 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; + struct lttng_evaluation *evaluation = NULL; + struct notification_client_list client_list = { 0 }; + struct notification_client_list_element client_list_element = { 0 }; + + assert(trigger); + assert(condition); + assert(client); + assert(state); + + /* Find the channel associated with the trigger. */ + 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) { + struct lttng_condition *current_condition = + lttng_trigger_get_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 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_condition(condition, &evaluation, state, NULL, + last_sample, channel_info->capacity); + if (ret) { + WARN("[notification-thread] Fatal error occurred while evaluating a newly subscribed-to condition"); + 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_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, channel_info->uid, channel_info->gid); + +end: + return ret; +} + static int notification_thread_client_subscribe(struct notification_client *client, struct lttng_condition *condition, @@ -404,11 +564,6 @@ int notification_thread_client_subscribe(struct notification_client *client, condition_list_element->condition = condition; cds_list_add(&condition_list_element->node, &client->condition_list); - /* - * Add the client to the list of clients interested in a given trigger - * if a "notification" trigger with a corresponding condition was - * added prior. - */ cds_lfht_lookup(state->notification_trigger_clients_ht, lttng_condition_hash(condition), match_client_list_condition, @@ -422,6 +577,19 @@ int notification_thread_client_subscribe(struct notification_client *client, client_list = caa_container_of(node, struct notification_client_list, notification_trigger_ht_node); + 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_unlock; + } + + /* + * 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); cds_list_add(&client_list_element->node, &client_list->list); @@ -520,6 +688,12 @@ end: 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) @@ -542,7 +716,7 @@ void notification_client_destroy(struct notification_client *client, } lttng_dynamic_buffer_reset(&client->communication.inbound.buffer); lttng_dynamic_buffer_reset(&client->communication.outbound.buffer); - free(client); + call_rcu(&client->rcu_node, free_notification_client_rcu); } /* @@ -642,13 +816,6 @@ bool trigger_applies_to_client(struct lttng_trigger *trigger, return applies; } -static -unsigned long hash_channel_key(struct channel_key *key) -{ - return hash_key_u64(&key->key, lttng_ht_seed) ^ hash_key_ulong( - (void *) (unsigned long) key->domain, lttng_ht_seed); -} - static int handle_notification_thread_command_add_channel( struct notification_thread_state *state, @@ -676,6 +843,7 @@ int handle_notification_thread_command_add_channel( channel_key = &new_channel_info->key; + 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) { @@ -688,7 +856,7 @@ int handle_notification_thread_command_add_channel( new_element = zmalloc(sizeof(*new_element)); if (!new_element) { - goto error; + goto error_unlock; } CDS_INIT_LIST_HEAD(&new_element->node); new_element->trigger = trigger_ht_element->trigger; @@ -700,14 +868,13 @@ int handle_notification_thread_command_add_channel( trigger_count); channel_trigger_list = zmalloc(sizeof(*channel_trigger_list)); if (!channel_trigger_list) { - goto error; + goto error_unlock; } channel_trigger_list->channel_key = *channel_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(channel_key), @@ -722,12 +889,28 @@ int handle_notification_thread_command_add_channel( rcu_read_unlock(); *cmd_result = LTTNG_OK; return 0; +error_unlock: + rcu_read_unlock(); error: /* Empty trigger list */ channel_info_destroy(new_channel_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, @@ -770,7 +953,7 @@ int handle_notification_thread_command_remove_channel( free(trigger_list_element); } cds_lfht_del(state->channel_triggers_ht, node); - free(trigger_list); + call_rcu(&trigger_list->rcu_node, free_channel_trigger_list_rcu); /* Free sampled channel state. */ cds_lfht_lookup(state->channel_state_ht, @@ -789,7 +972,7 @@ int handle_notification_thread_command_remove_channel( channel_state_ht_node); cds_lfht_del(state->channel_state_ht, node); - free(sample); + call_rcu(&sample->rcu_node, free_channel_state_sample_rcu); } /* Remove the channel from the channels_ht and free it. */ @@ -974,6 +1157,7 @@ int handle_notification_thread_command_register_trigger( channels_ht_node) { struct lttng_trigger_list_element *trigger_list_element; struct lttng_channel_trigger_list *trigger_list; + struct cds_lfht_iter lookup_iter; if (!trigger_applies_to_channel(trigger, channel)) { continue; @@ -983,10 +1167,9 @@ int handle_notification_thread_command_register_trigger( hash_channel_key(&channel->key), match_channel_trigger_list, &channel->key, - &iter); - node = cds_lfht_iter_get_node(&iter); + &lookup_iter); + node = cds_lfht_iter_get_node(&lookup_iter); assert(node); - /* Free the list of triggers associated with this channel. */ trigger_list = caa_container_of(node, struct lttng_channel_trigger_list, channel_triggers_ht_node); @@ -999,6 +1182,7 @@ int handle_notification_thread_command_register_trigger( CDS_INIT_LIST_HEAD(&trigger_list_element->node); trigger_list_element->trigger = trigger; cds_list_add(&trigger_list_element->node, &trigger_list->list); + /* A trigger can only apply to one channel. */ break; } @@ -1026,6 +1210,20 @@ error: return ret; } +static +void free_notification_client_list_rcu(struct rcu_head *node) +{ + free(caa_container_of(node, struct notification_client_list, + rcu_node)); +} + +static +void free_lttng_trigger_ht_element_rcu(struct rcu_head *node) +{ + free(caa_container_of(node, struct lttng_trigger_ht_element, + rcu_node)); +} + static int handle_notification_thread_command_unregister_trigger( struct notification_thread_state *state, @@ -1077,6 +1275,8 @@ int handle_notification_thread_command_unregister_trigger( DBG("[notification-thread] Removed trigger from channel_triggers_ht"); cds_list_del(&trigger_element->node); + /* A trigger can only appear once per channel */ + break; } } @@ -1098,7 +1298,7 @@ int handle_notification_thread_command_unregister_trigger( free(client_list_element); } cds_lfht_del(state->notification_trigger_clients_ht, node); - free(client_list); + call_rcu(&client_list->rcu_node, free_notification_client_list_rcu); /* Remove trigger from triggers_ht. */ trigger_ht_element = caa_container_of(triggers_ht_node, @@ -1110,7 +1310,7 @@ int handle_notification_thread_command_unregister_trigger( action = lttng_trigger_get_action(trigger_ht_element->trigger); lttng_action_destroy(action); lttng_trigger_destroy(trigger_ht_element->trigger); - free(trigger_ht_element); + call_rcu(&trigger_ht_element->rcu_node, free_lttng_trigger_ht_element_rcu); end: rcu_read_unlock(); if (_cmd_reply) { @@ -1129,7 +1329,7 @@ int handle_notification_thread_command( struct notification_thread_command *cmd; /* Read event_fd to put it back into a quiescent state. */ - ret = read(handle->cmd_queue.event_fd, &counter, sizeof(counter)); + ret = read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter, sizeof(counter)); if (ret == -1) { goto error; }