return ret;
}
+/*
+ * Receive command from session daemon and process it.
+ *
+ * Return 1 on success else a negative value or 0.
+ */
int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
int sock, struct pollfd *consumer_sockpoll)
{
goto end_nosignal;
}
- if (ctx->on_recv_stream != NULL) {
- ret = ctx->on_recv_stream(new_stream);
- if (ret == 0) {
- consumer_add_stream(new_stream);
- } else if (ret < 0) {
- goto end_nosignal;
+ /* Send stream to the metadata thread */
+ if (new_stream->metadata_flag) {
+ if (ctx->on_recv_stream) {
+ ret = ctx->on_recv_stream(new_stream);
+ if (ret < 0) {
+ goto end_nosignal;
+ }
+ }
+
+ do {
+ ret = write(ctx->consumer_metadata_pipe[1], new_stream,
+ sizeof(struct lttng_consumer_stream));
+ } while (ret < 0 && errno == EINTR);
+ if (ret < 0) {
+ PERROR("write metadata pipe");
}
} else {
+ if (ctx->on_recv_stream) {
+ ret = ctx->on_recv_stream(new_stream);
+ if (ret < 0) {
+ goto end_nosignal;
+ }
+ }
consumer_add_stream(new_stream);
}
ret = write(ctx->consumer_poll_pipe[1], "", 1);
} while (ret < 0 && errno == EINTR);
end_nosignal:
- /* XXX: At some point we might want to return something else than zero */
rcu_read_unlock();
- return 0;
+
+ /*
+ * Return 1 to indicate success since the 0 value can be a socket
+ * shutdown during the recv() or send() call.
+ */
+ return 1;
}
int lttng_ustconsumer_allocate_channel(struct lttng_consumer_channel *chan)
int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
struct lttng_consumer_local_data *ctx)
{
- unsigned long len;
+ unsigned long len, subbuf_size, padding;
int err;
long ret = 0;
struct lttng_ust_shm_handle *handle;
/* Get the next subbuffer */
err = ustctl_get_next_subbuf(handle, buf);
if (err != 0) {
- ret = -ret; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
+ ret = err; /* ustctl_get_next_subbuf returns negative, caller expect positive. */
/*
* This is a debug message even for single-threaded consumer,
* because poll() have more relaxed criterions than get subbuf,
goto end;
}
assert(stream->output == LTTNG_EVENT_MMAP);
- /* read the used subbuffer size */
+ /* Get the full padded subbuffer size */
err = ustctl_get_padded_subbuf_size(handle, buf, &len);
assert(err == 0);
+
+ /* Get subbuffer data size (without padding) */
+ err = ustctl_get_subbuf_size(handle, buf, &subbuf_size);
+ assert(err == 0);
+
+ /* Make sure we don't get a subbuffer size bigger than the padded */
+ assert(len >= subbuf_size);
+
+ padding = len - subbuf_size;
/* write the subbuffer to the tracefile */
- ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
- if (ret != len) {
+ ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size, padding);
+ if (ret != subbuf_size) {
/*
* display the error but continue processing to try
* to release the subbuffer