X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=89b946896fc2858844be1f63c77bbd55c9e7642a;hb=fc6432477fe8a5d3c4494acccdbe94c5912420bb;hp=431b94626736d042d92db89db0774e7e941cc3e7;hpb=331744e34f56a5aec69b05d356d6901e67926acc;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 431b94626..89b946896 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -38,6 +38,7 @@ #include #include #include +#include #include "ust-consumer.h" @@ -112,13 +113,14 @@ error: */ static struct lttng_consumer_channel *allocate_channel(uint64_t session_id, const char *pathname, const char *name, uid_t uid, gid_t gid, - int relayd_id, uint64_t key, enum lttng_event_output output) + int relayd_id, uint64_t key, enum lttng_event_output output, + uint64_t tracefile_size, uint64_t tracefile_count) { assert(pathname); assert(name); return consumer_allocate_channel(key, session_id, pathname, name, uid, gid, - relayd_id, output); + relayd_id, output, tracefile_size, tracefile_count); } /* @@ -221,7 +223,9 @@ static int send_stream_to_relayd(struct lttng_consumer_stream *stream) pthread_mutex_lock(&relayd->ctrl_sock_mutex); /* Add stream on the relayd */ ret = relayd_add_stream(&relayd->control_sock, stream->name, - stream->chan->pathname, &stream->relayd_stream_id); + stream->chan->pathname, &stream->relayd_stream_id, + stream->chan->tracefile_size, + stream->chan->tracefile_count); pthread_mutex_unlock(&relayd->ctrl_sock_mutex); if (ret < 0) { goto error; @@ -552,7 +556,7 @@ int lttng_ustconsumer_push_metadata(struct lttng_consumer_channel *metadata, ret = ustctl_write_metadata_to_channel(metadata->uchan, metadata_str + target_offset, len); if (ret < 0) { - ERR("ustctl write metadata fail with ret %d, len %ld", ret, len); + ERR("ustctl write metadata fail with ret %d, len %" PRIu64, ret, len); goto error; } @@ -575,11 +579,11 @@ static int flush_channel(uint64_t chan_key) struct lttng_ht *ht; struct lttng_ht_iter iter; - DBG("UST consumer flush channel key %lu", chan_key); + DBG("UST consumer flush channel key %" PRIu64, chan_key); channel = consumer_find_channel(chan_key); if (!channel) { - ERR("UST consumer flush channel %lu not found", chan_key); + ERR("UST consumer flush channel %" PRIu64 " not found", chan_key); ret = LTTNG_ERR_UST_CHAN_NOT_FOUND; goto error; } @@ -609,11 +613,11 @@ static int close_metadata(uint64_t chan_key) int ret; struct lttng_consumer_channel *channel; - DBG("UST consumer close metadata key %lu", chan_key); + DBG("UST consumer close metadata key %" PRIu64, chan_key); channel = consumer_find_channel(chan_key); if (!channel) { - ERR("UST consumer close metadata %lu not found", chan_key); + ERR("UST consumer close metadata %" PRIu64 " not found", chan_key); ret = LTTNG_ERR_UST_CHAN_NOT_FOUND; goto error; } @@ -644,7 +648,7 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key) int ret; struct lttng_consumer_channel *metadata; - DBG("UST consumer setup metadata key %lu", key); + DBG("UST consumer setup metadata key %" PRIu64, key); metadata = consumer_find_channel(key); if (!metadata) { @@ -697,7 +701,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, int ret, ret_code = LTTNG_OK; char *metadata_str; - DBG("UST consumer push metadata key %lu of len %lu", key, len); + DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len); metadata_str = zmalloc(len * sizeof(char)); if (!metadata_str) { @@ -848,7 +852,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, msg.u.ask_channel.pathname, msg.u.ask_channel.name, msg.u.ask_channel.uid, msg.u.ask_channel.gid, msg.u.ask_channel.relayd_id, msg.u.ask_channel.key, - (enum lttng_event_output) msg.u.ask_channel.output); + (enum lttng_event_output) msg.u.ask_channel.output, + msg.u.ask_channel.tracefile_size, + msg.u.ask_channel.tracefile_count); if (!channel) { goto end_channel_error; } @@ -892,10 +898,22 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_channel_error; } + if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) { + ret = consumer_metadata_cache_allocate(channel); + if (ret < 0) { + ERR("Allocating metadata cache"); + goto end_channel_error; + } + consumer_timer_switch_start(channel, attr.switch_timer_interval); + attr.switch_timer_interval = 0; + } + /* * Add the channel to the internal state AFTER all streams were created * and successfully sent to session daemon. This way, all streams must * be ready before this channel is visible to the threads. + * If add_channel succeeds, ownership of the channel is + * passed to consumer_thread_channel_poll(). */ ret = add_channel(channel, ctx); if (ret < 0) { @@ -917,16 +935,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } - if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) { - ret = consumer_metadata_cache_allocate(channel); - if (ret < 0) { - ERR("Allocating metadata cache"); - goto end_channel_error; - } - consumer_timer_switch_start(channel, attr.switch_timer_interval); - attr.switch_timer_interval = 0; - } - break; } case LTTNG_CONSUMER_GET_CHANNEL: @@ -937,7 +945,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, channel = consumer_find_channel(key); if (!channel) { - ERR("UST consumer get channel key %lu not found", key); + ERR("UST consumer get channel key %" PRIu64 " not found", key); ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND; goto end_msg_sessiond; } @@ -988,7 +996,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, channel = consumer_find_channel(key); if (!channel) { - ERR("UST consumer get channel key %lu not found", key); + ERR("UST consumer get channel key %" PRIu64 " not found", key); ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND; goto end_msg_sessiond; } @@ -1027,11 +1035,12 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, uint64_t offset = msg.u.push_metadata.target_offset; struct lttng_consumer_channel *channel; - DBG("UST consumer push metadata key %lu of len %lu", key, len); + DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, + len); channel = consumer_find_channel(key); if (!channel) { - ERR("UST consumer push metadata %lu not found", key); + ERR("UST consumer push metadata %" PRIu64 " not found", key); ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND; } @@ -1268,7 +1277,7 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream, * happen and it is OK with the code flow. */ DBG("Error writing to tracefile " - "(ret: %zd != len: %lu != subbuf_size: %lu)", + "(ret: %ld != len: %lu != subbuf_size: %lu)", ret, len, subbuf_size); } err = ustctl_put_next_subbuf(ustream); @@ -1280,35 +1289,25 @@ end: /* * Called when a stream is created. + * + * Return 0 on success or else a negative value. */ int lttng_ustconsumer_on_recv_stream(struct lttng_consumer_stream *stream) { int ret; - char full_path[PATH_MAX]; - - /* Opening the tracefile in write mode */ - if (stream->net_seq_idx != (uint64_t) -1ULL) { - goto end; - } - - ret = snprintf(full_path, sizeof(full_path), "%s/%s", - stream->chan->pathname, stream->name); - if (ret < 0) { - PERROR("snprintf on_recv_stream"); - goto error; - } - ret = run_as_open(full_path, O_WRONLY | O_CREAT | O_TRUNC, - S_IRWXU | S_IRWXG | S_IRWXO, stream->uid, stream->gid); - if (ret < 0) { - PERROR("open stream path %s", full_path); - goto error; + /* Don't create anything if this is set for streaming. */ + if (stream->net_seq_idx == (uint64_t) -1ULL) { + ret = utils_create_stream_file(stream->chan->pathname, stream->name, + stream->chan->tracefile_size, stream->tracefile_count_current, + stream->uid, stream->gid); + if (ret < 0) { + goto error; + } + stream->out_fd = ret; + stream->tracefile_size_current = 0; } - stream->out_fd = ret; - -end: - /* we return 0 to let the library handle the FD internally */ - return 0; + ret = 0; error: return ret; @@ -1437,7 +1436,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, ret = lttcomm_recv_unix_sock(ctx->consumer_metadata_socket, &msg, sizeof(msg)); if (ret != sizeof(msg)) { - DBG("Consumer received unexpected message size %d (expects %lu)", + DBG("Consumer received unexpected message size %d (expects %zu)", ret, sizeof(msg)); lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD); /*