[Perfstress] Multiprocess support (Azure#22997)

* Draft multiproc support

* Improve barrier reliability

* Spelling

* Run global setup per proc

* Update cspell

* Add parameters to SleepTest
- Aligns with .NET and supports more scenarios

* Add LogTest to help observe and debug the framework

* Subprocess spawn instead of fork

* Fix status reporting

* Improved error handling

* Updated docs

* Apply suggestions from code review

Co-authored-by: Paul Van Eck <[email protected]>

* Cspell fixes

* Typo

Co-authored-by: Mike Harder <[email protected]>
Co-authored-by: Paul Van Eck <[email protected]>

Author: 3 people

doc/dev/perfstress_tests.md

@@ -1,9 +1,11 @@
 # Table of Contents
 1. [The perfstress framework](#the-perfstress-framework)
-    - [The PerfStressTest base](#the-perfstresstest-base)
+    - [The framework baseclasses](#the-provided-baseclasses)
+        - [PerfStressTest](#the-perfstresstest-baseclass)
+        - [BatchPerfTest](#the-batchperftest-baseclass)
+        - [EventPerfTest](#the-eventperftest-baseclass)
     - [Default command options](#default-command-options)
     - [Running with test proxy](#running-with-the-test-proxy)
-    - [The BatchPerfTest base](#the-batchperftest-base)
 2. [Adding performance tests to an SDK](#adding-performance-tests-to-an-sdk)
     - [Writing a test](#writing-a-test)
     - [Writing a batch test](#writing-a-batch-test)
@@ -12,6 +14,8 @@
     - [Running the system tests](#running-the-system-tests)
 4. [Readme](#readme)
 
+[comment]: # ( cspell:ignore perfstresstest batchperftest )
+
 # The perfstress framework
 
 The perfstress framework has been added to azure-devtools module. The code can be found [here](https://github.com/Azure/azure-sdk-for-python/tree/main/tools/azure-devtools/src/azure_devtools/perfstress_tests).
@@ -22,52 +26,69 @@ the tests. To start using the framework, make sure that `azure-devtools` is incl
 ```
 The perfstress framework offers the following:
 - The `perfstress` commandline tool.
-- The `PerfStressTest` baseclass.
-- The `BatchPerfTest` baseclass.
+- The `PerfStressTest` baseclass (each test run counted as a single data point).
+- The `BatchPerfTest` baseclass (each test run counted as 1 or more data points).
+- The `EventPerfTest` baseclass (supports a callback based event handler).
 - Stream utilities for uploading/downloading without storing in memory: `RandomStream`, `AsyncRandomStream`, `WriteStream`.
 - A `get_random_bytes` utility for returning randomly generated data.
 - A series of "system tests" to test the perfstress framework along with the performance of the raw transport layers (requests, aiohttp, etc).
 
-## The PerfStressTest base
-The `PerfStressTest` base class is what will be used for all perf test implementations. It provides the following API:
+## The provided baseclasses:
+The perf framework provides three baseclasses to accommodate testing different SDK scenarios. Depending on which baseclass you select, different
+methods will need to be implemented. All of them have a common base API (`_PerfTestBase`), defined below:
+
 ```python
-class PerfStressTest:
+class _PerfTestBase:
     args = {}  # Command line arguments
 
+    @property
+    def completed_operations(self) -> int:
+        # Total number of operations completed by run_all(). Reset after warmup.
+
+    @property
+    def last_completion_time(self) -> float:
+        # Elapsed time between start of warmup/run and last completed operation. Reset after warmup.
+
     def __init__(self, arguments):
         # The command line args can be accessed on construction.
 
-    async def global_setup(self):
-        # Can be optionally defined. Only run once, regardless of parallelism.
+    async def global_setup(self) -> None:
+        # Can be optionally defined. Only run once per process, regardless of multi-threading.
+        # The baseclasses will also define logic here, so if you override this method, make sure you include a call to super().
 
-    async def global_cleanup(self):
-        # Can be optionally defined. Only run once, regardless of parallelism.
+    async def global_cleanup(self) -> None:
+        # Can be optionally defined. Only run once per process, regardless of multi-threading.
+        # The baseclasses will also define logic here, so if you override this method, make sure you include a call to super().
 
-    async def post_setup(self):
+    async def post_setup(self) -> None:
         # Post-setup called once per parallel test instance.
         # Used by base classes to setup state (like test-proxy) after all derived class setup is complete.
         # There should be no need to overwrite this function.
 
-    async def pre_cleanup(self):
+    async def pre_cleanup(self) -> None:
         # Pre-cleanup called once per parallel test instance.
         # Used by base classes to cleanup state (like test-proxy) before all derived class cleanup runs.
         # There should be no need to overwrite this function.
 
-    async def setup(self):
+    async def setup(self) -> None:
         # Can be optionally defined. Run once per test instance, after global_setup.
+        # The baseclasses will also define logic here, so if you override this method, make sure you include a call to super().
 
-    async def cleanup(self):
+    async def cleanup(self) -> None:
         # Can be optionally defined. Run once per test instance, before global_cleanup.
+        # The baseclasses will also define logic here, so if you override this method, make sure you include a call to super().
 
-    async def close(self):
+    async def close(self) -> None:
         # Can be optionally defined. Run once per test instance, after cleanup and global_cleanup.
+        # The baseclasses will also define logic here, so if you override this method, make sure you include a call to super().
 
-    def run_sync(self):
-        # Must be implemented. This will be the perf test to be run synchronously.
+    def run_all_sync(self, duration: int) -> None:
+        # Run all sync tests, including both warmup and duration. This method is implemented by the provided base
+        # classes, there should be no need to overwrite this function.
 
-    async def run_async(self):
-        # Must be implemented. This will be the perf test to be run asynchronously.
-        # If writing a test for a T1 legacy SDK with no async, implement this method and raise an exception.
+    async def run_all_async(self, duration: int) -> None:
+        # Run all async tests, including both warmup and duration. This method is implemented by the provided base
+        # classes, there should be no need to overwrite this function.
 
     @staticmethod
     def add_arguments(parser):
@@ -78,17 +99,92 @@ class PerfStressTest:
     def get_from_env(variable):
         # Get the value of an env var. If empty or not found, a ValueError will be raised.
 ```
+### The PerfStressTest baseclass
+This is probably the most common test baseclass, and should be used where each test run represents 1 logical successful result.
+For example, 1 successful service request, 1 file uploaded, 1 output downloaded, etc.
+Along with the above base API, the following methods will need to be implemented:
+
+```python
+class PerfStressTest:
+    def run_sync(self) -> None:
+        # Must be implemented. This will be the perf test to be run synchronously.
+
+    async def run_async(self) -> None:
+        # Must be implemented. This will be the perf test to be run asynchronously.
+        # If writing a test for an SDK without async support (e.g. a T1 legacy SDK), implement this method and raise an exception.
+
+```
+### The BatchPerfTest baseclass
+The `BatchPerfTest` class is the parent class of the above `PerfStressTest` class that is further abstracted to allow for more flexible testing of SDKs that don't conform to a 1:1 ratio of operations to results.
+This baseclass should be used where each test run represent a more than a single result. For example, results that are uploaded
+or downloaded in batches.
+Along with the above base API, the following methods will need to be implemented:
+
+```python
+class BatchPerfTest:
+    def run_batch_sync(self) -> int:
+        # Run cumulative operation(s) synchronously - i.e. an operation that results in more than a single logical result.
+        # When inheriting from BatchPerfTest, this method will need to be implemented.
+        # Must return the number of completed results represented by a single successful test run.
+
+    async def run_batch_async(self) -> int:
+        # Run cumulative operation(s) asynchronously - i.e. an operation that results in more than a single logical result.
+        # When inheriting from BatchPerfTest, this method will need to be implemented.
+        # Must return the number of completed results represented by a single successful test run.
+        # If writing a test for an SDK without async support (e.g. a T1 legacy SDK), implement this method and raise an exception.
+
+```
+### The EventPerfTest baseclass
+This baseclass should be used when SDK operation to be tested requires starting up a process that acts on events via callback.
+Along with the above base API, the following methods will need to be implemented:
+```python
+class EventPerfTest:
+    def event_raised_sync(self) -> None:
+        # This method should not be overwritten, instead it should be called in your sync callback implementation
+        # to register a single successful event.
+
+    def error_raised_sync(self, error):
+        # This method should not be overwritten, instead it should be called in your sync callback implementation
+        # to register a failure in the event handler. This will result in the test being shutdown.
+
+    async def event_raised_async(self):
+        # This method should not be overwritten, instead it should be called in your async callback implementation
+        # to register a single successful event.
+
+    async def error_raised_async(self, error):
+        # This method should not be overwritten, instead it should be called in your async callback implementation
+        # to register a failure in the event handler. This will result in the test being shutdown.
+
+    def start_events_sync(self) -> None:
+        # Must be implemented - starts the synchronous process for receiving events.
+        # This can be blocking for the duration of the test as it will be run during setup() in a thread.
+
+    def stop_events_sync(self) -> None:
+        # Stop the synchronous process for receiving events. Must be implemented. Will be called during cleanup.
+
+    async def start_events_async(self) -> None:
+        # Must be implemented - starts the asynchronous process for receiving events.
+        # This can be blocking for the duration of the test as it will be scheduled in the eventloop during setup().
+
+    async def stop_events_async(self) -> None:
+        # Stop the asynchronous process for receiving events. Must be implemented. Will be called during cleanup.
+
+```
+
 ## Default command options
 The framework has a series of common command line options built in:
 - `-d --duration=10` Number of seconds to run as many operations (the "run" function) as possible. Default is 10.
-- `-i --iterations=1` Number of test iterations to run. Default is 1.
 - `-p --parallel=1` Number of tests to run in parallel. Default is 1.
+- `--processes=multiprocessing.cpu_count()` Number of concurrent processes that the parallel test runs should be distributed over. This is used
+  together with `--parallel` to distribute the number of concurrent tests first between available processes, then between threads within each
+  process. For example if `--parallel=16 --processes=4`, 4 processes will be started, each running 4 concurrent threaded test instances.
+  Best effort will be made to distribute evenly, for example if `--parallel=10 --processes=4`, 4 processes will be start, two of which run 3 threads, and two that run 2 threads. It's therefore recommended that the value of `parallel` be less than, or a multiple of, the value of `processes`.
 - `-w --warm-up=5` Number of seconds to spend warming up the connection before measuring begins. Default is 5.
 - `--sync` Whether to run the tests in sync or async. Default is False (async).
 - `--no-cleanup` Whether to keep newly created resources after test run. Default is False (resources will be deleted).
 - `--insecure` Whether to run without SSL validation. Default is False.
 - `-x --test-proxies` Whether to run the tests against the test proxy server. Specify the URL(s) for the proxy endpoint(s) (e.g. "https://localhost:5001"). Multiple values should be semi-colon-separated.
-- `--profile` Whether to run the perftest with cProfile. If enabled (default is False), the output file of a single iteration will be written to the current working directory in the format `"cProfile-<TestClassName>-<TestID>-<sync|async>.pstats"`.
+- `--profile` Whether to run the perftest with cProfile. If enabled (default is False), the output file of a single iteration will be written to the current working directory in the format `"cProfile-<TestClassName>-<TestID>-<sync|async>.pstats"`. **Note:** The profiler is not currently supported for the `EventPerfTest` baseclass.
 
 ## Running with the test proxy
 Follow the instructions here to install and run the test proxy server:
@@ -98,29 +194,6 @@ Once running, in a separate process run the perf test in question, combined with
 ```cmd
 (env) ~/azure-storage-blob/tests> perfstress DownloadTest -x "https://localhost:5001"
 ```
-## The BatchPerfTest base
-The `BatchPerfTest` class is the parent class of the above `PerfStressTest` class that is further abstracted to allow for more flexible testing of SDKs that don't conform to a 1:1 ratio of operations to results.
-An example of this is a messaging SDK that streams multiple messages for a period of time.
-This base class uses the same setup/cleanup/close functions described above, however instead of `run_sync` and `run_async`, it has `run_batch_sync` and `run_batch_async`:
-```python
-class BatchPerfTest:
-
-    def run_batch_sync(self) -> int:
-        """
-        Run cumultive operation(s) - i.e. an operation that results in more than a single logical result.
-        :returns: The number of completed results.
-        :rtype: int
-        """
-
-    async def run_batch_async(self) -> int:
-        """
-        Run cumultive operation(s) - i.e. an operation that results in more than a single logical result.
-        :returns: The number of completed results.
-        :rtype: int
-        """
-
-```
-An example test case using the `BatchPerfTest` base can be found below.
 
 # Adding performance tests to an SDK
 The performance tests will be in a submodule called `perfstress_tests` within the `tests` directory in an SDK project.
@@ -131,7 +204,7 @@ sdk/storage/azure-storage-blob/tests/perfstress_tests
 This `perfstress_tests` directory is a module, and so must contain an `__init__.py` file. This can be empty.
 
 ## Writing a test
-To add a test, import and inherit from `PerfStressTest` and populate the functions as needed.
+To add a test, import and inherit from one of the provided baseclasses and populate the functions as needed.
 The name of the class will be the name of the perf test, and is what will be passed into the command line to execute that test.
 ```python
 from azure_devtools.perfstress_tests import PerfStressTest
@@ -182,7 +255,7 @@ class ListContainersTest(PerfStressTest):
         """The synchronous perf test.
 
         Try to keep this minimal and focused. Using only a single client API.
-        Avoid putting any ancilliary logic (e.g. generating UUIDs), and put this in the setup/init instead
+        Avoid putting any ancillary logic (e.g. generating UUIDs), and put this in the setup/init instead
         so that we're only measuring the client API call.
         """
         for _ in self.client.list_containers():
@@ -192,7 +265,7 @@ class ListContainersTest(PerfStressTest):
         """The asynchronous perf test.
 
         Try to keep this minimal and focused. Using only a single client API.
-        Avoid putting any ancilliary logic (e.g. generating UUIDs), and put this in the setup/init instead
+        Avoid putting any ancillary logic (e.g. generating UUIDs), and put this in the setup/init instead
         so that we're only measuring the client API call.
         """
         async for _ in self.async_client.list_containers():
@@ -218,7 +291,7 @@ class _StorageStreamTestBase(PerfStressTest):
         super().__init__(arguments)
 
         # Any common attributes
-        self.container_name = 'streamperftests'
+        self.container_name = 'stream-perf-tests'
 
         # Auth configuration
         connection_string = self.get_from_env("AZURE_STORAGE_CONNECTION_STRING")

tools/azure-devtools/src/azure_devtools/perfstress_tests/_perf_stress_base.py

@@ -5,7 +5,7 @@
 
 import os
 import abc
-import threading
+import multiprocessing
 import argparse
 
 
@@ -106,17 +106,15 @@ class _PerfTestBase(_PerfTestABC):
     """Base class for implementing a python perf test."""
 
     args = {}
-    _global_parallel_index_lock = threading.Lock()
+    _global_parallel_index_lock = multiprocessing.Lock()
     _global_parallel_index = 0
 
     def __init__(self, arguments):
         self.args = arguments
         self._completed_operations = 0
         self._last_completion_time = 0.0
-
-        with _PerfTestBase._global_parallel_index_lock:
-            self._parallel_index = _PerfTestBase._global_parallel_index
-            _PerfTestBase._global_parallel_index += 1
+        self._parallel_index = _PerfTestBase._global_parallel_index
+        _PerfTestBase._global_parallel_index += 1
 
     @property
     def completed_operations(self) -> int:
@@ -136,14 +134,14 @@ def last_completion_time(self) -> float:
 
     async def global_setup(self) -> None:
         """
-        Setup called once across all parallel test instances.
+        Setup called once per process across all threaded test instances.
         Used to setup state that can be used by all test instances.
         """
         return
 
     async def global_cleanup(self) -> None:
         """
-        Cleanup called once across all parallel test instances.
+        Cleanup called once per process across all threaded test instances.
         Used to cleanup state that can be used by all test instances.
         """
         return