X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=5f1f93b0e1024c6c1eee422d80930b71ccaaf1a8;hp=ce2e17f742eae728649ebdfd8632565eae160ac6;hb=b1316da1ffbd276fc8271e7a9438e683ad352781;hpb=577eea73132dee4da47752590ed535206678eb34 diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index ce2e17f74..5f1f93b0e 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -409,7 +409,8 @@ static int open_ust_stream_fd(struct lttng_consumer_channel *channel, int cpu, } return run_as_open(shm_path, O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR, - session_credentials->uid, session_credentials->gid); + lttng_credentials_get_uid(session_credentials), + lttng_credentials_get_gid(session_credentials)); error_shm_path: return -1; @@ -487,8 +488,10 @@ error_open: ERR("Cannot get stream shm path"); } closeret = run_as_unlink(shm_path, - channel->buffer_credentials.value.uid, - channel->buffer_credentials.value.gid); + lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR( + channel->buffer_credentials)), + lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR( + channel->buffer_credentials))); if (closeret) { PERROR("unlink %s", shm_path); } @@ -497,8 +500,10 @@ error_open: /* Try to rmdir all directories under shm_path root. */ if (channel->root_shm_path[0]) { (void) run_as_rmdir_recursive(channel->root_shm_path, - channel->buffer_credentials.value.uid, - channel->buffer_credentials.value.gid, + lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR( + channel->buffer_credentials)), + lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR( + channel->buffer_credentials)), LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG); } free(stream_fds); @@ -1278,6 +1283,17 @@ error_unlock: return ret; } +static +void metadata_stream_reset_cache_consumed_position( + struct lttng_consumer_stream *stream) +{ + ASSERT_LOCKED(stream->lock); + + DBG("Reset metadata cache of session %" PRIu64, + stream->chan->session_id); + stream->ust_metadata_pushed = 0; +} + /* * Receive the metadata updates from the sessiond. Supports receiving * overlapping metadata, but is needs to always belong to a contiguous @@ -1292,6 +1308,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, { int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS; char *metadata_str; + enum consumer_metadata_cache_write_status cache_write_status; DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len); @@ -1315,9 +1332,40 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, health_code_update(); pthread_mutex_lock(&channel->metadata_cache->lock); - ret = consumer_metadata_cache_write(channel, offset, len, version, - metadata_str); - if (ret < 0) { + cache_write_status = consumer_metadata_cache_write( + channel, offset, len, version, metadata_str); + pthread_mutex_unlock(&channel->metadata_cache->lock); + switch (cache_write_status) { + case CONSUMER_METADATA_CACHE_WRITE_STATUS_NO_CHANGE: + /* + * The write entirely overlapped with existing contents of the + * same metadata version (same content); there is nothing to do. + */ + break; + case CONSUMER_METADATA_CACHE_WRITE_STATUS_INVALIDATED: + /* + * The metadata cache was invalidated (previously pushed + * content has been overwritten). Reset the stream's consumed + * metadata position to ensure the metadata poll thread consumes + * the whole cache. + */ + pthread_mutex_lock(&channel->metadata_stream->lock); + metadata_stream_reset_cache_consumed_position( + channel->metadata_stream); + pthread_mutex_unlock(&channel->metadata_stream->lock); + /* Fall-through. */ + case CONSUMER_METADATA_CACHE_WRITE_STATUS_APPENDED_CONTENT: + /* + * In both cases, the metadata poll thread has new data to + * consume. + */ + ret = consumer_metadata_wakeup_pipe(channel); + if (ret) { + ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA; + goto end_free; + } + break; + case CONSUMER_METADATA_CACHE_WRITE_STATUS_ERROR: /* Unable to handle metadata. Notify session daemon. */ ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA; /* @@ -1325,10 +1373,10 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, * not have been updated which could create an infinite loop below when * waiting for the metadata cache to be flushed. */ - pthread_mutex_unlock(&channel->metadata_cache->lock); goto end_free; + default: + abort(); } - pthread_mutex_unlock(&channel->metadata_cache->lock); if (!wait) { goto end_free; @@ -1461,8 +1509,8 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, struct ustctl_consumer_channel_attr attr; const uint64_t chunk_id = msg.u.ask_channel.chunk_id.value; const struct lttng_credentials buffer_credentials = { - .uid = msg.u.ask_channel.buffer_credentials.uid, - .gid = msg.u.ask_channel.buffer_credentials.gid, + .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.ask_channel.buffer_credentials.uid), + .gid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.ask_channel.buffer_credentials.gid), }; /* Create a plain object and reserve a channel key. */ @@ -2072,8 +2120,8 @@ end_rotate_channel_nosignal: case LTTNG_CONSUMER_CREATE_TRACE_CHUNK: { const struct lttng_credentials credentials = { - .uid = msg.u.create_trace_chunk.credentials.value.uid, - .gid = msg.u.create_trace_chunk.credentials.value.gid, + .uid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.uid), + .gid = LTTNG_OPTIONAL_INIT_VALUE(msg.u.create_trace_chunk.credentials.value.gid), }; const bool is_local_trace = !msg.u.create_trace_chunk.relayd_id.is_set; @@ -2181,6 +2229,28 @@ end_rotate_channel_nosignal: msg.u.trace_chunk_exists.chunk_id); goto end_msg_sessiond; } + case LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS: + { + const uint64_t key = msg.u.open_channel_packets.key; + struct lttng_consumer_channel *channel = + consumer_find_channel(key); + + if (channel) { + pthread_mutex_lock(&channel->lock); + ret_code = lttng_consumer_open_channel_packets(channel); + pthread_mutex_unlock(&channel->lock); + } else { + /* + * The channel could have disappeared in per-pid + * buffering mode. + */ + DBG("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } + + health_code_update(); + goto end_msg_sessiond; + } default: break; } @@ -2380,8 +2450,10 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan) ERR("Cannot get stream shm path"); } ret = run_as_unlink(shm_path, - chan->buffer_credentials.value.uid, - chan->buffer_credentials.value.gid); + lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR( + chan->buffer_credentials)), + lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR( + chan->buffer_credentials))); if (ret) { PERROR("unlink %s", shm_path); } @@ -2400,8 +2472,10 @@ void lttng_ustconsumer_free_channel(struct lttng_consumer_channel *chan) /* Try to rmdir all directories under shm_path root. */ if (chan->root_shm_path[0]) { (void) run_as_rmdir_recursive(chan->root_shm_path, - chan->buffer_credentials.value.uid, - chan->buffer_credentials.value.gid, + lttng_credentials_get_uid(LTTNG_OPTIONAL_GET_PTR( + chan->buffer_credentials)), + lttng_credentials_get_gid(LTTNG_OPTIONAL_GET_PTR( + chan->buffer_credentials)), LTTNG_DIRECTORY_HANDLE_SKIP_NON_EMPTY_FLAG); } free(chan->stream_fds); @@ -2434,15 +2508,6 @@ int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream) return ustctl_stream_close_wakeup_fd(stream->ustream); } -static -void metadata_stream_reset_cache_consumed_position( - struct lttng_consumer_stream *stream) -{ - DBG("Reset metadata cache of session %" PRIu64, - stream->chan->session_id); - stream->ust_metadata_pushed = 0; -} - /* * Write up to one packet from the metadata cache to the channel. * @@ -2935,6 +3000,7 @@ static int put_next_subbuffer(struct lttng_consumer_stream *stream, static int signal_metadata(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx) { + ASSERT_LOCKED(stream->metadata_rdv_lock); return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0; } @@ -3020,6 +3086,7 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) assert(stream); assert(stream->ustream); + ASSERT_LOCKED(stream->lock); DBG("UST consumer checking data pending"); @@ -3032,7 +3099,9 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) uint64_t contiguous, pushed; /* Ease our life a bit. */ + pthread_mutex_lock(&stream->chan->metadata_cache->lock); contiguous = stream->chan->metadata_cache->max_offset; + pthread_mutex_unlock(&stream->chan->metadata_cache->lock); pushed = stream->ust_metadata_pushed; /*