+ssize_t lttng_consumer_on_read_subbuffer_splice(
+ struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream, unsigned long len,
+ unsigned long padding)
+{
+ ssize_t ret = 0, written = 0, ret_splice = 0;
+ loff_t offset = 0;
+ off_t orig_offset = stream->out_fd_offset;
+ int fd = stream->wait_fd;
+ /* Default is on the disk */
+ int outfd = stream->out_fd;
+ struct consumer_relayd_sock_pair *relayd = NULL;
+ int *splice_pipe;
+
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ break;
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ /* Not supported for user space tracing */
+ return -ENOSYS;
+ default:
+ ERR("Unknown consumer_data type");
+ assert(0);
+ }
+
+ /* RCU lock for the relayd pointer */
+ rcu_read_lock();
+
+ pthread_mutex_lock(&stream->lock);
+
+ /* Flag that the current stream if set for network streaming. */
+ if (stream->net_seq_idx != -1) {
+ relayd = consumer_find_relayd(stream->net_seq_idx);
+ if (relayd == NULL) {
+ goto end;
+ }
+ }
+
+ /*
+ * Choose right pipe for splice. Metadata and trace data are handled by
+ * different threads hence the use of two pipes in order not to race or
+ * corrupt the written data.
+ */
+ if (stream->metadata_flag) {
+ splice_pipe = ctx->consumer_splice_metadata_pipe;
+ } else {
+ splice_pipe = ctx->consumer_thread_pipe;
+ }
+
+ /* Write metadata stream id before payload */
+ if (relayd) {
+ int total_len = len;
+
+ if (stream->metadata_flag) {
+ /*
+ * Lock the control socket for the complete duration of the function
+ * since from this point on we will use the socket.
+ */
+ pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+
+ ret = write_relayd_metadata_id(splice_pipe[1], stream, relayd,
+ padding);
+ if (ret < 0) {
+ written = ret;
+ goto end;
+ }
+
+ total_len += sizeof(struct lttcomm_relayd_metadata_payload);
+ }
+
+ ret = write_relayd_stream_header(stream, total_len, padding, relayd);
+ if (ret >= 0) {
+ /* Use the returned socket. */
+ outfd = ret;
+ } else {
+ ERR("Remote relayd disconnected. Stopping");
+ goto end;
+ }
+ } else {
+ /* No streaming, we have to set the len with the full padding */
+ len += padding;
+ }
+
+ while (len > 0) {
+ DBG("splice chan to pipe offset %lu of len %lu (fd : %d, pipe: %d)",
+ (unsigned long)offset, len, fd, splice_pipe[1]);
+ ret_splice = splice(fd, &offset, splice_pipe[1], NULL, len,
+ SPLICE_F_MOVE | SPLICE_F_MORE);
+ DBG("splice chan to pipe, ret %zd", ret_splice);
+ if (ret_splice < 0) {
+ PERROR("Error in relay splice");
+ if (written == 0) {
+ written = ret_splice;
+ }
+ ret = errno;
+ goto splice_error;
+ }
+
+ /* Handle stream on the relayd if the output is on the network */
+ if (relayd) {
+ if (stream->metadata_flag) {
+ size_t metadata_payload_size =
+ sizeof(struct lttcomm_relayd_metadata_payload);
+
+ /* Update counter to fit the spliced data */
+ ret_splice += metadata_payload_size;
+ len += metadata_payload_size;
+ /*
+ * We do this so the return value can match the len passed as
+ * argument to this function.
+ */
+ written -= metadata_payload_size;
+ }
+ }
+
+ /* Splice data out */
+ ret_splice = splice(splice_pipe[0], NULL, outfd, NULL,
+ ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
+ DBG("Consumer splice pipe to file, ret %zd", ret_splice);
+ if (ret_splice < 0) {
+ PERROR("Error in file splice");
+ if (written == 0) {
+ written = ret_splice;
+ }
+ ret = errno;
+ goto splice_error;
+ } else if (ret_splice > len) {
+ errno = EINVAL;
+ PERROR("Wrote more data than requested %zd (len: %lu)",
+ ret_splice, len);
+ written += ret_splice;
+ ret = errno;
+ goto splice_error;
+ }
+ len -= ret_splice;
+
+ /* This call is useless on a socket so better save a syscall. */
+ if (!relayd) {
+ /* This won't block, but will start writeout asynchronously */
+ lttng_sync_file_range(outfd, stream->out_fd_offset, ret_splice,
+ SYNC_FILE_RANGE_WRITE);
+ stream->out_fd_offset += ret_splice;
+ }
+ written += ret_splice;
+ }
+ lttng_consumer_sync_trace_file(stream, orig_offset);
+
+ ret = ret_splice;
+
+ goto end;
+
+splice_error:
+ /* send the appropriate error description to sessiond */
+ switch (ret) {
+ case EBADF:
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EBADF);
+ break;
+ case EINVAL:
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_EINVAL);
+ break;
+ case ENOMEM:
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ENOMEM);
+ break;
+ case ESPIPE:
+ lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_SPLICE_ESPIPE);
+ break;
+ }
+
+end:
+ pthread_mutex_unlock(&stream->lock);
+ if (relayd && stream->metadata_flag) {
+ pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+ }
+
+ rcu_read_unlock();
+ return written;
+}
+
+/*
+ * Take a snapshot for a specific fd
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx,
+ struct lttng_consumer_stream *stream)
+{
+ switch (consumer_data.type) {
+ case LTTNG_CONSUMER_KERNEL:
+ return lttng_kconsumer_take_snapshot(ctx, stream);
+ case LTTNG_CONSUMER32_UST:
+ case LTTNG_CONSUMER64_UST:
+ return lttng_ustconsumer_take_snapshot(ctx, stream);
+ default:
+ ERR("Unknown consumer_data type");
+ assert(0);
+ return -ENOSYS;
+ }
+
+}
+
+/*
+ * Get the produced position
+ *
+ * Returns 0 on success, < 0 on error
+ */
+int lttng_consumer_get_produced_snapshot(