var queue =
this.queueStorage.GetQueueReference(CloudUtility.GetSafeContainerName(queueName));
IEnumerable<CloudQueueMessage> queueMessages =
this.retryPolicy.ExecuteAction<IEnumerable<CloudQueueMessage>>(() =>
{
return queue.GetMessages(Math.Min(count, MaxDequeueMessageCount), visibilityTimeout);
});
// ... There is more code after this point ...
And finally, we are not going to run the dequeue task indefinitely. We provisioned an explicit checkpoint implemented as a QueueEmpty event which is raised whenever a queue becomes empty. At that point, we consult to a QueueEmpty event handler to determine whether or not it permits us to finish the running dequeue task. A well-designed implementation of the
QueueEmpty event handler allows supporting the “auto scale-down” capability as explained in the following section.
Auto Scaling Down Dequeue Tasks
The purpose of QueueEmpty event handler is a two-fold. First, it is responsible for providing feedback to the source dequeue task instructing it to enter a sleep state for a given time interval (as defined in the delay output parameter in the event delegate). Secondly, it indicates to the dequeue task whether or not it must gracefully shut itself down (as prescribed by the Boolean return parameter).
The following implementation of the QueueEmpty event handler solves the two challenges highlighted earlier in this whitepaper. It calculates a random exponential back-off interval and tells the dequeue task to exponentially increase the delay between queue polling requests. Note that the back-off delay will not exceed 1 second as configured in our solution, since it really isn’t necessary to have a long delay between polling when auto-scaling is implemented well enough.
In addition, it queries the queue listener state to determine the number of active dequeue tasks.
Should this number be in excess of 1, the event handler advise the originating dequeue task to complete its polling loop provided the back-off interval has also reached its specified maximum.
Otherwise, the dequeue task will not be terminated, leaving exactly 1 polling thread running at a time per a single instance of the queue listener. This approach helps reduce the number of storage transactions and therefore decrease the transaction costs as explained earlier.
private bool HandleQueueEmptyEvent(object sender, int idleCount, out TimeSpan delay) {
// The sender is an instance of the ICloudQueueServiceWorkerRoleExtension, we can safely perform type casting.
ICloudQueueServiceWorkerRoleExtension queueService = sender as ICloudQueueServiceWorkerRoleExtension;
// Find out which extension is responsible for retrieving the worker role configuration settings.
IWorkItemProcessorConfigurationExtension config = Extensions.Find<IWorkItemProcessorConfigurationExtension>();
// Get the current state of the queue listener to determine point-in-time load characteristics.
CloudQueueListenerInfo queueServiceState = queueService.QueryState();
// Set up the initial parameters, read configuration settings.
int deltaBackoffMs = 100;
int minimumIdleIntervalMs =
Convert.ToInt32(config.Settings.MinimumIdleInterval.TotalMilliseconds);
int maximumIdleIntervalMs =
Convert.ToInt32(config.Settings.MaximumIdleInterval.TotalMilliseconds);
// Calculate a new sleep interval value that will follow a random exponential back-off curve.
int delta = (int)((Math.Pow(2.0, (double)idleCount) - 1.0) * (new Random()).Next((int)(deltaBackoffMs * 0.8), (int)(deltaBackoffMs * 1.2)));
int interval = Math.Min(minimumIdleIntervalMs + delta, maximumIdleIntervalMs);
// Pass the calculated interval to the dequeue task to enable it to enter into a sleep state for the specified duration.
delay = TimeSpan.FromMilliseconds((double)interval);
// As soon as interval reaches its maximum, tell the source dequeue task that it must gracefully terminate itself
// unless this is a last deqeueue task. If so, we are not going to keep it running and continue polling the queue.
return delay.TotalMilliseconds >= maximumIdleIntervalMs &&
queueServiceState.ActiveDequeueTasks > 1;
}
At a higher level, the above described "dequeue task scale-down" capability can be explained as follows:
1. Whenever there is anything in the queue, the dequeue tasks will ensure that the workload will be processed as soon as possible. There will be no delay between requests to dequeue messages from a queue.
2. As soon as the source queue becomes empty, each dequeue task will raise a QueueEmpty event.
3. The QueueEmpty event handler will calculate a random exponential back-off delay and instruct the dequeue task to suspend its activity for a given interval.
4. The dequeue tasks will continue polling the source queue at computed intervals until the idle duration exceeds its allowed maximum.
5. Upon reaching the maximum idle interval, and provided that the source queue is still empty, all active dequeue tasks will start gracefully shutting themselves down – this will not occur all at once, since the dequeue tasks are backing off at different points in the back off algorithm.
6. At some point in time, there will be only one active dequeue task waiting for work. As the result, no idle polling transactions will occur against a queue except only by that single task.
To elaborate on the process of collecting point-in-time load characteristics, it is worth mentioning the relevant source code artifacts. First, there is a structure holding the relevant metrics that measure the result of the load that is being applied to the solution. For the purposes of simplicity, we included a small subset of metrics that will be used further in the sample code.
/// Implements a structure containing point-in-time load characteristics for a given queue listener.
public struct CloudQueueListenerInfo {
/// Returns the approximate number of items in the Windows Azure queue.
public int CurrentQueueDepth { get; internal set; }
/// Returns the number of dequeue tasks that are actively performing work or waiting for work.
public int ActiveDequeueTasks { get; internal set; }
/// Returns the maximum number of dequeue tasks that were active at a time.
public int TotalDequeueTasks { get; internal set; } }
Secondly, there is a method implemented by a queue listener which returns its load metrics as depicted in the following example:
/// Returns the current state of the queue listener to determine point-in-time load characteristics.
public CloudQueueListenerInfo QueryState() {
return new CloudQueueListenerInfo() {
CurrentQueueDepth = this.queueStorage.GetCount(this.queueLocation.QueueName), ActiveDequeueTasks = (from task in this.dequeueTasks where task.Status !=
TaskStatus.Canceled && task.Status != TaskStatus.Faulted && task.Status !=
TaskStatus.RanToCompletion select task).Count(), TotalDequeueTasks = this.dequeueTasks.Count
};
}