[Bug]Concurrent upload (Azure#11668)

rakshith91 · Jun 6, 2020 · 0245fce · 0245fce
1 parent c35c3a5
commit 0245fce
Show file tree

Hide file tree

Showing 8 changed files with 24 additions and 24 deletions.
diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/uploads.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/uploads.py
@@ -32,11 +32,11 @@ def _parallel_uploads(executor, uploader, pending, running):
         done, running = futures.wait(running, return_when=futures.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(executor.submit(with_current_context(uploader), next_chunk))
         except StopIteration:
             break
-        else:
-            running.add(executor.submit(with_current_context(uploader), next_chunk))
 
     # Wait for the remaining uploads to finish
     done, _running = futures.wait(running)

diff --git a/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/uploads_async.py b/sdk/storage/azure-storage-blob/azure/storage/blob/_shared/uploads_async.py
@@ -32,11 +32,11 @@ async def _parallel_uploads(uploader, pending, running):
         done, running = await asyncio.wait(running, return_when=asyncio.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(asyncio.ensure_future(uploader(next_chunk)))
         except StopIteration:
             break
-        else:
-            running.add(asyncio.ensure_future(uploader(next_chunk)))
 
     # Wait for the remaining uploads to finish
     if running:

diff --git a/sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/uploads.py b/sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/uploads.py
@@ -32,11 +32,11 @@ def _parallel_uploads(executor, uploader, pending, running):
         done, running = futures.wait(running, return_when=futures.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(executor.submit(with_current_context(uploader), next_chunk))
         except StopIteration:
             break
-        else:
-            running.add(executor.submit(with_current_context(uploader), next_chunk))
 
     # Wait for the remaining uploads to finish
     done, _running = futures.wait(running)

diff --git a/sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/uploads_async.py b/sdk/storage/azure-storage-file-datalake/azure/storage/filedatalake/_shared/uploads_async.py
@@ -32,11 +32,11 @@ async def _parallel_uploads(uploader, pending, running):
         done, running = await asyncio.wait(running, return_when=asyncio.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(asyncio.ensure_future(uploader(next_chunk)))
         except StopIteration:
             break
-        else:
-            running.add(asyncio.ensure_future(uploader(next_chunk)))
 
     # Wait for the remaining uploads to finish
     if running:

diff --git a/sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/uploads.py b/sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/uploads.py
@@ -32,11 +32,11 @@ def _parallel_uploads(executor, uploader, pending, running):
         done, running = futures.wait(running, return_when=futures.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(executor.submit(with_current_context(uploader), next_chunk))
         except StopIteration:
             break
-        else:
-            running.add(executor.submit(with_current_context(uploader), next_chunk))
 
     # Wait for the remaining uploads to finish
     done, _running = futures.wait(running)

diff --git a/sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/uploads_async.py b/sdk/storage/azure-storage-file-share/azure/storage/fileshare/_shared/uploads_async.py
@@ -32,11 +32,11 @@ async def _parallel_uploads(uploader, pending, running):
         done, running = await asyncio.wait(running, return_when=asyncio.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(asyncio.ensure_future(uploader(next_chunk)))
         except StopIteration:
             break
-        else:
-            running.add(asyncio.ensure_future(uploader(next_chunk)))
 
     # Wait for the remaining uploads to finish
     if running:

diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/uploads.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/uploads.py
@@ -32,11 +32,11 @@ def _parallel_uploads(executor, uploader, pending, running):
         done, running = futures.wait(running, return_when=futures.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(executor.submit(with_current_context(uploader), next_chunk))
         except StopIteration:
             break
-        else:
-            running.add(executor.submit(with_current_context(uploader), next_chunk))
 
     # Wait for the remaining uploads to finish
     done, _running = futures.wait(running)

diff --git a/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/uploads_async.py b/sdk/storage/azure-storage-queue/azure/storage/queue/_shared/uploads_async.py
@@ -32,11 +32,11 @@ async def _parallel_uploads(uploader, pending, running):
         done, running = await asyncio.wait(running, return_when=asyncio.FIRST_COMPLETED)
         range_ids.extend([chunk.result() for chunk in done])
         try:
-            next_chunk = next(pending)
+            for _ in range(0, len(done)):
+                next_chunk = next(pending)
+                running.add(asyncio.ensure_future(uploader(next_chunk)))
         except StopIteration:
             break
-        else:
-            running.add(asyncio.ensure_future(uploader(next_chunk)))
 
     # Wait for the remaining uploads to finish
     if running: