health_code_update();
/* Try to send the stream to the relayd if one is available. */
- ret = consumer_send_relayd_stream(stream, stream->chan->pathname);
+ ret = consumer_send_relayd_stream(stream, stream->chan->pathname,
+ LTTNG_DOMAIN_UST);
if (ret < 0) {
/*
* Flag that the relayd was the problem here probably due to a
/* Send metadata stream to relayd if needed. */
if (metadata->metadata_stream->net_seq_idx != (uint64_t) -1ULL) {
ret = consumer_send_relayd_stream(metadata->metadata_stream,
- metadata->pathname);
+ metadata->pathname, LTTNG_DOMAIN_UST);
if (ret < 0) {
ret = LTTCOMM_CONSUMERD_ERROR_METADATA;
goto error;
if (relayd_id != (uint64_t) -1ULL) {
metadata_stream->net_seq_idx = relayd_id;
- ret = consumer_send_relayd_stream(metadata_stream, path);
+ ret = consumer_send_relayd_stream(metadata_stream, path,
+ LTTNG_DOMAIN_UST);
if (ret < 0) {
goto error_stream;
}
stream->net_seq_idx = relayd_id;
if (use_relayd) {
- ret = consumer_send_relayd_stream(stream, path);
+ ret = consumer_send_relayd_stream(stream, path,
+ LTTNG_DOMAIN_UST);
if (ret < 0) {
goto error_unlock;
}
msg.u.rotate_channel.pathname,
msg.u.rotate_channel.relayd_id,
msg.u.rotate_channel.metadata,
+ msg.u.rotate_channel.new_chunk_id,
ctx);
if (ret < 0) {
ERR("Rotate channel failed");
}
break;
}
+ case LTTNG_CONSUMER_ROTATE_RENAME:
+ {
+ DBG("Consumer rename session %" PRIu64 " after rotation",
+ msg.u.rotate_rename.session_id);
+ ret = lttng_consumer_rotate_rename(msg.u.rotate_rename.current_path,
+ msg.u.rotate_rename.new_path,
+ msg.u.rotate_rename.uid,
+ msg.u.rotate_rename.gid,
+ msg.u.rotate_rename.relayd_id);
+ if (ret < 0) {
+ ERR("Rotate rename failed");
+ ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND;
+ }
+
+ health_code_update();
+
+ ret = consumer_send_status_msg(sock, ret_code);
+ if (ret < 0) {
+ /* Somehow, the session daemon is not responding anymore. */
+ goto end_nosignal;
+ }
+
+ }
default:
break;
}
struct lttng_consumer_local_data *ctx)
{
unsigned long len, subbuf_size, padding;
- int err, write_index = 1, rotation_ret;
+ int err, write_index = 1, rotation_ret, rotate_ready;
long ret = 0;
struct ustctl_consumer_stream *ustream;
struct ctf_packet_index index;
readlen = lttng_read(stream->wait_fd, &dummy, 1);
if (readlen < 0 && errno != EAGAIN && errno != EWOULDBLOCK) {
ret = readlen;
- goto end;
+ goto error;
}
}
if (stream->metadata_flag) {
ret = commit_one_metadata_packet(stream);
if (ret <= 0) {
- goto end;
+ goto error;
}
ustctl_flush_buffer(stream->ustream, 1);
goto retry;
*/
DBG("Reserving sub buffer failed (everything is normal, "
"it is due to concurrency) [ret: %d]", err);
- goto end;
+ goto error;
}
assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
if (ret < 0) {
err = ustctl_put_subbuf(ustream);
assert(err == 0);
- goto end;
+ goto error;
}
/* Update the stream's sequence and discarded events count. */
PERROR("kernctl_get_events_discarded");
err = ustctl_put_subbuf(ustream);
assert(err == 0);
- goto end;
+ goto error;
}
} else {
write_index = 0;
assert(len >= subbuf_size);
padding = len - subbuf_size;
+
+ rotate_ready = lttng_consumer_stream_is_rotate_ready(stream, len);
+ if (rotate_ready < 0) {
+ ERR("Failed to check if stream is ready for rotation");
+ ret = -1;
+ err = ustctl_put_subbuf(ustream);
+ assert(err == 0);
+ goto error;
+ }
+ stream->rotate_ready = rotate_ready;
+
/* write the subbuffer to the tracefile */
ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding, &index);
/*
if (!stream->metadata_flag) {
ret = notify_if_more_data(stream, ctx);
if (ret < 0) {
- goto end;
+ goto error;
}
}
/* Write index if needed. */
if (!write_index) {
- goto end;
+ goto rotate;
}
if (stream->chan->live_timer_interval && !stream->metadata_flag) {
}
if (err < 0) {
- goto end;
+ goto error;
}
}
assert(!stream->metadata_flag);
err = consumer_stream_write_index(stream, &index);
if (err < 0) {
- goto end;
+ goto error;
}
-end:
- /* FIXME: do we need this lock, it causes deadlocks when called
- * at the same time with lttng_ustconsumer_rotate_channel ? */
-// pthread_mutex_lock(&stream->chan->lock);
- rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
- if (rotation_ret < 0) {
-// pthread_mutex_unlock(&stream->chan->lock);
- ret = -1;
- ERR("Stream rotation error");
- goto end;
+rotate:
+ if (stream->rotate_ready) {
+ rotation_ret = lttng_consumer_rotate_stream(ctx, stream);
+ if (rotation_ret < 0) {
+ ret = -1;
+ ERR("Stream rotation error");
+ goto error;
+ }
}
-// pthread_mutex_unlock(&stream->chan->lock);
+error:
return ret;
}