[PATCH 20/25] perf cs-etm: Drain pending packets before finishing trace blocks

[Leo Yan <leo.yan@arm.com>]

OpenCSD can consume the final input byte while returning WAIT with trace
output still pending. Stopping when buf_len reaches zero leaves that
output unprocessed before the next block resets the decoder.

Add cs_etm_decoder__drain_packets() to resume a waiting decoder with
OCSD_OP_FLUSH without supplying more input. Call it from
cs_etm__decode_data_block() after all input has been consumed and keep the
buffer active until draining completes.

Return a positive value while input or pending output remains, zero on
completion, and a negative error on failure. Update timeless decoding,
timestamp seeding and timestamped queue processing to use this contract.
Process queued packets even on the final call. Also drain pending output
before resetting the decoder when dumping raw packets.

Assisted-by: Codex:gpt-6
Signed-off-by: Leo Yan <leo.yan@arm.com>
---
 tools/perf/util/cs-etm-decoder/cs-etm-decoder.c | 17 +++++++
 tools/perf/util/cs-etm-decoder/cs-etm-decoder.h |  7 +++
 tools/perf/util/cs-etm.c                        | 59 ++++++++++++++++---------
 3 files changed, 61 insertions(+), 22 deletions(-)

diff --git a/tools/perf/util/cs-etm-decoder/cs-etm-decoder.c b/tools/perf/util/cs-etm-decoder/cs-etm-decoder.c
index 26940f1f1b0bf44c8671d80020597c672c361c1a..a7a6410481e5c99be14a951493092c145a8512ba 100644
--- a/tools/perf/util/cs-etm-decoder/cs-etm-decoder.c
+++ b/tools/perf/util/cs-etm-decoder/cs-etm-decoder.c
@@ -793,6 +793,23 @@ int cs_etm_decoder__process_data_block(struct cs_etm_decoder *decoder,
 	return ret;
 }
 
