X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fkernel-consumer%2Fkernel-consumer.c;h=4e000fbd167e294c6b9139b6b7bdd2cf0098d6f1;hp=8ed279f44e16db7e73a14d8947c9e87b143f6946;hb=c8f59ee5fc11492ef472dc5cfd2fd2c4926b1787;hpb=00e2e675d54dc726a7c8f8887c889cc8ef022003

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 8ed279f44..4e000fbd1 100644
--- a/src/common/kernel-consumer/kernel-consumer.c
+++ b/src/common/kernel-consumer/kernel-consumer.c
@@ -25,6 +25,7 @@
 #include <sys/mman.h>
 #include <sys/socket.h>
 #include <sys/types.h>
+#include <inttypes.h>
 #include <unistd.h>
 #include <sys/stat.h>
 
@@ -41,285 +42,6 @@ extern struct lttng_consumer_global_data consumer_data;
 extern int consumer_poll_timeout;
 extern volatile int consumer_quit;
 
-/*
- * Mmap the ring buffer, read it and write the data to the tracefile.
- *
- * Returns the number of bytes written
- */
-ssize_t lttng_kconsumer_on_read_subbuffer_mmap(
-		struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream, unsigned long len)
-{
-	unsigned long mmap_offset;
-	ssize_t ret = 0, written = 0;
-	off_t orig_offset = stream->out_fd_offset;
-	int fd = stream->wait_fd;
-	/* Default is on the disk */
-	int outfd = stream->out_fd;
-	uint64_t metadata_id;
-	struct consumer_relayd_sock_pair *relayd = NULL;
-
-	/* 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;
-		}
-	}
-
-	/* get the offset inside the fd to mmap */
-	ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
-	if (ret != 0) {
-		errno = -ret;
-		perror("kernctl_get_mmap_read_offset");
-		written = ret;
-		goto end;
-	}
-
-	/* RCU lock for the relayd pointer */
-	rcu_read_lock();
-
-	/* Handle stream on the relayd if the output is on the network */
-	if (relayd) {
-		/*
-		 * Lock the control socket for the complete duration of the function
-		 * since from this point on we will use the socket.
-		 */
-		if (stream->metadata_flag) {
-			/* Metadata requires the control socket. */
-			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-		}
-
-		ret = consumer_handle_stream_before_relayd(stream, len);
-		if (ret >= 0) {
-			/* Use the returned socket. */
-			outfd = ret;
-
-			/* Write metadata stream id before payload */
-			if (stream->metadata_flag) {
-				metadata_id = htobe64(stream->relayd_stream_id);
-				do {
-					ret = write(outfd, (void *) &metadata_id,
-							sizeof(stream->relayd_stream_id));
-					if (ret < 0) {
-						PERROR("write metadata stream id");
-						written = ret;
-						goto end;
-					}
-				} while (errno == EINTR);
-				DBG("Metadata stream id %zu written before data",
-						stream->relayd_stream_id);
-				/*
-				 * We do this so the return value can match the len passed as
-				 * argument to this function.
-				 */
-				written -= sizeof(stream->relayd_stream_id);
-			}
-		}
-		/* Else, use the default set before which is the filesystem. */
-	}
-
-	while (len > 0) {
-		ret = write(outfd, stream->mmap_base + mmap_offset, len);
-		if (ret < 0) {
-			if (errno == EINTR) {
-				/* restart the interrupted system call */
-				continue;
-			} else {
-				perror("Error in file write");
-				if (written == 0) {
-					written = ret;
-				}
-				goto end;
-			}
-		} else if (ret > len) {
-			perror("Error in file write");
-			written += ret;
-			goto end;
-		} else {
-			len -= ret;
-			mmap_offset += ret;
-		}
-
-		/* 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,
-					SYNC_FILE_RANGE_WRITE);
-			stream->out_fd_offset += ret;
-		}
-		written += ret;
-	}
-	lttng_consumer_sync_trace_file(stream, orig_offset);
-
-end:
-	/* Unlock only if ctrl socket used */
-	if (relayd && stream->metadata_flag) {
-		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-	}
-
-	rcu_read_unlock();
-
-	return written;
-}
-
-/*
- * Splice the data from the ring buffer to the tracefile.
- *
- * Returns the number of bytes spliced.
- */
-ssize_t lttng_kconsumer_on_read_subbuffer_splice(
-		struct lttng_consumer_local_data *ctx,
-		struct lttng_consumer_stream *stream, unsigned long len)
-{
-	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;
-	uint64_t metadata_id;
-	struct consumer_relayd_sock_pair *relayd = NULL;
-
-	/* 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;
-		}
-	}
-
-	/* RCU lock for the relayd pointer */
-	rcu_read_lock();
-
-	/* Write metadata stream id before payload */
-	if (stream->metadata_flag && relayd) {
-		/*
-		 * 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);
-
-		do {
-			metadata_id = htobe64(stream->relayd_stream_id);
-			ret = write(ctx->consumer_thread_pipe[1],
-					(void *) &metadata_id,
-					sizeof(stream->relayd_stream_id));
-			if (ret < 0) {
-				PERROR("write metadata stream id");
-				written = ret;
-				goto end;
-			}
-		} while (errno == EINTR);
-		DBG("Metadata stream id %zu written before data",
-				stream->relayd_stream_id);
-	}
-
-	while (len > 0) {
-		DBG("splice chan to pipe offset %lu of len %lu (fd : %d)",
-				(unsigned long)offset, len, fd);
-		ret_splice = splice(fd, &offset, ctx->consumer_thread_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) {
-				/* Update counter to fit the spliced data */
-				ret_splice += sizeof(stream->relayd_stream_id);
-				len += sizeof(stream->relayd_stream_id);
-				/*
-				 * We do this so the return value can match the len passed as
-				 * argument to this function.
-				 */
-				written -= sizeof(stream->relayd_stream_id);
-			}
-
-			ret = consumer_handle_stream_before_relayd(stream, ret_splice);
-			if (ret >= 0) {
-				/* Use the returned socket. */
-				outfd = ret;
-			} else {
-				if (outfd == -1) {
-					ERR("Remote relayd disconnected. Stopping");
-					goto end;
-				}
-			}
-		}
-
-		DBG3("Kernel consumer splice data in %d to out %d",
-				ctx->consumer_thread_pipe[0], outfd);
-		ret_splice = splice(ctx->consumer_thread_pipe[0], NULL, outfd, NULL,
-				ret_splice, SPLICE_F_MOVE | SPLICE_F_MORE);
-		DBG("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;
-		}
-		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, CONSUMERD_SPLICE_EBADF);
-		break;
-	case EINVAL:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_EINVAL);
-		break;
-	case ENOMEM:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ENOMEM);
-		break;
-	case ESPIPE:
-		lttng_consumer_send_error(ctx, CONSUMERD_SPLICE_ESPIPE);
-		break;
-	}
-
-end:
-	if (relayd && stream->metadata_flag) {
-		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-	}
-
-	rcu_read_unlock();
-
-	return written;
-}
-
 /*
  * Take a snapshot for a specific fd
  *
@@ -370,91 +92,22 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 	ret = lttcomm_recv_unix_sock(sock, &msg, sizeof(msg));
 	if (ret != sizeof(msg)) {
-		lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_CMD);
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_CMD);
 		return ret;
 	}
 	if (msg.cmd_type == LTTNG_CONSUMER_STOP) {
 		return -ENOENT;
 	}
 
+	/* relayd needs RCU read-side protection */
+	rcu_read_lock();
+
 	switch (msg.cmd_type) {
 	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 	{
-		int fd;
-		struct consumer_relayd_sock_pair *relayd;
-
-		DBG("Consumer adding relayd socket");
-
-		/* Get relayd reference if exists. */
-		relayd = consumer_find_relayd(msg.u.relayd_sock.net_index);
-		if (relayd == NULL) {
-			/* Not found. Allocate one. */
-			relayd = consumer_allocate_relayd_sock_pair(
-					msg.u.relayd_sock.net_index);
-			if (relayd == NULL) {
-				lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
-				goto end_nosignal;
-			}
-		}
-
-		/* Poll on consumer socket. */
-		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
-			return -EINTR;
-		}
-
-		/* Get relayd socket from session daemon */
-		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
-		if (ret != sizeof(fd)) {
-			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
-			goto end_nosignal;
-		}
-
-		/* Copy socket information and received FD */
-		switch (msg.u.relayd_sock.type) {
-		case LTTNG_STREAM_CONTROL:
-			/* Copy received lttcomm socket */
-			lttcomm_copy_sock(&relayd->control_sock, &msg.u.relayd_sock.sock);
-
-			ret = lttcomm_create_sock(&relayd->control_sock);
-			if (ret < 0) {
-				goto end_nosignal;
-			}
-
-			/* Close the created socket fd which is useless */
-			close(relayd->control_sock.fd);
-
-			/* Assign new file descriptor */
-			relayd->control_sock.fd = fd;
-			break;
-		case LTTNG_STREAM_DATA:
-			/* Copy received lttcomm socket */
-			lttcomm_copy_sock(&relayd->data_sock, &msg.u.relayd_sock.sock);
-			ret = lttcomm_create_sock(&relayd->data_sock);
-			if (ret < 0) {
-				goto end_nosignal;
-			}
-
-			/* Close the created socket fd which is useless */
-			close(relayd->data_sock.fd);
-
-			/* Assign new file descriptor */
-			relayd->data_sock.fd = fd;
-			break;
-		default:
-			ERR("Unknown relayd socket type");
-			goto end_nosignal;
-		}
-
-		DBG("Consumer %s socket created successfully with net idx %d (fd: %d)",
-				msg.u.relayd_sock.type == LTTNG_STREAM_CONTROL ? "control" : "data",
-				relayd->net_seq_idx, fd);
-
-		/*
-		 * Add relayd socket pair to consumer data hashtable. If object already
-		 * exists or on error, the function gracefully returns.
-		 */
-		consumer_add_relayd(relayd);
-
+		ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
+				msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
+				&msg.u.relayd_sock.sock);
 		goto end_nosignal;
 	}
 	case LTTNG_CONSUMER_ADD_CHANNEL:
