|  | @@ -0,0 +1,336 @@
 | 
	
		
			
				|  |  | +/*
 | 
	
		
			
				|  |  | + *
 | 
	
		
			
				|  |  | + * 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;
 |