dt_aggregate.c

Sun Solaris 10 中的 DTrace 组件的源代码。请参看: http://www.sun.com/software/solaris/observability.jsp

/*
 * Copyright 2005 Sun Microsystems, Inc.  All rights reserved.
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License, Version 1.0 only.
 * See the file usr/src/LICENSING.NOTICE in this distribution or
 * http://www.opensolaris.org/license/ for details.
 */

#pragma ident	"@(#)dt_aggregate.c	1.6	04/12/18 SMI"

#include <stdlib.h>
#include <strings.h>
#include <errno.h>
#include <unistd.h>
#include <dt_impl.h>
#include <assert.h>

#define	DTRACE_AHASHSIZE	32779		/* big 'ol prime */

static void
dt_aggregate_count(uint64_t *existing, uint64_t *new, size_t size)
{
	int i;

	for (i = 0; i < size / sizeof (uint64_t); i++)
		existing[i] = existing[i] + new[i];
}

static int
dt_aggregate_countcmp(uint64_t *lhs, uint64_t *rhs)
{
	uint64_t lvar = *lhs;
	uint64_t rvar = *rhs;

	if (lvar > rvar)
		return (1);

	if (lvar < rvar)
		return (-1);

	return (0);
}

/*ARGSUSED*/
static void
dt_aggregate_min(uint64_t *existing, uint64_t *new, size_t size)
{
	if (*new < *existing)
		*existing = *new;
}

/*ARGSUSED*/
static void
dt_aggregate_max(uint64_t *existing, uint64_t *new, size_t size)
{
	if (*new > *existing)
		*existing = *new;
}

static int
dt_aggregate_averagecmp(uint64_t *lhs, uint64_t *rhs)
{
	uint64_t lavg = lhs[0] ? (lhs[1] / lhs[0]) : 0;
	uint64_t ravg = rhs[0] ? (rhs[1] / rhs[0]) : 0;

	if (lavg > ravg)
		return (1);

	if (lavg < ravg)
		return (-1);

	return (0);
}

/*ARGSUSED*/
static void
dt_aggregate_lquantize(uint64_t *existing, uint64_t *new, size_t size)
{
	uint64_t arg = *existing++;
	uint16_t levels = DTRACE_LQUANTIZE_LEVELS(arg);
	int i;

	for (i = 0; i <= levels + 1; i++)
		existing[i] = existing[i] + new[i + 1];
}

static int64_t
dt_aggregate_lquantizedsum(uint64_t *lquanta)
{
	uint64_t arg = *lquanta++;
	int32_t base = DTRACE_LQUANTIZE_BASE(arg);
	uint16_t step = DTRACE_LQUANTIZE_STEP(arg);
	uint16_t levels = DTRACE_LQUANTIZE_LEVELS(arg), i;
	int64_t total = lquanta[0] * (base - 1);

	for (i = 0; i < levels; base += step, i++)
		total += lquanta[i + 1] * base;

	return (total + lquanta[levels + 1] * (base + 1));
}

static int
dt_aggregate_lquantizedcmp(uint64_t *lhs, uint64_t *rhs)
{
	int64_t lsum = dt_aggregate_lquantizedsum(lhs);
	int64_t rsum = dt_aggregate_lquantizedsum(rhs);

	if (lsum > rsum)
		return (1);

	if (lsum < rsum)
		return (-1);

	return (0);
}

static int
dt_aggregate_quantizedcmp(uint64_t *lhs, uint64_t *rhs)
{
	int nbuckets = DTRACE_QUANTIZE_NBUCKETS, i;
	int64_t ltotal = 0, rtotal = 0;

	for (i = 0; i < nbuckets; i++) {
		int64_t bucketval = DTRACE_QUANTIZE_BUCKETVAL(i);

		ltotal += bucketval * lhs[i];
		rtotal += bucketval * rhs[i];
	}

	if (ltotal > rtotal)
		return (1);

	if (ltotal < rtotal)
		return (-1);

	return (0);
}

