| /* |
| * |
| * Copyright 2015 gRPC authors. |
| * |
| * Licensed under the Apache License, Version 2.0 (the "License"); |
| * you may not use this file except in compliance with the License. |
| * You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| * |
| */ |
| |
| /** |
| * Benchmark client module |
| * @module |
| */ |
| |
| 'use strict'; |
| |
| var fs = require('fs'); |
| var path = require('path'); |
| var util = require('util'); |
| var EventEmitter = require('events'); |
| |
| var async = require('async'); |
| 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 = { |
| "grpc.max_receive_message_length": -1, |
| "grpc.max_send_message_length": -1 |
| }; |
| 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 every client in the list of clients by waiting for each to be ready, |
| * then starting outstanding_rpcs_per_channel calls on each of them |
| * @param {Array<grpc.Client>} client_list The list of clients |
| * @param {Number} outstanding_rpcs_per_channel The number of calls to start |
| * on each client |
| * @param {function(grpc.Client)} makeCall Function to make a single call on |
| * a single client |
| * @param {EventEmitter} emitter The event emitter to send errors on, if |
| * necessary |
| */ |
| function startAllClients(client_list, outstanding_rpcs_per_channel, makeCall, |
| emitter) { |
| var ready_wait_funcs = _.map(client_list, function(client) { |
| return _.partial(grpc.waitForClientReady, client, Infinity); |
| }); |
| async.parallel(ready_wait_funcs, function(err) { |
| if (err) { |
| emitter.emit('error', err); |
| return; |
| } |
| |
| _.each(client_list, function(client) { |
| _.times(outstanding_rpcs_per_channel, function() { |
| makeCall(client); |
| }); |
| }); |
| }); |
| } |
| |
| /** |
| * 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(); |
| |
| self.last_usage = process.cpuUsage(); |
| |
| 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 call = client.streamingCall(); |
| var start_time = process.hrtime(); |
| call.write(argument); |
| call.on('data', function() { |
| var time_diff = process.hrtime(start_time); |
| self.histogram.add(timeDiffToNanos(time_diff)); |
| self.pending_calls--; |
| if (self.running) { |
| self.pending_calls++; |
| start_time = process.hrtime(); |
| call.write(argument); |
| } else { |
| call.end(); |
| if (self.pending_calls == 0) { |
| self.emit('finished'); |
| } |
| } |
| }); |
| call.on('error', function(error) { |
| self.emit('error', new Error('Client error: ' + error.message)); |
| self.running = false; |
| }); |
| } |
| }; |
| } |
| |
| startAllClients(client_list, outstanding_rpcs_per_channel, makeCall, self); |
| }; |
| |
| /** |
| * 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(); |
| |
| self.last_usage = process.cpuUsage(); |
| |
| 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 { |
| self.emit('error', new Error('Streaming Poisson benchmarks not supported')); |
| return; |
| } |
| |
| var averageIntervalMs = (1 / offered_load) * 1000; |
| |
| startAllClients(client_list, outstanding_rpcs_per_channel, function(client){ |
| var p = PoissonProcess.create(averageIntervalMs, function() { |
| makeCall(client, p); |
| }); |
| p.start(); |
| }, self); |
| }; |
| |
| /** |
| * 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 usage_diff = process.cpuUsage(this.last_usage); |
| var histogram = this.histogram; |
| if (reset) { |
| this.last_wall_time = process.hrtime(); |
| this.last_usage = process.cpuUsage(); |
| 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, |
| time_user: usage_diff.user / 1000000, |
| time_system: usage_diff.system / 1000000 |
| }; |
| }; |
| |
| /** |
| * 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; |