From 464fc7028f7a0ba34dacc0a512cab988f825709a Mon Sep 17 00:00:00 2001 From: amdeel <52223332+amdeel@users.noreply.github.com> Date: Thu, 28 Oct 2021 10:28:44 -0700 Subject: [PATCH 1/4] added request throttling --- .../Storage/AzureStorageClient.cs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs b/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs index 4a7d675a9..1ab64a84e 100644 --- a/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs +++ b/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs @@ -29,6 +29,7 @@ class AzureStorageClient readonly CloudBlobClient blobClient; readonly CloudQueueClient queueClient; readonly CloudTableClient tableClient; + readonly SemaphoreSlim semaphore; public AzureStorageClient(AzureStorageOrchestrationServiceSettings settings) : this(settings.StorageAccountDetails == null ? @@ -61,6 +62,8 @@ public AzureStorageClient(CloudStorageAccount account, AzureStorageOrchestration } this.tableClient.BufferManager = SimpleBufferManager.Shared; + + this.semaphore = new SemaphoreSlim(this.Settings.MaxStorageOperationConcurrency); } public AzureStorageOrchestrationServiceSettings Settings { get; } @@ -101,14 +104,21 @@ public Table GetTableReference(string tableName) public async Task MakeStorageRequest(Func> storageRequest, string operationName, string? clientRequestId = null) { + await semaphore.WaitAsync(); + try { - return await TimeoutHandler.ExecuteWithTimeout(operationName, this.StorageAccountName, this.Settings, storageRequest, this.Stats, clientRequestId); + var result = await TimeoutHandler.ExecuteWithTimeout(operationName, this.StorageAccountName, this.Settings, storageRequest, this.Stats, clientRequestId); + return result; } catch (StorageException ex) { throw new DurableTaskStorageException(ex); } + finally + { + semaphore.Release(); + } } public async Task MakeStorageRequest(Func storageRequest, string operationName, string? clientRequestId = null) From 6be2e265d7216ea62958ea529c75d900f701da45 Mon Sep 17 00:00:00 2001 From: amdeel <52223332+amdeel@users.noreply.github.com> Date: Thu, 28 Oct 2021 10:38:31 -0700 Subject: [PATCH 2/4] removed dispatch queue --- .../OrchestrationSessionManager.cs | 74 +++++++++---------- 1 file changed, 35 insertions(+), 39 deletions(-) diff --git a/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs b/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs index 276f0abca..9f04e3290 100644 --- a/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs +++ b/src/DurableTask.AzureStorage/OrchestrationSessionManager.cs @@ -434,7 +434,7 @@ internal void AddMessageToPendingOrchestration( // Before the batch of messages can be processed, we need to download the latest execution state. // This is done beforehand in the background as a performance optimization. - this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken); + Task.Run(() => this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken)); } // New messages are added; duplicate messages are replaced @@ -454,55 +454,51 @@ internal void AddMessageToPendingOrchestration( } // This method runs on a background task thread - void ScheduleOrchestrationStatePrefetch( + async Task ScheduleOrchestrationStatePrefetch( LinkedListNode node, Guid traceActivityId, CancellationToken cancellationToken) { PendingMessageBatch batch = node.Value; - // Do the fetch in a background thread - this.fetchRuntimeStateQueue.EnqueueAndDispatch(async delegate + AnalyticsEventSource.SetLogicalTraceActivityId(traceActivityId); + + try { - AnalyticsEventSource.SetLogicalTraceActivityId(traceActivityId); - - try + if (batch.OrchestrationState == null) { - if (batch.OrchestrationState == null) - { - OrchestrationHistory history = await this.trackingStore.GetHistoryEventsAsync( - batch.OrchestrationInstanceId, - batch.OrchestrationExecutionId, - cancellationToken); - - batch.OrchestrationState = new OrchestrationRuntimeState(history.Events); - batch.ETag = history.ETag; - batch.LastCheckpointTime = history.LastCheckpointTime; - } - - this.readyForProcessingQueue.Enqueue(node); + OrchestrationHistory history = await this.trackingStore.GetHistoryEventsAsync( + batch.OrchestrationInstanceId, + batch.OrchestrationExecutionId, + cancellationToken); + + batch.OrchestrationState = new OrchestrationRuntimeState(history.Events); + batch.ETag = history.ETag; + batch.LastCheckpointTime = history.LastCheckpointTime; } - catch (OperationCanceledException) - { - // shutting down - } - catch (Exception e) - { - this.settings.Logger.OrchestrationProcessingFailure( - this.storageAccountName, - this.settings.TaskHubName, - batch.OrchestrationInstanceId, - batch.OrchestrationExecutionId, - e.ToString()); - // Sleep briefly to avoid a tight failure loop. - await Task.Delay(TimeSpan.FromSeconds(5)); + this.readyForProcessingQueue.Enqueue(node); + } + catch (OperationCanceledException) + { + // shutting down + } + catch (Exception e) + { + this.settings.Logger.OrchestrationProcessingFailure( + this.storageAccountName, + this.settings.TaskHubName, + batch.OrchestrationInstanceId, + batch.OrchestrationExecutionId, + e.ToString()); - // This is a background operation so failure is not an option. All exceptions must be handled. - // To avoid starvation, we need to re-enqueue this async operation instead of retrying in a loop. - this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken); - } - }); + // Sleep briefly to avoid a tight failure loop. + await Task.Delay(TimeSpan.FromSeconds(5)); + + // This is a background operation so failure is not an option. All exceptions must be handled. + // To avoid starvation, we need to re-enqueue this async operation instead of retrying in a loop. + await Task.Run(() => this.ScheduleOrchestrationStatePrefetch(node, traceActivityId, cancellationToken)); + } } public async Task GetNextSessionAsync(CancellationToken cancellationToken) From 780368e083bf6eb2ed27fa3c2deafc6dbb90f800 Mon Sep 17 00:00:00 2001 From: amdeel <52223332+amdeel@users.noreply.github.com> Date: Thu, 28 Oct 2021 10:40:38 -0700 Subject: [PATCH 3/4] reverted accidental change --- src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs b/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs index 1ab64a84e..d18d143fc 100644 --- a/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs +++ b/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs @@ -108,8 +108,7 @@ public async Task MakeStorageRequest(Func(operationName, this.StorageAccountName, this.Settings, storageRequest, this.Stats, clientRequestId); - return result; + return await TimeoutHandler.ExecuteWithTimeout(operationName, this.StorageAccountName, this.Settings, storageRequest, this.Stats, clientRequestId); } catch (StorageException ex) { From d628b2556caf565e59b65bc1290a6d6dae653335 Mon Sep 17 00:00:00 2001 From: amdeel <52223332+amdeel@users.noreply.github.com> Date: Thu, 28 Oct 2021 14:19:02 -0700 Subject: [PATCH 4/4] changed semaphore variable name --- .../Storage/AzureStorageClient.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs b/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs index d18d143fc..b5a0672af 100644 --- a/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs +++ b/src/DurableTask.AzureStorage/Storage/AzureStorageClient.cs @@ -29,7 +29,7 @@ class AzureStorageClient readonly CloudBlobClient blobClient; readonly CloudQueueClient queueClient; readonly CloudTableClient tableClient; - readonly SemaphoreSlim semaphore; + readonly SemaphoreSlim requestThrottleSemaphore; public AzureStorageClient(AzureStorageOrchestrationServiceSettings settings) : this(settings.StorageAccountDetails == null ? @@ -63,7 +63,7 @@ public AzureStorageClient(CloudStorageAccount account, AzureStorageOrchestration this.tableClient.BufferManager = SimpleBufferManager.Shared; - this.semaphore = new SemaphoreSlim(this.Settings.MaxStorageOperationConcurrency); + this.requestThrottleSemaphore = new SemaphoreSlim(this.Settings.MaxStorageOperationConcurrency); } public AzureStorageOrchestrationServiceSettings Settings { get; } @@ -104,7 +104,7 @@ public Table GetTableReference(string tableName) public async Task MakeStorageRequest(Func> storageRequest, string operationName, string? clientRequestId = null) { - await semaphore.WaitAsync(); + await requestThrottleSemaphore.WaitAsync(); try { @@ -116,7 +116,7 @@ public async Task MakeStorageRequest(Func