+int cs_etm_decoder__drain_packets(struct cs_etm_decoder *decoder)
+{
+	if (OCSD_DATA_RESP_IS_WAIT(decoder->prev_return))
+		decoder->prev_return = ocsd_dt_process_data(decoder->dcd_tree,
+							    OCSD_OP_FLUSH,
+							    0,
+							    0,
+							    NULL,
+							    NULL);
+
+	if (OCSD_DATA_RESP_IS_WAIT(decoder->prev_return))
+		return 1;
+
+	/* No further WAIT driven flushing is needed */
+	return OCSD_DATA_RESP_IS_CONT(decoder->prev_return) ? 0 : -EINVAL;
+}
+
 void cs_etm_decoder__free(struct cs_etm_decoder *decoder)
 {
 	if (!decoder)
diff --git a/tools/perf/util/cs-etm-decoder/cs-etm-decoder.h b/tools/perf/util/cs-etm-decoder/cs-etm-decoder.h
index 12c782fa6db285852f9f901735aa1f0e55b955e8..c187f3809ba0a350ca6ee549d2c0c67ed8e16d1e 100644
--- a/tools/perf/util/cs-etm-decoder/cs-etm-decoder.h
+++ b/tools/perf/util/cs-etm-decoder/cs-etm-decoder.h
@@ -91,6 +91,13 @@ int cs_etm_decoder__process_data_block(struct cs_etm_decoder *decoder,
 				       u64 indx, const u8 *buf,
 				       size_t len, size_t *consumed);
 
+/*
+ * Drain pending packets after consuming a data block.
+ * Process queued packets after each call, including the final call.
+ * Returns 1 for WAIT, 0 when done, or a negative error.
+ */
+int cs_etm_decoder__drain_packets(struct cs_etm_decoder *decoder);
+
 struct cs_etm_decoder *
 cs_etm_decoder__new(int num_cpu,
 		    struct cs_etm_decoder_params *d_params,
diff --git a/tools/perf/util/cs-etm.c b/tools/perf/util/cs-etm.c
index 8340b8a16fed363795fc78ffbc419661a61442a3..e78c8e089f38eae34faa12bbd2f899f1f72acce4 100644
--- a/tools/perf/util/cs-etm.c
+++ b/tools/perf/util/cs-etm.c
@@ -129,6 +129,7 @@ struct cs_etm_queue {
 	u8 pending_timestamp_chan_id;
 	enum cs_etm_format format;
 	u64 offset;
+	/* Cleared once the current block is fully decoded and drained */
 	const unsigned char *buf;
 	size_t buf_len, buf_used;
 	/* Conversion between traceID and index in traceid_queues array */
@@ -943,6 +944,12 @@ static void cs_etm__dump_event(struct cs_etm_queue *etmq,
 		buffer_used += consumed;
 	} while (buffer_used < buffer->size);
 
+	if (!ret) {
+		do {
+			ret = cs_etm_decoder__drain_packets(etmq->decoder);
+		} while (ret > 0);
+	}
+
 	cs_etm_decoder__reset(etmq->decoder);
 }
 
@@ -1319,7 +1326,7 @@ static int cs_etm__queue_first_cs_timestamp(struct cs_etm_auxtrace *etm,
 		 * trace for that block.
 		 */
 		ret = cs_etm__decode_data_block(etmq);
-		if (ret)
+		if (ret < 0)
 			goto out;
 
 		/*
@@ -2171,14 +2178,17 @@ static void cs_etm__flush_all_stack(struct cs_etm_queue *etmq)
  *			   if need be.
  * Returns:	< 0	if error
  *		= 0	if no more auxtrace_buffer to read
- *		> 0	if the current buffer isn't empty yet
+ *		> 0	if input or decoder output remains in the current block
  */
 static int cs_etm__get_data_block(struct cs_etm_queue *etmq)
 {
 	int ret;
 
-	/* The current block is not finished */
-	if (etmq->buf_len)
+	/*
+	 * A non-NULL buf indicates that packets are still pending.
+	 * Drain them before resetting for a new block.
+	 */
+	if (etmq->buf)
 		return 1;
 
 	ret = cs_etm__get_trace(etmq);
@@ -2555,11 +2565,22 @@ static int cs_etm__set_sample_flags(struct cs_etm_queue *etmq,
 	return 0;
 }
 
+/*
+ * Return 0 when decoding and draining are complete, 1 if another call is
+ * needed, or a negative error. Process queued packets even on return 0.
+ */
 static int cs_etm__decode_data_block(struct cs_etm_queue *etmq)
 {
-	int ret = 0;
+	int ret;
 	size_t processed = 0;
 
+	if (!etmq->buf_len) {
+		ret = cs_etm_decoder__drain_packets(etmq->decoder);
+		if (!ret)
+			etmq->buf = NULL;
+		return ret;
+	}
+
 	/*
 	 * Packets are decoded and added to the decoder's packet queue
 	 * until the decoder packet processing callback has requested that
@@ -2573,14 +2594,13 @@ static int cs_etm__decode_data_block(struct cs_etm_queue *etmq)
 						 etmq->buf_len,
 						 &processed);
 	if (ret)
-		goto out;
+		return ret;
 
 	etmq->offset += processed;
 	etmq->buf_used += processed;
 	etmq->buf_len -= processed;
 
-out:
-	return ret;
+	return 1;
 }
 
 static int cs_etm__process_traceid_queue(struct cs_etm_queue *etmq,
@@ -2689,7 +2709,7 @@ static int cs_etm__clear_all_traceid_queues(struct cs_etm_queue *etmq)
 
 static int cs_etm__run_timeless_decoder(struct cs_etm_queue *etmq)
 {
-	int idx, err;
+	int idx, err, pending;
 	struct cs_etm_traceid_queue *tidq;
 	struct int_node *inode;
 
@@ -2701,9 +2721,9 @@ static int cs_etm__run_timeless_decoder(struct cs_etm_queue *etmq)
 
 		/* Run trace decoder until the input buffer is consumed. */
 		do {
-			err = cs_etm__decode_data_block(etmq);
-			if (err)
-				return err;
+			pending = cs_etm__decode_data_block(etmq);
+			if (pending < 0)
+				return pending;
 
 			/*
 			 * Per-thread decoding uses a single traceID queue;
@@ -2717,7 +2737,7 @@ static int cs_etm__run_timeless_decoder(struct cs_etm_queue *etmq)
 				if (err)
 					return err;
 			}
-		} while (etmq->buf_len);
+		} while (pending);
 
 		intlist__for_each_entry(inode, etmq->traceid_queues_list) {
 			idx = (int)(intptr_t)inode->priv;
@@ -2884,26 +2904,21 @@ static int cs_etm__process_timestamped_queues(struct cs_etm_auxtrace *etm,
 		}
 
 		ret = cs_etm__decode_data_block(etmq);
-		if (ret)
+		if (ret < 0)
 			goto out;
 
 		cs_timestamp = cs_etm__etmq_get_timestamp(etmq, &trace_chan_id);
 
 		if (!cs_timestamp) {
 			/*
-			 * Function cs_etm__decode_data_block() returns when
-			 * there is no more traces to decode in the current
-			 * auxtrace_buffer OR when a timestamp has been
-			 * encountered on any of the traceID queues.  Since we
-			 * did not get a timestamp, there is no more traces to
-			 * process in this auxtrace_buffer.  As such empty and
-			 * flush all traceID queues.
+			 * No timestamp is available yet. Process the queued packets
+			 * before resuming input or draining the current block.
 			 */
 			ret = cs_etm__clear_all_traceid_queues(etmq);
 			if (ret)
 				goto out;
 
-			/* Fetch another auxtrace_buffer for this etmq */
+			/* Draining and fetch another auxtrace_buffer for this etmq */
 			goto refetch;
 		}
 

-- 
2.34.1