Remove per-message Task.Run from Service Bus inline message deserialization

What is the issue

ServiceBusUtils.LoadMessageStreamAsync queues synchronous, in-memory work to the thread pool for every message whose body is stored inline:

durabletask/src/DurableTask.ServiceBus/Common/ServiceBusUtils.cs

Lines 290 to 307 in 5217032

    
                   static Task<Stream> LoadMessageStreamAsync(Message message, IOrchestrationServiceBlobStore orchestrationServiceBlobStore) 
        
                   { 
        
                       string blobKey = string.Empty; 
        
                       if (message.UserProperties.TryGetValue(ServiceBusConstants.MessageBlobKey, out object blobKeyObj)) 
        
                       { 
        
                           blobKey = (string)blobKeyObj; 
        
                       } 
        
                       if (string.IsNullOrWhiteSpace(blobKey)) 
        
                       { 
        
                           // load the stream from the message directly if the blob key property is not set, 
        
                           // i.e., it is not stored externally 
        
           #if NETSTANDARD2_0 
        
                           return Task.Run(() => new System.IO.MemoryStream(message.Body) as Stream); 
        
           #else 
        
                           return Task.Run(() => message.GetBody<Stream>()); 
        
           #endif

For netstandard2.0, the work is only new MemoryStream(message.Body). For net48, message.GetBody<Stream>() reads the already-received brokered-message body. Neither branch performs asynchronous I/O, but both use Task.Run.

The orchestration and tracking receivers deserialize whole batches through this method using Task.WhenAll:

durabletask/src/DurableTask.ServiceBus/ServiceBusOrchestrationService.cs

Lines 543 to 561 in 5217032

    
                       // TODO : Here and elsewhere, consider standard retry block instead of our own hand rolled version 
        
                       IList<Message> newMessages = 
        
                           (await Utils.ExecuteWithRetries(() => session.ReceiveAsync(this.Settings.PrefetchCount), 
        
                               session.SessionId, "Receive Session Message Batch", this.Settings.MaxRetries, this.Settings.IntervalBetweenRetriesSecs)).Cast<Message>().ToList(); 
        
                       this.ServiceStats.OrchestrationDispatcherStats.MessagesReceived.Increment(newMessages.Count); 
        
                       TraceHelper.TraceSession( 
        
                           TraceEventType.Information, 
        
                           "ServiceBusOrchestrationService-LockNextTaskOrchestrationWorkItem-MessageToProcess", 
        
                           session.SessionId, 
        
                           GetFormattedLog( 
        
                               $@"{newMessages.Count} new messages to process: { 
        
                                       string.Join(",", newMessages.Select(m => m.MessageId))}, max latency: { 
        
                                       newMessages.Max(message => message.DeliveryLatency())}ms")); 
        
                       ServiceBusUtils.CheckAndLogDeliveryCount(session.SessionId, newMessages, this.Settings.MaxTaskOrchestrationDeliveryCount); 
        
                       IList<TaskMessage> newTaskMessages = await Task.WhenAll( 
        
                           newMessages.Select(async message => await ServiceBusUtils.GetObjectFromBrokeredMessageAsync<TaskMessage>(message, this.BlobStore)));

durabletask/src/DurableTask.ServiceBus/ServiceBusOrchestrationService.cs

Lines 1358 to 1372 in 5217032

    
           IList<Message> newMessages = 
        
               (await Utils.ExecuteWithRetries(() => session.ReceiveAsync(this.Settings.PrefetchCount), 
        
                   session.SessionId, "Receive Tracking Session Message Batch", this.Settings.MaxRetries, this.Settings.IntervalBetweenRetriesSecs)).Cast<Message>().ToList(); 
        
           this.ServiceStats.TrackingDispatcherStats.MessagesReceived.Increment(newMessages.Count); 
        
           TraceHelper.TraceSession( 
        
               TraceEventType.Information, 
        
               "ServiceBusOrchestrationService-FetchTrackingWorkItem-Messages", 
        
               session.SessionId, 
        
               GetFormattedLog($"{newMessages.Count} new tracking messages to process: {string.Join(",", newMessages.Select(m => m.MessageId))}")); 
        
           ServiceBusUtils.CheckAndLogDeliveryCount(newMessages, this.Settings.MaxTrackingDeliveryCount); 
        
           IList<TaskMessage> newTaskMessages = await Task.WhenAll( 
        
               newMessages.Select(async message => await ServiceBusUtils.GetObjectFromBrokeredMessageAsync<TaskMessage>(message, this.BlobStore)));

The configured prefetch count is 50, so a full batch can enqueue 50 trivial thread-pool work items at once:

durabletask/src/DurableTask.ServiceBus/Settings/ServiceBusOrchestrationServiceSettings.cs

Lines 70 to 73 in 5217032

    
           /// <summary> 
        
           /// Gets the message prefetch count 
        
           /// </summary> 
        
           public int PrefetchCount { get; } = 50;

Performance impact

Each inline message creates and schedules an unnecessary work item plus its task/delegate state. Under sustained load, batches from multiple dispatchers create bursts of thread-pool queueing that add scheduling latency, consume worker threads, and increase short-lived allocations. Thread-pool ramp-up or contention can amplify first-batch and tail latency even though there is no I/O to overlap.

The overhead scales with message rate and is paid before every inline task-message deserialization. The external-blob path is genuinely asynchronous and is not affected by this concern.

Proposed backward-compatible solution

Keep the existing private Task<Stream> signature and return an already-completed task for inline bodies:

#if NETSTANDARD2_0
return Task.FromResult<Stream>(new MemoryStream(message.Body));
#else
return Task.FromResult(message.GetBody<Stream>());
#endif

This preserves the same stream construction and downstream deserialization behavior while removing the thread-pool hop. Leave the blob-store load path unchanged.

Validation

Add coverage for both target-framework branches confirming that inline bodies deserialize identically and the returned task is already complete.
Retain integration coverage for external blob-backed messages.
Benchmark batch deserialization at the default prefetch size, comparing elapsed time, allocations, thread-pool work-item count, and tail latency.

Navigation Menu

Repository navigation

Remove per-message Task.Run from Service Bus inline message deserialization #1376

What is the issue

Performance impact

Proposed backward-compatible solution

Validation

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	static Task<Stream> LoadMessageStreamAsync(Message message, IOrchestrationServiceBlobStore orchestrationServiceBlobStore)
	{
	string blobKey = string.Empty;

	if (message.UserProperties.TryGetValue(ServiceBusConstants.MessageBlobKey, out object blobKeyObj))
	{
	blobKey = (string)blobKeyObj;
	}

	if (string.IsNullOrWhiteSpace(blobKey))
	{
	// load the stream from the message directly if the blob key property is not set,
	// i.e., it is not stored externally
	#if NETSTANDARD2_0
	return Task.Run(() => new System.IO.MemoryStream(message.Body) as Stream);
	#else
	return Task.Run(() => message.GetBody<Stream>());
	#endif

	// TODO : Here and elsewhere, consider standard retry block instead of our own hand rolled version
	IList<Message> newMessages =
	(await Utils.ExecuteWithRetries(() => session.ReceiveAsync(this.Settings.PrefetchCount),
	session.SessionId, "Receive Session Message Batch", this.Settings.MaxRetries, this.Settings.IntervalBetweenRetriesSecs)).Cast<Message>().ToList();

	this.ServiceStats.OrchestrationDispatcherStats.MessagesReceived.Increment(newMessages.Count);
	TraceHelper.TraceSession(
	TraceEventType.Information,
	"ServiceBusOrchestrationService-LockNextTaskOrchestrationWorkItem-MessageToProcess",
	session.SessionId,
	GetFormattedLog(
	$@"{newMessages.Count} new messages to process: {
	string.Join(",", newMessages.Select(m => m.MessageId))}, max latency: {
	newMessages.Max(message => message.DeliveryLatency())}ms"));

	ServiceBusUtils.CheckAndLogDeliveryCount(session.SessionId, newMessages, this.Settings.MaxTaskOrchestrationDeliveryCount);

	IList<TaskMessage> newTaskMessages = await Task.WhenAll(
	newMessages.Select(async message => await ServiceBusUtils.GetObjectFromBrokeredMessageAsync<TaskMessage>(message, this.BlobStore)));

	IList<Message> newMessages =
	(await Utils.ExecuteWithRetries(() => session.ReceiveAsync(this.Settings.PrefetchCount),
	session.SessionId, "Receive Tracking Session Message Batch", this.Settings.MaxRetries, this.Settings.IntervalBetweenRetriesSecs)).Cast<Message>().ToList();
	this.ServiceStats.TrackingDispatcherStats.MessagesReceived.Increment(newMessages.Count);

	TraceHelper.TraceSession(
	TraceEventType.Information,
	"ServiceBusOrchestrationService-FetchTrackingWorkItem-Messages",
	session.SessionId,
	GetFormattedLog($"{newMessages.Count} new tracking messages to process: {string.Join(",", newMessages.Select(m => m.MessageId))}"));

	ServiceBusUtils.CheckAndLogDeliveryCount(newMessages, this.Settings.MaxTrackingDeliveryCount);

	IList<TaskMessage> newTaskMessages = await Task.WhenAll(
	newMessages.Select(async message => await ServiceBusUtils.GetObjectFromBrokeredMessageAsync<TaskMessage>(message, this.BlobStore)));

	/// <summary>
	/// Gets the message prefetch count
	/// </summary>
	public int PrefetchCount { get; } = 50;

Remove per-message Task.Run from Service Bus inline message deserialization #1376

Description

What is the issue

Performance impact

Proposed backward-compatible solution

Validation

Activity

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions