#include "lttng-sessiond.hpp"
#include "kernel.hpp"
-#define CLIENT_POLL_MASK_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
-#define CLIENT_POLL_MASK_IN_OUT (CLIENT_POLL_MASK_IN | LPOLLOUT)
+#define CLIENT_POLL_EVENTS_IN (LPOLLIN | LPOLLERR | LPOLLHUP | LPOLLRDHUP)
+#define CLIENT_POLL_EVENTS_IN_OUT (CLIENT_POLL_EVENTS_IN | LPOLLOUT)
/* The tracers currently limit the capture size to PIPE_BUF (4kb on linux). */
#define MAX_CAPTURE_SIZE (PIPE_BUF)
LTTNG_OBJECT_TYPE_SESSION,
};
-struct lttng_trigger_list_element {
- /* No ownership of the trigger object is assumed. */
- struct lttng_trigger *trigger;
- struct cds_list_head node;
-};
-
struct lttng_channel_trigger_list {
struct channel_key channel_key;
/* List of struct lttng_trigger_list_element. */
struct rcu_head rcu_node;
};
+namespace {
+struct lttng_trigger_list_element {
+ /* No ownership of the trigger object is assumed. */
+ struct lttng_trigger *trigger;
+ struct cds_list_head node;
+};
+
struct lttng_trigger_ht_element {
struct lttng_trigger *trigger;
struct cds_lfht_node node;
/* call_rcu delayed reclaim. */
struct rcu_head rcu_node;
};
+} /* namespace */
static unsigned long hash_channel_key(struct channel_key *key);
static int evaluate_buffer_condition(const struct lttng_condition *condition,
static
void session_info_put(struct session_info *session_info);
static
-struct session_info *session_info_create(const char *name,
- uid_t uid, gid_t gid,
+struct session_info *session_info_create(uint64_t id,
+ const char *name,
+ uid_t uid,
+ gid_t gid,
struct lttng_session_trigger_list *trigger_list,
struct cds_lfht *sessions_ht);
-static
-void session_info_add_channel(struct session_info *session_info,
- struct channel_info *channel_info);
+static void session_info_add_channel(
+ struct session_info *session_info, struct channel_info *channel_info);
static
void session_info_remove_channel(struct session_info *session_info,
struct channel_info *channel_info);
{
/* This double-cast is intended to supress pointer-to-cast warning. */
const notification_client_id id = *((notification_client_id *) key);
- const struct notification_client *client = caa_container_of(
- node, struct notification_client, client_id_ht_node);
+ const struct notification_client *client = lttng::utils::container_of(
+ node, ¬ification_client::client_id_ht_node);
return client->id == id;
}
}
static
-int match_session(struct cds_lfht_node *node, const void *key)
+int match_session_info(struct cds_lfht_node *node, const void *key)
+{
+ const auto session_id = *((uint64_t *) key);
+ const auto *session_info = lttng::utils::container_of(
+ node, &session_info::sessions_ht_node);
+
+ return session_id == session_info->id;
+}
+
+static
+unsigned long hash_session_info_id(uint64_t id)
+{
+ return hash_key_u64(&id, lttng_ht_seed);
+}
+
+static
+unsigned long hash_session_info(const struct session_info *session_info)
+{
+ return hash_session_info_id(session_info->id);
+}
+
+static
+struct session_info *get_session_info_by_id(
+ const struct notification_thread_state *state, uint64_t id)
{
- const char *name = (const char *) key;
- struct session_info *session_info = caa_container_of(
- node, struct session_info, sessions_ht_node);
+ struct cds_lfht_iter iter;
+ struct cds_lfht_node *node;
+ lttng::urcu::read_lock_guard read_lock_guard;
+
+ cds_lfht_lookup(state->sessions_ht,
+ hash_session_info_id(id),
+ match_session_info,
+ &id,
+ &iter);
+ node = cds_lfht_iter_get_node(&iter);
+
+ if (node) {
+ auto session_info = lttng::utils::container_of(node, &session_info::sessions_ht_node);
- return !strcmp(session_info->name, name);
+ session_info_get(session_info);
+ return session_info;
+ }
+
+ return NULL;
+}
+
+static
+struct session_info *get_session_info_by_name(
+ const struct notification_thread_state *state, const char *name)
+{
+ uint64_t session_id;
+ const auto found = sample_session_id_by_name(name, &session_id);
+
+ return found ? get_session_info_by_id(state, session_id) : NULL;
}
static
return "ADD_CHANNEL";
case NOTIFICATION_COMMAND_TYPE_REMOVE_CHANNEL:
return "REMOVE_CHANNEL";
+ case NOTIFICATION_COMMAND_TYPE_ADD_SESSION:
+ return "ADD_SESSION";
+ case NOTIFICATION_COMMAND_TYPE_REMOVE_SESSION:
+ return "REMOVE_SESSION";
case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
return "SESSION_ROTATION_ONGOING";
case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
static
void free_channel_info_rcu(struct rcu_head *node)
{
- free(caa_container_of(node, struct channel_info, rcu_node));
+ free(lttng::utils::container_of(node, &channel_info::rcu_node));
}
static
static
void free_session_info_rcu(struct rcu_head *node)
{
- free(caa_container_of(node, struct session_info, rcu_node));
+ free(lttng::utils::container_of(node, &session_info::rcu_node));
}
/* Don't call directly, use the ref-counting mechanism. */
}
static
-struct session_info *session_info_create(const char *name, uid_t uid, gid_t gid,
+struct session_info *session_info_create(uint64_t id,
+ const char *name,
+ uid_t uid,
+ gid_t gid,
struct lttng_session_trigger_list *trigger_list,
struct cds_lfht *sessions_ht)
{
if (!session_info) {
goto end;
}
+
lttng_ref_init(&session_info->ref, session_info_destroy);
session_info->channel_infos_ht = cds_lfht_new(DEFAULT_HT_SIZE,
}
cds_lfht_node_init(&session_info->sessions_ht_node);
+ session_info->id = id;
session_info->name = strdup(name);
if (!session_info->name) {
goto error;
}
+
session_info->uid = uid;
session_info->gid = gid;
session_info->trigger_list = trigger_list;
void notification_client_list_release(struct urcu_ref *list_ref)
{
struct notification_client_list *list =
- container_of(list_ref, typeof(*list), ref);
+ lttng::utils::container_of(list_ref, ¬ification_client_list::ref);
struct notification_client_list_element *client_list_element, *tmp;
lttng_condition_put(list->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 = lttng::utils::container_of(node,
+ ¬ification_client_list::notification_trigger_clients_ht_node);
list = notification_client_list_get(list) ? list : NULL;
}
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("No known session matching name \"%s\"",
+ /* Find the session associated with the condition. */
+ session_info = get_session_info_by_name(state, session_name);
+ if (!session_info) {
+ DBG("Unknown session while evaluating session condition for client: 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.
static
void free_notification_client_rcu(struct rcu_head *node)
{
- free(caa_container_of(node, struct notification_client, rcu_node));
+ free(lttng::utils::container_of(node, ¬ification_client::rcu_node));
}
static
}
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 *create_and_publish_session_info(struct notification_thread_state *state,
+ uint64_t id,
+ 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("Found session info of session \"%s\" (uid = %i, gid = %i)",
- name, uid, gid);
- session = caa_container_of(node, struct session_info,
- sessions_ht_node);
- LTTNG_ASSERT(session->uid == uid);
- LTTNG_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,
+ session = session_info_create(id, name, uid, gid, trigger_list,
state->sessions_ht);
if (!session) {
ERR("Failed to allocation session info for session \"%s\" (uid = %i, gid = %i)",
lttng_session_trigger_list_destroy(trigger_list);
goto error;
}
+
+ /* Transferred ownership to the new session. */
trigger_list = NULL;
- cds_lfht_add(state->sessions_ht, hash_key_str(name, lttng_ht_seed),
- &session->sessions_ht_node);
-end:
+ if (cds_lfht_add_unique(state->sessions_ht, hash_session_info(session), match_session_info,
+ &id, &session->sessions_ht_node) != &session->sessions_ht_node) {
+ ERR("Duplicate session found: name = `%s`, id = %" PRIu64, name, id);
+ goto error;
+ }
+
rcu_read_unlock();
return session;
error:
}
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,
+int handle_notification_thread_command_add_channel(struct notification_thread_state *state,
+ uint64_t session_id,
+ 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 cds_lfht_iter iter;
struct session_info *session_info = NULL;
- DBG("Adding channel %s from session %s, channel key = %" PRIu64 " in %s domain",
- channel_name, session_name, channel_key_int,
+ DBG("Adding channel: channel name = `%s`, session id = %" PRIu64 ", channel key = %" PRIu64 ", domain = %s",
+ channel_name, session_id, 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);
+ session_info = get_session_info_by_id(state, session_id);
if (!session_info) {
- /* Allocation error or an internal error occurred. */
+ /* Fatal logic error. */
+ ERR("Failed to find session while adding channel: session id = %" PRIu64,
+ session_id);
goto error;
}
return 1;
}
+static
+int handle_notification_thread_command_add_session(struct notification_thread_state *state,
+ uint64_t session_id,
+ const char *session_name,
+ uid_t session_uid,
+ gid_t session_gid,
+ enum lttng_error_code *cmd_result)
+{
+ int ret;
+
+ DBG("Adding session: session name = `%s`, session id = %" PRIu64 ", session uid = %d, session gid = %d",
+ session_name, session_id, session_uid, session_gid);
+
+ auto session = create_and_publish_session_info(state, session_id, session_name, session_uid, session_gid);
+ if (!session) {
+ PERROR("Failed to add session: session name = `%s`, session id = %" PRIu64 ", session uid = %d, session gid = %d",
+ session_name, session_id, session_uid, session_gid);
+ ret = -1;
+ *cmd_result = LTTNG_ERR_NOMEM;
+ goto end;
+ }
+
+ /*
+ * Note that the reference to `session` is not released; this reference is
+ * the "global" reference that is used to allow look-ups. This reference will
+ * only be released when the session is removed. See
+ * handle_notification_thread_command_remove_session.
+ */
+ ret = 0;
+ *cmd_result = LTTNG_OK;
+end:
+ return ret;
+}
+
+static
+int handle_notification_thread_command_remove_session(
+ struct notification_thread_state *state,
+ uint64_t session_id,
+ enum lttng_error_code *cmd_result)
+{
+ int ret;
+
+ DBG("Removing session: session id = %" PRIu64, session_id);
+
+ auto session = get_session_info_by_id(state, session_id);
+ if (!session) {
+ ERR("Failed to remove session: session id = %" PRIu64, session_id);
+ ret = -1;
+ *cmd_result = LTTNG_ERR_NO_SESSION;
+ goto end;
+ }
+
+ /* Release the reference returned by the look-up, and then release the global reference. */
+ session_info_put(session);
+ session_info_put(session);
+ ret = 0;
+ *cmd_result = LTTNG_OK;
+end:
+ return ret;
+}
+
static
void free_channel_trigger_list_rcu(struct rcu_head *node)
{
int handle_notification_thread_command_session_rotation(
struct notification_thread_state *state,
enum notification_thread_command_type cmd_type,
- const char *session_name, uid_t session_uid, gid_t session_gid,
+ uint64_t session_id,
uint64_t trace_archive_chunk_id,
struct lttng_trace_archive_location *location,
enum lttng_error_code *_cmd_result)
struct lttng_session_trigger_list *trigger_list;
struct lttng_trigger_list_element *trigger_list_element;
struct session_info *session_info;
- const struct lttng_credentials session_creds = {
- .uid = LTTNG_OPTIONAL_INIT_VALUE(session_uid),
- .gid = LTTNG_OPTIONAL_INIT_VALUE(session_gid),
- };
+ struct lttng_credentials session_creds;
rcu_read_lock();
- session_info = find_or_create_session_info(state, session_name,
- session_uid, session_gid);
+ session_info = get_session_info_by_id(state, session_id);
if (!session_info) {
- /* Allocation error or an internal error occurred. */
+ /* Fatal logic error. */
+ ERR("Failed to find session while handling rotation state change: session id = %" PRIu64,
+ session_id);
ret = -1;
- cmd_result = LTTNG_ERR_NOMEM;
+ cmd_result = LTTNG_ERR_FATAL;
goto end;
}
+ session_creds = {
+ .uid = LTTNG_OPTIONAL_INIT_VALUE(session_info->uid),
+ .gid = LTTNG_OPTIONAL_INIT_VALUE(session_info->gid),
+ };
+
session_info->rotation.ongoing =
cmd_type == NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING;
session_info->rotation.id = trace_archive_chunk_id;
- trigger_list = get_session_trigger_list(state, session_name);
+ trigger_list = get_session_trigger_list(state, session_info->name);
if (!trigger_list) {
- DBG("No triggers applying to session \"%s\" found",
- session_name);
+ DBG("No triggers apply to session: session name = `%s` ",
+ session_info->name);
goto end;
}
return 0;
}
+static
+int pop_cmd_queue(struct notification_thread_handle *handle,
+ struct notification_thread_command **cmd)
+{
+ int ret;
+ uint64_t counter;
+
+ pthread_mutex_lock(&handle->cmd_queue.lock);
+ ret = lttng_read(handle->cmd_queue.event_fd, &counter, sizeof(counter));
+ if (ret != sizeof(counter)) {
+ ret = -1;
+ goto error_unlock;
+ }
+
+ *cmd = cds_list_first_entry(&handle->cmd_queue.list,
+ struct notification_thread_command, cmd_list_node);
+ cds_list_del(&((*cmd)->cmd_list_node));
+ ret = 0;
+
+error_unlock:
+ pthread_mutex_unlock(&handle->cmd_queue.lock);
+ return ret;
+}
+
/* Returns 0 on success, 1 on exit requested, negative value on error. */
int handle_notification_thread_command(
struct notification_thread_handle *handle,
struct notification_thread_state *state)
{
int ret;
- uint64_t counter;
struct notification_thread_command *cmd;
- /* Read the event pipe to put it back into a quiescent state. */
- ret = lttng_read(lttng_pipe_get_readfd(handle->cmd_queue.event_pipe), &counter,
- sizeof(counter));
- if (ret != sizeof(counter)) {
+ ret = pop_cmd_queue(handle, &cmd);
+ if (ret) {
goto error;
}
- pthread_mutex_lock(&handle->cmd_queue.lock);
- cmd = cds_list_first_entry(&handle->cmd_queue.list,
- struct notification_thread_command, cmd_list_node);
- cds_list_del(&cmd->cmd_list_node);
- pthread_mutex_unlock(&handle->cmd_queue.lock);
-
DBG("Received `%s` command",
notification_command_type_str(cmd->type));
switch (cmd->type) {
case NOTIFICATION_COMMAND_TYPE_ADD_CHANNEL:
ret = handle_notification_thread_command_add_channel(
state,
- cmd->parameters.add_channel.session.name,
- cmd->parameters.add_channel.session.uid,
- cmd->parameters.add_channel.session.gid,
+ cmd->parameters.add_channel.session.id,
cmd->parameters.add_channel.channel.name,
cmd->parameters.add_channel.channel.domain,
cmd->parameters.add_channel.channel.key,
cmd->parameters.remove_channel.domain,
&cmd->reply_code);
break;
+ case NOTIFICATION_COMMAND_TYPE_ADD_SESSION:
+ ret = handle_notification_thread_command_add_session(state,
+ cmd->parameters.add_session.session_id,
+ cmd->parameters.add_session.session_name,
+ cmd->parameters.add_session.session_uid,
+ cmd->parameters.add_session.session_gid, &cmd->reply_code);
+ break;
+ case NOTIFICATION_COMMAND_TYPE_REMOVE_SESSION:
+ ret = handle_notification_thread_command_remove_session(
+ state, cmd->parameters.remove_session.session_id, &cmd->reply_code);
+ break;
case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_ONGOING:
case NOTIFICATION_COMMAND_TYPE_SESSION_ROTATION_COMPLETED:
ret = handle_notification_thread_command_session_rotation(
state,
cmd->type,
- cmd->parameters.session_rotation.session_name,
- cmd->parameters.session_rotation.uid,
- cmd->parameters.session_rotation.gid,
+ cmd->parameters.session_rotation.session_id,
cmd->parameters.session_rotation.trace_archive_chunk_id,
cmd->parameters.session_rotation.location,
&cmd->reply_code);
goto error;
}
+ client->communication.current_poll_events = CLIENT_POLL_EVENTS_IN;
ret = lttng_poll_add(&state->events, client->socket,
- LPOLLIN | LPOLLERR |
- LPOLLHUP | LPOLLRDHUP);
+ client->communication.current_poll_events);
if (ret < 0) {
ERR("Failed to add notification channel client socket to poll set");
ret = 0;
return error_occurred ? -1 : 0;
}
+static
+bool client_has_outbound_data_left(
+ const struct notification_client *client)
+{
+ const struct lttng_payload_view pv = lttng_payload_view_from_payload(
+ &client->communication.outbound.payload, 0, -1);
+ const bool has_data = pv.buffer.size != 0;
+ const bool has_fds = lttng_payload_view_get_fd_handle_count(&pv);
+
+ return has_data || has_fds;
+}
+
static
int client_handle_transmission_status(
struct notification_client *client,
switch (transmission_status) {
case CLIENT_TRANSMISSION_STATUS_COMPLETE:
- ret = lttng_poll_mod(&state->events, client->socket,
- CLIENT_POLL_MASK_IN);
- if (ret) {
- goto end;
- }
-
- break;
case CLIENT_TRANSMISSION_STATUS_QUEUED:
+ {
+ int current_poll_events;
+ int new_poll_events;
/*
* We want to be notified whenever there is buffer space
- * available to send the rest of the payload.
+ * available to send the rest of the payload if we are
+ * waiting to send data to the client.
+ *
+ * The state of the outbound queue being sampled here is
+ * fine since:
+ * - it is okay to wake-up "for nothing" in case we see
+ * that data is left, but another thread succeeds in
+ * flushing it before us when handling the client "out"
+ * event. We will simply stop monitoring that event the next
+ * time it wakes us up and we see no data left to be sent,
+ * - if another thread fails to flush the entire client
+ * outgoing queue, it will issue a "communication update"
+ * command and cause the client's (e)poll mask to be
+ * re-evaluated.
+ *
+ * The situation we seek to avoid would be to disable the
+ * monitoring of "out" client events indefinitely when there is
+ * data to be sent, which can't happen because of the
+ * aforementioned "communication update" mechanism.
*/
- ret = lttng_poll_mod(&state->events, client->socket,
- CLIENT_POLL_MASK_IN_OUT);
- if (ret) {
- goto end;
+ pthread_mutex_lock(&client->lock);
+ current_poll_events = client->communication.current_poll_events;
+ new_poll_events = client_has_outbound_data_left(client) ?
+ CLIENT_POLL_EVENTS_IN_OUT :
+ CLIENT_POLL_EVENTS_IN;
+ client->communication.current_poll_events = new_poll_events;
+ pthread_mutex_unlock(&client->lock);
+
+ /* Update the monitored event set only if it changed. */
+ if (current_poll_events != new_poll_events) {
+ ret = lttng_poll_mod(&state->events, client->socket,
+ new_poll_events);
+ if (ret) {
+ goto end;
+ }
}
+
break;
+ }
case CLIENT_TRANSMISSION_STATUS_FAIL:
ret = notification_thread_client_disconnect(client, state);
if (ret) {
return CLIENT_TRANSMISSION_STATUS_ERROR;
}
-static
-bool client_has_outbound_data_left(
- const struct notification_client *client)
-{
- const struct lttng_payload_view pv = lttng_payload_view_from_payload(
- &client->communication.outbound.payload, 0, -1);
- const bool has_data = pv.buffer.size != 0;
- const bool has_fds = lttng_payload_view_get_fd_handle_count(&pv);
-
- return has_data || has_fds;
-}
-
/* Client lock must _not_ be held by the caller. */
static
int client_send_command_reply(struct notification_client *client,
struct lttng_notification_channel_command_reply reply = {
.status = (int8_t) status,
};
- struct lttng_notification_channel_message msg = {
- .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY,
- .size = sizeof(reply),
- .fds = 0,
- };
+ struct lttng_notification_channel_message msg;
char buffer[sizeof(msg) + sizeof(reply)];
enum client_transmission_status transmission_status;
+ msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_COMMAND_REPLY;
+ msg.size = sizeof(reply);
+ msg.fds = 0;
+
memcpy(buffer, &msg, sizeof(msg));
memcpy(buffer + sizeof(msg), &reply, sizeof(reply));
DBG("Send command reply (%i)", (int) status);
.major = LTTNG_NOTIFICATION_CHANNEL_VERSION_MAJOR,
.minor = LTTNG_NOTIFICATION_CHANNEL_VERSION_MINOR,
};
- const struct lttng_notification_channel_message msg_header = {
- .type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE,
- .size = sizeof(handshake_reply),
- .fds = 0,
- };
+ struct lttng_notification_channel_message msg_header;
enum lttng_notification_channel_status status =
LTTNG_NOTIFICATION_CHANNEL_STATUS_OK;
char send_buffer[sizeof(msg_header) + sizeof(handshake_reply)];
+ msg_header.type = LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_HANDSHAKE;
+ msg_header.size = sizeof(handshake_reply);
+ msg_header.fds = 0;
+
memcpy(send_buffer, &msg_header, sizeof(msg_header));
memcpy(send_buffer + sizeof(msg_header), &handshake_reply,
sizeof(handshake_reply));
}
pthread_mutex_lock(&client->lock);
- transmission_status = client_flush_outgoing_queue(client);
+ if (!client_has_outbound_data_left(client)) {
+ /*
+ * A client "out" event can be received when no payload is left
+ * to send under some circumstances.
+ *
+ * Many threads can flush a client's outgoing queue and, if they
+ * had to queue their message (socket was full), will use the
+ * "communication update" command to signal the (e)poll thread
+ * to monitor for space being made available in the socket.
+ *
+ * Commands are sent over an internal pipe serviced by the same
+ * thread as the client sockets.
+ *
+ * When space is made available in the socket, there is a race
+ * between the (e)poll thread and the other threads that may
+ * wish to use the client's socket to flush its outgoing queue.
+ *
+ * A non-(e)poll thread may attempt (and succeed) in flushing
+ * the queue before the (e)poll thread gets a chance to service
+ * the client's "out" event.
+ *
+ * In this situation, the (e)poll thread processing the client
+ * out event will see an empty payload: there is nothing to do
+ * except unsubscribing (e)poll "out" events.
+ *
+ * Note that this thread is the (e)poll thread so it can modify
+ * the (e)poll mask directly without using a communication
+ * update command. Other threads that flush the outgoing queue
+ * will use the "communication update" command to wake up this
+ * thread and force it to monitor "out" events.
+ *
+ * When other threads succeed in emptying the outgoing queue,
+ * they don't need to update the (e)poll mask: if the "out"
+ * event is monitored, it will fire once and the (e)poll
+ * thread will reach this condition, causing the event to
+ * stop being monitored.
+ */
+ transmission_status = CLIENT_TRANSMISSION_STATUS_COMPLETE;
+ } else {
+ transmission_status = client_flush_outgoing_queue(client);
+ }
pthread_mutex_unlock(&client->lock);
ret = client_handle_transmission_status(
bool result = false;
uint64_t threshold;
enum lttng_condition_type condition_type;
- const struct lttng_condition_buffer_usage *use_condition = container_of(
- condition, struct lttng_condition_buffer_usage,
- parent);
+ const struct lttng_condition_buffer_usage *use_condition = lttng::utils::container_of(
+ condition, <tng_condition_buffer_usage::parent);
if (use_condition->threshold_bytes.set) {
threshold = use_condition->threshold_bytes.value;
{
uint64_t threshold;
const struct lttng_condition_session_consumed_size *size_condition =
- container_of(condition,
- struct lttng_condition_session_consumed_size,
- parent);
+ lttng::utils::container_of(condition,
+ <tng_condition_session_consumed_size::parent);
threshold = size_condition->consumed_threshold_bytes.value;
DBG("Session consumed size condition being evaluated: threshold = %" PRIu64 ", current size = %" PRIu64,
int client_notification_overflow(struct notification_client *client)
{
int ret = 0;
- const struct lttng_notification_channel_message msg = {
- .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED,
- .size = 0,
- .fds = 0,
- };
+ struct lttng_notification_channel_message msg;
+
+ msg.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION_DROPPED;
+ msg.size = 0;
+ msg.fds = 0;
ASSERT_LOCKED(client->lock);
.trigger = (struct lttng_trigger *) trigger,
.evaluation = (struct lttng_evaluation *) evaluation,
};
- struct lttng_notification_channel_message msg_header = {
- .type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION,
- .size = 0,
- .fds = 0,
- };
+ struct lttng_notification_channel_message msg_header;
const struct lttng_credentials *trigger_creds =
lttng_trigger_get_credentials(trigger);
lttng_payload_init(&msg_payload);
+ msg_header.type = (int8_t) LTTNG_NOTIFICATION_CHANNEL_MESSAGE_TYPE_NOTIFICATION;
+ msg_header.size = 0;
+ msg_header.fds = 0;
+
ret = lttng_dynamic_buffer_append(&msg_payload.buffer, &msg_header,
sizeof(msg_header));
if (ret) {
}
evaluation = lttng_evaluation_event_rule_matches_create(
- container_of(lttng_trigger_get_const_condition(
+ lttng::utils::container_of(lttng_trigger_get_const_condition(
element->trigger),
- struct lttng_condition_event_rule_matches,
- parent),
+ <tng_condition_event_rule_matches::parent),
notification->capture_buffer,
notification->capture_buf_size, false);
projects / lttng-tools.git / blobdiff