llquantize() support; version() support; primordial printf() support

README.md

@@ -126,6 +126,10 @@ induce any additional data processing.
 `consumer.aggwalk()` does not iterate over aggregation data in any guaranteed
 order, and may interleave aggregation variables and/or keys.
 
+### `consumer.version()`
+
+Returns the version string, as returned from `dtrace -V`.
+
 Examples
 --------
 

libdtrace.cc

@@ -28,6 +28,43 @@
 
 #include <dtrace.h>
 
+/*
+ * This is a tad unsightly:  if we didn't find the definition of the
+ * llquantize() aggregating action, we're going to redefine it here (along
+ * with its support cast of macros).  This allows node-libdtrace to operate
+ * on a machine that has llquantize(), even if it was compiled on a machine
+ * without the support.
+ */
+#ifndef DTRACEAGG_LLQUANTIZE
+
+#define	DTRACEAGG_LLQUANTIZE			(DTRACEACT_AGGREGATION + 9)
+
+#define	DTRACE_LLQUANTIZE_FACTORSHIFT		48
+#define	DTRACE_LLQUANTIZE_FACTORMASK		((uint64_t)UINT16_MAX << 48)
+#define	DTRACE_LLQUANTIZE_LOWSHIFT		32
+#define	DTRACE_LLQUANTIZE_LOWMASK		((uint64_t)UINT16_MAX << 32)
+#define	DTRACE_LLQUANTIZE_HIGHSHIFT		16
+#define	DTRACE_LLQUANTIZE_HIGHMASK		((uint64_t)UINT16_MAX << 16)
+#define	DTRACE_LLQUANTIZE_NSTEPSHIFT		0
+#define	DTRACE_LLQUANTIZE_NSTEPMASK		UINT16_MAX
+
+#define DTRACE_LLQUANTIZE_FACTOR(x)             \
+	(uint16_t)(((x) & DTRACE_LLQUANTIZE_FACTORMASK) >> \
+	DTRACE_LLQUANTIZE_FACTORSHIFT)
+
+#define DTRACE_LLQUANTIZE_LOW(x)                \
+        (uint16_t)(((x) & DTRACE_LLQUANTIZE_LOWMASK) >> \
+        DTRACE_LLQUANTIZE_LOWSHIFT)
+
+#define DTRACE_LLQUANTIZE_HIGH(x)               \
+        (uint16_t)(((x) & DTRACE_LLQUANTIZE_HIGHMASK) >> \
+        DTRACE_LLQUANTIZE_HIGHSHIFT)
+
+#define DTRACE_LLQUANTIZE_NSTEP(x)              \
+        (uint16_t)(((x) & DTRACE_LLQUANTIZE_NSTEPMASK) >> \
+        DTRACE_LLQUANTIZE_NSTEPSHIFT)
+#endif
+
 using namespace v8;
 using std::string;
 using std::vector;
@@ -45,11 +82,13 @@ class DTraceConsumer : node::ObjectWrap {
 	boolean_t valid(const dtrace_recdesc_t *);
 	const char *action(const dtrace_recdesc_t *, char *, int);
 	Local<Value> record(const dtrace_recdesc_t *, caddr_t);
+	Local<Object> probedesc(const dtrace_probedesc_t *);
 
 	Local<Array> *ranges_cached(dtrace_aggvarid_t);
 	Local<Array> *ranges_cache(dtrace_aggvarid_t, Local<Array> *);
 	Local<Array> *ranges_quantize(dtrace_aggvarid_t);
 	Local<Array> *ranges_lquantize(dtrace_aggvarid_t, uint64_t);
+	Local<Array> *ranges_llquantize(dtrace_aggvarid_t, uint64_t, int);
 
 	static int consume(const dtrace_probedata_t *data,
 	    const dtrace_recdesc_t *rec, void *arg);
@@ -65,6 +104,7 @@ class DTraceConsumer : node::ObjectWrap {
 	static Handle<Value> Setopt(const Arguments& args);
 	static Handle<Value> Go(const Arguments& args);
 	static Handle<Value> Stop(const Arguments& args);
+	static Handle<Value> Version(const Arguments& args);
 
 private:
 	dtrace_hdl_t *dtc_handle;
@@ -93,7 +133,7 @@ DTraceConsumer::DTraceConsumer() : node::ObjectWrap()
 	(void) dtrace_setopt(dtp, "bufsize", "4m");
 	(void) dtrace_setopt(dtp, "aggsize", "4m");
 
-	if (dtrace_handle_buffered(dtp, DTraceConsumer::bufhandler, NULL) == -1)
+	if (dtrace_handle_buffered(dtp, DTraceConsumer::bufhandler, this) == -1)
 		throw (dtrace_errmsg(dtp, dtrace_errno(dtp)));
 
 	dtc_ranges = NULL;
@@ -126,11 +166,11 @@ DTraceConsumer::Initialize(Handle<Object> target)
 	    DTraceConsumer::Consume);
 	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "aggwalk",
 	    DTraceConsumer::Aggwalk);
