X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=kconsumerd%2Fkconsumerd.c;h=c51e4adff535f56fc039459f41693f6e1dd6af30;hp=5a1fe89a77cbd6772722cf45dbbab5f67732ef1e;hb=7d452e122923ade6f4dc7a6e21bef105851d79e3;hpb=9d26659ae085e96066881c70d93a4929a9726070

diff --git a/kconsumerd/kconsumerd.c b/kconsumerd/kconsumerd.c
index 5a1fe89a7..c51e4adff 100644
--- a/kconsumerd/kconsumerd.c
+++ b/kconsumerd/kconsumerd.c
@@ -35,6 +35,7 @@
 #include <urcu/list.h>
 #include <poll.h>
 #include <unistd.h>
+#include <sys/mman.h>
 
 #include "lttngerr.h"
 #include "libkernelctl.h"
@@ -239,6 +240,119 @@ static int set_signal_handler(void)
 	return ret;
 }
 
+/*
+ * on_read_subbuffer_mmap
+ *
+ * mmap the ring buffer, read it and write the data to the tracefile.
+ * Returns the number of bytes written
+ */
+static int on_read_subbuffer_mmap(struct ltt_kconsumerd_fd *kconsumerd_fd,
+		unsigned long len)
+{
+	unsigned long mmap_len;
+	unsigned long mmap_offset;
+	unsigned long padded_len;
+	unsigned long padding_len;
+	char *mmap_base;
+	char *padding = NULL;
+	long ret = 0;
+	off_t orig_offset = kconsumerd_fd->out_fd_offset;
+	int fd = kconsumerd_fd->consumerd_fd;
+	int outfd = kconsumerd_fd->out_fd;
+
+	/* get the padded subbuffer size to know the padding required */
+	ret = kernctl_get_padded_subbuf_size(fd, &padded_len);
+	if (ret != 0) {
+		ret = errno;
+		perror("kernctl_get_padded_subbuf_size");
+		goto end;
+	}
+	padding_len = padded_len - len;
+	padding = malloc(padding_len * sizeof(char));
+	memset(padding, '\0', padding_len);
+
+	/* get the len of the mmap region */
+	ret = kernctl_get_mmap_len(fd, &mmap_len);
+	if (ret != 0) {
+		ret = errno;
+		perror("kernctl_get_mmap_len");
+		goto end;
+	}
+
+	/* get the offset inside the fd to mmap */
+	ret = kernctl_get_mmap_read_offset(fd, &mmap_offset);
+	if (ret != 0) {
+		ret = errno;
+		perror("kernctl_get_mmap_read_offset");
+		goto end;
+	}
+
+	mmap_base = mmap(NULL, mmap_len, PROT_READ, MAP_PRIVATE, fd, mmap_offset);
+	if (mmap_base == MAP_FAILED) {
+		perror("Error mmaping");
+		ret = -1;
+		goto end;
+	}
+
+	while (len > 0) {
+		ret = write(outfd, mmap_base, len);
+		if (ret >= len) {
+			len = 0;
+		} else if (ret < 0) {
+			ret = errno;
+			perror("Error in file write");
+			goto end;
+		}
+		/* This won't block, but will start writeout asynchronously */
+		sync_file_range(outfd, kconsumerd_fd->out_fd_offset, ret,
+				SYNC_FILE_RANGE_WRITE);
+		kconsumerd_fd->out_fd_offset += ret;
+	}
+
+	/* once all the data is written, write the padding to disk */
+	ret = write(outfd, padding, padding_len);
+	if (ret < 0) {
+		ret = errno;
+		perror("Error writing padding to file");
+		goto end;
+	}
+
+	/*
+	 * This does a blocking write-and-wait on any page that belongs to the
+	 * subbuffer prior to the one we just wrote.
+	 * Don't care about error values, as these are just hints and ways to
+	 * limit the amount of page cache used.
+	 */
+	if (orig_offset >= kconsumerd_fd->max_sb_size) {
+		sync_file_range(outfd, orig_offset - kconsumerd_fd->max_sb_size,
+				kconsumerd_fd->max_sb_size,
+				SYNC_FILE_RANGE_WAIT_BEFORE
+				| SYNC_FILE_RANGE_WRITE
+				| SYNC_FILE_RANGE_WAIT_AFTER);
+		/*
+		 * Give hints to the kernel about how we access the file:
+		 * POSIX_FADV_DONTNEED : we won't re-access data in a near
+		 * future after we write it.
+		 * We need to call fadvise again after the file grows because
+		 * the kernel does not seem to apply fadvise to non-existing
+		 * parts of the file.
+		 * Call fadvise _after_ having waited for the page writeback to
+		 * complete because the dirty page writeback semantic is not
+		 * well defined. So it can be expected to lead to lower
+		 * throughput in streaming.
+		 */
+		posix_fadvise(outfd, orig_offset - kconsumerd_fd->max_sb_size,
+				kconsumerd_fd->max_sb_size, POSIX_FADV_DONTNEED);
+	}
+	goto end;
+
+end:
+	if (padding != NULL) {
+		free(padding);
+	}
+	return ret;
+}
+
 /*
  * on_read_subbuffer
  *
@@ -355,22 +469,48 @@ static int read_subbuffer(struct ltt_kconsumerd_fd *kconsumerd_fd)
 		goto end;
 	}
 
-	/* read the whole subbuffer */
-	err = kernctl_get_padded_subbuf_size(infd, &len);
-	if (err != 0) {
-		ret = errno;
-		perror("Getting sub-buffer len failed.");
-		goto end;
-	}
+	switch (DEFAULT_KERNEL_CHANNEL_OUTPUT) {
+	case LTTNG_KERNEL_SPLICE:
+		/* read the whole subbuffer */
+		err = kernctl_get_padded_subbuf_size(infd, &len);
+		if (err != 0) {
+			ret = errno;
+			perror("Getting sub-buffer len failed.");
+			goto end;
+		}
 
-	/* splice the subbuffer to the tracefile */
-	ret = on_read_subbuffer(kconsumerd_fd, len);
-	if (ret < 0) {
-		/*
-		 * display the error but continue processing to try
-		 * to release the subbuffer
-		 */
-		ERR("Error splicing to tracefile");
+		/* splice the subbuffer to the tracefile */
+		ret = on_read_subbuffer(kconsumerd_fd, len);
+		if (ret < 0) {
+			/*
+			 * display the error but continue processing to try
+			 * to release the subbuffer
+			 */
+			ERR("Error splicing to tracefile");
+		}
+		break;
+	case LTTNG_KERNEL_MMAP:
+		/* read the used subbuffer size */
+		err = kernctl_get_subbuf_size(infd, &len);
+		if (err != 0) {
+			ret = errno;
+			perror("Getting sub-buffer len failed.");
+			goto end;
+		}
+
+		/* write the subbuffer to the tracefile */
+		ret = on_read_subbuffer_mmap(kconsumerd_fd, len);
+		if (ret < 0) {
+			/*
+			 * 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;
 	}
 
 	err = kernctl_put_next_subbuf(infd);
@@ -418,82 +558,78 @@ static int consumerd_recv_fd(int sfd, int size,
 {
 	struct msghdr msg;
 	struct iovec iov[1];
-	int ret, i, tmp2;
+	int ret = 0, i, tmp2;
 	struct cmsghdr *cmsg;
 	int nb_fd;
-	char tmp[CMSG_SPACE(size)];
-	struct lttcomm_kconsumerd_msg *buf;
+	char recv_fd[CMSG_SPACE(sizeof(int))];
+	struct lttcomm_kconsumerd_msg lkm;
+
 	/* the number of fds we are about to receive */
-	nb_fd = size/sizeof(struct lttcomm_kconsumerd_msg);
+	nb_fd = size / sizeof(struct lttcomm_kconsumerd_msg);
 
-	buf = malloc(size);
+	for (i = 0; i < nb_fd; i++) {
+		memset(&msg, 0, sizeof(msg));
 
-	memset(&msg, 0, sizeof(msg));
+		/* Prepare to receive the structures */
+		iov[0].iov_base = &lkm;
+		iov[0].iov_len = sizeof(lkm);
+		msg.msg_iov = iov;
+		msg.msg_iovlen = 1;
 
-	/* Prepare to receive the structures */
-	iov[0].iov_base = buf;
-	iov[0].iov_len = size;
-	msg.msg_iov = iov;
-	msg.msg_iovlen = 1;
+		msg.msg_control = recv_fd;
+		msg.msg_controllen = sizeof(recv_fd);
 
-	msg.msg_control = tmp;
-	msg.msg_controllen = sizeof(tmp);
+		DBG("Waiting to receive fd");
+		if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
+			perror("recvmsg");
+			continue;
+		}
 
-	DBG("Waiting to receive fds");
-	if ((ret = recvmsg(sfd, &msg, 0)) < 0) {
-		perror("recvmsg");
-	}
-	if (ret != size) {
-		ERR("Received only %d, expected %d", ret, size);
-		send_error(KCONSUMERD_ERROR_RECV_FD);
-		goto end;
-	}
+		if (ret != (size / nb_fd)) {
+			ERR("Received only %d, expected %d", ret, size);
+			send_error(KCONSUMERD_ERROR_RECV_FD);
+			goto end;
+		}
 
-	cmsg = CMSG_FIRSTHDR(&msg);
-	if (!cmsg) {
-		ERR("Invalid control message header");
-		ret = -1;
-		send_error(KCONSUMERD_ERROR_RECV_FD);
-		goto end;
-	}
+		cmsg = CMSG_FIRSTHDR(&msg);
+		if (!cmsg) {
+			ERR("Invalid control message header");
+			ret = -1;
+			send_error(KCONSUMERD_ERROR_RECV_FD);
+			goto end;
+		}
 
-	/* if we received fds */
-	if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
-		DBG("Receive : expecting %d fds", nb_fd);
-		for (i = 0; i < nb_fd; i++) {
+		/* if we received fds */
+		if (cmsg->cmsg_level == SOL_SOCKET && cmsg->cmsg_type == SCM_RIGHTS) {
 			switch (cmd_type) {
 			case ADD_STREAM:
-				DBG("add_fd %s (%d)", buf[i].path_name, ((int *)CMSG_DATA(cmsg))[i]);
-				ret = add_fd(&buf[i], ((int *)CMSG_DATA(cmsg))[i]);
+				DBG("add_fd %s (%d)", lkm.path_name, (CMSG_DATA(cmsg)[0]));
+				ret = add_fd(&lkm, (CMSG_DATA(cmsg)[0]));
 				if (ret < 0) {
 					send_error(KCONSUMERD_OUTFD_ERROR);
 					goto end;
 				}
 				break;
 			case UPDATE_STREAM:
-				change_fd_state(buf[i].fd, buf[i].state);
+				change_fd_state(lkm.fd, lkm.state);
 				break;
 			default:
 				break;
 			}
+			/* flag to tell the polling thread to update its fd array */
+			update_fd_array = 1;
+			/* signal the poll thread */
+			tmp2 = write(poll_pipe[1], "4", 1);
+		} else {
+			ERR("Didn't received any fd");
+			send_error(KCONSUMERD_ERROR_RECV_FD);
+			ret = -1;
+			goto end;
 		}
-		/* flag to tell the polling thread to update its fd array */
-		update_fd_array = 1;
-		/* signal the poll thread */
-		tmp2 = write(poll_pipe[1], "4", 1);
-	} else {
-		ERR("Didn't received any fd");
-		send_error(KCONSUMERD_ERROR_RECV_FD);
-		ret = -1;
-		goto end;
 	}
 
 end:
 	DBG("consumerd_recv_fd thread exiting");
-	if (buf != NULL) {
-		free(buf);
-		buf = NULL;
-	}
 	return ret;
 }
 
@@ -544,7 +680,6 @@ static void *thread_receive_fds(void *data)
 		}
 		if (tmp.cmd_type == STOP) {
 			DBG("Received STOP command");
-			quit = 1;
 			goto end;
 		}
 		/* we received a command to add or update fds */
@@ -557,6 +692,11 @@ static void *thread_receive_fds(void *data)
 
 end:
 	DBG("thread_receive_fds exiting");
+	quit = 1;
+	ret = write(poll_pipe[1], "4", 1);
+	if (ret < 0) {
+		perror("poll pipe write");
+	}
 	return NULL;
 }
 
@@ -675,6 +815,11 @@ static void *thread_poll_fds(void *data)
 			goto end;
 		}
 
+		/* No FDs and quit, cleanup the thread */
+		if (nb_fd == 0 && quit == 1) {
+			goto end;
+		}
+
 		/*
 		 * if only the poll_pipe triggered poll to return just return to the
 		 * beginning of the loop to update the array