@@ -465,9 +118,10 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		new_channel = consumer_allocate_channel(msg.u.channel.channel_key,
 				-1, -1,
 				msg.u.channel.mmap_len,
-				msg.u.channel.max_sb_size);
+				msg.u.channel.max_sb_size,
+				msg.u.channel.nb_init_streams);
 		if (new_channel == NULL) {
-			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
+			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
 			goto end_nosignal;
 		}
 		if (ctx->on_recv_channel != NULL) {
@@ -484,19 +138,22 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	}
 	case LTTNG_CONSUMER_ADD_STREAM:
 	{
-		int fd;
+		int fd, stream_pipe;
 		struct consumer_relayd_sock_pair *relayd = NULL;
 		struct lttng_consumer_stream *new_stream;
+		int alloc_ret = 0;
 
 		/* block */
 		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
+			rcu_read_unlock();
 			return -EINTR;
 		}
 
 		/* Get stream file descriptor from socket */
 		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 		if (ret != sizeof(fd)) {
-			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+			lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
+			rcu_read_unlock();
 			return ret;
 		}
 
@@ -510,12 +167,36 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				msg.u.stream.uid,
 				msg.u.stream.gid,
 				msg.u.stream.net_index,
-				msg.u.stream.metadata_flag);
+				msg.u.stream.metadata_flag,
+				msg.u.stream.session_id,
+				&alloc_ret);
 		if (new_stream == NULL) {
-			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
-			goto end;
+			switch (alloc_ret) {
+			case -ENOMEM:
+			case -EINVAL:
+			default:
+				lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_OUTFD_ERROR);
+				break;
+			case -ENOENT:
+				/*
+				 * We could not find the channel. Can happen if cpu hotplug
+				 * happens while tearing down.
+				 */
+				DBG3("Could not find channel");
+				break;
+			}
+			goto end_nosignal;
 		}
 
