[DTrace-devel] [PATCH v2 43/61] Clear dt_aggregate each snapshot

[eugene.loh at oracle.com]

From: Eugene Loh <eugene.loh at oracle.com>

The producer (kernel) gathers aggregation data.  The consumer
(user space) snapshots (reads) the kernel data and maintains
its own copy in dt_aggregate.

In legacy DTrace, the consumer would snapshot incremental kernel
data and either initialize or update an entry in dt_aggregate,
depending on whether the corresponding entry already existed.

In the port to BPF, the kernel data is kept in BPF maps that are
up-to-date, not incremental since the last snapshot.  So a second
mechanism was added to decide whether to initialize or update a
dt_aggregate entry:  for the first CPU, even if an entry has been
found, new data is copied (rather than aggregated) in.

When we support aggregation keys, the new mechanism will not help.
It will be possible for a key to appear on some other CPU than
the first.

Nevertheless, we cannot blindly keep re-aggregating data that
is already in dt_aggregate.

Therefore, clear aggregation data in dt_aggregate's hash entries
at each snapshot.  Use dtrace_aggregate_clear() for this purpose.
Note that this function is apparently unused in both legacy DTrace
as well as the port to BPF.  Note also that the function needs to
be corrected to account for for min() and max()

Signed-off-by: Eugene Loh <eugene.loh at oracle.com>
---
 libdtrace/dt_aggregate.c | 41 ++++++++++++++++++++++++++--------------
 1 file changed, 27 insertions(+), 14 deletions(-)

diff --git a/libdtrace/dt_aggregate.c b/libdtrace/dt_aggregate.c
index db9fd11a..3cdd4ea4 100644
--- a/libdtrace/dt_aggregate.c
+++ b/libdtrace/dt_aggregate.c
@@ -417,13 +417,6 @@ typedef struct dt_snapstate {
 	agg_cpu_f	fun;
 } dt_snapstate_t;
 
-static void
-dt_agg_one_copy(dt_ident_t *aid, int64_t *dst, int64_t *src, uint_t realsz)
-{
-	src++;  /* skip latch sequence number */
-	memcpy(dst, src, realsz);
-}
-
 static void
 dt_agg_one_agg(dt_ident_t *aid, int64_t *dst, int64_t *src, uint_t realsz)
 {
@@ -475,6 +468,10 @@ dt_aggregate_snap_one(dt_idhash_t *dhp, dt_ident_t *aid, dt_snapstate_t *st)
 	/* point to the latch sequence number */
 	src = (int64_t *)(st->buf + aid->di_offset);
 
+	/* skip it if latch sequence number is 0 */
+	if (*src == 0)
+		return 0;
+
 	datasz = agg->dtagd_recs[nrecs - 1].dtrd_size;
 
 	/* See if we already have an entry for this aggregation. */
@@ -610,13 +607,15 @@ dtrace_aggregate_snap(dtrace_hdl_t *dtp)
 	if (agp->dtat_cpu_buf == NULL)
 		return 0;
 
+	dtrace_aggregate_clear(dtp);
+
 	rval = dt_bpf_map_lookup(dtp->dt_aggmap_fd, &key, agp->dtat_buf);
 	if (rval != 0)
 		return dt_set_errno(dtp, -rval);
 
 	for (i = 0; i < dtp->dt_conf.num_online_cpus; i++) {
 		rval = dt_aggregate_snap_cpu(dtp, dtp->dt_conf.cpus[i].cpu_id,
-		    i == 0 ? dt_agg_one_copy : dt_agg_one_agg);
+		    dt_agg_one_agg);
 		if (rval != 0)
 			return rval;
 	}
@@ -1788,13 +1787,27 @@ dtrace_aggregate_clear(dtrace_hdl_t *dtp)
 		rec = &aggdesc->dtagd_recs[aggdesc->dtagd_nrecs - 1];
 		data = &h->dtahe_data;
 
-		memset(&data->dtada_data[rec->dtrd_offset], 0, rec->dtrd_size);
-
-		if (data->dtada_percpu == NULL)
-			continue;
+		switch (rec->dtrd_action) {
+		case DT_AGG_MIN:
+			*((uint64_t*)(&data->dtada_data[rec->dtrd_offset])) = INT64_MAX;
+			if (data->dtada_percpu)
+				for (i = 0; i < max_cpus; i++)
+					*((uint64_t*)data->dtada_percpu[i]) = INT64_MAX;
+			break;
+		case DT_AGG_MAX:
+			*((uint64_t*)(&data->dtada_data[rec->dtrd_offset])) = INT64_MIN;
+			if (data->dtada_percpu)
+				for (i = 0; i < max_cpus; i++)
+					*((uint64_t*)data->dtada_percpu[i]) = INT64_MIN;
+			break;
+		default:
+			memset(&data->dtada_data[rec->dtrd_offset], 0, rec->dtrd_size);
+			if (data->dtada_percpu)
+				for (i = 0; i < max_cpus; i++)
+					memset(data->dtada_percpu[i], 0, rec->dtrd_size);
+			break;
+		}
 
-		for (i = 0; i < max_cpus; i++)
-			memset(data->dtada_percpu[i], 0, rec->dtrd_size);
 	}
 }
 
-- 
2.18.4