[Perf] Added EventPerfTest base class + sample (Azure#22002)

* Added Event base class

* Added event sample test

* Updated event counter

* Some import cleanup

* Removed extra import

Author: annatisch

tools/azure-devtools/setup.py

@@ -58,7 +58,7 @@
             "GitPython",
             "requests>=2.0",
         ],
-        "systemperf": ["aiohttp>=3.0", "requests>=2.0", "tornado==6.0.3", "pycurl==7.43.0.5", "httpx>=0.21", "azure-core"],
+        "systemperf": ["aiohttp>=3.0", "requests>=2.0", "tornado==6.0.3", "httpx>=0.21", "azure-core"],
     },
     package_dir={"": "src"},
     install_requires=DEPENDENCIES,

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

@@ -11,10 +11,12 @@
 from ._random_stream import RandomStream, WriteStream, get_random_bytes
 from ._async_random_stream import AsyncRandomStream
 from ._batch_perf_test import BatchPerfTest
+from ._event_perf_test import EventPerfTest
 
 __all__ = [
     "PerfStressTest",
     "BatchPerfTest",
+    "EventPerfTest",
     "RandomStream",
     "WriteStream",
     "AsyncRandomStream",

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

@@ -5,7 +5,6 @@
 
 import cProfile
 import os
-import threading
 import aiohttp
 import time
 from typing import Optional, Any, Dict, List

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

@@ -0,0 +1,137 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+import asyncio
+import threading
+import time
+
+from ._repeated_timer import AtomicCounter
+from ._perf_stress_base import _PerfTestBase
+
+
+class EventPerfTest(_PerfTestBase):
+
+    def __init__(self, arguments):
+        super().__init__(arguments)
+        if self.args.profile:
+            raise NotImplementedError("Profiler support for event tests pending.")
+
+        if self.args.sync:
+            self._condition = threading.Condition()
+        else:
+            self._condition = asyncio.Condition()
+        self._start_time = time.time()
+        self._error = None
+        self._processing = None
+        self._completed_operations = AtomicCounter()
+
+    @property
+    def completed_operations(self) -> int:
+        """
+        Total number of operations completed by run_all().
+        Reset after warmup.
+        """
+        return self._completed_operations.value()
+
+    @property
+    def last_completion_time(self) -> float:
+        """
+        Elapsed time between start of warmup/run and last completed operation.
+        Reset after warmup.
+        """
+        return self._last_completion_time - self._start_time
+
+    def event_raised_sync(self):
+        self._completed_operations.increment()
+        self._last_completion_time = time.time()
+
+    def error_raised_sync(self, error):
+        with self._condition:
+            self._error = error
+            self._condition.notify_all()
+
+    async def event_raised_async(self):
+        self._completed_operations.increment()
+        self._last_completion_time = time.time()
+
+    async def error_raised_async(self, error):
+        async with self._condition:
+            self._error = error
+            self._condition.notify_all()
+
+    async def setup(self) -> None:
+        """
+        Setup called once per parallel test instance.
+        Used to setup state specific to this test instance.
+        """
+        if self.args.sync:
+            self._processing = threading.Thread(target=self.start_events_sync)
+            self._processing.daemon = True
+            self._processing.start()
+        else:
+            self._processing = asyncio.ensure_future(self.start_events_async())
+
+    async def cleanup(self) -> None:
+        """
+        Cleanup called once per parallel test instance.
+        Used to cleanup state specific to this test instance.
+        """
+        if self.args.sync:
+            self.stop_events_sync()
+            self._processing.join()
+        else:
+            await self.stop_events_async()
+            await self._processing
+        try:
+            raise self._error
+        except TypeError:
+            pass
+
+    def run_all_sync(self, duration: int) -> None:
+        """
+        Run all sync tests, including both warmup and duration.
+        """
+        with self._condition:
+            self._completed_operations.reset()
+            self._last_completion_time = 0.0
+            self._start_time = time.time()
+            self._condition.wait(timeout=duration)
+
+    async def run_all_async(self, duration: int) -> None:
+        """
+        Run all async tests, including both warmup and duration.
+        """
+        async with self._condition:
+            self._completed_operations.reset()
+            self._last_completion_time = 0.0
+            self._start_time = time.time()
+            try:
+                await asyncio.wait_for(self._condition.wait(), timeout=duration)
+            except asyncio.TimeoutError:
+                pass
+
+    def start_events_sync(self) -> None:
+        """
+        Start the process for receiving events.
+        """
+        raise NotImplementedError("start_events_sync must be implemented for {}".format(self.__class__.__name__))
+
+    def stop_events_sync(self) -> None:
+        """
+        Stop the process for receiving events.
+        """
+        raise NotImplementedError("stop_events_sync must be implemented for {}".format(self.__class__.__name__))
+
+    async def start_events_async(self) -> None:
+        """
+        Start the process for receiving events.
+        """
+        raise NotImplementedError("start_events_async must be implemented for {}".format(self.__class__.__name__))
+
+    async def stop_events_async(self) -> None:
+        """
+        Stop the process for receiving events.
+        """
+        raise NotImplementedError("stop_events_async must be implemented for {}".format(self.__class__.__name__))

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

@@ -155,7 +155,7 @@ async def setup(self) -> None:
         """
         return
 
-    async def cleanup(self):
+    async def cleanup(self) -> None:
         """
         Cleanup called once per parallel test instance.
         Used to cleanup state specific to this test instance.

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

@@ -16,6 +16,7 @@
 
 from ._perf_stress_base import _PerfTestABC
 from ._batch_perf_test import BatchPerfTest
+from ._event_perf_test import EventPerfTest
 from ._perf_stress_test import PerfStressTest
 from ._repeated_timer import RepeatedTimer
 
@@ -120,6 +121,7 @@ def _parse_args(self):
         self.logger.info("")
 
     def _discover_tests(self, test_folder_path):
+        base_classes = [PerfStressTest, BatchPerfTest, EventPerfTest]
         self._test_classes = {}
         if os.path.isdir(os.path.join(test_folder_path, 'tests')):
             test_folder_path = os.path.join(test_folder_path, 'tests')
@@ -136,7 +138,7 @@ def _discover_tests(self, test_folder_path):
                 if name.startswith("_"):
                     continue
                 if inspect.isclass(value):
-                    if issubclass(value, _PerfTestABC) and value not in [PerfStressTest, BatchPerfTest]:
+                    if issubclass(value, _PerfTestABC) and value not in base_classes:
                         self.logger.info("Loaded test class: {}".format(name))
                         self._test_classes[name] = value
 

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

@@ -3,7 +3,8 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
-from threading import Timer
+import itertools
+from threading import Timer, Lock
 
 # Credit to https://stackoverflow.com/questions/3393612/run-certain-code-every-n-seconds
 class RepeatedTimer(object):
@@ -32,3 +33,26 @@ def start(self):
     def stop(self):
         self._timer.cancel()
         self.is_running = False
+
+
+# Credit to https://julien.danjou.info/atomic-lock-free-counters-in-python/
+class AtomicCounter(object):
+
+    def __init__(self):
+        self._number_of_read = 0
+        self._counter = itertools.count()
+        self._read_lock = Lock()
+
+    def increment(self):
+        next(self._counter)
+    
+    def reset(self):
+        with self._read_lock:
+            self._number_of_read = 0
+            self._counter = itertools.count()
+
+    def value(self):
+        with self._read_lock:
+            value = next(self._counter) - self._number_of_read
+            self._number_of_read += 1
+        return value