-	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "aggmin",
-	    DTraceConsumer::Aggmin);
-	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "aggmax",
-	    DTraceConsumer::Aggmax);
+	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "aggmin", DTraceConsumer::Aggmin);
+	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "aggmax", DTraceConsumer::Aggmax);
 	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "stop", DTraceConsumer::Stop);
+	NODE_SET_PROTOTYPE_METHOD(dtc_templ, "version",
+	    DTraceConsumer::Version);
 
 	target->Set(String::NewSymbol("Consumer"), dtc_templ->GetFunction());
 }
@@ -185,6 +225,7 @@ DTraceConsumer::action(const dtrace_recdesc_t *rec, char *buf, int size)
 		{ DTRACEAGG_STDDEV,	"stddev()" },
 		{ DTRACEAGG_QUANTIZE,	"quantize()" },
 		{ DTRACEAGG_LQUANTIZE,	"lquantize()" },
+		{ DTRACEAGG_LLQUANTIZE,	"llquantize()" },
 		{ DTRACEACT_NONE,	NULL },
 	};
 
@@ -410,13 +451,35 @@ DTraceConsumer::Stop(const Arguments& args)
 	return (Undefined());
 }
 
+Local<Object>
+DTraceConsumer::probedesc(const dtrace_probedesc_t *pd)
+{
+	Local<Object> probe = Object::New();
+	probe->Set(String::New("provider"), String::New(pd->dtpd_provider));
+	probe->Set(String::New("module"), String::New(pd->dtpd_mod));
+	probe->Set(String::New("function"), String::New(pd->dtpd_func));
+	probe->Set(String::New("name"), String::New(pd->dtpd_name));
+
+	return (probe);
+}
+
 int
 DTraceConsumer::bufhandler(const dtrace_bufdata_t *bufdata, void *arg)
 {
-	/*
-	 * We do nothing here -- but should we wish to ever support complete
-	 * dtrace(1) compatibility via node.js, we will need to do work here.
-	 */
+	dtrace_probedata_t *data = bufdata->dtbda_probe;
+	const dtrace_recdesc_t *rec = bufdata->dtbda_recdesc;
+	DTraceConsumer *dtc = (DTraceConsumer *)arg;
+
+	if (rec == NULL || rec->dtrd_action != DTRACEACT_PRINTF)
+		return (DTRACE_HANDLE_OK);
+
+	Local<Object> probe = dtc->probedesc(data->dtpda_pdesc);
+	Local<Object> record = Object::New();
+	record->Set(String::New("data"), String::New(bufdata->dtbda_buffered));
+	Local<Value> argv[2] = { probe, record };
+
+	dtc->dtc_callback->Call(dtc->dtc_args->This(), 2, argv);
+
 	return (DTRACE_HANDLE_OK);
 }
 
