Merge pull request #2523 from murgatroid99/benchmark_improvements

Fix some issues with the benchmark code

test/any_grpc.js

@@ -25,7 +25,7 @@ function getImplementation(globalField) {
   const impl = global[globalField];
 
   if (impl === 'js') {
-    return require(`../packages/grpc-${impl}`);
+    return require('../packages/grpc-js');
   } else if (impl === 'native') {
     return require('grpc');
   }

test/package.json

@@ -16,8 +16,10 @@
   "dependencies": {
     "express": "^4.16.3",
     "google-auth-library": "^6.1.0",
-    "grpc": "^1.24.2",
     "lodash": "^4.17.4",
     "poisson-process": "^1.0.0"
+  },
+  "optionalDependencies": {
+    "grpc": "^1.24.2"
   }
 }

test/performance/benchmark_server.js

@@ -154,8 +154,9 @@ util.inherits(BenchmarkServer, EventEmitter);
  * Start the benchmark server.
  */
 BenchmarkServer.prototype.start = function() {
-  this.server.bindAsync(this.host + ':' + this.port, this.creds, (err) => {
+  this.server.bindAsync(this.host + ':' + this.port, this.creds, (err, port) => {
     assert.ifError(err);
+    this.port = port;
     this.server.start();
     this.last_wall_time = process.hrtime();
     this.last_usage = process.cpuUsage();

test/performance/driver.js

@@ -0,0 +1,82 @@
+const grpc = require('../any_grpc').server;
+const protoLoader = require('../../packages/proto-loader');
+const protoPackage = protoLoader.loadSync(
+    'src/proto/grpc/testing/worker_service.proto',
+    {keepCase: true,
+     defaults: true,
+     enums: String,
+     oneofs: true,
+     includeDirs: [__dirname + '/../proto/']});
+const serviceProto = grpc.loadPackageDefinition(protoPackage).grpc.testing;
+
+function main() {
+  const parseArgs = require('minimist');
+  const argv = parseArgs(process.argv, {
+    string: ['client_worker_port', 'server_worker_port']
+  });
+  const clientWorker = new serviceProto.WorkerService(`localhost:${argv.client_worker_port}`, grpc.credentials.createInsecure());
+  const serverWorker = new serviceProto.WorkerService(`localhost:${argv.server_worker_port}`, grpc.credentials.createInsecure());
+  const serverWorkerStream = serverWorker.runServer();
+  const clientWorkerStream = clientWorker.runClient();
+  let firstServerResponseReceived = false;
+  let markCount = 0;
+  serverWorkerStream.on('data', (response) => {
+    console.log('Server stats:', response.stats);
+    if (!firstServerResponseReceived) {
+      firstServerResponseReceived = true;
+      clientWorkerStream.write({
+        setup: {
+          server_targets: [`localhost:${response.port}`],
+          client_channels: 1,
+          outstanding_rpcs_per_channel: 1,
+          histogram_params: {
+            resolution: 0.01,
+            max_possible:60000000000
+          },
+          payload_config: {
+            bytebuf_params: {
+              req_size: 10,
+              resp_size: 10
+            }
+          },
+          load_params: {
+            closed_loop: {}
+          }
+        }
+      });
+      clientWorkerStream.on('status', (status) => {
+        console.log('Received client worker status ' + JSON.stringify(status));
+        serverWorkerStream.end();
+      });
+      const markInterval = setInterval(() => {
+        if (markCount >= 5) {
+          clientWorkerStream.end();
+          clearInterval(markInterval);
+        } else {
+          clientWorkerStream.write({
+            mark: {}
+          });
+          serverWorkerStream.write({
+            mark: {}
+          });
+        }
+        markCount += 1;
+      }, 1000);
+    }
+  });
+  clientWorkerStream.on('data', (response) => {
+    console.log('Client stats:', response.stats);
+  });
+  serverWorkerStream.write({
+    setup: {
+      port: 0
+    }
+  });
+  serverWorkerStream.on('status', (status) => {
+    console.log('Received server worker status ' + JSON.stringify(status));
+  });
+}
+
+if (require.main === module) {
+  main();
+}

test/performance/worker.js

@@ -39,13 +39,13 @@ function runServer(port, benchmark_impl, callback) {
   server.addService(serviceProto.WorkerService.service,
                     new WorkerServiceImpl(benchmark_impl, server));
   var address = '0.0.0.0:' + port;
-  server.bindAsync(address, server_creds, (err) => {
+  server.bindAsync(address, server_creds, (err, port) => {
     if (err) {
       return callback(err);
     }
 
     server.start();
-    console.log('running QPS worker on %s', address);
+    console.log('running QPS worker on 0.0.0.0:%s', port);
     callback(null, server);
   });
 }