node.js echo/flood test

NetTestClient.js

var Net = require('net');
var Assert = require('assert');

var socket;

function randomInt(max) {
  return Math.min(Math.floor(Math.random() * max), max - 1);
}

console.log('Initializing test data...');
var payload_data = new Array(1024);
var i;
var s = '####';
for (i = 0; i < payload_data.length; i++) {
  payload_data[i] = new Buffer(s);

  s += '....';
}
console.log('Done.');

function mb(bytes) {
  return (bytes / 1024 / 1024).toFixed(1) + ' MB';
}

var send_count = 0;
var bytes_sent = 0;
var bytes_recv = 0;
var num_waiting_for_cb = 0;
var num_cbs = 0;

var last_time = Date.now();
var last_count = 0;
function printStats(force) {
  var now = Date.now();
  var dt = (now - last_time) || 1;
  var dc = send_count - last_count;
  if (dt > 1000 || dc > 5000 || force) {
    console.log(send_count + ' pkts, '
      + (dc * 1000 / dt).toFixed(1) + ' pkts/s, '
      + num_waiting_for_cb + ' waiting for cb, '
      + mb(bytes_sent - bytes_recv) + ' un-echo\'d, '
      + socket.bufferSize + ' buffered, '
      + mb(bytes_sent) + ' sent, '
      + mb(bytes_recv) + ' received, '
      + mb(process.memoryUsage().rss) + ' RSS');
    last_time = now;
    last_count = send_count;
  }
}

function sendRandomPayload(socket) {
  var data = payload_data[randomInt(payload_data.length)];
  var callback_called = false;
  ++num_waiting_for_cb;
  socket.write(data, function() {
    Assert.ok(!callback_called);
    callback_called = true;
    --num_waiting_for_cb;
    ++num_cbs;
  });
  bytes_sent += data.length;
  ++send_count;
  if (send_count % 100 === 0) {
    printStats();
  }
}

socket = new Net.Socket();
socket.connect(8080, '127.0.0.1');
socket.on('connect', function() {
  socket.setNoDelay(true);
  var events_received = 0;
  socket.on('error', function(e) {
    console.log('socket error ' + e);
    printStats(true);
  });
  socket.on('data', function(data) {
    ++events_received;
    if (events_received % 100 === 0) {
      printStats();
    }
    bytes_recv += data.length;
  });
  console.log('connected.');

  var stopped = 0;

  setInterval(function() {
    var ii;
    // Limit rate so we are also seeing server processing speed
    if (socket.bufferSize > 1000000) {
      stopped = 'buffer full';
      return;
    }
    // server so swamped it's not sending anything, that's bad.
    if (bytes_sent - bytes_recv > 5000000) {
      stopped = 'too much unecho\'d data';
      return;
    }
    stopped = 0;
    for (ii = 0; ii < 200; ++ii) {
      sendRandomPayload(socket);
    }
  }, 10);

  setInterval(function() {
    printStats(true);
    if (stopped) {
      console.log('Stopped = ' + stopped);
    }
    //global && global.gc && global.gc();
  }, 3333);
});

NetTestServer.js

var Net = require('net');
var Assert = require('assert');

var send_count = 0;
var num_waiting_for_cb = 0;
var num_cb = 0;
var last_time = Date.now();
var last_count = 0;
var last_cb = 0;

function printStats(force) {
  var now = Date.now();
  var dt = (now - last_time) || 1;
  var dc = send_count - last_count;
  var dcb = num_cb - last_cb;
  if (dt > 1000 || (dc + dcb > 1000) || force) {
    console.log(send_count + ' data events, '
      + (dc * 1000 / dt).toFixed(1) + ' events/s, '
      + num_waiting_for_cb + ' waiting for cb, '
      + (process.memoryUsage().rss / 1024 / 1024).toFixed(1) + ' MB RSS');
    last_time = now;
    last_count = send_count;
    last_cb = num_cb;
  }
}

function decNumWaiting() {
  --num_waiting_for_cb;
  ++num_cb;
  if (num_cb % 100 === 0) {
    printStats();
  }
}

function packetHandler(socket, data) {
  // just sending echo here
  ++send_count;
  ++num_waiting_for_cb;
  socket.write(data, decNumWaiting);
  if (send_count % 100 === 0) {
    printStats();
  }
  // do some work
  var ii;
  for (ii = 0; ii < 50*data.length; ++ii) {
  }
}

var net_server = Net.createServer();
net_server.on('connection', function(socket) {
  var addr = socket.remoteAddress + ':' + socket.remotePort;
  console.log('new connection from ' + addr);
  socket.setNoDelay(true);
  socket.on('data', function(data) {
    packetHandler(socket, data);
  });
  socket.on('close', function(had_error) {
    console.log('close from ' + addr);
    printStats(true);
  });
  socket.on('error', function(e) {
    console.log('socket error ' + e);
    printStats(true);
  });
});
net_server.on('error', function(e) {
  console.log('Socket error : ' + e);
});
net_server.listen(8080);

setInterval(printStats, 5000);