@@ -428,11 +491,7 @@ DTraceConsumer::consume(const dtrace_probedata_t *data,
 	dtrace_probedesc_t *pd = data->dtpda_pdesc;
 	Local<Value> datum;
 
-	Local<Object> probe = Object::New();
-	probe->Set(String::New("provider"), String::New(pd->dtpd_provider));
-	probe->Set(String::New("module"), String::New(pd->dtpd_mod));
-	probe->Set(String::New("function"), String::New(pd->dtpd_func));
-	probe->Set(String::New("name"), String::New(pd->dtpd_name));
+	Local<Object> probe = dtc->probedesc(data->dtpda_pdesc);
 
 	if (rec == NULL) {
 		Local<Value> argv[1] = { probe };
@@ -443,6 +502,12 @@ DTraceConsumer::consume(const dtrace_probedata_t *data,
 	if (!dtc->valid(rec)) {
 		char errbuf[256];
 
+		/*
+		 * If this is a printf(), we'll defer to the bufhandler.
+		 */
+		if (rec->dtrd_action == DTRACEACT_PRINTF)
+			return (DTRACE_CONSUME_THIS);
+
 		dtc->dtc_error = dtc->error("unsupported action %s "
 		    "in record for %s:%s:%s:%s\n",
 		    dtc->action(rec, errbuf, sizeof (errbuf)),
@@ -452,14 +517,12 @@ DTraceConsumer::consume(const dtrace_probedata_t *data,
 	}
 
 	Local<Object> record = Object::New();
-
 	record->Set(String::New("data"), dtc->record(rec, data->dtpda_data));
-
 	Local<Value> argv[2] = { probe, record };
 
 	dtc->dtc_callback->Call(dtc->dtc_args->This(), 2, argv);
 
-	return (rec == NULL ? DTRACE_CONSUME_NEXT : DTRACE_CONSUME_THIS);
+	return (DTRACE_CONSUME_THIS);
 }
 
 Handle<Value>
@@ -584,6 +647,59 @@ DTraceConsumer::ranges_lquantize(dtrace_aggvarid_t varid,
 	return (ranges_cache(varid, ranges));
 }
 
+Local<Array> *
+DTraceConsumer::ranges_llquantize(dtrace_aggvarid_t varid,
+    const uint64_t arg, int nbuckets)
+{
+	int64_t value = 1, next, step;
+	Local<Array> *ranges;
+	int bucket = 0, order;
+	uint16_t factor, low, high, nsteps;
+
+	if ((ranges = ranges_cached(varid)) != NULL)
+		return (ranges);
+
+	factor = DTRACE_LLQUANTIZE_FACTOR(arg);
+	low = DTRACE_LLQUANTIZE_LOW(arg);
+	high = DTRACE_LLQUANTIZE_HIGH(arg);
+	nsteps = DTRACE_LLQUANTIZE_NSTEP(arg);
+
+	ranges = new Local<Array>[nbuckets];
+
+	for (order = 0; order < low; order++)
+		value *= factor;
+
+	ranges[bucket] = Array::New(2);
+	ranges[bucket]->Set(0, Number::New(0));
+	ranges[bucket]->Set(1, Number::New(value - 1));
+	bucket++;
+
+	next = value * factor;
+	step = next > nsteps ? next / nsteps : 1;
+
+	while (order <= high) {
+		ranges[bucket] = Array::New(2);
+		ranges[bucket]->Set(0, Number::New(value));
+		ranges[bucket]->Set(1, Number::New(value + step - 1));
+		bucket++;
+
+		if ((value += step) != next)
+			continue;
+
+		next = value * factor;
+		step = next > nsteps ? next / nsteps : 1;
+		order++;
+	}
+
+	ranges[bucket] = Array::New(2);
+	ranges[bucket]->Set(0, Number::New(value));
+	ranges[bucket]->Set(1, Number::New(INT64_MAX));
+
+	assert(bucket + 1 == nbuckets);
+
+	return (ranges_cache(varid, ranges));
+}
+
 int
 DTraceConsumer::aggwalk(const dtrace_aggdata_t *agg, void *arg)
 {
@@ -665,19 +781,22 @@ DTraceConsumer::aggwalk(const dtrace_aggdata_t *agg, void *arg)
 		break;
 	}
 
-	case DTRACEAGG_LQUANTIZE: {
+	case DTRACEAGG_LQUANTIZE:
+	case DTRACEAGG_LLQUANTIZE: {
 		Local<Array> lquantize = Array::New();
 		const int64_t *data = (int64_t *)(agg->dtada_data +
 		    aggrec->dtrd_offset);
 		Local<Array> *ranges, datum;
 		int i, j = 0;
 
 		uint64_t arg = *data++;
-		uint16_t levels = DTRACE_LQUANTIZE_LEVELS(arg);
+		int levels = (aggrec->dtrd_size / sizeof (uint64_t)) - 1;
 
-		ranges = dtc->ranges_lquantize(aggdesc->dtagd_varid, arg);
+		ranges = (aggrec->dtrd_action == DTRACEAGG_LQUANTIZE ?
+		    dtc->ranges_lquantize(aggdesc->dtagd_varid, arg) :
+		    dtc->ranges_llquantize(aggdesc->dtagd_varid, arg, levels));
 
-		for (i = 0; i <= levels + 1; i++) {
+		for (i = 0; i < levels; i++) {
 			if (!data[i])
 				continue;
 
@@ -762,6 +881,12 @@ DTraceConsumer::Aggmax(const Arguments& args)
 	return (Number::New(INT64_MAX));
 }
 
+Handle<Value>
+DTraceConsumer::Version(const Arguments& args)
+{
+	return (String::New(_dtrace_version));
+}
+
 extern "C" void
 init (Handle<Object> target) 
 {

tests/test-llquantize.js

@@ -0,0 +1,101 @@
+var sys = require('sys');
+var libdtrace = require('libdtrace');
+var assert = require('assert');
+
+dtp = new libdtrace.Consumer();
+
+ver = dtp.version().split(' ')[2].split('.');
+
+if (parseInt(ver[0]) == 1 && parseInt(ver[1]) <= 8) {
+	sys.puts('llquantize() not present in version ' + dtp.version() +
+	    '; not testing.');
+	process.exit(0);
+}
+
+prog = 'BEGIN\n{\n';
+
+for (i = 0; i < 101; i++)
+	prog += '\t@ = llquantize(' + i + ', 10, 0, 1, 20);\n';
+
+prog += '}\n';
+
+dtp.strcompile(prog);
+
+dtp.go();
+
+dtp.aggwalk(function (varid,key, val) {
+	var expected = [
+	    [ [ 0, 0 ], 1 ],
+	    [ [ 1, 1 ], 1 ],
+	    [ [ 2, 2 ], 1 ],
+	    [ [ 3, 3 ], 1 ],
+	    [ [ 4, 4 ], 1 ],
+	    [ [ 5, 5 ], 1 ],
+	    [ [ 6, 6 ], 1 ],
+	    [ [ 7, 7 ], 1 ],
+	    [ [ 8, 8 ], 1 ],
+	    [ [ 9, 9 ], 1 ],
+	    [ [ 10, 14 ], 5 ],
+	    [ [ 15, 19 ], 5 ],
+	    [ [ 20, 24 ], 5 ],
+	    [ [ 25, 29 ], 5 ],
+	    [ [ 30, 34 ], 5 ],
+	    [ [ 35, 39 ], 5 ],
+	    [ [ 40, 44 ], 5 ],
+	    [ [ 45, 49 ], 5 ],
+	    [ [ 50, 54 ], 5 ],
+	    [ [ 55, 59 ], 5 ],
+	    [ [ 60, 64 ], 5 ],
+	    [ [ 65, 69 ], 5 ],
+	    [ [ 70, 74 ], 5 ],
+	    [ [ 75, 79 ], 5 ],
+	    [ [ 80, 84 ], 5 ],
+	    [ [ 85, 89 ], 5 ],
+	    [ [ 90, 94 ], 5 ],
+	    [ [ 95, 99 ], 5 ],
+	    [ [ 100, 9223372036854776000 ], 1 ],
+	];
+
+	assert.deepEqual(val, expected);
+});
+
+dtp = new libdtrace.Consumer();
+
+prog = 'BEGIN\n{\n';
+
+for (i = 0; i < 10100; i += 50)
+	prog += '\t@ = llquantize(' + i + ', 10, 2, 3, 10);\n';
+
+prog += '}\n';
+
+dtp.strcompile(prog);
+
+dtp.go();
+
+dtp.aggwalk(function (varid, key, val) {
+	var expected =
+	    [ [ [ 0, 99 ], 2 ],
+	    [ [ 100, 199 ], 2 ],
+	    [ [ 200, 299 ], 2 ],
+	    [ [ 300, 399 ], 2 ],
+	    [ [ 400, 499 ], 2 ],
+	    [ [ 500, 599 ], 2 ],
+	    [ [ 600, 699 ], 2 ],
+	    [ [ 700, 799 ], 2 ],
+	    [ [ 800, 899 ], 2 ],
+	    [ [ 900, 999 ], 2 ],
+	    [ [ 1000, 1999 ], 20 ],
+	    [ [ 2000, 2999 ], 20 ],
+	    [ [ 3000, 3999 ], 20 ],
+	    [ [ 4000, 4999 ], 20 ],
+	    [ [ 5000, 5999 ], 20 ],
+	    [ [ 6000, 6999 ], 20 ],
+	    [ [ 7000, 7999 ], 20 ],
+	    [ [ 8000, 8999 ], 20 ],
+	    [ [ 9000, 9999 ], 20 ],
+	    [ [ 10000, 9223372036854776000 ], 2 ]
+	];
+
+	assert.deepEqual(val, expected);
+});
+