X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=dbec177b7328468b278d0205598e04c574609b53;hb=f66c074ce00abc984ec7cb57a4fa31b1d9cffd72;hp=41f8ae53fe67adde7552b54dfb31aa89d4abedcb;hpb=6d805429e9cb049eb0c9205fcf742a53e3166caf;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 41f8ae53f..dbec177b7 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -47,9 +47,6 @@ struct lttng_consumer_global_data consumer_data = {
 	.type = LTTNG_CONSUMER_UNKNOWN,
 };
 
-/* timeout parameter, to control the polling thread grace period. */
-int consumer_poll_timeout = -1;
-
 /*
  * Flag to inform the polling thread to quit when all fd hung up. Updated by
  * the consumer_thread_receive_fds when it notices that all fds has hung up.
@@ -59,15 +56,12 @@ int consumer_poll_timeout = -1;
 volatile int consumer_quit;
 
 /*
- * The following two hash tables are visible by all threads which are separated
- * in different source files.
- *
  * Global hash table containing respectively metadata and data streams. The
  * stream element in this ht should only be updated by the metadata poll thread
  * for the metadata and the data poll thread for the data.
  */
-struct lttng_ht *metadata_ht;
-struct lttng_ht *data_ht;
+static struct lttng_ht *metadata_ht;
+static struct lttng_ht *data_ht;
 
 /*
  * Notify a thread pipe to poll back again. This usually means that some global
@@ -288,6 +282,8 @@ static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
 
 	assert(relayd);
 
+	DBG("Cleaning up relayd sockets");
+
 	/* Save the net sequence index before destroying the object */
 	netidx = relayd->net_seq_idx;
 
@@ -1557,7 +1553,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 				written = ret_splice;
 			}
 			/* Socket operation failed. We consider the relayd dead */