static int
dt_aggregate_snap_cpu(dtrace_hdl_t *dtp, processorid_t cpu)
{
	dtrace_epid_t id;
	uint64_t hashval;
	size_t offs, roffs, size, ndx;
	int i, j, rval;
	caddr_t addr, data;
	dtrace_recdesc_t *rec;
	dt_aggregate_t *agp = &dtp->dt_aggregate;
	dtrace_aggdesc_t *agg;
	dt_ahash_t *hash = &agp->dtat_hash;
	dt_ahashent_t *h;
	dtrace_bufdesc_t b = agp->dtat_buf, *buf = &b;
	dtrace_aggdata_t *aggdata;
	int flags = agp->dtat_flags;

	buf->dtbd_cpu = cpu;

	if (dt_ioctl(dtp, DTRACEIOC_AGGSNAP, buf) == -1) {
		if (errno == ENOENT) {
			/*
			 * If that failed with ENOENT, it may be because the
			 * CPU was unconfigured.  This is okay; we'll just
			 * do nothing but return success.
			 */
			return (0);
		}

		return (dt_set_errno(dtp, errno));
	}

	if (buf->dtbd_drops != 0) {
		if (dt_handle_cpudrop(dtp, cpu,
		    DTRACEDROP_AGGREGATION, buf->dtbd_drops) == -1)
			return (-1);
	}

	if (buf->dtbd_size == 0)
		return (0);

	if (hash->dtah_hash == NULL) {
		size_t size;

		hash->dtah_size = DTRACE_AHASHSIZE;
		size = hash->dtah_size * sizeof (dt_ahashent_t *);

		if ((hash->dtah_hash = malloc(size)) == NULL)
			return (dt_set_errno(dtp, EDT_NOMEM));

		bzero(hash->dtah_hash, size);
	}

	for (offs = 0; offs < buf->dtbd_size; ) {
		/*
		 * We're guaranteed to have an ID.
		 */
		id = *((dtrace_epid_t *)((uintptr_t)buf->dtbd_data +
		    (uintptr_t)offs));

		if (id == DTRACE_AGGIDNONE) {
			/*
			 * This is filler to assure proper alignment of the
			 * next record; we simply ignore it.
			 */
			offs += sizeof (id);
			continue;
		}

		if ((rval = dt_aggid_lookup(dtp, id, &agg)) != 0)
			return (rval);

		addr = buf->dtbd_data + offs;
		size = agg->dtagd_size;
		hashval = 0;

		for (j = 0; j < agg->dtagd_nrecs - 1; j++) {
			rec = &agg->dtagd_rec[j];
			roffs = rec->dtrd_offset;

			for (i = 0; i < rec->dtrd_size; i++)
				hashval += addr[roffs + i];
		}

		ndx = hashval % hash->dtah_size;

		for (h = hash->dtah_hash[ndx]; h != NULL; h = h->dtahe_next) {
			if (h->dtahe_hashval != hashval)
				continue;

			if (h->dtahe_size != size)
				continue;

			aggdata = &h->dtahe_data;
			data = aggdata->dtada_data;

			for (j = 0; j < agg->dtagd_nrecs - 1; j++) {
				rec = &agg->dtagd_rec[j];
				roffs = rec->dtrd_offset;

				for (i = 0; i < rec->dtrd_size; i++)
					if (addr[roffs + i] != data[roffs + i])
						goto hashnext;
			}

			/*
			 * We found it.  Now we need to apply the aggregating
			 * action on the data here.
			 */
			rec = &agg->dtagd_rec[agg->dtagd_nrecs - 1];
			roffs = rec->dtrd_offset;
			/* LINTED - alignment */
			h->dtahe_aggregate((uint64_t *)&data[roffs],
			    /* LINTED - alignment */
			    (uint64_t *)&addr[roffs], rec->dtrd_size);

			/*
			 * If we're keeping per CPU data, apply the aggregating
			 * action there as well.
			 */
			if (aggdata->dtada_percpu != NULL) {
				data = aggdata->dtada_percpu[cpu];

				/* LINTED - alignment */
				h->dtahe_aggregate((uint64_t *)data,
				    /* LINTED - alignment */
				    (uint64_t *)&addr[roffs], rec->dtrd_size);
			}

			goto bufnext;
hashnext:
			continue;
		}

		/*
		 * If we're here, we couldn't find an entry for this record.
		 */
		if ((h = malloc(sizeof (dt_ahashent_t))) == NULL)
			return (dt_set_errno(dtp, EDT_NOMEM));
		bzero(h, sizeof (dt_ahashent_t));
		aggdata = &h->dtahe_data;

		if ((aggdata->dtada_data = malloc(size)) == NULL) {
			free(h);
			return (dt_set_errno(dtp, EDT_NOMEM));
		}

		bcopy(addr, aggdata->dtada_data, size);
		aggdata->dtada_size = size;
		aggdata->dtada_desc = agg;
		aggdata->dtada_handle = dtp;
		(void) dt_epid_lookup(dtp, agg->dtagd_epid,
		    &aggdata->dtada_edesc, &aggdata->dtada_pdesc);
		aggdata->dtada_normal = 1;

		h->dtahe_hashval = hashval;
		h->dtahe_size = size;

		rec = &agg->dtagd_rec[agg->dtagd_nrecs - 1];

		if (flags & DTRACE_A_PERCPU) {
			int max_cpus = agp->dtat_maxcpu;
			caddr_t *percpu = malloc(max_cpus * sizeof (caddr_t));

			if (percpu == NULL) {
				free(aggdata->dtada_data);
				free(h);
				return (dt_set_errno(dtp, EDT_NOMEM));
			}

			for (j = 0; j < max_cpus; j++) {
				percpu[j] = malloc(rec->dtrd_size);

				if (percpu[j] == NULL) {
					while (--j >= 0)
						free(percpu[j]);

					free(aggdata->dtada_data);
					free(h);
					return (dt_set_errno(dtp, EDT_NOMEM));
				}

				if (j == cpu) {
					bcopy(&addr[rec->dtrd_offset],
					    percpu[j], rec->dtrd_size);
				} else {
					bzero(percpu[j], rec->dtrd_size);
				}
			}

			aggdata->dtada_percpu = percpu;
		}

		switch (rec->dtrd_action) {
		case DTRACEAGG_MIN:
			h->dtahe_aggregate = dt_aggregate_min;
			break;

		case DTRACEAGG_MAX:
			h->dtahe_aggregate = dt_aggregate_max;
			break;

		case DTRACEAGG_LQUANTIZE:
			h->dtahe_aggregate = dt_aggregate_lquantize;
			break;

		case DTRACEAGG_COUNT:
		case DTRACEAGG_SUM:
		case DTRACEAGG_AVG:
		case DTRACEAGG_QUANTIZE:
			h->dtahe_aggregate = dt_aggregate_count;
			break;

		default:
			return (dt_set_errno(dtp, EDT_BADAGG));
		}

		if (hash->dtah_hash[ndx] != NULL)
			hash->dtah_hash[ndx]->dtahe_prev = h;

		h->dtahe_next = hash->dtah_hash[ndx];
		hash->dtah_hash[ndx] = h;

		if (hash->dtah_all != NULL)
			hash->dtah_all->dtahe_prevall = h;

		h->dtahe_nextall = hash->dtah_all;
		hash->dtah_all = h;
bufnext:
		offs += agg->dtagd_size;
	}

	return (0);
}

