[PATCH 5/9] perf cs-etm: Split up cs_etm__process_timestamped_queues()

[Amir Ayupov <aaupov@fb.com>]

cs_etm__process_timestamped_queues() currently does three things: it seeds
the auxtrace heap with one entry per queue, it decodes until the heap is
empty, and it then walks every traceID queue to flush whatever is left in
the branch stacks. That is fine while the only caller is
cs_etm__flush_events(), which runs once, but it does not survive the
function being called repeatedly.

Seeding cannot be repeated because a queue that still holds a heap slot
would be seeded again, adding duplicate entries and growing the heap
without bound. Flushing cannot be repeated either, because ending a block
finalises state that later trace still needs.

Move both out. Seeding becomes cs_etm__update_queues(), gated on
queues.new_data so it only runs when new AUX data has been queued, with
etmq->on_heap tracking whether a queue currently occupies a heap slot;
this mirrors intel_pt_update_queues() and intel_pt_queue::on_heap.
Flushing becomes cs_etm__flush_timestamped_queues(). What remains is the
decode loop on its own, which a later patch can then drive incrementally.

No functional change: the sole caller performs the same three steps in the
same order.

Signed-off-by: Amir Ayupov <aaupov@fb.com>
---
 tools/perf/util/cs-etm.c | 71 ++++++++++++++++++++++++++++++++++------
 1 file changed, 61 insertions(+), 10 deletions(-)

diff --git a/tools/perf/util/cs-etm.c b/tools/perf/util/cs-etm.c
index 114b3cd2da495..4d895f11deb7f 100644
--- a/tools/perf/util/cs-etm.c
+++ b/tools/perf/util/cs-etm.c
@@ -136,9 +136,13 @@ struct cs_etm_queue {
 	 */
 	struct intlist *own_traceid_list;
 	u32 sink_id;
+	/* Whether this queue currently occupies a slot in etm->heap */
+	bool on_heap;
 };
 
+static int cs_etm__update_queues(struct cs_etm_auxtrace *etm);
 static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm);
+static int cs_etm__flush_timestamped_queues(struct cs_etm_auxtrace *etm);
 static int cs_etm__process_timeless_queues(struct cs_etm_auxtrace *etm,
 					   pid_t tid);
 static int cs_etm__get_data_block(struct cs_etm_queue *etmq);
@@ -939,6 +943,8 @@ static int cs_etm__flush_events(struct perf_session *session,
 	struct cs_etm_auxtrace *etm = container_of(session->auxtrace,
 						   struct cs_etm_auxtrace,
 						   auxtrace);
+	int ret;
+
 	if (dump_trace)
 		return 0;
 
@@ -953,7 +959,15 @@ static int cs_etm__flush_events(struct perf_session *session,
 		return cs_etm__process_timeless_queues(etm, -1);
 	}
 
-	return cs_etm__process_timestamped_queues(etm);
+	ret = cs_etm__update_queues(etm);
+	if (ret)
+		return ret;
+
+	ret = cs_etm__process_timestamped_queues(etm);
+	if (ret)
+		return ret;
+
+	return cs_etm__flush_timestamped_queues(etm);
 }
 
 static void cs_etm__free_traceid_queues(struct cs_etm_queue *etmq)
@@ -1330,6 +1344,8 @@ static int cs_etm__queue_first_cs_timestamp(struct cs_etm_auxtrace *etm,
 	 */
 	cs_queue_nr = TO_CS_QUEUE_NR(queue_nr, trace_chan_id);
 	ret = auxtrace_heap__add(&etm->heap, cs_queue_nr, cs_timestamp);
+	if (!ret)
+		etmq->on_heap = true;
 out:
 	return ret;
 }
@@ -2767,23 +2783,30 @@ static int cs_etm__process_timeless_queues(struct cs_etm_auxtrace *etm,
 	return 0;
 }
 
-static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm)
+/*
+ * Seed the heap with one entry from each queue that is not already
+ * represented in it, so that decoding proceeds in time order across all
+ * queues. Only queues that have newly queued data need to be considered.
+ */
+static int cs_etm__update_queues(struct cs_etm_auxtrace *etm)
 {
 	int ret = 0;
-	unsigned int cs_queue_nr, queue_nr, i;
-	u8 trace_chan_id;
-	u64 cs_timestamp;
-	struct auxtrace_queue *queue;
+	unsigned int i;
 	struct cs_etm_queue *etmq;
-	struct cs_etm_traceid_queue *tidq;
+
+	if (!etm->queues.new_data)
+		return 0;
+
+	etm->queues.new_data = false;
 
 	/*
 	 * Pre-populate the heap with one entry from each queue so that we can
-	 * start processing in time order across all queues.
+	 * start processing in time order across all queues. Skip queues that
+	 * already occupy a heap slot, otherwise they would be added twice.
 	 */
 	for (i = 0; i < etm->queues.nr_queues; i++) {
 		etmq = etm->queues.queue_array[i].priv;
-		if (!etmq)
+		if (!etmq || etmq->on_heap)
 			continue;
 
 		ret = cs_etm__queue_first_cs_timestamp(etm, etmq, i);
@@ -2791,6 +2814,19 @@ static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm)
 			return ret;
 	}
 
+	return ret;
+}
+
+static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm)
+{
+	int ret = 0;
+	unsigned int cs_queue_nr, queue_nr;
+	u8 trace_chan_id;
+	u64 cs_timestamp;
+	struct auxtrace_queue *queue;
+	struct cs_etm_queue *etmq;
+	struct cs_etm_traceid_queue *tidq;
+
 	while (1) {
 		if (!etm->heap.heap_cnt)
 			break;
@@ -2807,6 +2843,7 @@ static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm)
 		 * to process it.
 		 */
 		auxtrace_heap__pop(&etm->heap);
+		etmq->on_heap = false;
 
 		tidq  = cs_etm__etmq_get_traceid_queue(etmq, trace_chan_id);
 		if (!tidq) {
@@ -2874,7 +2911,21 @@ static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm)
 		 */
 		cs_queue_nr = TO_CS_QUEUE_NR(queue_nr, trace_chan_id);
 		ret = auxtrace_heap__add(&etm->heap, cs_queue_nr, cs_timestamp);
+		if (ret)
+			goto out;
+		etmq->on_heap = true;
 	}
+out:
+	return ret;
+}
+
+/* Flush any branch stack entries left over once all trace is decoded */
+static int cs_etm__flush_timestamped_queues(struct cs_etm_auxtrace *etm)
+{
+	int ret = 0;
+	unsigned int i;
+	struct cs_etm_queue *etmq;
+	struct cs_etm_traceid_queue *tidq;
 
 	for (i = 0; i < etm->queues.nr_queues; i++) {
 		struct int_node *inode;
@@ -2893,7 +2944,7 @@ static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm)
 				return ret;
 		}
 	}
-out:
+
 	return ret;
 }
 
-- 
2.52.0