Merge pull request #4321 from murgatroid99/node_benchmark_service

Node benchmark service

package.json

@@ -39,7 +39,8 @@
     "minimist": "^1.1.0",
     "mocha": "~1.21.0",
     "mocha-jenkins-reporter": "^0.1.9",
-    "mustache": "^2.0.0"
+    "mustache": "^2.0.0",
+    "poisson-process": "^0.2.1"
   },
   "engines": {
     "node": ">=0.10.13"

src/node/performance/benchmark_client.js

+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/**
+ * Benchmark client module
+ * @module
+ */
+
+'use strict';
+
+var fs = require('fs');
+var path = require('path');
+var util = require('util');
+var EventEmitter = require('events');
+var _ = require('lodash');
+var PoissonProcess = require('poisson-process');
+var Histogram = require('./histogram');
+var grpc = require('../../../');
+var serviceProto = grpc.load({
+  root: __dirname + '/../../..',
+  file: 'test/proto/benchmarks/services.proto'}).grpc.testing;
+
+/**
+ * Create a buffer filled with size zeroes
+ * @param {number} size The length of the buffer
+ * @return {Buffer} The new buffer
+ */
+function zeroBuffer(size) {
+  var zeros = new Buffer(size);
+  zeros.fill(0);
+  return zeros;
+}
+
+/**
+ * Convert a time difference, as returned by process.hrtime, to a number of
+ * nanoseconds.
+ * @param {Array.<number>} time_diff The time diff, represented as
+ *     [seconds, nanoseconds]
+ * @return {number} The total number of nanoseconds
+ */
+function timeDiffToNanos(time_diff) {
+  return time_diff[0] * 1e9 + time_diff[1];
+}
+
+/**
+ * The BenchmarkClient class. Opens channels to servers and makes RPCs based on
+ * parameters from the driver, and records statistics about those RPCs.
+ * @param {Array.<string>} server_targets List of servers to connect to
+ * @param {number} channels The total number of channels to open
+ * @param {Object} histogram_params Options for setting up the histogram
+ * @param {Object=} security_params Options for TLS setup. If absent, don't use
+ *     TLS
+ */
+function BenchmarkClient(server_targets, channels, histogram_params,
+    security_params) {
+  var options = {};
+  var creds;
+  if (security_params) {
+    var ca_path;
+    if (security_params.use_test_ca) {
+      ca_path = path.join(__dirname, '../test/data/ca.pem');
+      var ca_data = fs.readFileSync(ca_path);
+      creds = grpc.credentials.createSsl(ca_data);
+    } else {
+      creds = grpc.credentials.createSsl();
+    }
+    if (security_params.server_host_override) {
+      var host_override = security_params.server_host_override;
+      options['grpc.ssl_target_name_override'] = host_override;
+      options['grpc.default_authority'] = host_override;
+    }
+  } else {
+    creds = grpc.credentials.createInsecure();
+  }
+
+  this.clients = [];
+
+  for (var i = 0; i < channels; i++) {
+    this.clients[i] = new serviceProto.BenchmarkService(
+        server_targets[i % server_targets.length], creds, options);
+  }
+
+  this.histogram = new Histogram(histogram_params.resolution,
+                                 histogram_params.max_possible);
+
+  this.running = false;
+
+  this.pending_calls = 0;
+};
+
+util.inherits(BenchmarkClient, EventEmitter);
+
+/**
+ * Start a closed-loop test. For each channel, start
+ * outstanding_rpcs_per_channel RPCs. Then, whenever an RPC finishes, start
+ * another one.
+ * @param {number} outstanding_rpcs_per_channel Number of RPCs to start per
+ *     channel
+ * @param {string} rpc_type Which method to call. Should be 'UNARY' or
+ *     'STREAMING'
+ * @param {number} req_size The size of the payload to send with each request
+ * @param {number} resp_size The size of payload to request be sent in responses
+ */
+BenchmarkClient.prototype.startClosedLoop = function(
+    outstanding_rpcs_per_channel, rpc_type, req_size, resp_size) {
+  var self = this;
+
+  self.running = true;
+
+  self.last_wall_time = process.hrtime();
+
+  var makeCall;
+
+  var argument = {
+    response_size: resp_size,
+    payload: {
+      body: zeroBuffer(req_size)
+    }
+  };
+
+  if (rpc_type == 'UNARY') {
+    makeCall = function(client) {
+      if (self.running) {
+        self.pending_calls++;
+        var start_time = process.hrtime();
+        client.unaryCall(argument, function(error, response) {
+          if (error) {
+            self.emit('error', new Error('Client error: ' + error.message));
+            self.running = false;
+            return;
+          }
+          var time_diff = process.hrtime(start_time);
+          self.histogram.add(timeDiffToNanos(time_diff));
+          makeCall(client);
+          self.pending_calls--;
+          if ((!self.running) && self.pending_calls == 0) {
+            self.emit('finished');
+          }
+        });
+      }
+    };
+  } else {
+    makeCall = function(client) {
+      if (self.running) {
+        self.pending_calls++;
+        var start_time = process.hrtime();
+        var call = client.streamingCall();
+        call.write(argument);
+        call.on('data', function() {
+        });
+        call.on('end', function() {
+          var time_diff = process.hrtime(start_time);
+          self.histogram.add(timeDiffToNanos(time_diff));
+          makeCall(client);
+          self.pending_calls--;
+          if ((!self.running) && self.pending_calls == 0) {
+            self.emit('finished');
+          }
+        });
+        call.on('error', function(error) {
+          self.emit('error', new Error('Client error: ' + error.message));
+          self.running = false;
+        });
+      }
+    };
+  }
+
+  _.each(self.clients, function(client) {
+    _.times(outstanding_rpcs_per_channel, function() {
+      makeCall(client);
+    });
+  });
+};
+
+/**
+ * Start a poisson test. For each channel, this initiates a number of Poisson
+ * processes equal to outstanding_rpcs_per_channel, where each Poisson process
+ * has the load parameter offered_load.
+ * @param {number} outstanding_rpcs_per_channel Number of RPCs to start per
+ *     channel
+ * @param {string} rpc_type Which method to call. Should be 'UNARY' or
+ *     'STREAMING'
+ * @param {number} req_size The size of the payload to send with each request
+ * @param {number} resp_size The size of payload to request be sent in responses
+ * @param {number} offered_load The load parameter for the Poisson process
+ */
+BenchmarkClient.prototype.startPoisson = function(
+    outstanding_rpcs_per_channel, rpc_type, req_size, resp_size, offered_load) {
+  var self = this;
+
+  self.running = true;
+
+  self.last_wall_time = process.hrtime();
+
+  var makeCall;
+
+  var argument = {
+    response_size: resp_size,
+    payload: {
+      body: zeroBuffer(req_size)
+    }
+  };
+
+  if (rpc_type == 'UNARY') {
+    makeCall = function(client, poisson) {
+      if (self.running) {
+        self.pending_calls++;
+        var start_time = process.hrtime();
+        client.unaryCall(argument, function(error, response) {
+          if (error) {
+            self.emit('error', new Error('Client error: ' + error.message));
+            self.running = false;
+            return;
+          }
+          var time_diff = process.hrtime(start_time);
+          self.histogram.add(timeDiffToNanos(time_diff));
+          self.pending_calls--;
+          if ((!self.running) && self.pending_calls == 0) {
+            self.emit('finished');
+          }
+        });
+      } else {
+        poisson.stop();
+      }
+    };
+  } else {
+    makeCall = function(client, poisson) {
+      if (self.running) {
+        self.pending_calls++;
+        var start_time = process.hrtime();
+        var call = client.streamingCall();
+        call.write(argument);
+        call.on('data', function() {
+        });
+        call.on('end', function() {
+          var time_diff = process.hrtime(start_time);
+          self.histogram.add(timeDiffToNanos(time_diff));
+          self.pending_calls--;
+          if ((!self.running) && self.pending_calls == 0) {
+            self.emit('finished');
+          }
+        });
+        call.on('error', function(error) {
+          self.emit('error', new Error('Client error: ' + error.message));
+          self.running = false;
+        });
+      } else {
+        poisson.stop();
+      }
+    };
+  }
+
+  var averageIntervalMs = (1 / offered_load) * 1000;
+
+  _.each(self.clients, function(client) {
+    _.times(outstanding_rpcs_per_channel, function() {
+      var p = PoissonProcess.create(averageIntervalMs, function() {
+        makeCall(client, p);
+      });
+      p.start();
+    });
+  });
+};
+
+/**
+ * Return curent statistics for the client. If reset is set, restart
+ * statistic collection.
+ * @param {boolean} reset Indicates that statistics should be reset
+ * @return {object} Client statistics
+ */
+BenchmarkClient.prototype.mark = function(reset) {
+  var wall_time_diff = process.hrtime(this.last_wall_time);
+  var histogram = this.histogram;
+  if (reset) {
+    this.last_wall_time = process.hrtime();
+    this.histogram = new Histogram(histogram.resolution,
+                                   histogram.max_possible);
+  }
+
+  return {
+    latencies: {
+      bucket: histogram.getContents(),
+      min_seen: histogram.minimum(),
+      max_seen: histogram.maximum(),
+      sum: histogram.getSum(),
+      sum_of_squares: histogram.sumOfSquares(),
+      count: histogram.getCount()
+    },
+    time_elapsed: wall_time_diff[0] + wall_time_diff[1] / 1e9,
+    // Not sure how to measure these values
+    time_user: 0,
+    time_system: 0
+  };
+};
+
+/**
+ * Stop the clients.
+ * @param {function} callback Called when the clients have finished shutting
+ *     down
+ */
+BenchmarkClient.prototype.stop = function(callback) {
+  this.running = false;
+  this.on('finished', callback);
+};
+
+module.exports = BenchmarkClient;

src/node/performance/benchmark_server.js

+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/**
+ * Benchmark server module
+ * @module
+ */
+
+'use strict';
+
+var fs = require('fs');
+var path = require('path');
+
+var grpc = require('../../../');
+var serviceProto = grpc.load({
+  root: __dirname + '/../../..',
+  file: 'test/proto/benchmarks/services.proto'}).grpc.testing;
+
+/**
+ * Create a buffer filled with size zeroes
+ * @param {number} size The length of the buffer
+ * @return {Buffer} The new buffer
+ */
+function zeroBuffer(size) {
+  var zeros = new Buffer(size);
+  zeros.fill(0);
+  return zeros;
+}
+
+/**
+ * Handler for the unary benchmark method. Simply responds with a payload
+ * containing the requested number of zero bytes.
+ * @param {Call} call The call object to be handled
+ * @param {function} callback The callback to call with the response
+ */
+function unaryCall(call, callback) {
+  var req = call.request;
+  var payload = {body: zeroBuffer(req.response_size)};
+  callback(null, {payload: payload});
+}
+
+/**
+ * Handler for the streaming benchmark method. Simply responds to each request
+ * with a payload containing the requested number of zero bytes.
+ * @param {Call} call The call object to be handled
+ */
+function streamingCall(call) {
+  call.on('data', function(value) {
+    var payload = {body: zeroBuffer(value.repsonse_size)};
+    call.write({payload: payload});
+  });
+  call.on('end', function() {
+    call.end();
+  });
+}
+
+/**
+ * BenchmarkServer class. Constructed based on parameters from the driver and
+ * stores statistics.
+ * @param {string} host The host to serve on
+ * @param {number} port The port to listen to
+ * @param {tls} Indicates whether TLS should be used
+ */
+function BenchmarkServer(host, port, tls) {
+  var server_creds;
+  var host_override;
+  if (tls) {
+    var key_path = path.join(__dirname, '../test/data/server1.key');
+    var pem_path = path.join(__dirname, '../test/data/server1.pem');
+
+    var key_data = fs.readFileSync(key_path);
+    var pem_data = fs.readFileSync(pem_path);
+    server_creds = grpc.ServerCredentials.createSsl(null,
+                                                    [{private_key: key_data,
+                                                      cert_chain: pem_data}]);
+  } else {
+    server_creds = grpc.ServerCredentials.createInsecure();
+  }
+
+  var server = new grpc.Server();
+  this.port = server.bind(host + ':' + port, server_creds);
+  server.addProtoService(serviceProto.BenchmarkService.service, {
+    unaryCall: unaryCall,
+    streamingCall: streamingCall
+  });
+  this.server = server;
+}
+
+/**
+ * Start the benchmark server.
+ */
+BenchmarkServer.prototype.start = function() {
+  this.server.start();
+  this.last_wall_time = process.hrtime();
+};
+
+/**
+ * Return the port number that the server is bound to.
+ * @return {Number} The port number
+ */
+BenchmarkServer.prototype.getPort = function() {
+  return this.port;
+};
+
+/**
+ * Return current statistics for the server. If reset is set, restart
+ * statistic collection.
+ * @param {boolean} reset Indicates that statistics should be reset
+ * @return {object} Server statistics
+ */
+BenchmarkServer.prototype.mark = function(reset) {
+  var wall_time_diff = process.hrtime(this.last_wall_time);
+  if (reset) {
+    this.last_wall_time = process.hrtime();
+  }
+  return {
+    time_elapsed: wall_time_diff[0] + wall_time_diff[1] / 1e9,
+    // Not sure how to measure these values
+    time_user: 0,
+    time_system: 0
+  };
+};
+
+/**
+ * Stop the server.
+ * @param {function} callback Called when the server has finished shutting down
+ */
+BenchmarkServer.prototype.stop = function(callback) {
+  this.server.tryShutdown(callback);
+};
+
+module.exports = BenchmarkServer;

src/node/performance/histogram.js

+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/**
+ * Histogram module. Exports the Histogram class
+ * @module
+ */
+
+'use strict';
+
+/**
+ * Histogram class. Collects data and exposes a histogram and other statistics.
+ * This data structure is taken directly from src/core/support/histogram.c, but
+ * pared down to the statistics needed for client stats in
+ * test/proto/benchmarks/stats.proto.
+ * @constructor
+ * @param {number} resolution The histogram's bucket resolution. Must be positive
+ * @param {number} max_possible The maximum allowed value. Must be greater than 1
+ */
+function Histogram(resolution, max_possible) {
+  this.resolution = resolution;
+  this.max_possible = max_possible;
+
+  this.sum = 0;
+  this.sum_of_squares = 0;
+  this.multiplier = 1 + resolution;
+  this.count = 0;
+  this.min_seen = max_possible;
+  this.max_seen = 0;
+  this.buckets = [];
+  for (var i = 0; i < this.bucketFor(max_possible) + 1; i++) {
+    this.buckets[i] = 0;
+  }
+}
+
+/**
+ * Get the bucket index for a given value.
+ * @param {number} value The value to check
+ * @return {number} The bucket index
+ */
+Histogram.prototype.bucketFor = function(value) {
+  return Math.floor(Math.log(value) / Math.log(this.multiplier));
+};
+
+/**
+ * Get the minimum value for a given bucket index
+ * @param {number} The bucket index to check
+ * @return {number} The minimum value for that bucket
+ */
+Histogram.prototype.bucketStart = function(index) {
+  return Math.pow(this.multiplier, index);
+};
+
+/**
+ * Add a value to the histogram. This updates all statistics with the new
+ * value. Those statistics should not be modified except with this function
+ * @param {number} value The value to add
+ */
+Histogram.prototype.add = function(value) {
+  // Ensure value is a number
+  value = +value;
+  this.sum += value;
+  this.sum_of_squares += value * value;
+  this.count++;
+  if (value < this.min_seen) {
+    this.min_seen = value;
+  }
+  if (value > this.max_seen) {
+    this.max_seen = value;
+  }
+  this.buckets[this.bucketFor(value)]++;
+};
+
+/**
+ * Get the mean of all added values
+ * @return {number} The mean
+ */
+Histogram.prototype.mean = function() {
+  return this.sum / this.count;
+};
+
+/**
+ * Get the variance of all added values. Used to calulate the standard deviation
+ * @return {number} The variance
+ */
+Histogram.prototype.variance = function() {
+  if (this.count == 0) {
+    return 0;
+  }
+  return (this.sum_of_squares * this.count - this.sum * this.sum) /
+      (this.count * this.count);
+};
+
+/**
+ * Get the standard deviation of all added values
+ * @return {number} The standard deviation
+ */
+Histogram.prototype.stddev = function() {
+  return Math.sqrt(this.variance);
+};
+
+/**
+ * Get the maximum among all added values
+ * @return {number} The maximum
+ */
+Histogram.prototype.maximum = function() {
+  return this.max_seen;
+};
+
+/**
+ * Get the minimum among all added values
+ * @return {number} The minimum
+ */
+Histogram.prototype.minimum = function() {
+  return this.min_seen;
+};
+
+/**
+ * Get the number of all added values
+ * @return {number} The count
+ */
+Histogram.prototype.getCount = function() {
+  return this.count;
+};
+
+/**
+ * Get the sum of all added values
+ * @return {number} The sum
+ */
+Histogram.prototype.getSum = function() {
+  return this.sum;
+};
+
+/**
+ * Get the sum of squares of all added values
+ * @return {number} The sum of squares
+ */
+Histogram.prototype.sumOfSquares = function() {
+  return this.sum_of_squares;
+};
+
+/**
+ * Get the raw histogram as a list of bucket sizes
+ * @return {Array.<number>} The buckets
+ */
+Histogram.prototype.getContents = function() {
+  return this.buckets;
+};
+
+module.exports = Histogram;

src/node/performance/perf_test.js → src/node/performance/worker_server.js

@@ -33,87 +33,31 @@
 
 'use strict';
 
-var grpc = require('..');
-var testProto = grpc.load(__dirname + '/../interop/test.proto').grpc.testing;
-var _ = require('lodash');
-var interop_server = require('../interop/interop_server.js');
-
-function runTest(iterations, callback) {
-  var testServer = interop_server.getServer(0, false);
-  testServer.server.start();
-  var client = new testProto.TestService('localhost:' + testServer.port,
-                                         grpc.credentials.createInsecure());
-
-  function runIterations(finish) {
-    var start = process.hrtime();
-    var intervals = [];
-    function next(i) {
-      if (i >= iterations) {
-        testServer.server.shutdown();
-        var totalDiff = process.hrtime(start);
-        finish({
-          total: totalDiff[0] * 1000000 + totalDiff[1] / 1000,
-          intervals: intervals
-        });
-      } else{
-        var deadline = new Date();
-        deadline.setSeconds(deadline.getSeconds() + 3);
-        var startTime = process.hrtime();
-        client.emptyCall({}, function(err, resp) {
-          var timeDiff = process.hrtime(startTime);
-          intervals[i] = timeDiff[0] * 1000000 + timeDiff[1] / 1000;
-          next(i+1);
-        }, {}, {deadline: deadline});
-      }
-    }
-    next(0);
-  }
-
-  function warmUp(num) {
-    var pending = num;
-    function startCall() {
-      client.emptyCall({}, function(err, resp) {
-        pending--;
-        if (pending === 0) {
-          runIterations(callback);
-        }
-      });
-    }
-    for (var i = 0; i < num; i++) {
-      startCall();
-    }
-  }
-  warmUp(100);
-}
-
-function percentile(arr, pct) {
-  if (pct > 99) {
-    pct = 99;
-  }
-  if (pct < 0) {
-    pct = 0;
-  }
-  var index = Math.floor(arr.length * pct / 100);
-  return arr[index];
+var worker_service_impl = require('./worker_service_impl');
+
+var grpc = require('../../../');
+var serviceProto = grpc.load({
+  root: __dirname + '/../../..',
+  file: 'test/proto/benchmarks/services.proto'}).grpc.testing;
+
+function runServer(port) {
+  var server_creds = grpc.ServerCredentials.createInsecure();
+  var server = new grpc.Server();
+  server.addProtoService(serviceProto.WorkerService.service,
+                         worker_service_impl);
+  var address = '0.0.0.0:' + port;
+  server.bind(address, server_creds);
+  server.start();
+  return server;
 }
 
 if (require.main === module) {
-  var count;
-  if (process.argv.length >= 3) {
-    count = process.argv[2];
-  } else {
-    count = 100;
-  }
-  runTest(count, function(results) {
-    var sorted_intervals = _.sortBy(results.intervals, _.identity);
-    console.log('count:', count);
-    console.log('total time:', results.total, 'us');
-    console.log('median:', percentile(sorted_intervals, 50), 'us');
-    console.log('90th percentile:', percentile(sorted_intervals, 90), 'us');
-    console.log('95th percentile:', percentile(sorted_intervals, 95), 'us');
-    console.log('99th percentile:', percentile(sorted_intervals, 99), 'us');
-    console.log('QPS:', (count / results.total) * 1000000);
+  Error.stackTraceLimit = Infinity;
+  var parseArgs = require('minimist');
+  var argv = parseArgs(process.argv, {
+    string: ['driver_port']
   });
+  runServer(argv.driver_port);
 }
 
-module.exports = runTest;
+exports.runServer = runServer;

src/node/performance/qps_test.js → src/node/performance/worker_service_impl.js

@@ -31,107 +31,102 @@
  *
  */
 
-/**
- * This script runs a QPS test. It sends requests for a specified length of time
- * with a specified number pending at any one time. It then outputs the measured
- * QPS. Usage:
- * node qps_test.js [--concurrent=count] [--time=seconds]
- * concurrent defaults to 100 and time defaults to 10
- */
-
 'use strict';
 
-var async = require('async');
-var parseArgs = require('minimist');
-
-var grpc = require('..');
-var testProto = grpc.load(__dirname + '/../interop/test.proto').grpc.testing;
-var interop_server = require('../interop/interop_server.js');
-
-/**
- * Runs the QPS test. Sends requests constantly for the given number of seconds,
- * and keeps concurrent_calls requests pending at all times. When the test ends,
- * the callback is called with the number of calls that completed within the
- * time limit.
- * @param {number} concurrent_calls The number of calls to have pending
- *     simultaneously
- * @param {number} seconds The number of seconds to run the test for
- * @param {function(Error, number)} callback Callback for test completion
- */
-function runTest(concurrent_calls, seconds, callback) {
-  var testServer = interop_server.getServer(0, false);
-  testServer.server.start();
-  var client = new testProto.TestService('localhost:' + testServer.port,
-                                         grpc.credentials.createInsecure());
-
-  var warmup_num = 100;
+var BenchmarkClient = require('./benchmark_client');
+var BenchmarkServer = require('./benchmark_server');
 
-  /**
-   * Warms up the client to avoid counting startup time in the test result
-   * @param {function(Error)} callback Called when warmup is complete
-   */
-  function warmUp(callback) {
-    var pending = warmup_num;
-    function startCall() {
-      client.emptyCall({}, function(err, resp) {
-        if (err) {
-          callback(err);
-          return;
-        }
-        pending--;
-        if (pending === 0) {
-          callback(null);
-        }
+exports.runClient = function runClient(call) {
+  var client;
+  call.on('data', function(request) {
+    var stats;
+    switch (request.argtype) {
+      case 'setup':
+      var setup = request.setup;
+      client = new BenchmarkClient(setup.server_targets,
+                                   setup.client_channels,
+                                   setup.histogram_params,
+                                   setup.security_params);
+      client.on('error', function(error) {
+        call.emit('error', error);
       });
-    }
-    for (var i = 0; i < warmup_num; i++) {
-      startCall();
-    }
-  }
-  /**
-   * Run the QPS test. Starts concurrent_calls requests, then starts a new
-   * request whenever one completes until time runs out.
-   * @param {function(Error, number)} callback Called when the test is complete.
-   *     The second argument is the number of calls that finished within the
-   *     time limit
-   */
-  function run(callback) {
-    var running = 0;
-    var count = 0;
-    var start = process.hrtime();
-    function responseCallback(err, resp) {
-      if (process.hrtime(start)[0] < seconds) {
-        count += 1;
-        client.emptyCall({}, responseCallback);
+      switch (setup.load_params.load) {
+        case 'closed_loop':
+        client.startClosedLoop(setup.outstanding_rpcs_per_channel,
+                               setup.rpc_type,
+                               setup.payload_config.simple_params.req_size,
+                               setup.payload_config.simple_params.resp_size);
+        break;
+        case 'poisson':
+        client.startPoisson(setup.outstanding_rpcs_per_channel,
+                            setup.rpc_type, setup.payload_config.req_size,
+                            setup.payload_config.resp_size,
+                            setup.load_params.poisson.offered_load);
+        break;
+        default:
+        call.emit('error', new Error('Unsupported LoadParams type' +
+            setup.load_params.load));
+      }
+      stats = client.mark();
+      call.write({
+        stats: stats
+      });
+      break;
+      case 'mark':
+      if (client) {
+        stats = client.mark(request.mark.reset);
+        call.write({
+          stats: stats
+        });
       } else {
-        running -= 1;
-        if (running <= 0) {
-          callback(null, count);
-        }
+        call.emit('error', new Error('Got Mark before ClientConfig'));
       }
+      break;
+      default:
+      throw new Error('Nonexistent client argtype option: ' + request.argtype);
     }
-    for (var i = 0; i < concurrent_calls; i++) {
-      running += 1;
-      client.emptyCall({}, responseCallback);
-    }
-  }
-  async.waterfall([warmUp, run], function(err, count) {
-      testServer.server.shutdown();
-      callback(err, count);
+  });
+  call.on('end', function() {
+    client.stop(function() {
+      call.end();
     });
-}
-
-if (require.main === module) {
-  var argv = parseArgs(process.argv.slice(2), {
-    default: {'concurrent': 100,
-              'time': 10}
   });
-  runTest(argv.concurrent, argv.time, function(err, count) {
-    if (err) {
-      throw err;
+};
+
+exports.runServer = function runServer(call) {
+  var server;
+  call.on('data', function(request) {
+    var stats;
+    switch (request.argtype) {
+      case 'setup':
+      server = new BenchmarkServer(request.setup.host, request.setup.port,
+                                   request.setup.security_params);
+      server.start();
+      stats = server.mark();
+      call.write({
+        stats: stats,
+        port: server.getPort()
+      });
+      break;
+      case 'mark':
+      if (server) {
+        stats = server.mark(request.mark.reset);
+        call.write({
+          stats: stats,
+          port: server.getPort(),
+          cores: 1
+        });
+      } else {
+        call.emit('error', new Error('Got Mark before ServerConfig'));
+      }
+      break;
+      default:
+      throw new Error('Nonexistent server argtype option');
     }
-    console.log('Concurrent calls:', argv.concurrent);
-    console.log('Time:', argv.time, 'seconds');
-    console.log('QPS:', (count/argv.time));
   });
-}
+  call.on('end', function() {
+    server.stop(function() {
+      call.end();
+    });
+  });
+};