X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=0005b1039cb27d6ced5dfac3537e72bbf737fce0;hp=116441171e5aa73d198c2bbd90175da0d9fb03fa;hb=10a5031171c7bca5b4498c871b119e5a88b6a3fb;hpb=f2a444f17e07f805109c01ab4c7f53cc98b1adf3 diff --git a/src/common/consumer.c b/src/common/consumer.c index 116441171..0005b1039 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -41,6 +41,7 @@ #include #include "consumer.h" +#include "consumer-stream.h" struct lttng_consumer_global_data consumer_data = { .stream_count = 0, @@ -164,20 +165,20 @@ static struct lttng_consumer_stream *find_stream(uint64_t key, return stream; } -static void steal_stream_key(int key, struct lttng_ht *ht) +static void steal_stream_key(uint64_t key, struct lttng_ht *ht) { struct lttng_consumer_stream *stream; rcu_read_lock(); stream = find_stream(key, ht); if (stream) { - stream->key = -1ULL; + stream->key = (uint64_t) -1ULL; /* * We don't want the lookup to match, but we still need * to iterate on this stream when iterating over the hash table. Just * change the node key. */ - stream->node.key = -1ULL; + stream->node.key = (uint64_t) -1ULL; } rcu_read_unlock(); } @@ -254,10 +255,8 @@ static void free_relayd_rcu(struct rcu_head *head) /* * Destroy and free relayd socket pair object. - * - * This function MUST be called with the consumer_data lock acquired. */ -static void destroy_relayd(struct consumer_relayd_sock_pair *relayd) +void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd) { int ret; struct lttng_ht_iter iter; @@ -302,8 +301,11 @@ void consumer_del_channel(struct lttng_consumer_channel *channel) cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head, send_node) { cds_list_del(&stream->send_node); - lttng_ustconsumer_del_stream(stream); - free(stream); + /* + * Once a stream is added to this list, the buffers were created so + * we have a guarantee that this call will succeed. + */ + consumer_stream_destroy(stream, NULL); } lttng_ustconsumer_del_channel(channel); break; @@ -337,7 +339,7 @@ static void cleanup_relayd_ht(void) cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd, node.node) { - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); } rcu_read_unlock(); @@ -352,13 +354,13 @@ static void cleanup_relayd_ht(void) * It's atomically set without having the stream mutex locked which is fine * because we handle the write/read race with a pipe wakeup for each thread. */ -static void update_endpoint_status_by_netidx(int net_seq_idx, +static void update_endpoint_status_by_netidx(uint64_t net_seq_idx, enum consumer_endpoint_status status) { struct lttng_ht_iter iter; struct lttng_consumer_stream *stream; - DBG("Consumer set delete flag on stream by idx %d", net_seq_idx); + DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx); rcu_read_lock(); @@ -391,7 +393,7 @@ static void update_endpoint_status_by_netidx(int net_seq_idx, static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd, struct lttng_consumer_local_data *ctx) { - int netidx; + uint64_t netidx; assert(relayd); @@ -404,7 +406,7 @@ static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd, * Delete the relayd from the relayd hash table, close the sockets and free * the object in a RCU call. */ - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); /* Set inactive endpoint to all streams */ update_endpoint_status_by_netidx(netidx, CONSUMER_ENDPOINT_INACTIVE); @@ -436,130 +438,20 @@ void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd) /* Destroy the relayd if refcount is 0 */ if (uatomic_read(&relayd->refcount) == 0) { - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); } } /* - * Remove a stream from the global list protected by a mutex. This - * function is also responsible for freeing its data structures. + * Completly destroy stream from every visiable data structure and the given + * hash table if one. + * + * One this call returns, the stream object is not longer usable nor visible. */ void consumer_del_stream(struct lttng_consumer_stream *stream, struct lttng_ht *ht) { - int ret; - struct lttng_ht_iter iter; - struct lttng_consumer_channel *free_chan = NULL; - struct consumer_relayd_sock_pair *relayd; - - assert(stream); - - DBG("Consumer del stream %d", stream->wait_fd); - - if (ht == NULL) { - /* Means the stream was allocated but not successfully added */ - goto free_stream_rcu; - } - - pthread_mutex_lock(&consumer_data.lock); - pthread_mutex_lock(&stream->lock); - - switch (consumer_data.type) { - case LTTNG_CONSUMER_KERNEL: - if (stream->mmap_base != NULL) { - ret = munmap(stream->mmap_base, stream->mmap_len); - if (ret != 0) { - PERROR("munmap"); - } - } - - if (stream->wait_fd >= 0) { - ret = close(stream->wait_fd); - if (ret) { - PERROR("close"); - } - } - break; - case LTTNG_CONSUMER32_UST: - case LTTNG_CONSUMER64_UST: - lttng_ustconsumer_del_stream(stream); - break; - default: - ERR("Unknown consumer_data type"); - assert(0); - goto end; - } - - rcu_read_lock(); - iter.iter.node = &stream->node.node; - ret = lttng_ht_del(ht, &iter); - assert(!ret); - - iter.iter.node = &stream->node_channel_id.node; - ret = lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter); - assert(!ret); - - iter.iter.node = &stream->node_session_id.node; - ret = lttng_ht_del(consumer_data.stream_list_ht, &iter); - assert(!ret); - rcu_read_unlock(); - - assert(consumer_data.stream_count > 0); - consumer_data.stream_count--; - - if (stream->out_fd >= 0) { - ret = close(stream->out_fd); - if (ret) { - PERROR("close"); - } - } - - /* Check and cleanup relayd */ - rcu_read_lock(); - relayd = consumer_find_relayd(stream->net_seq_idx); - if (relayd != NULL) { - uatomic_dec(&relayd->refcount); - assert(uatomic_read(&relayd->refcount) >= 0); - - /* Closing streams requires to lock the control socket. */ - pthread_mutex_lock(&relayd->ctrl_sock_mutex); - ret = relayd_send_close_stream(&relayd->control_sock, - stream->relayd_stream_id, - stream->next_net_seq_num - 1); - pthread_mutex_unlock(&relayd->ctrl_sock_mutex); - if (ret < 0) { - DBG("Unable to close stream on the relayd. Continuing"); - /* - * Continue here. There is nothing we can do for the relayd. - * Chances are that the relayd has closed the socket so we just - * continue cleaning up. - */ - } - - /* Both conditions are met, we destroy the relayd. */ - if (uatomic_read(&relayd->refcount) == 0 && - uatomic_read(&relayd->destroy_flag)) { - destroy_relayd(relayd); - } - } - rcu_read_unlock(); - - if (!uatomic_sub_return(&stream->chan->refcount, 1) - && !uatomic_read(&stream->chan->nb_init_stream_left)) { - free_chan = stream->chan; - } - -end: - consumer_data.need_update = 1; - pthread_mutex_unlock(&stream->lock); - pthread_mutex_unlock(&consumer_data.lock); - - if (free_chan) { - consumer_del_channel(free_chan); - } - -free_stream_rcu: - call_rcu(&stream->node.head, free_stream_rcu); + consumer_stream_destroy(stream, ht); } struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, @@ -568,7 +460,7 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, const char *channel_name, uid_t uid, gid_t gid, - int relayd_id, + uint64_t relayd_id, uint64_t session_id, int cpu, int *alloc_ret, @@ -620,8 +512,10 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, /* Init session id node with the stream session id */ lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id); - DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64 " relayd_id %" PRIu64 ", session_id %" PRIu64, - stream->name, stream->key, channel_key, stream->net_seq_idx, stream->session_id); + DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64 + " relayd_id %" PRIu64 ", session_id %" PRIu64, + stream->name, stream->key, channel_key, + stream->net_seq_idx, stream->session_id); rcu_read_unlock(); return stream; @@ -727,12 +621,12 @@ end: * Allocate and return a consumer relayd socket. */ struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair( - int net_seq_idx) + uint64_t net_seq_idx) { struct consumer_relayd_sock_pair *obj = NULL; - /* Negative net sequence index is a failure */ - if (net_seq_idx < 0) { + /* net sequence index of -1 is a failure */ + if (net_seq_idx == (uint64_t) -1ULL) { goto error; } @@ -783,6 +677,66 @@ error: return relayd; } +/* + * Find a relayd and send the stream + * + * Returns 0 on success, < 0 on error + */ +int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, + char *path) +{ + int ret = 0; + struct consumer_relayd_sock_pair *relayd; + + assert(stream); + assert(stream->net_seq_idx != -1ULL); + assert(path); + + /* The stream is not metadata. Get relayd reference if exists. */ + rcu_read_lock(); + relayd = consumer_find_relayd(stream->net_seq_idx); + if (relayd != NULL) { + /* Add stream on the relayd */ + pthread_mutex_lock(&relayd->ctrl_sock_mutex); + ret = relayd_add_stream(&relayd->control_sock, stream->name, + path, &stream->relayd_stream_id, + stream->chan->tracefile_size, stream->chan->tracefile_count); + pthread_mutex_unlock(&relayd->ctrl_sock_mutex); + if (ret < 0) { + goto end; + } + uatomic_inc(&relayd->refcount); + } else { + ERR("Stream %" PRIu64 " relayd ID %" PRIu64 " unknown. Can't send it.", + stream->key, stream->net_seq_idx); + ret = -1; + goto end; + } + + DBG("Stream %s with key %" PRIu64 " sent to relayd id %" PRIu64, + stream->name, stream->key, stream->net_seq_idx); + +end: + rcu_read_unlock(); + return ret; +} + +/* + * Find a relayd and close the stream + */ +void close_relayd_stream(struct lttng_consumer_stream *stream) +{ + struct consumer_relayd_sock_pair *relayd; + + /* The stream is not metadata. Get relayd reference if exists. */ + rcu_read_lock(); + relayd = consumer_find_relayd(stream->net_seq_idx); + if (relayd) { + consumer_stream_relayd_close(stream, relayd); + } + rcu_read_unlock(); +} + /* * Handle stream for relayd transmission if the stream applies for network * streaming where the net sequence index is set. @@ -855,10 +809,12 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, const char *name, uid_t uid, gid_t gid, - int relayd_id, + uint64_t relayd_id, enum lttng_event_output output, uint64_t tracefile_size, - uint64_t tracefile_count) + uint64_t tracefile_count, + uint64_t session_id_per_pid, + unsigned int monitor) { struct lttng_consumer_channel *channel; @@ -871,12 +827,27 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, channel->key = key; channel->refcount = 0; channel->session_id = session_id; + channel->session_id_per_pid = session_id_per_pid; channel->uid = uid; channel->gid = gid; channel->relayd_id = relayd_id; channel->output = output; channel->tracefile_size = tracefile_size; channel->tracefile_count = tracefile_count; + channel->monitor = monitor; + + /* + * In monitor mode, the streams associated with the channel will be put in + * a special list ONLY owned by this channel. So, the refcount is set to 1 + * here meaning that the channel itself has streams that are referenced. + * + * On a channel deletion, once the channel is no longer visible, the + * refcount is decremented and checked for a zero value to delete it. With + * streams in no monitor mode, it will now be safe to destroy the channel. + */ + if (!channel->monitor) { + channel->refcount = 1; + } strncpy(channel->pathname, pathname, sizeof(channel->pathname)); channel->pathname[sizeof(channel->pathname) - 1] = '\0'; @@ -928,7 +899,7 @@ end: pthread_mutex_unlock(&consumer_data.lock); if (!ret && channel->wait_fd != -1 && - channel->metadata_stream == NULL) { + channel->type == CONSUMER_CHANNEL_TYPE_DATA) { notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD); } return ret; @@ -1331,7 +1302,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( rcu_read_lock(); /* Flag that the current stream if set for network streaming. */ - if (stream->net_seq_idx != -1) { + if (stream->net_seq_idx != (uint64_t) -1ULL) { relayd = consumer_find_relayd(stream->net_seq_idx); if (relayd == NULL) { goto end; @@ -1530,7 +1501,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice( rcu_read_lock(); /* Flag that the current stream if set for network streaming. */ - if (stream->net_seq_idx != -1) { + if (stream->net_seq_idx != (uint64_t) -1ULL) { relayd = consumer_find_relayd(stream->net_seq_idx); if (relayd == NULL) { goto end; @@ -1894,7 +1865,6 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, PERROR("munmap metadata stream"); } } - if (stream->wait_fd >= 0) { ret = close(stream->wait_fd); if (ret < 0) { @@ -1957,7 +1927,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, /* Both conditions are met, we destroy the relayd. */ if (uatomic_read(&relayd->refcount) == 0 && uatomic_read(&relayd->destroy_flag)) { - destroy_relayd(relayd); + consumer_destroy_relayd(relayd); } } rcu_read_unlock(); @@ -2817,7 +2787,6 @@ restart: lttng_poll_del(&events, chan->wait_fd); ret = lttng_ht_del(channel_ht, &iter); assert(ret == 0); - assert(cds_list_empty(&chan->streams.head)); consumer_close_channel_streams(chan); /* Release our own refcount */ @@ -3069,7 +3038,7 @@ void lttng_consumer_init(void) * This will create a relayd socket pair and add it to the relayd hash table. * The caller MUST acquire a RCU read side lock before calling it. */ -int consumer_add_relayd_socket(int net_seq_idx, int sock_type, +int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll, struct lttcomm_relayd_sock *relayd_sock, unsigned int sessiond_id) @@ -3081,48 +3050,67 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, assert(ctx); assert(relayd_sock); - DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx); - - /* First send a status message before receiving the fds. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { - /* Somehow, the session daemon is not responding anymore. */ - goto error; - } + DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx); /* Get relayd reference if exists. */ relayd = consumer_find_relayd(net_seq_idx); if (relayd == NULL) { + assert(sock_type == LTTNG_STREAM_CONTROL); /* Not found. Allocate one. */ relayd = consumer_allocate_relayd_sock_pair(net_seq_idx); if (relayd == NULL) { - lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR); - ret = -1; + ret = -ENOMEM; + ret_code = LTTCOMM_CONSUMERD_ENOMEM; goto error; + } else { + relayd->sessiond_session_id = (uint64_t) sessiond_id; + relayd_created = 1; } - relayd->sessiond_session_id = (uint64_t) sessiond_id; - relayd_created = 1; + + /* + * This code path MUST continue to the consumer send status message to + * we can notify the session daemon and continue our work without + * killing everything. + */ + } else { + /* + * relayd key should never be found for control socket. + */ + assert(sock_type != LTTNG_STREAM_CONTROL); + } + + /* First send a status message before receiving the fds. */ + ret = consumer_send_status_msg(sock, LTTNG_OK); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL); + goto error_nosignal; } /* Poll on consumer socket. */ if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) { + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR); ret = -EINTR; - goto error; + goto error_nosignal; } /* Get relayd socket from session daemon */ ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1); if (ret != sizeof(fd)) { - lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD); ret = -1; fd = -1; /* Just in case it gets set with an invalid value. */ - goto error_close; - } - /* We have the fds without error. Send status back. */ - ret = consumer_send_status_msg(sock, ret_code); - if (ret < 0) { - /* Somehow, the session daemon is not responding anymore. */ + /* + * Failing to receive FDs might indicate a major problem such as + * reaching a fd limit during the receive where the kernel returns a + * MSG_CTRUNC and fails to cleanup the fd in the queue. Any case, we + * don't take any chances and stop everything. + * + * XXX: Feature request #558 will fix that and avoid this possible + * issue when reaching the fd limit. + */ + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD); + ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD; goto error; } @@ -3132,19 +3120,23 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, /* Copy received lttcomm socket */ lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock); ret = lttcomm_create_sock(&relayd->control_sock.sock); - /* Immediately try to close the created socket if valid. */ - if (relayd->control_sock.sock.fd >= 0) { - if (close(relayd->control_sock.sock.fd)) { - PERROR("close relayd control socket"); - } - } /* Handle create_sock error. */ if (ret < 0) { + ret_code = LTTCOMM_CONSUMERD_ENOMEM; goto error; } + /* + * Close the socket created internally by + * lttcomm_create_sock, so we can replace it by the one + * received from sessiond. + */ + if (close(relayd->control_sock.sock.fd)) { + PERROR("close"); + } /* Assign new file descriptor */ relayd->control_sock.sock.fd = fd; + fd = -1; /* For error path */ /* Assign version values. */ relayd->control_sock.major = relayd_sock->major; relayd->control_sock.minor = relayd_sock->minor; @@ -3169,6 +3161,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, */ (void) relayd_close(&relayd->control_sock); (void) relayd_close(&relayd->data_sock); + ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL; goto error; } @@ -3177,19 +3170,23 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, /* Copy received lttcomm socket */ lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock); ret = lttcomm_create_sock(&relayd->data_sock.sock); - /* Immediately try to close the created socket if valid. */ - if (relayd->data_sock.sock.fd >= 0) { - if (close(relayd->data_sock.sock.fd)) { - PERROR("close relayd data socket"); - } - } /* Handle create_sock error. */ if (ret < 0) { + ret_code = LTTCOMM_CONSUMERD_ENOMEM; goto error; } + /* + * Close the socket created internally by + * lttcomm_create_sock, so we can replace it by the one + * received from sessiond. + */ + if (close(relayd->data_sock.sock.fd)) { + PERROR("close"); + } /* Assign new file descriptor */ relayd->data_sock.sock.fd = fd; + fd = -1; /* for eventual error paths */ /* Assign version values. */ relayd->data_sock.major = relayd_sock->major; relayd->data_sock.minor = relayd_sock->minor; @@ -3197,6 +3194,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, default: ERR("Unknown relayd socket type (%d)", sock_type); ret = -1; + ret_code = LTTCOMM_CONSUMERD_FATAL; goto error; } @@ -3204,6 +3202,14 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, sock_type == LTTNG_STREAM_CONTROL ? "control" : "data", relayd->net_seq_idx, fd); + /* We successfully added the socket. Send status back. */ + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL); + goto error_nosignal; + } + /* * Add relayd socket pair to consumer data hashtable. If object already * exists or on error, the function gracefully returns. @@ -3214,6 +3220,11 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type, return 0; error: + if (consumer_send_status_msg(sock, ret_code) < 0) { + lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL); + } + +error_nosignal: /* Close received socket if valid. */ if (fd >= 0) { if (close(fd)) { @@ -3221,7 +3232,6 @@ error: } } -error_close: if (relayd_created) { free(relayd); }