-			if (errno == EBADF) {
+			if (errno == EBADF || errno == EPIPE) {
 				WARN("Remote relayd disconnected. Stopping");
 				relayd_hang_up = 1;
 				goto write_error;
@@ -1941,7 +1937,7 @@ static void validate_endpoint_status_data_stream(void)
 	rcu_read_lock();
 	cds_lfht_for_each_entry(data_ht->ht, &iter.iter, stream, node.node) {
 		/* Validate delete flag of the stream */
-		if (!stream->endpoint_status) {
+		if (stream->endpoint_status != CONSUMER_ENDPOINT_INACTIVE) {
 			continue;
 		}
 		/* Delete it right now */
@@ -2055,7 +2051,10 @@ restart:
 					 * since their might be data to consume.
 					 */
 					lttng_poll_del(&events, ctx->consumer_metadata_pipe[0]);
-					close(ctx->consumer_metadata_pipe[0]);
+					ret = close(ctx->consumer_metadata_pipe[0]);
+					if (ret < 0) {
+						PERROR("close metadata pipe");
+					}
 					continue;
 				} else if (revents & LPOLLIN) {
 					do {
@@ -2248,7 +2247,7 @@ void *consumer_thread_data_poll(void *data)
 		/* poll on the array of fds */
 	restart:
 		DBG("polling on %d fd", nb_fd + 1);
-		num_rdy = poll(pollfd, nb_fd + 1, consumer_poll_timeout);
+		num_rdy = poll(pollfd, nb_fd + 1, -1);
 		DBG("poll num_rdy : %d", num_rdy);
 		if (num_rdy == -1) {
 			/*
@@ -2307,6 +2306,9 @@ void *consumer_thread_data_poll(void *data)
 
 		/* Take care of high priority channels first. */
 		for (i = 0; i < nb_fd; i++) {
+			if (local_stream[i] == NULL) {
+				continue;
+			}
 			if (pollfd[i].revents & POLLPRI) {
 				DBG("Urgent read on fd %d", pollfd[i].fd);
 				high_prio = 1;
@@ -2315,6 +2317,7 @@ void *consumer_thread_data_poll(void *data)
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean the stream and free it. */
 					consumer_del_stream(local_stream[i], data_ht);
+					local_stream[i] = NULL;
 				} else if (len > 0) {
 					local_stream[i]->data_read = 1;
 				}
@@ -2331,6 +2334,9 @@ void *consumer_thread_data_poll(void *data)
 
 		/* Take care of low priority channels. */
 		for (i = 0; i < nb_fd; i++) {
+			if (local_stream[i] == NULL) {
+				continue;
+			}
 			if ((pollfd[i].revents & POLLIN) ||
 					local_stream[i]->hangup_flush_done) {
 				DBG("Normal read on fd %d", pollfd[i].fd);
@@ -2339,6 +2345,7 @@ void *consumer_thread_data_poll(void *data)
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean the stream and free it. */
 					consumer_del_stream(local_stream[i], data_ht);
+					local_stream[i] = NULL;
 				} else if (len > 0) {
 					local_stream[i]->data_read = 1;
 				}
@@ -2347,12 +2354,15 @@ void *consumer_thread_data_poll(void *data)
 
 		/* Handle hangup and errors */
 		for (i = 0; i < nb_fd; i++) {
+			if (local_stream[i] == NULL) {
+				continue;
+			}
 			if (!local_stream[i]->hangup_flush_done
 					&& (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
 					&& (consumer_data.type == LTTNG_CONSUMER32_UST
 						|| consumer_data.type == LTTNG_CONSUMER64_UST)) {
 				DBG("fd %d is hup|err|nval. Attempting flush and read.",
-					pollfd[i].fd);
+						pollfd[i].fd);
 				lttng_ustconsumer_on_stream_hangup(local_stream[i]);
 				/* Attempt read again, for the data we just flushed. */
 				local_stream[i]->data_read = 1;
@@ -2366,22 +2376,27 @@ void *consumer_thread_data_poll(void *data)
 				DBG("Polling fd %d tells it has hung up.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
 					consumer_del_stream(local_stream[i], data_ht);
+					local_stream[i] = NULL;
 					num_hup++;
 				}
 			} else if (pollfd[i].revents & POLLERR) {
 				ERR("Error returned in polling fd %d.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
 					consumer_del_stream(local_stream[i], data_ht);
+					local_stream[i] = NULL;
 					num_hup++;
 				}
 			} else if (pollfd[i].revents & POLLNVAL) {
 				ERR("Polling fd %d tells fd is not open.", pollfd[i].fd);
 				if (!local_stream[i]->data_read) {
 					consumer_del_stream(local_stream[i], data_ht);
+					local_stream[i] = NULL;
 					num_hup++;
 				}
 			}
-			local_stream[i]->data_read = 0;
+			if (local_stream[i] != NULL) {
+				local_stream[i]->data_read = 0;
+			}
 		}
 	}
 end:
@@ -2403,7 +2418,10 @@ end:
 	 * only tracked fd in the poll set. The thread will take care of closing
 	 * the read side.
 	 */
-	close(ctx->consumer_metadata_pipe[1]);
+	ret = close(ctx->consumer_metadata_pipe[1]);
+	if (ret < 0) {
+		PERROR("close data pipe");
+	}
 
 	if (data_ht) {
 		destroy_data_stream_ht(data_ht);
@@ -2516,13 +2534,6 @@ end:
 	 */
 	consumer_quit = 1;
 
-	/*
-	 * 2s of grace period, if no polling events occur during
-	 * this period, the polling thread will exit even if there
-	 * are still open FDs (should not happen, but safety mechanism).
-	 */
-	consumer_poll_timeout = LTTNG_CONSUMER_POLL_TIMEOUT;
-
 	/*
 	 * Notify the data poll thread to poll back again and test the
 	 * consumer_quit state that we just set so to quit gracefully.
@@ -2630,7 +2641,10 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		}
 
 		/* Close the created socket fd which is useless */
-		close(relayd->control_sock.fd);
+		ret = close(relayd->control_sock.fd);
+		if (ret < 0) {
+			PERROR("close relayd control socket");
+		}
 
 		/* Assign new file descriptor */
 		relayd->control_sock.fd = fd;
@@ -2644,7 +2658,10 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		}
 
 		/* Close the created socket fd which is useless */
-		close(relayd->data_sock.fd);
+		ret = close(relayd->data_sock.fd);
+		if (ret < 0) {
+			PERROR("close relayd control socket");
+		}
 
 		/* Assign new file descriptor */
 		relayd->data_sock.fd = fd;
@@ -2736,7 +2753,7 @@ int consumer_data_pending(uint64_t id)
 	ht = consumer_data.stream_list_ht;
 
 	cds_lfht_for_each_entry_duplicate(ht->ht,
-			ht->hash_fct((void *)((unsigned long) id), 0x42UL),
+			ht->hash_fct((void *)((unsigned long) id), lttng_ht_seed),
 			ht->match_fct, (void *)((unsigned long) id),
 			&iter.iter, stream, node_session_id.node) {
 		/* If this call fails, the stream is being used hence data pending. */