X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=27e96007c920a94ef94aaae8fefa21d7aa771ab8;hp=2c82836558287418a171f7e312c45d1fe362f459;hb=d9f0c7c7ce520c309f8e7148ecaff9aacbe52b55;hpb=45f1d9a160baa4e8245124fe9ebb3295376e07b8

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 2c8283655..27e96007c 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -3673,34 +3673,6 @@ error_nosignal:
 	}
 }
 
-/*
- * Try to lock the stream mutex.
- *
- * On success, 1 is returned else 0 indicating that the mutex is NOT lock.
- */
-static int stream_try_lock(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) {
-		/* For both EBUSY and EINVAL error, the mutex is NOT locked. */
-		ret = 0;
-		goto end;
-	}
-
-	ret = 1;
-
-end:
-	return ret;
-}
-
 /*
  * Search for a relayd associated to the session id and return the reference.
  *
@@ -3767,30 +3739,11 @@ int consumer_data_pending(uint64_t id)
 	/* Ease our life a bit */
 	ht = consumer_data.stream_list_ht;
 
-	relayd = find_relayd_by_session_id(id);
-	if (relayd) {
-		/* Send init command for data pending. */
-		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-		ret = relayd_begin_data_pending(&relayd->control_sock,
-				relayd->relayd_session_id);
-		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-		if (ret < 0) {
-			/* Communication error thus the relayd so no data pending. */
-			ERR("Relayd begin data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
-			lttng_consumer_cleanup_relayd(relayd);
-			goto data_not_pending;
-		}
-	}
-
 	cds_lfht_for_each_entry_duplicate(ht->ht,
 			ht->hash_fct(&id, lttng_ht_seed),
 			ht->match_fct, &id,
 			&iter.iter, stream, node_session_id.node) {
-		/* If this call fails, the stream is being used hence data pending. */
-		ret = stream_try_lock(stream);
-		if (!ret) {
-			goto data_pending;
-		}
+		pthread_mutex_lock(&stream->lock);
 
 		/*
 		 * A removed node from the hash table indicates that the stream has
@@ -3808,9 +3761,27 @@ int consumer_data_pending(uint64_t id)
 			}
 		}
 
-		/* Relayd check */
-		if (relayd) {
-			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		pthread_mutex_unlock(&stream->lock);
+	}
+
+	relayd = find_relayd_by_session_id(id);
+	if (relayd) {
+		unsigned int is_data_inflight = 0;
+
+		/* Send init command for data pending. */
+		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		ret = relayd_begin_data_pending(&relayd->control_sock,
+				relayd->relayd_session_id);
+		if (ret < 0) {
+			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+			/* Communication error thus the relayd so no data pending. */
+			goto data_not_pending;
+		}
+
+		cds_lfht_for_each_entry_duplicate(ht->ht,
+				ht->hash_fct(&id, lttng_ht_seed),
+				ht->match_fct, &id,
+				&iter.iter, stream, node_session_id.node) {
 			if (stream->metadata_flag) {
 				ret = relayd_quiescent_control(&relayd->control_sock,
 						stream->relayd_stream_id);
@@ -3819,27 +3790,19 @@ int consumer_data_pending(uint64_t id)
 						stream->relayd_stream_id,
 						stream->next_net_seq_num - 1);
 			}
-			if (ret < 0) {
+
+			if (ret == 1) {
+				pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+				goto data_pending;
+			} else if (ret < 0) {
 				ERR("Relayd data pending failed. Cleaning up relayd %" PRIu64".", relayd->net_seq_idx);
 				lttng_consumer_cleanup_relayd(relayd);
 				pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-				pthread_mutex_unlock(&stream->lock);
 				goto data_not_pending;
 			}
-			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-			if (ret == 1) {
-				pthread_mutex_unlock(&stream->lock);
-				goto data_pending;
-			}
 		}
-		pthread_mutex_unlock(&stream->lock);
-	}
 
-	if (relayd) {
-		unsigned int is_data_inflight = 0;
-
-		/* Send init command for data pending. */
-		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		/* Send end command for data pending. */
 		ret = relayd_end_data_pending(&relayd->control_sock,
 				relayd->relayd_session_id, &is_data_inflight);
 		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);