[Perf] corehttp - updating async stream upload httpx test

sdk/core/azure-core/perf-tests.yml

@@ -14,16 +14,16 @@ Tests:
   Arguments:
   - --size 1024 --parallel 64 --duration 60 --policies all
   - --size 1024 --parallel 64 --duration 60 --policies all --use-entra-id
-  - --size 10240 --parallel 32 --duration 60
-  - --size 10240 --parallel 32 --duration 60 --transport requests
+  - --size 10240 --parallel 64 --duration 60
+  - --size 10240 --parallel 64 --duration 60 --transport requests
 
 - Test: download-binary
   Class: DownloadBinaryDataTest
   Arguments:
   - --size 1024 --parallel 64 --duration 60
   - --size 1024 --parallel 64 --duration 60 --transport requests
   - --size 1024 --parallel 64 --duration 60 --use-entra-id
-  - --size 10240 --parallel 32 --duration 60 --policies all
+  - --size 10240 --parallel 64 --duration 60 --policies all
 
 - Test: update-entity
   Class: UpdateEntityJSONTest

sdk/core/corehttp/perf-tests.yml

@@ -14,16 +14,16 @@ Tests:
   Arguments:
   - --size 1024 --parallel 64 --duration 60 --policies all
   - --size 1024 --parallel 64 --duration 60 --policies all --use-entra-id
-  - --size 10240 --parallel 32 --duration 60
-  - --size 10240 --parallel 32 --duration 60 --transport httpx
+  - --size 10240 --parallel 64 --duration 60
+  - --size 10240 --parallel 64 --duration 60 --transport httpx
 
 - Test: download-binary
   Class: DownloadBinaryDataTest
   Arguments:
   - --size 1024 --parallel 64 --duration 60
   - --size 1024 --parallel 64 --duration 60 --transport httpx
   - --size 1024 --parallel 64 --duration 60 --use-entra-id
-  - --size 10240 --parallel 32 --duration 60 --policies all
+  - --size 10240 --parallel 64 --duration 60 --policies all
 
 - Test: update-entity
   Class: UpdateEntityJSONTest

sdk/core/corehttp/tests/perf_tests/_async_iterator_random_stream.py

@@ -0,0 +1,71 @@
+# --------------------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License. See License.txt in the project root for license information.
+# --------------------------------------------------------------------------------------------
+
+
+from devtools_testutils.perfstress_tests._random_stream import get_random_bytes, _DEFAULT_LENGTH
+
+
+# The AsyncIteratorRandomStream is used specifically for the AsyncHttpXTransport upload stream scenario, since the
+# async httpx transport requires the request body stream to be type AsyncIterator (i.e. have an __aiter__ and __anext__ method rather than __iter__).
+# Specific check in httpx here: https://github.com/encode/httpx/blob/7df47ce4d93a06f2c3310cd692b4c2336d7663ba/httpx/_content.py#L116.
+
+class AsyncIteratorRandomStream():
+    def __init__(self, length, initial_buffer_length=_DEFAULT_LENGTH):
+        super().__init__()
+        self._base_data = get_random_bytes(initial_buffer_length)
+        self._data_length = length
+        self._base_buffer_length = initial_buffer_length
+        self._position = 0
+        self._remaining = length
+        self._closed = False
+
+    def __len__(self):
+        return self._remaining
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if self._remaining == 0:
+            raise StopAsyncIteration
+        return self.read()
+
+    def reset(self):
+        self._position = 0
+        self._remaining = self._data_length
+        self._closed = False
+
+    def read(self, size=None):
+        if self._remaining == 0:
+            return b""
+
+        if size is None:
+            e = self._base_buffer_length
+        else:
+            e = size
+        e = min(e, self._remaining)
+        if e > self._base_buffer_length:
+            self._base_data = get_random_bytes(e)
+            self._base_buffer_length = e
+        self._remaining = self._remaining - e
+        self._position += e
+        return self._base_data[:e]
+
+    def seek(self, index, whence=0):
+        if whence == 0:
+            self._position = index
+        elif whence == 1:
+            self._position = self._position + index
+        elif whence == 2:
+            self._position = self._data_length - 1 + index
+
+    def tell(self):
+        return self._position
+
+    def remaining(self):
+        return self._remaining
+
+    def close(self):
+        self._closed = True

sdk/core/corehttp/tests/perf_tests/upload_binary.py

@@ -6,6 +6,7 @@
 from time import time
 from wsgiref.handlers import format_date_time
 from devtools_testutils.perfstress_tests import RandomStream, AsyncRandomStream
+from ._async_iterator_random_stream import AsyncIteratorRandomStream
 
 from corehttp.rest import HttpRequest
 from corehttp.exceptions import (
@@ -29,7 +30,10 @@ def __init__(self, arguments):
         blob_name = "uploadtest"
         self.blob_endpoint = f"{self.account_endpoint}{self.container_name}/{blob_name}"
         self.upload_stream = RandomStream(self.args.size)
-        self.upload_stream_async = AsyncRandomStream(self.args.size)
+        if self.args.transport == "httpx":
+            self.upload_stream_async = AsyncIteratorRandomStream(self.args.size)
+        else:
+            self.upload_stream_async = AsyncRandomStream(self.args.size)
 
     def run_sync(self):
         self.upload_stream.reset()