src/node/performance/benchmark_client.js - platform/external/grpc-grpc - Gitiles

 /*
  *
  * Copyright 2015-2016, 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 genericService = require('./generic_service');

 var grpc = require('../../../');
 var serviceProto = grpc.load({
   root: __dirname + '/../../..',
   file: 'src/proto/grpc/testing/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 = [];
   var GenericClient = grpc.makeGenericClientConstructor(genericService);
   this.genericClients = [];

   for (var i = 0; i < channels; i++) {
     this.clients[i] = new serviceProto.BenchmarkService(
         server_targets[i % server_targets.length], creds, options);
     this.genericClients[i] = new GenericClient(
         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
  * @param {boolean} generic Indicates that the generic (non-proto) clients
  *     should be used
  */
 BenchmarkClient.prototype.startClosedLoop = function(
     outstanding_rpcs_per_channel, rpc_type, req_size, resp_size, generic) {
   var self = this;

   self.running = true;

   self.last_wall_time = process.hrtime();

   var makeCall;

   var argument;
   var client_list;
   if (generic) {
     argument = zeroBuffer(req_size);
     client_list = self.genericClients;
   } else {
     argument = {
       response_size: resp_size,
       payload: {
         body: zeroBuffer(req_size)
       }
     };
     client_list = self.clients;
   }

   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(client_list, 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
  * @param {boolean} generic Indicates that the generic (non-proto) clients
  *     should be used
  */
 BenchmarkClient.prototype.startPoisson = function(
     outstanding_rpcs_per_channel, rpc_type, req_size, resp_size, offered_load,
     generic) {
   var self = this;

   self.running = true;

   self.last_wall_time = process.hrtime();

   var makeCall;

   var argument;
   var client_list;
   if (generic) {
     argument = zeroBuffer(req_size);
     client_list = self.genericClients;
   } else {
     argument = {
       response_size: resp_size,
       payload: {
         body: zeroBuffer(req_size)
       }
     };
     client_list = self.clients;
   }

   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(client_list, 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;
	/*
	*
	* Copyright 2015-2016, 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 genericService = require('./generic_service');

	var grpc = require('../../../');
	var serviceProto = grpc.load({
	root: __dirname + '/../../..',
	file: 'src/proto/grpc/testing/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 = [];
	var GenericClient = grpc.makeGenericClientConstructor(genericService);
	this.genericClients = [];

	for (var i = 0; i < channels; i++) {
	this.clients[i] = new serviceProto.BenchmarkService(
	server_targets[i % server_targets.length], creds, options);
	this.genericClients[i] = new GenericClient(
	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
	* @param {boolean} generic Indicates that the generic (non-proto) clients
	* should be used
	*/
	BenchmarkClient.prototype.startClosedLoop = function(
	outstanding_rpcs_per_channel, rpc_type, req_size, resp_size, generic) {
	var self = this;

	self.running = true;

	self.last_wall_time = process.hrtime();

	var makeCall;

	var argument;
	var client_list;
	if (generic) {
	argument = zeroBuffer(req_size);
	client_list = self.genericClients;
	} else {
	argument = {
	response_size: resp_size,
	payload: {
	body: zeroBuffer(req_size)
	}
	};
	client_list = self.clients;
	}

	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(client_list, 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
	* @param {boolean} generic Indicates that the generic (non-proto) clients
	* should be used
	*/
	BenchmarkClient.prototype.startPoisson = function(
	outstanding_rpcs_per_channel, rpc_type, req_size, resp_size, offered_load,
	generic) {
	var self = this;

	self.running = true;

	self.last_wall_time = process.hrtime();

	var makeCall;

	var argument;
	var client_list;
	if (generic) {
	argument = zeroBuffer(req_size);
	client_list = self.genericClients;
	} else {
	argument = {
	response_size: resp_size,
	payload: {
	body: zeroBuffer(req_size)
	}
	};
	client_list = self.clients;
	}

	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(client_list, 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;