X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=deebd2e2b353cd1cc4365f5ede45e78203e9653e;hp=1863cddc5757e6e1788642dd788629303a890881;hb=a4b92340642035d1eafeb1eead0ad01f64d2007d;hpb=f02e1e8a5820da2eda835add020f92ca8d32c973

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 1863cddc5..deebd2e2b 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -278,8 +278,12 @@ void consumer_del_stream(struct lttng_consumer_stream *stream)
 				stream->relayd_stream_id,
 				stream->next_net_seq_num - 1);
 		if (ret < 0) {
-			ERR("Unable to close stream on the relayd. Continuing");
-			/* Continue here. There is nothing we can do for the relayd.*/
+			DBG("Unable to close stream on the relayd. Continuing");
+			/*
+			 * Continue here. There is nothing we can do for the relayd.
+			 * Chances are that the relayd has closed the socket so we just
+			 * continue cleaning up.
+			 */
 		}
 
 		/* Both conditions are met, we destroy the relayd. */
@@ -529,28 +533,19 @@ error:
  *
  * Return destination file descriptor or negative value on error.
  */
-int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
-		size_t data_size)
+static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
+		size_t data_size, struct consumer_relayd_sock_pair *relayd)
 {
 	int outfd = -1, ret;
-	struct consumer_relayd_sock_pair *relayd;
 	struct lttcomm_relayd_data_hdr data_hdr;
 
 	/* Safety net */
 	assert(stream);
+	assert(relayd);
 
 	/* Reset data header */
 	memset(&data_hdr, 0, sizeof(data_hdr));
 
-	rcu_read_lock();
-	/* Get relayd reference of the stream. */
-	relayd = consumer_find_relayd(stream->net_seq_idx);
-	if (relayd == NULL) {
-		/* Stream is either local or corrupted */
-		goto error;
-	}
-
-	DBG("Consumer found relayd socks with index %d", stream->net_seq_idx);
 	if (stream->metadata_flag) {
 		/* Caller MUST acquire the relayd control socket lock */
 		ret = relayd_send_metadata(&relayd->control_sock, data_size);
@@ -578,7 +573,6 @@ int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream,
 	}
 
 error:
-	rcu_read_unlock();
 	return outfd;
 }
 
@@ -1079,7 +1073,37 @@ void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx)
 }
 
 /*
- * Mmap the ring buffer, read it and write the data to the tracefile.
+ * Write the metadata stream id on the specified file descriptor.
+ */
+static int write_relayd_metadata_id(int fd,
+		struct lttng_consumer_stream *stream,
+		struct consumer_relayd_sock_pair *relayd)
+{
+	int ret;
+	uint64_t metadata_id;
+
+	metadata_id = htobe64(stream->relayd_stream_id);
+	do {
+		ret = write(fd, (void *) &metadata_id,
+				sizeof(stream->relayd_stream_id));
+	} while (ret < 0 && errno == EINTR);
+	if (ret < 0) {
+		PERROR("write metadata stream id");
+		goto end;
+	}
+	DBG("Metadata stream id %zu written before data",
+			stream->relayd_stream_id);
+
+end:
+	return ret;
+}
+
+/*
+ * Mmap the ring buffer, read it and write the data to the tracefile. This is a
+ * core function for writing trace buffers to either the local filesystem or
+ * the network.
+ *
+ * Careful review MUST be put if any changes occur!
  *
  * Returns the number of bytes written
  */
@@ -1092,7 +1116,6 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	off_t orig_offset = stream->out_fd_offset;
 	/* Default is on the disk */
 	int outfd = stream->out_fd;
-	uint64_t metadata_id;
 	struct consumer_relayd_sock_pair *relayd = NULL;
 
 	/* RCU lock for the relayd pointer */
@@ -1141,30 +1164,18 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			netlen += sizeof(stream->relayd_stream_id);
 		}
 
-		ret = consumer_handle_stream_before_relayd(stream, netlen);
+		ret = write_relayd_stream_header(stream, netlen, relayd);
 		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));
-				} while (ret < 0 && errno == EINTR);
+				ret = write_relayd_metadata_id(outfd, stream, relayd);
 				if (ret < 0) {
-					PERROR("write metadata stream id");
 					written = ret;
 					goto end;
 				}
-				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. */
@@ -1226,7 +1237,6 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 	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;
 
 	switch (consumer_data.type) {
@@ -1260,18 +1270,12 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		 */
 		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
 
-		metadata_id = htobe64(stream->relayd_stream_id);
-		do {
-			ret = write(ctx->consumer_thread_pipe[1], (void *) &metadata_id,
-					sizeof(stream->relayd_stream_id));
-		} while (ret < 0 && errno == EINTR);
+		ret = write_relayd_metadata_id(ctx->consumer_thread_pipe[1],
+				stream, relayd);
 		if (ret < 0) {
-			PERROR("write metadata stream id");
 			written = ret;
 			goto end;
 		}
-		DBG("Metadata stream id %zu written before data",
-				stream->relayd_stream_id);
 	}
 
 	while (len > 0) {
@@ -1302,15 +1306,13 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 				written -= sizeof(stream->relayd_stream_id);
 			}
 
-			ret = consumer_handle_stream_before_relayd(stream, ret_splice);
+			ret = write_relayd_stream_header(stream, ret_splice, relayd);
 			if (ret >= 0) {
 				/* Use the returned socket. */
 				outfd = ret;
 			} else {
-				if (outfd == -1) {
-					ERR("Remote relayd disconnected. Stopping");
-					goto end;
-				}
+				ERR("Remote relayd disconnected. Stopping");
+				goto end;
 			}
 		}