int
dtrace_aggregate_snap(dtrace_hdl_t *dtp)
{
	int i, rval;
	dt_aggregate_t *agp = &dtp->dt_aggregate;
	hrtime_t now = gethrtime();
	dtrace_optval_t interval = dtp->dt_options[DTRACEOPT_AGGRATE];

	if (dtp->dt_lastagg != 0) {
		if (now - dtp->dt_lastagg < interval)
			return (0);

		dtp->dt_lastagg += interval;
	} else {
		dtp->dt_lastagg = now;
	}

	if (!dtp->dt_active)
		return (dt_set_errno(dtp, EINVAL));

	if (agp->dtat_buf.dtbd_size == 0)
		return (0);

	for (i = 0; i < agp->dtat_ncpus; i++) {
		if (rval = dt_aggregate_snap_cpu(dtp, agp->dtat_cpus[i]))
			return (rval);
	}

	return (0);
}

static int
dt_aggregate_hashcmp(const void *lhs, const void *rhs)
{
	dt_ahashent_t *lh = *((dt_ahashent_t **)lhs);
	dt_ahashent_t *rh = *((dt_ahashent_t **)rhs);
	dtrace_aggdesc_t *lagg = lh->dtahe_data.dtada_desc;
	dtrace_aggdesc_t *ragg = rh->dtahe_data.dtada_desc;

	if (lagg->dtagd_nrecs < ragg->dtagd_nrecs)
		return (-1);

	if (lagg->dtagd_nrecs > ragg->dtagd_nrecs)
		return (1);

	return (0);
}