+		/*
+		 * The buffer flush is done on the session daemon side for the kernel
+		 * so no need for the stream "hangup_flush_done" variable to be
+		 * tracked. This is important for a kernel stream since we don't rely
+		 * on the flush state of the stream to read data. It's not the case for
+		 * user space tracing.
+		 */
+		new_stream->hangup_flush_done = 0;
+
 		/* The stream is not metadata. Get relayd reference if exists. */
 		relayd = consumer_find_relayd(msg.u.stream.net_index);
 		if (relayd != NULL) {
@@ -526,64 +207,107 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 					&new_stream->relayd_stream_id);
 			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
 			if (ret < 0) {
-				goto end;
+				consumer_del_stream(new_stream, NULL);
+				goto end_nosignal;
 			}
 		} else if (msg.u.stream.net_index != -1) {
 			ERR("Network sequence index %d unknown. Not adding stream.",
 					msg.u.stream.net_index);
-			free(new_stream);
-			goto end;
+			consumer_del_stream(new_stream, NULL);
+			goto end_nosignal;
 		}
 
-		if (ctx->on_recv_stream != NULL) {
+		if (ctx->on_recv_stream) {
 			ret = ctx->on_recv_stream(new_stream);
-			if (ret == 0) {
-				consumer_add_stream(new_stream);
-			} else if (ret < 0) {
-				goto end;
+			if (ret < 0) {
+				consumer_del_stream(new_stream, NULL);
+				goto end_nosignal;
 			}
+		}
+
+		/* Get the right pipe where the stream will be sent. */
+		if (new_stream->metadata_flag) {
+			stream_pipe = ctx->consumer_metadata_pipe[1];
 		} else {
-			consumer_add_stream(new_stream);
+			stream_pipe = ctx->consumer_data_pipe[1];
+		}
+
+		do {
+			ret = write(stream_pipe, &new_stream, sizeof(new_stream));
+		} while (ret < 0 && errno == EINTR);
+		if (ret < 0) {
+			PERROR("Consumer write %s stream to pipe %d",
+					new_stream->metadata_flag ? "metadata" : "data",
+					stream_pipe);
+			consumer_del_stream(new_stream, NULL);
+			goto end_nosignal;
 		}
 
-		DBG("Kernel consumer_add_stream (%d)", fd);
+		DBG("Kernel consumer ADD_STREAM %s (fd: %d) with relayd id %" PRIu64,
+				msg.u.stream.path_name, fd, new_stream->relayd_stream_id);
 		break;
 	}
 	case LTTNG_CONSUMER_UPDATE_STREAM:
 	{
-		if (ctx->on_update_stream != NULL) {
-			ret = ctx->on_update_stream(msg.u.stream.stream_key, msg.u.stream.state);
-			if (ret == 0) {
-				consumer_change_stream_state(msg.u.stream.stream_key, msg.u.stream.state);
-			} else if (ret < 0) {
-				goto end;
-			}
-		} else {
-			consumer_change_stream_state(msg.u.stream.stream_key,
-				msg.u.stream.state);
+		rcu_read_unlock();
+		return -ENOSYS;
+	}
+	case LTTNG_CONSUMER_DESTROY_RELAYD:
+	{
+		uint64_t index = msg.u.destroy_relayd.net_seq_idx;
+		struct consumer_relayd_sock_pair *relayd;
+
+		DBG("Kernel consumer destroying relayd %" PRIu64, index);
+
+		/* Get relayd reference if exists. */
+		relayd = consumer_find_relayd(index);
+		if (relayd == NULL) {
+			ERR("Unable to find relayd %" PRIu64, index);
+			goto end_nosignal;
+		}
+
+		/*
+		 * Each relayd socket pair has a refcount of stream attached to it
+		 * which tells if the relayd is still active or not depending on the
+		 * refcount value.
+		 *
+		 * This will set the destroy flag of the relayd object and destroy it
+		 * if the refcount reaches zero when called.
+		 *
+		 * The destroy can happen either here or when a stream fd hangs up.
+		 */
+		consumer_flag_relayd_for_destroy(relayd);
+
+		goto end_nosignal;
+	}
+	case LTTNG_CONSUMER_DATA_AVAILABLE:
+	{
+		int32_t ret;
+		uint64_t id = msg.u.data_available.session_id;
+
+		DBG("Kernel consumer data available command for id %" PRIu64, id);
+
+		ret = consumer_data_available(id);
+
+		/* Send back returned value to session daemon */
+		ret = lttcomm_send_unix_sock(sock, &ret, sizeof(ret));
+		if (ret < 0) {
+			PERROR("send data available ret code");
 		}
 		break;
 	}
 	default:
-		break;
+		goto end_nosignal;
 	}
-end:
+
+end_nosignal:
+	rcu_read_unlock();
+
 	/*
-	 * Wake-up the other end by writing a null byte in the pipe
-	 * (non-blocking). Important note: Because writing into the
-	 * pipe is non-blocking (and therefore we allow dropping wakeup
-	 * data, as long as there is wakeup data present in the pipe
-	 * buffer to wake up the other end), the other end should
-	 * perform the following sequence for waiting:
-	 * 1) empty the pipe (reads).
-	 * 2) perform update operation.
-	 * 3) wait on the pipe (poll).
+	 * Return 1 to indicate success since the 0 value can be a socket
+	 * shutdown during the recv() or send() call.
 	 */
-	do {
-		ret = write(ctx->consumer_poll_pipe[1], "", 1);
-	} while (ret == -1UL && errno == EINTR);
-end_nosignal:
-	return 0;
+	return 1;
 }
 
 /*
@@ -592,7 +316,7 @@ end_nosignal:
 ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		struct lttng_consumer_local_data *ctx)
 {
-	unsigned long len;
+	unsigned long len, subbuf_size, padding;
 	int err;
 	ssize_t ret = 0;
 	int infd = stream->wait_fd;
@@ -601,6 +325,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 	/* Get the next subbuffer */
 	err = kernctl_get_next_subbuf(infd);
 	if (err != 0) {
+		ret = err;
 		/*
 		 * This is a debug message even for single-threaded consumer,
 		 * because poll() have more relaxed criterions than get subbuf,
@@ -612,60 +337,92 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		goto end;
 	}
 
+	/* Get the full subbuffer size including padding */
+	err = kernctl_get_padded_subbuf_size(infd, &len);
+	if (err != 0) {
+		errno = -err;
+		perror("Getting sub-buffer len failed.");
+		ret = err;
+		goto end;
+	}
+
 	switch (stream->output) {
-		case LTTNG_EVENT_SPLICE:
-			/* read the whole subbuffer */
-			err = kernctl_get_padded_subbuf_size(infd, &len);
-			if (err != 0) {
-				errno = -ret;
-				perror("Getting sub-buffer len failed.");
-				goto end;
-			}
+	case LTTNG_EVENT_SPLICE:
 
-			/* splice the subbuffer to the tracefile */
-			ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, len);
-			if (ret != len) {
-				/*
-				 * display the error but continue processing to try
-				 * to release the subbuffer
-				 */
-				ERR("Error splicing to tracefile (ret: %ld != len: %ld)",
-						ret, len);
-			}
+		/*
+		 * XXX: The lttng-modules splice "actor" does not handle copying
+		 * partial pages hence only using the subbuffer size without the
+		 * padding makes the splice fail.
+		 */
+		subbuf_size = len;
+		padding = 0;
 
-			break;
-		case LTTNG_EVENT_MMAP:
-			/* read the used subbuffer size */
-			err = kernctl_get_padded_subbuf_size(infd, &len);
-			if (err != 0) {
-				errno = -ret;
-				perror("Getting sub-buffer len failed.");
-				goto end;
-			}
-			/* write the subbuffer to the tracefile */
-			ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, len);
-			if (ret != len) {
-				/*
-				 * display the error but continue processing to try
-				 * to release the subbuffer
-				 */
-				ERR("Error writing to tracefile");
-			}
-			break;
-		default:
-			ERR("Unknown output method");
-			ret = -1;
+		/* splice the subbuffer to the tracefile */
+		ret = lttng_consumer_on_read_subbuffer_splice(ctx, stream, subbuf_size,
+				padding);
+		/*
+		 * XXX: Splice does not support network streaming so the return value
+		 * is simply checked against subbuf_size and not like the mmap() op.
+		 */
+		if (ret != subbuf_size) {
+			/*
+			 * display the error but continue processing to try
+			 * to release the subbuffer
+			 */
+			ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
+					ret, subbuf_size);
+		}
+		break;
+	case LTTNG_EVENT_MMAP:
+		/* Get subbuffer size without padding */
+		err = kernctl_get_subbuf_size(infd, &subbuf_size);
+		if (err != 0) {
+			errno = -err;
+			perror("Getting sub-buffer len failed.");
+			ret = err;
+			goto end;
+		}
+
+		/* Make sure the tracer is not gone mad on us! */
+		assert(len >= subbuf_size);
+
+		padding = len - subbuf_size;
+
+		/* write the subbuffer to the tracefile */
+		ret = lttng_consumer_on_read_subbuffer_mmap(ctx, stream, subbuf_size,
+				padding);
+		/*
+		 * The mmap operation should write subbuf_size amount of data when
+		 * network streaming or the full padding (len) size when we are _not_
+		 * streaming.
+		 */
+		if ((ret != subbuf_size && stream->net_seq_idx != -1) ||
+				(ret != len && stream->net_seq_idx == -1)) {
+			/*
+			 * Display the error but continue processing to try to release the
+			 * subbuffer
+			 */
+			ERR("Error writing to tracefile "
+					"(ret: %zd != len: %lu != subbuf_size: %lu)",
+					ret, len, subbuf_size);
+		}
+		break;
+	default:
+		ERR("Unknown output method");
+		ret = -1;
 	}
 
 	err = kernctl_put_next_subbuf(infd);
 	if (err != 0) {
-		errno = -ret;
+		errno = -err;
 		if (errno == EFAULT) {
 			perror("Error in unreserving sub buffer\n");
 		} else if (errno == EIO) {
 			/* Should never happen with newer LTTng versions */
 			perror("Reader has been pushed by the writer, last sub-buffer corrupted.");
 		}
+
+		ret = -err;
 		goto end;
 	}
 
@@ -726,3 +483,42 @@ error:
 	return ret;
 }
 
+/*
+ * Check if data is still being extracted from the buffers for a specific
+ * stream. Consumer data lock MUST be acquired before calling this function.
+ *
+ * Return 0 if the traced data are still getting read else 1 meaning that the
+ * data is available for trace viewer reading.
+ */
+int lttng_kconsumer_data_available(struct lttng_consumer_stream *stream)
+{
+	int ret;
+
+	assert(stream);
+
+	/*
+	 * Try to lock the stream mutex. On failure, we know that the stream is
+	 * being used else where hence there is data still being extracted.
+	 */
+	ret = pthread_mutex_trylock(&stream->lock);
+	if (ret == EBUSY) {
+		goto data_not_available;
+	}
+	/* The stream is now locked so we can do our ustctl calls */
+
+	ret = kernctl_get_next_subbuf(stream->wait_fd);
+	if (ret == 0) {
+		/* There is still data so let's put back this subbuffer. */
+		ret = kernctl_put_subbuf(stream->wait_fd);
+		assert(ret == 0);
+		pthread_mutex_unlock(&stream->lock);
+		goto data_not_available;
+	}
+
+	/* Data is available to be read for this stream. */
+	pthread_mutex_unlock(&stream->lock);
+	return 1;
+
+data_not_available:
+	return 0;
+}