X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=366f8550bf03774db161b6d70bd34765f9005869;hp=7006e1cb435fb47a4b628e863d24aa338d1728ae;hb=2527bf859786192dd530b408717960495f48d208;hpb=b35308203f7844f48542978fd00aab21f2057c17

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 7006e1cb4..366f8550b 100644
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -51,7 +51,6 @@
 
 extern struct lttng_consumer_global_data consumer_data;
 extern int consumer_poll_timeout;
-extern volatile int consumer_quit;
 
 /*
  * Free channel object and all streams associated with it. This MUST be used
@@ -1077,9 +1076,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 	DBG("UST consumer snapshot channel %" PRIu64, key);
 
 	cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
-		/* Are we at a position _before_ the first available packet ? */
-		bool before_first_packet = true;
-
 		health_code_update();
 
 		/* Lock stream because we are about to change its state. */
@@ -1151,7 +1147,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 		while (consumed_pos < produced_pos) {
 			ssize_t read_len;
 			unsigned long len, padded_len;
-			int lost_packet = 0;
 
 			health_code_update();
 
@@ -1165,15 +1160,7 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 				}
 				DBG("UST consumer get subbuf failed. Skipping it.");
 				consumed_pos += stream->max_sb_size;
-
-				/*
-				 * Start accounting lost packets only when we
-				 * already have extracted packets (to match the
-				 * content of the final snapshot).
-				 */
-				if (!before_first_packet) {
-					lost_packet = 1;
-				}
+				stream->chan->lost_packets++;
 				continue;
 			}
 
@@ -1209,16 +1196,6 @@ static int snapshot_channel(uint64_t key, char *path, uint64_t relayd_id,
 				goto error_close_stream;
 			}
 			consumed_pos += stream->max_sb_size;
-
-			/*
-			 * Only account lost packets located between
-			 * succesfully extracted packets (do not account before
-			 * and after since they are not visible in the
-			 * resulting snapshot).
-			 */
-			stream->chan->lost_packets += lost_packet;
-			lost_packet = 0;
-			before_first_packet = false;
 		}
 
 		/* Simply close the stream so we can use it on the next snapshot. */
@@ -1355,7 +1332,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 	case LTTNG_CONSUMER_ADD_RELAYD_SOCKET:
 	{
 		/* Session daemon status message are handled in the following call. */
-		ret = consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
+		consumer_add_relayd_socket(msg.u.relayd_sock.net_index,
 				msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll,
 				&msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id,
 				msg.u.relayd_sock.relayd_session_id);
@@ -1456,6 +1433,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		attr.read_timer_interval = msg.u.ask_channel.read_timer_interval;
 		attr.chan_id = msg.u.ask_channel.chan_id;
 		memcpy(attr.uuid, msg.u.ask_channel.uuid, sizeof(attr.uuid));
+		attr.blocking_timeout= msg.u.ask_channel.blocking_timeout;
 
 		/* Match channel buffer type to the UST abi. */
 		switch (msg.u.ask_channel.output) {
@@ -1502,8 +1480,17 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			consumer_timer_switch_start(channel, attr.switch_timer_interval);
 			attr.switch_timer_interval = 0;
 		} else {
+			int monitor_start_ret;
+
 			consumer_timer_live_start(channel,
 					msg.u.ask_channel.live_timer_interval);
+			monitor_start_ret = consumer_timer_monitor_start(
+					channel,
+					msg.u.ask_channel.monitor_timer_interval);
+			if (monitor_start_ret < 0) {
+				ERR("Starting channel monitoring timer failed");
+				goto end_channel_error;
+			}
 		}
 
 		health_code_update();
@@ -1526,6 +1513,9 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 			if (channel->live_timer_enabled == 1) {
 				consumer_timer_live_stop(channel);
 			}
+			if (channel->monitor_timer_enabled == 1) {
+				consumer_timer_monitor_stop(channel);
+			}
 			goto end_channel_error;
 		}
 
@@ -1984,7 +1974,7 @@ void *lttng_ustctl_get_mmap_base(struct lttng_consumer_stream *stream)
 }
 
 /*
- * Take a snapshot for a specific fd
+ * Take a snapshot for a specific stream.
  *
  * Returns 0 on success, < 0 on error
  */
@@ -1996,6 +1986,20 @@ int lttng_ustconsumer_take_snapshot(struct lttng_consumer_stream *stream)
 	return ustctl_snapshot(stream->ustream);
 }
 
+/*
+ * Sample consumed and produced positions for a specific stream.
+ *
+ * Returns 0 on success, < 0 on error.
+ */
+int lttng_ustconsumer_sample_snapshot_positions(
+		struct lttng_consumer_stream *stream)
+{
+	assert(stream);
+	assert(stream->ustream);
+
+	return ustctl_snapshot_sample_positions(stream->ustream);
+}
+
 /*
  * Get the produced position
  *