static int
dt_aggregate_varcmp(const void *lhs, const void *rhs)
{
	dt_ahashent_t *lh = *((dt_ahashent_t **)lhs);
	dt_ahashent_t *rh = *((dt_ahashent_t **)rhs);
	dtrace_aggdesc_t *lagg = lh->dtahe_data.dtada_desc;
	dtrace_aggdesc_t *ragg = rh->dtahe_data.dtada_desc;
	caddr_t ldata = lh->dtahe_data.dtada_data;
	caddr_t rdata = rh->dtahe_data.dtada_data;
	dtrace_recdesc_t *lrec, *rrec;
	uint64_t lid, rid;

	/*
	 * We know that we have a compiler-generated ID as the first record.
	 */
	lrec = lagg->dtagd_rec;
	rrec = ragg->dtagd_rec;

	lid = *((uint64_t *)(uintptr_t)(ldata + lrec->dtrd_offset));
	rid = *((uint64_t *)(uintptr_t)(rdata + rrec->dtrd_offset));

	if (lid < rid)
		return (-1);

	if (lid > rid)
		return (1);

	return (0);
}

static int
dt_aggregate_valcmp(const void *lhs, const void *rhs)
{
	dt_ahashent_t *lh = *((dt_ahashent_t **)lhs);
	dt_ahashent_t *rh = *((dt_ahashent_t **)rhs);
	dtrace_aggdesc_t *lagg = lh->dtahe_data.dtada_desc;
	dtrace_aggdesc_t *ragg = rh->dtahe_data.dtada_desc;
	caddr_t ldata = lh->dtahe_data.dtada_data;
	caddr_t rdata = rh->dtahe_data.dtada_data;
	dtrace_recdesc_t *lrec, *rrec;
	uint64_t *laddr, *raddr;
	int rval;

	if ((rval = dt_aggregate_hashcmp(lhs, rhs)) != 0)
		return (rval);

	lrec = &lagg->dtagd_rec[lagg->dtagd_nrecs - 1];
	rrec = &ragg->dtagd_rec[ragg->dtagd_nrecs - 1];

	if (lrec->dtrd_offset < rrec->dtrd_offset)
		return (-1);

	if (lrec->dtrd_offset > rrec->dtrd_offset)
		return (1);

	if (lrec->dtrd_action < rrec->dtrd_action)
		return (-1);

	if (lrec->dtrd_action > rrec->dtrd_action)
		return (1);

	laddr = (uint64_t *)(uintptr_t)(ldata + lrec->dtrd_offset);
	raddr = (uint64_t *)(uintptr_t)(rdata + rrec->dtrd_offset);

	switch (lrec->dtrd_action) {
	case DTRACEAGG_AVG:
		return (dt_aggregate_averagecmp(laddr, raddr));

	case DTRACEAGG_QUANTIZE:
		return (dt_aggregate_quantizedcmp(laddr, raddr));

	case DTRACEAGG_LQUANTIZE:
		return (dt_aggregate_lquantizedcmp(laddr, raddr));

	case DTRACEAGG_COUNT:
	case DTRACEAGG_SUM: