[Service Bus] Perf tests for track 2 and update track 1 ones to latest

sdk/servicebus/service-bus/test/perf/service-bus-v1/package.json

@@ -0,0 +1,16 @@
+{
+  "name": "service-bus-v1",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@azure/service-bus": "^1.1.10",
+    "dotenv": "^8.2.0"
+  }
+}

sdk/servicebus/service-bus/test/perf/service-bus/receive.ts → sdk/servicebus/service-bus/test/perf/service-bus-v1/receive.ts

@@ -8,47 +8,62 @@ Measures the maximum throughput of `receiver.receive()` in package `@azure/servi
 3. Set env vars `SERVICEBUS_CONNECTION_STRING` and `SERVICE_BUS_QUEUE_NAME`.
 4. This test presumes that there are messages in the queue.
 4. `ts-node receive.ts [totalMessages]`
-5. Example: `ts-node receive.ts 1000000`
+5. Example: `ts-node receive.ts 1000 1000000`
  */
 
-import { ReceiveMode, OnError, OnMessage } from "../../../src";
-import { ServiceBusClient } from "../../../src/old/serviceBusClient";
+import {
+  ServiceBusClient,
+  ReceiveMode,
+  OnError,
+  OnMessage,
+  ServiceBusMessage
+} from "@azure/service-bus";
 import delay from "delay";
 import moment from "moment";
 
+// Load the .env file if it exists
+import * as dotenv from "dotenv";
+dotenv.config();
+
 const _start = moment();
 
 let _messages = 0;
 
 async function main(): Promise<void> {
   // Endpoint=sb://<your-namespace>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<shared-access-key>
   const connectionString = process.env.SERVICEBUS_CONNECTION_STRING as string;
-  const entityPath = process.env.SERVICE_BUS_QUEUE_NAME as string;
+  const entityPath = process.env.SERVICEBUS_QUEUE_NAME as string;
 
   const maxConcurrentCalls = process.argv.length > 2 ? parseInt(process.argv[2]) : 10;
   const messages = process.argv.length > 3 ? parseInt(process.argv[3]) : 100;
+  const isReceiveAndDelete = process.argv.length > 4 ? !(process.argv[4] === "false") : true;
   log(`Maximum Concurrent Calls: ${maxConcurrentCalls}`);
   log(`Total messages: ${messages}`);
+  log(`isReceiveAndDelete: ${isReceiveAndDelete}`);
 
   const writeResultsPromise = WriteResults(messages);
 
-  await RunTest(connectionString, entityPath, maxConcurrentCalls, messages);
+  await RunTest(connectionString, entityPath, maxConcurrentCalls, messages, isReceiveAndDelete);
   await writeResultsPromise;
 }
 
 async function RunTest(
   connectionString: string,
   entityPath: string,
   maxConcurrentCalls: number,
-  messages: number
+  messages: number,
+  isReceiveAndDelete: boolean
 ): Promise<void> {
-  const ns = new ServiceBusClient(connectionString);
+  const ns = ServiceBusClient.createFromConnectionString(connectionString);
 
   const client = ns.createQueueClient(entityPath);
-  const receiver = client.createReceiver(ReceiveMode.receiveAndDelete);
+  const receiver = client.createReceiver(
+    isReceiveAndDelete ? ReceiveMode.receiveAndDelete : ReceiveMode.peekLock
+  );
 
-  const onMessageHandler: OnMessage = async () => {
+  const onMessageHandler: OnMessage = async (msg: ServiceBusMessage) => {
     _messages++;
+    if (!isReceiveAndDelete) await msg.complete();
     if (_messages === messages) {
       await receiver.close();
       await client.close();
@@ -106,11 +121,14 @@ function WriteResult(
   maxMessages: number,
   maxElapsed: number
 ): void {
+  const memoryUsage = process.memoryUsage();
   log(
     `\tTot Msg\t${totalMessages}` +
       `\tCur MPS\t${Math.round((currentMessages * 1000) / currentElapsed)}` +
       `\tAvg MPS\t${Math.round((totalMessages * 1000) / totalElapsed)}` +
-      `\tMax MPS\t${Math.round((maxMessages * 1000) / maxElapsed)}`
+      `\tMax MPS\t${Math.round((maxMessages * 1000) / maxElapsed)}` +
+      `\tRSS\t${memoryUsage.rss}` +
+      `\tHeapUsed\t${memoryUsage.heapUsed}`
   );
 }
 

sdk/servicebus/service-bus/test/perf/service-bus-v1/receiveBatch.ts

@@ -0,0 +1,128 @@
+/*
+# Overview
+Measures the maximum throughput of `receiver.receive()` in package `@azure/service-bus`.
+
+# Instructions
+1. Create a Service Bus namespace with `Tier=Premium` and `Messaging Units=4`.  It is recommended to use the largest possible namespace to allow maximum client throughput.
+2. Create a queue inside the namespace.
+3. Set env vars `SERVICE_BUS_CONNECTION_STRING` and `SERVICE_BUS_QUEUE_NAME`.
+4. This test presumes that there are messages in the queue.
+4. `ts-node receive.ts [maxMsgCount] [totalMessages] [isReceiveAndDelete]`
+5. Example: `ts-node receive.ts 10 1000000 false`
+ */
+
+import { ReceiveMode, ServiceBusClient } from "@azure/service-bus";
+import delay from "delay";
+import moment from "moment";
+// Load the .env file if it exists
+import * as dotenv from "dotenv";
+dotenv.config();
+
+const _start = moment();
+
+let _messages = 0;
+
+async function main(): Promise<void> {
+  // Endpoint=sb://<your-namespace>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<shared-access-key>
+  const connectionString = process.env.SERVICEBUS_CONNECTION_STRING as string;
+  const entityPath = process.env.SERVICEBUS_QUEUE_NAME as string;
+
+  const maxMsgCount = process.argv.length > 2 ? parseInt(process.argv[2]) : 10;
+  const messages = process.argv.length > 3 ? parseInt(process.argv[3]) : 100;
+  const isReceiveAndDelete = process.argv.length > 4 ? !(process.argv[4] === "false") : true;
+  log(`Maximum Concurrent Calls: ${maxMsgCount}`);
+  log(`Total messages: ${messages}`);
+  log(`isReceiveAndDelete: ${isReceiveAndDelete}`);
+
+  const writeResultsPromise = WriteResults(messages);
+
+  await RunTest(connectionString, entityPath, maxMsgCount, messages, isReceiveAndDelete);
+  await writeResultsPromise;
+}
+
+async function RunTest(
+  connectionString: string,
+  entityPath: string,
+  maxMsgCount: number,
+  messages: number,
+  isReceiveAndDelete: boolean
+): Promise<void> {
+  const ns = ServiceBusClient.createFromConnectionString(connectionString);
+
+  const client = ns.createQueueClient(entityPath);
+  const receiver = client.createReceiver(
+    isReceiveAndDelete ? ReceiveMode.receiveAndDelete : ReceiveMode.peekLock
+  );
+
+  while (_messages < messages) {
+    const msgs = await receiver.receiveMessages(maxMsgCount, 10000);
+    _messages += msgs.length;
+    if (!isReceiveAndDelete) {
+      for (const msg of msgs) {
+        await msg.complete();
+      }
+    }
+  }
+
+  await ns.close();
+}
+
+async function WriteResults(messages: number): Promise<void> {
+  let lastMessages = 0;
+  let lastElapsed = 0;
+  let maxMessages = 0;
+  let maxElapsed = Number.MAX_SAFE_INTEGER;
+
+  do {
+    await delay(1000);
+
+    const receivedMessages = _messages;
+    const currentMessages = receivedMessages - lastMessages;
+    lastMessages = receivedMessages;
+
+    const elapsed = moment().diff(_start);
+    const currentElapsed = elapsed - lastElapsed;
+    lastElapsed = elapsed;
+
+    if (currentMessages / currentElapsed > maxMessages / maxElapsed) {
+      maxMessages = currentMessages;
+      maxElapsed = currentElapsed;
+    }
+
+    WriteResult(
+      receivedMessages,
+      elapsed,
+      currentMessages,
+      currentElapsed,
+      maxMessages,
+      maxElapsed
+    );
+  } while (_messages < messages);
+}
+
+function WriteResult(
+  totalMessages: number,
+  totalElapsed: number,
+  currentMessages: number,
+  currentElapsed: number,
+  maxMessages: number,
+  maxElapsed: number
+): void {
+  const memoryUsage = process.memoryUsage();
+  log(
+    `\tTot Msg\t${totalMessages}` +
+      `\tCur MPS\t${Math.round((currentMessages * 1000) / currentElapsed)}` +
+      `\tAvg MPS\t${Math.round((totalMessages * 1000) / totalElapsed)}` +
+      `\tMax MPS\t${Math.round((maxMessages * 1000) / maxElapsed)}` +
+      `\tRSS\t${memoryUsage.rss}` +
+      `\tHeapUsed\t${memoryUsage.heapUsed}`
+  );
+}
+
+function log(message: string): void {
+  console.log(`[${moment().format("hh:mm:ss.SSS")}] ${message}`);
+}
+
+main().catch((err) => {
+  log(`Error occurred: ${err}`);
+});

sdk/servicebus/service-bus/test/perf/service-bus/send.ts → sdk/servicebus/service-bus/test/perf/service-bus-v1/send.ts

@@ -10,10 +10,12 @@ Measures the maximum throughput of `sender.send()` in package `@azure/service-bu
 5. Example: `ts-node app.ts 1000 1000000`
  */
 
-import { ServiceBusClient } from "../../../src/old/serviceBusClient";
-import { Sender } from "../../../src/sender";
+import { ServiceBusClient, Sender } from "@azure/service-bus";
 import delay from "delay";
 import moment from "moment";
+// Load the .env file if it exists
+import * as dotenv from "dotenv";
+dotenv.config();
 
 const _payload = Buffer.alloc(1024);
 const _start = moment();
@@ -23,7 +25,7 @@ let _messages = 0;
 async function main(): Promise<void> {
   // Endpoint=sb://<your-namespace>.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<shared-access-key>
   const connectionString = process.env.SERVICEBUS_CONNECTION_STRING as string;
-  const entityPath = process.env.SERVICE_BUS_QUEUE_NAME as string;
+  const entityPath = process.env.SERVICEBUS_QUEUE_NAME as string;
 
   const maxInflight = process.argv.length > 2 ? parseInt(process.argv[2]) : 1;
   const messages = process.argv.length > 3 ? parseInt(process.argv[3]) : 10;
@@ -43,7 +45,7 @@ async function RunTest(
   maxInflight: number,
   messages: number
 ): Promise<void> {
-  const ns = new ServiceBusClient(connectionString);
+  const ns = ServiceBusClient.createFromConnectionString(connectionString);
 
   const client = ns.createQueueClient(entityPath);
   const sender = client.createSender();
@@ -62,12 +64,10 @@ async function RunTest(
 }
 
 async function ExecuteSendsAsync(sender: Sender, messages: number): Promise<void> {
-  while (++_messages <= messages) {
+  while (_messages <= messages) {
     await sender.send({ body: _payload });
+    _messages++;
   }
-
-  // Undo last increment, since a message was never sent on the final loop iteration
-  _messages--;
 }
 
 async function WriteResults(messages: number): Promise<void> {
@@ -104,11 +104,14 @@ function WriteResult(
   maxMessages: number,
   maxElapsed: number
 ): void {
+  const memoryUsage = process.memoryUsage();
   log(
     `\tTot Msg\t${totalMessages}` +
       `\tCur MPS\t${Math.round((currentMessages * 1000) / currentElapsed)}` +
       `\tAvg MPS\t${Math.round((totalMessages * 1000) / totalElapsed)}` +
-      `\tMax MPS\t${Math.round((maxMessages * 1000) / maxElapsed)}`
+      `\tMax MPS\t${Math.round((maxMessages * 1000) / maxElapsed)}` +
+      `\tRSS\t${memoryUsage.rss}` +
+      `\tHeapUsed\t${memoryUsage.heapUsed}`
   );
 }
 

sdk/servicebus/service-bus/test/perf/service-bus-v7/package.json

@@ -0,0 +1,15 @@
+{
+  "name": "service-bus-v7",
+  "version": "1.0.0",
+  "description": "",
+  "main": "index.js",
+  "scripts": {
+    "test": "echo \"Error: no test specified\" && exit 1"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "@azure/service-bus": "^7.0.0-preview.7"
+  }
+}