mcavage#13 fast could provide tracing hooks around message decoding

CHANGES.md

@@ -1,5 +1,9 @@
 # node-fast Changelog
 
+## 0.5.0
+
+- Add profiling hooks triggered when message decoding starts and stops
+
 ## 0.4.1
 
 - Update dtrace-provider to 0.3.0

lib/protocol/message_decoder.js

@@ -1,4 +1,5 @@
 // Copyright 2013 Mark Cavage.  All rights reserved.
+// Copyright 2015 Joyent, Inc.
 
 var Writable = require('readable-stream').Writable;
 var util = require('util');
@@ -78,6 +79,8 @@ function MessageDecoder() {
 
     this._buf = null;
     this._msg = null;
+    this._cbStart = null;
+    this._cbStop = null;
 }
 util.inherits(MessageDecoder, Writable);
 
@@ -97,7 +100,15 @@ MessageDecoder.prototype._write = function _write(buf, encoding, cb) {
     msg = this._msg || {};
 
     while (buf.length > 0) {
+        if (this._cbStart !== null) {
+            this._cbStart();
+        }
+
         if (!parseBuffer(buf, msg)) {
+            if (this._cbStop !== null) {
+                this._cbStop();
+            }
+
             this._buf = buf;
             this._msg = msg;
             cb();
@@ -116,6 +127,10 @@ MessageDecoder.prototype._write = function _write(buf, encoding, cb) {
             self.emit('error', new InvalidContentError(parse_err));
         }
 
+        if (this._cbStop !== null) {
+            this._cbStop();
+        }
+
         msg.start = process.hrtime();
         this.emit('message', msg);
 

lib/server.js

@@ -1,4 +1,5 @@
 // Copyright 2013 Mark Cavage.  All rights reserved.
+// Copyright 2015 Joyent, Inc.
 
 var domain = require('domain');
 var EventEmitter = require('events').EventEmitter;
@@ -67,6 +68,13 @@ function Server(options) {
     this._rpc = null;
     this.srv = net.createServer(this.onConnection.bind(this));
 
+    // Optional hooks to allow the consumer to account for time spent decoding
+    // messages:
+    assert.optionalFunc(options.cbStartDecode, 'cbStartDecode');
+    this._cbStartDecode = options.cbStartDecode || null;
+    assert.optionalFunc(options.cbStopDecode, 'cbStopDecode');
+    this._cbStopDecode = options.cbStopDecode || null;
+
     //-- Properties
     ['connections', 'maxConnections'].forEach(function (p) {
         self.__defineSetter__(p, function (v) {
@@ -97,6 +105,9 @@ Server.prototype.onConnection = function onConnection(conn) {
         encoder: messageEncoder
     });
 
+    messageDecoder._cbStart = this._cbStartDecode;
+    messageDecoder._cbStop = this._cbStopDecode;
+
     conn.msgs = {};
     conn.rpcDecoder = rpcDecoder;
 

package.json

@@ -4,11 +4,12 @@
         "Mike Harsch",
         "Nate Fitch",
         "Yunong Xiao",
-        "Patrick Mooney"
+        "Patrick Mooney",
+        "Joshua M. Clulow"
     ],
     "name": "fast",
     "homepage": "http://mcavage.github.com/node-fast",
-    "version": "0.4.1",
+    "version": "0.5.0",
     "repository": {
         "type": "git",
         "url": "git://github.com/mcavage/node-fast.git"

test/hooks.test.js

@@ -0,0 +1,84 @@
+// Copyright 2015 Joyent, Inc.
+
+var fast = require('../lib');
+var test = require('tape').test;
+
+
+
+////--- Globals
+
+var PORT = process.env.TEST_PORT || 12345;
+
+var client;
+var server;
+
+var start_count = 0;
+var stop_count = 0;
+
+
+///--- Tests
+
+test('hooks: setup', function (t) {
+    server = fast.createServer({
+        cbStartDecode: function () {
+            start_count++;
+        },
+        cbStopDecode: function () {
+            stop_count++;
+        }
+    });
+    server.rpc('testTiming', function (input, res) {
+        res.end({});
+    });
+    server.listen(PORT, function () {
+        client = fast.createClient({
+            host: 'localhost',
+            port: PORT
+        });
+        client.on('connect', function () {
+            t.end();
+        });
+    });
+});
+
+test('hooks: decode hooks must be called', function (t) {
+    t.plan(5);
+
+    var msg = [];
+    while (msg.length < 100000) {
+        msg.push(Math.floor(Math.random() * 1000));
+    }
+
+    var orig = start_count;
+    t.equal(start_count, stop_count);
+    t.equal(start_count, 0);
+
+    var req = client.rpc('testTiming', msg);
+    t.ok(req);
+
+    req.on('end', function () {
+        t.equal(start_count, stop_count);
+        t.ok(start_count > orig);
+        t.end();
+    });
+});
+
+test('hooks: teardown', function (t) {
+    var serverClosed = false;
+    var clientClosed = false;
+    function tryEnd() {
+        if (serverClosed && clientClosed) {
+            t.end();
+        }
+    }
+    server.on('close', function () {
+        serverClosed = true;
+        tryEnd();
+    });
+    client.on('close', function () {
+        clientClosed = true;
+        tryEnd();
+    });
+    client.close();
+    server.close();
+});