Implement aggrate-based aggregation snapshots

Some aggregation operations like clear() depend on aggregation snapshot
frequency.

Signed-off-by: Kris Van Hees <kris.van.hees@oracle.com>
Reviewed-by: Eugene Loh <eugene.loh@oracle.com>

libdtrace/dt_aggregate.c

@@ -469,7 +469,7 @@ dt_aggregate_clear_one(const dtrace_aggdata_t *agd, void *arg)
 		break;
 	}
 
-	return 0;
+	return DTRACE_AGGWALK_NEXT;
 }
 
 static int
@@ -608,15 +608,33 @@ dt_aggregate_snap_cpu(dtrace_hdl_t *dtp, processorid_t cpu, int fd)
 int
 dtrace_aggregate_snap(dtrace_hdl_t *dtp)
 {
+	dtrace_optval_t	interval = dtp->dt_options[DTRACEOPT_AGGRATE];
 	dt_aggregate_t	*agp = &dtp->dt_aggregate;
 	int		i, rval;
 
+	/* Has tracing started yet? */
+	if (!dtp->dt_active)
+		return dt_set_errno(dtp, EINVAL);
+
 	/*
 	 * If we do not have a buffer initialized, we will not be processing
 	 * aggregations, so there is nothing to be done here.
 	 */
 	if (agp->dtat_buf == NULL)
-		return 0;
+		return DTRACE_WORKSTATUS_OKAY;
+
+	/* Do not retrieve at a rate faster than 'aggrate'. */
+	if (interval > 0) {
+		hrtime_t	now = gethrtime();
+
+		if (dtp->dt_lastagg != 0) {
+			if (now - dtp->dt_lastagg < interval)
+				return DTRACE_WORKSTATUS_OKAY;
+
+			dtp->dt_lastagg += interval;
+		} else
+			dtp->dt_lastagg = now;
+	}
 
 	dtrace_aggregate_clear(dtp);
 
@@ -1726,12 +1744,7 @@ dtrace_aggregate_print(dtrace_hdl_t *dtp, FILE *fp,
 void
 dtrace_aggregate_clear(dtrace_hdl_t *dtp)
 {
-	dt_aggregate_t		*agp = &dtp->dt_aggregate;
-	dt_ahash_t		*hash = &agp->dtat_hash;
-	dt_ahashent_t		*h;
-
-	for (h = hash->dtah_all; h != NULL; h = h->dtahe_nextall)
-		dt_aggregate_clear_one(&h->dtahe_data, dtp);
+	dtrace_aggregate_walk(dtp, dt_aggregate_clear_one, dtp);
 }
 
 void

libdtrace/dt_consume.c

@@ -2987,7 +2987,6 @@ dtrace_consume(dtrace_hdl_t *dtp, FILE *fp, dtrace_consume_probe_f *pf,
 	       dtrace_consume_rec_f *rf, void *arg)
 {
 	dtrace_optval_t		interval = dtp->dt_options[DTRACEOPT_SWITCHRATE];
-	hrtime_t		now = gethrtime();
 	struct epoll_event	events[dtp->dt_conf.num_online_cpus];
 	int			drained = 0;
 	int			i, cnt;
@@ -2998,6 +2997,8 @@ dtrace_consume(dtrace_hdl_t *dtp, FILE *fp, dtrace_consume_probe_f *pf,
 		return dt_set_errno(dtp, EINVAL);
 
 	if (interval > 0) {
+		hrtime_t	now = gethrtime();
+
 		if (dtp->dt_lastswitch != 0) {
 			if (now - dtp->dt_lastswitch < interval)
 				return DTRACE_WORKSTATUS_OKAY;
@@ -3045,6 +3046,8 @@ dtrace_consume(dtrace_hdl_t *dtp, FILE *fp, dtrace_consume_probe_f *pf,
 		}
 	}
 
+	if (dtrace_aggregate_snap(dtp) == DTRACE_WORKSTATUS_ERROR)
+		return DTRACE_WORKSTATUS_ERROR;
 
 	/*
 	 * If dtp->dt_beganon is not -1, we did not process the BEGIN probe
@@ -3058,6 +3061,7 @@ dtrace_consume(dtrace_hdl_t *dtp, FILE *fp, dtrace_consume_probe_f *pf,
 			return rval;
 
 		/* Force data retrieval since BEGIN was processed. */
+		dtp->dt_lastagg = 0;
 		dtp->dt_lastswitch = 0;
 	}
 

libdtrace/dt_impl.h

@@ -392,6 +392,7 @@ struct dtrace_hdl {
 	dtrace_status_t dt_status[2]; /* status cache */
 	int dt_statusgen;	/* current status generation */
 	hrtime_t dt_lastswitch;	/* last switch of buffer data */
+	hrtime_t dt_lastagg;	/* last snapshot of aggregation data */
 	dt_list_t dt_spec_bufs_draining; /* List of spec bufs being drained */
 	dt_htab_t *dt_spec_bufs;/* spec ID -> list of dt_spec_bufs_head_t */
 	char *dt_sprintf_buf;	/* buffer for dtrace_sprintf() */

libdtrace/dt_open.c

@@ -800,6 +800,7 @@ dt_vopen(int version, int flags, int *errp,
 	 * Set the default data rates.
 	 */
 	dtp->dt_options[DTRACEOPT_SWITCHRATE] = 0;
+	dtp->dt_options[DTRACEOPT_AGGRATE] = 0;
 
 	dtp->dt_cpp_argv[0] = (char *)strbasename(dtp->dt_cpp_path);
 

libdtrace/dt_work.c

@@ -174,6 +174,7 @@ dtrace_work(dtrace_hdl_t *dtp, FILE *fp, dtrace_consume_probe_f *pfunc,
 	case DTRACE_STATUS_EXITED:
 	case DTRACE_STATUS_STOPPED:
 		dtp->dt_lastswitch = 0;
+		dtp->dt_lastagg = 0;
 		rval = DTRACE_WORKSTATUS_DONE;
 		break;
 	case DTRACE_STATUS_NONE: