#include <common/compat/fcntl.h>
#include <common/consumer-metadata-cache.h>
#include <common/consumer-timer.h>
+#include <common/utils.h>
#include "ust-consumer.h"
*/
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);
}
/*
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;
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;
}
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;
}
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;
}
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) {
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) {
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;
}
+ /*
+ * Set refcount to 1 for owner. Below, we will pass
+ * ownership to the consumer_thread_channel_poll()
+ * thread.
+ */
+ channel->refcount = 1;
+
/* Build channel attributes from received message. */
attr.subbuf_size = msg.u.ask_channel.subbuf_size;
attr.num_subbuf = msg.u.ask_channel.num_subbuf;
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) {
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:
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;
}
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;
}
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;
+ goto end_msg_sessiond;
}
/* Tell session daemon we are ready to receive the metadata. */
* 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);
/*
* 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;
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);
/*