X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=ef0d4b7bb3f2edaf073b2215843cf7fc17d9a6e3;hb=79d4ffb7115aa6bd499caeadc3e6c58c36d86864;hp=260779ae6d1ace6d9a186c40c2202a654c85dbea;hpb=7608208897f23f1419759e786175247a39703a79;p=lttng-tools.git diff --git a/src/common/consumer.c b/src/common/consumer.c index 260779ae6..ef0d4b7bb 100644 --- a/src/common/consumer.c +++ b/src/common/consumer.c @@ -127,6 +127,12 @@ void consumer_steal_stream_key(int key, struct lttng_ht *ht) rcu_read_unlock(); } +/* + * Return a channel object for the given key. + * + * RCU read side lock MUST be acquired before calling this function and + * protects the channel ptr. + */ static struct lttng_consumer_channel *consumer_find_channel(int key) { struct lttng_ht_iter iter; @@ -138,8 +144,6 @@ static struct lttng_consumer_channel *consumer_find_channel(int key) return NULL; } - rcu_read_lock(); - lttng_ht_lookup(consumer_data.channel_ht, (void *)((unsigned long) key), &iter); node = lttng_ht_iter_get_node_ulong(&iter); @@ -147,8 +151,6 @@ static struct lttng_consumer_channel *consumer_find_channel(int key) channel = caa_container_of(node, struct lttng_consumer_channel, node); } - rcu_read_unlock(); - return channel; } @@ -348,8 +350,8 @@ void consumer_del_stream(struct lttng_consumer_stream *stream, goto free_stream; } - pthread_mutex_lock(&stream->lock); pthread_mutex_lock(&consumer_data.lock); + pthread_mutex_lock(&stream->lock); switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: @@ -441,8 +443,8 @@ void consumer_del_stream(struct lttng_consumer_stream *stream, end: consumer_data.need_update = 1; - pthread_mutex_unlock(&consumer_data.lock); pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&consumer_data.lock); if (free_chan) { consumer_del_channel(free_chan); @@ -475,6 +477,8 @@ struct lttng_consumer_stream *consumer_allocate_stream( goto end; } + rcu_read_lock(); + /* * Get stream's channel reference. Needed when adding the stream to the * global hash table. @@ -531,9 +535,12 @@ struct lttng_consumer_stream *consumer_allocate_stream( stream->path_name, stream->key, stream->shm_fd, stream->wait_fd, (unsigned long long) stream->mmap_len, stream->out_fd, stream->net_seq_idx, stream->session_id); + + rcu_read_unlock(); return stream; error: + rcu_read_unlock(); free(stream); end: return NULL; @@ -910,7 +917,7 @@ static int consumer_update_poll_array( * changed where this function will be called back again. */ if (stream->state != LTTNG_CONSUMER_ACTIVE_STREAM || - stream->endpoint_status) { + stream->endpoint_status == CONSUMER_ENDPOINT_INACTIVE) { continue; } DBG("Active FD %d", stream->wait_fd); @@ -1260,6 +1267,8 @@ end: * core function for writing trace buffers to either the local filesystem or * the network. * + * It must be called with the stream lock held. + * * Careful review MUST be put if any changes occur! * * Returns the number of bytes written @@ -1280,8 +1289,6 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap( /* RCU lock for the relayd pointer */ rcu_read_lock(); - pthread_mutex_lock(&stream->lock); - /* Flag that the current stream if set for network streaming. */ if (stream->net_seq_idx != -1) { relayd = consumer_find_relayd(stream->net_seq_idx); @@ -1406,7 +1413,6 @@ end: if (relayd && stream->metadata_flag) { pthread_mutex_unlock(&relayd->ctrl_sock_mutex); } - pthread_mutex_unlock(&stream->lock); rcu_read_unlock(); return written; @@ -1415,6 +1421,8 @@ end: /* * Splice the data from the ring buffer to the tracefile. * + * It must be called with the stream lock held. + * * Returns the number of bytes spliced. */ ssize_t lttng_consumer_on_read_subbuffer_splice( @@ -1447,8 +1455,6 @@ ssize_t lttng_consumer_on_read_subbuffer_splice( /* RCU lock for the relayd pointer */ rcu_read_lock(); - pthread_mutex_lock(&stream->lock); - /* Flag that the current stream if set for network streaming. */ if (stream->net_seq_idx != -1) { relayd = consumer_find_relayd(stream->net_seq_idx); @@ -1616,7 +1622,6 @@ end: if (relayd && stream->metadata_flag) { pthread_mutex_unlock(&relayd->ctrl_sock_mutex); } - pthread_mutex_unlock(&stream->lock); rcu_read_unlock(); return written; @@ -1762,9 +1767,9 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, goto free_stream; } + pthread_mutex_lock(&consumer_data.lock); pthread_mutex_lock(&stream->lock); - pthread_mutex_lock(&consumer_data.lock); switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: if (stream->mmap_base != NULL) { @@ -1854,8 +1859,8 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, } end: - pthread_mutex_unlock(&consumer_data.lock); pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&consumer_data.lock); if (free_chan) { consumer_del_channel(free_chan); @@ -1949,7 +1954,7 @@ static void validate_endpoint_status_data_stream(void) rcu_read_lock(); cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) { /* Validate delete flag of the stream */ - if (stream->endpoint_status != CONSUMER_ENDPOINT_INACTIVE) { + if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) { continue; } /* Delete it right now */ @@ -1974,7 +1979,7 @@ static void validate_endpoint_status_metadata_stream( rcu_read_lock(); cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) { /* Validate delete flag of the stream */ - if (!stream->endpoint_status) { + if (stream->endpoint_status == CONSUMER_ENDPOINT_ACTIVE) { continue; } /* @@ -2559,17 +2564,27 @@ end: ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx) { + ssize_t ret; + + pthread_mutex_lock(&stream->lock); + switch (consumer_data.type) { case LTTNG_CONSUMER_KERNEL: - return lttng_kconsumer_read_subbuffer(stream, ctx); + ret = lttng_kconsumer_read_subbuffer(stream, ctx); + break; case LTTNG_CONSUMER32_UST: case LTTNG_CONSUMER64_UST: - return lttng_ustconsumer_read_subbuffer(stream, ctx); + ret = lttng_ustconsumer_read_subbuffer(stream, ctx); + break; default: ERR("Unknown consumer_data type"); assert(0); - return -ENOSYS; + ret = -ENOSYS; + break; } + + pthread_mutex_unlock(&stream->lock); + return ret; } int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)