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

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 8ed279f44..86ccddfd0 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
  *
@@ -377,84 +99,15 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		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:
@@ -490,6 +143,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 
 		/* block */
 		if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
+			rcu_read_unlock();
 			return -EINTR;
 		}
 
@@ -497,6 +151,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 		if (ret != sizeof(fd)) {
 			lttng_consumer_send_error(ctx, CONSUMERD_ERROR_RECV_FD);
+			rcu_read_unlock();
 			return ret;
 		}
 
@@ -513,7 +168,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 				msg.u.stream.metadata_flag);
 		if (new_stream == NULL) {
 			lttng_consumer_send_error(ctx, CONSUMERD_OUTFD_ERROR);
-			goto end;
+			goto end_nosignal;
 		}
 
 		/* The stream is not metadata. Get relayd reference if exists. */
@@ -526,13 +181,13 @@ 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;
+				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;
+			goto end_nosignal;
 		}
 
 		if (ctx->on_recv_stream != NULL) {
@@ -540,7 +195,7 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			if (ret == 0) {
 				consumer_add_stream(new_stream);
 			} else if (ret < 0) {
-				goto end;
+				goto end_nosignal;
 			}
 		} else {
 			consumer_add_stream(new_stream);
@@ -551,38 +206,53 @@ int lttng_kconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	}
 	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:
+	{
+		struct consumer_relayd_sock_pair *relayd;
+
+		DBG("Kernel consumer destroying relayd %" PRIu64,
+				msg.u.destroy_relayd.net_seq_idx);
+
+		/* Get relayd reference if exists. */
+		relayd = consumer_find_relayd(msg.u.destroy_relayd.net_seq_idx);
+		if (relayd == NULL) {
+			ERR("Unable to find relayd %" PRIu64,
+					msg.u.destroy_relayd.net_seq_idx);
+			goto end_nosignal;
 		}
-		break;
+
+		/* Set destroy flag for this object */
+		uatomic_set(&relayd->destroy_flag, 1);
+
+		/* Destroy the relayd if refcount is 0 else set the destroy flag. */
+		if (uatomic_read(&relayd->refcount) == 0) {
+			consumer_destroy_relayd(relayd);
+		}
+		goto end_nosignal;
 	}
 	default:
-		break;
+		goto end_nosignal;
 	}
-end:
+
 	/*
-	 * 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:
+	 * 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).
 	 */
 	do {
 		ret = write(ctx->consumer_poll_pipe[1], "", 1);
-	} while (ret == -1UL && errno == EINTR);
+	} while (ret < 0 && errno == EINTR);
 end_nosignal:
+	rcu_read_unlock();
 	return 0;
 }
 
@@ -629,7 +299,7 @@ ssize_t lttng_kconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
 				 * display the error but continue processing to try
 				 * to release the subbuffer
 				 */
-				ERR("Error splicing to tracefile (ret: %ld != len: %ld)",
+				ERR("Error splicing to tracefile (ret: %zd != len: %lu)",
 						ret, len);
 			}