Software Design Blog - Queues

Push queue processing boundaries with 3x greater throughput

Sat, 26 Dec 2015 10:54:00 +1300

The problem is how can we measure the queue processing throughput? Or in a production environment, how can we add a performance counter without modifying our existing infrastructure?

The solution is to add a decorator pattern that will act as a proxy between the QueueMonitor class and the actual command processing class to intercept the communication and add instrumentation.

The previous post used the competing consumer pattern to process a queue concurrently. The message queue strategies post used a single processor to process a queue.

In this post, we will add instrumentation to the single queue processor and the competing consumer queue processor to compare performance.

Download Source Code

Setup

The experiment will add 512KB messages to a queue to simulate a decent workload. The intention is to read the messages from the queue and call a File SMTP mailer server that will write the emails to disk as fast as possible.

The core components such as the Queue Monitor were covered in the Message Queue Delivery Strategies post. The queue monitor is responsible for reading messages from the queue and pushing the message content to a command to execute.

The classes that will be used in this experiment are listed below.

public class FileSmtpMailer : IMailer
{
private readonly string _path;
public FileSmtpMailer(string path)
{
if (path == null) throw new ArgumentNullException("path");
_path = path;
}
public void Send(string message, string sender, string recipients, string subject)
{
using (var client = new SmtpClient())
{
// This can be configured in the app.config
client.DeliveryMethod = SmtpDeliveryMethod.SpecifiedPickupDirectory;
client.PickupDirectoryLocation = _path;
using (var mailMessage = new MailMessage(sender, recipients,
subject, message))
{
mailMessage.IsBodyHtml = true;
client.Send(mailMessage);
}
}
}
}
public class MailProcessorCommand : ICommand<OrderMessage>
{
private readonly IMailer _mailer;
public MailProcessorCommand(IMailer mailer)
{
if (mailer == null) throw new ArgumentNullException("mailer");
_mailer = mailer;
}
public void Execute(OrderMessage message)
{
if (message == null) throw new ArgumentNullException("message");
_mailer.Send(message.Body, message.Sender,
message.Recipients, message.Subject);
}
}

Instrumentation

The TimerCommandDecorator decorator will be placed in between the QueueMonitor and the MailProcessorComamnd in order to measure the performance throughput.

public class TimerCommandDecorator<T> : ICommand<T>
{
private readonly ICommand<T> _next;
private int _messages;
private Stopwatch _stopwatch;
public TimerCommandDecorator(ICommand<T> next)
{
if (next == null) throw new ArgumentNullException("next");
_next = next;
}
public void Execute(T message)
{
if (_stopwatch == null) _stopwatch = Stopwatch.StartNew();
_next.Execute(message);
_messages += 1;
Console.WriteLine("Processing {0} messages took {1} sec",
_messages, _stopwatch.Elapsed.TotalSeconds);
}
}

The TimerCommandDecorator implementation can easily be swapped out with a perfmon version that will increment the performance counter for each message.

Single vs Competing Consumers

Let's run the experiment by observing the throughput difference between the single processor and competing consumer receivers by processing 1000 messages.

The single processor version:

using (var queue = new MessageQueue(QueuePath))
{
queue.Formatter = new BinaryMessageFormatter();
var message = new OrderMessage()
{
Recipients = "[email protected]",
Sender = "[email protected]",
Subject = "Hello World",
Body = emailBody // loaded from a file
};
// Send messages to the queue
for (var i = 0; i < 1000; i++)
{
queue.Send(message);
}
var fileMailer = new FileSmtpMailer(@"C:\temp");
var command = new MailProcessorCommand(fileMailer);
var timerCommand = new TimerCommandDecorator<OrderMessage>(command);
var queueMonitor = new QueueMonitor<OrderMessage>(queue, timerCommand);
queueMonitor.Start();
}

Processing 1000 messages took 47.3004132 sec

The competing consumer processor version with 10 processors:

// Create 10 processors
var commands = new List<ICommand<OrderMessage>>();
for (var i = 0; i < 10; i++)
{
commands.Add(new MailProcessorCommand(fileMailer));
}
var command = new CompositeCompetingCommandProcessor<OrderMessage>(commands);
var timerCommand = new TimerCommandDecorator<OrderMessage>(command);
var queueMonitor = new QueueMonitor<OrderMessage>(queue, timerCommand);
queueMonitor.Start();

Processing 1000 messages took 16.241723 sec

Success! The competing consumer pattern processed the mail queue 3 times faster.

We could experiment using various numbers of concurrent receivers to determine how performance will be affected.

Note: The disk was saturated with request and was running at 100% capacity.

The competing consumer pattern is a great option to fully utilize resources but it may have a negative impact on other tenants of the system. For example, if we call an external API with too many requests, it may appear like a denial-of-service attack.

Summary

The advantage of the competing consumers pattern is the flexibility to configure the number of concurrent consumers to produce a balanced outcome. It is however difficult to determine the throughput and performance impact without instrumentation in place to measure resource utilization.

An approach was covered to add performance instrumentation. The performance metrics revealed that the queue processing throughput can be improved significantly using the competing consumer queue processor compared to the single queue processor.

Dispatching messages with the Competing Consumers Pattern

Thu, 24 Dec 2015 23:27:00 +1300

The previous post focused on fanning out the same queued message to multiple consumers.

The problem is that a single worker is great at processing a single message at a time but how do we increase the throughput by processing messages faster?

The solution is to use the Competing Consumers Pattern, which enables multiple concurrent workers to process messages received in the same queue. This approach will optimise throughput and balance the workload.

For example, travellers often queue at the airport check-in. Checking in would be a lengthy process if there was a long queue but only one check-in operator. The solution is to employee multiple operators to service the same queue to process it faster.

Download Source Code

Setup

The OrderProcessorCommand class used in the previous post was modified on line 19 below to simulate each worker taking a different duration to process a request. For example, worker 1 will take 1 second to process a request whereas worker 2 will take 2 seconds to process a request.

public interface ICommand<in T>
{
void Execute(T message);
}
public class OrderProcessorCommand : ICommand<OrderModel>
{
private readonly int _id;
public OrderProcessorCommand(int id)
{
_id = id;
}
public void Execute(OrderModel message)
{
if (message == null) throw new ArgumentNullException("message");
var start = DateTime.Now;
Thread.Sleep(TimeSpan.FromSeconds(_id));
Console.WriteLine("Processed {0} by worker {1} from {2} to {3}",
message.Name, _id, start.ToString("h:mm:ss"),
DateTime.Now.ToString("h:mm:ss"));
}
}

Competing Consumers

The pool of available commands are orchestrated using the CompositeCompetingCommandProcessor class.

The class will automatically block the queue monitor on line 33 from retrieving and pushing in the next message when all of the command processors are busy. The queue monitor will be unblocked as soon as the first command processor becomes available.

The WorkingCommandModel is used to associate a command with a task.

public class WorkingCommandModel<T>
{
public ICommand<T> Command { get; set; }
public Task Task { get; set; }
}
public class CompositeCompetingCommandProcessor<T> : ICommand<T>
{
private readonly IEnumerable<WorkingCommandModel<T>> _workers;
public CompositeCompetingCommandProcessor(IEnumerable<ICommand<T>> commands)
{
if (commands == null) throw new ArgumentNullException("commands");
_workers = commands.Select(
c => new WorkingCommandModel<T> { Command = c }).ToList();
}
public void Execute(T message)
{
WorkingCommandModel<T> worker = null;
while (worker == null)
{
worker = GetAvailableWorker();
if (worker == null) WaitForWorkerAvailability();
}
worker.Task = Task.Factory.StartNew(() =>
worker.Command.Execute(message),
TaskCreationOptions.LongRunning);
// Block new messages from arriving until a worker is available
if (GetAvailableWorker() == null) WaitForWorkerAvailability();
}
private void WaitForWorkerAvailability()
{
var tasks = (from w in _workers where w.Task != null select w.Task);
Task.WaitAny(tasks.ToArray());
}
private WorkingCommandModel<T> GetAvailableWorker()
{
var worker = _workers.FirstOrDefault(w =>
w.Task == null || w.Task.IsCompleted);
if (worker != null && worker.Task != null) worker.Task.Dispose();
return worker;
}
}

Let's run the solution using 3 workers to process 9 unique messages in the queue.

using (var queue = new MessageQueue(QueuePath))
{
queue.Formatter = new BinaryMessageFormatter();
// Write 9 messages to the queue
for (var orderId = 1; orderId <= 9; orderId++)
{
var order = new OrderModel()
{
Id = orderId,
Name = string.Format("Order {0}", orderId)
};
queue.Send(order);
}
// Create 3 workers
var workers = new List<ICommand<OrderModel>>();
for (var workerId = 1; workerId <= 3; workerId++)
{
workers.Add(new OrderProcessorCommand(workerId));
}
// Process the queue
var command = new CompositeCompetingCommandProcessor<OrderModel>(workers);
var queueMonitor = new QueueMonitor<OrderModel>(queue, command);
queueMonitor.Start();
}

Processed Order 1 by worker 1 from 6:42:48 to 6:42:49
Processed Order 2 by worker 2 from 6:42:48 to 6:42:50
Processed Order 4 by worker 1 from 6:42:49 to 6:42:50
Processed Order 3 by worker 3 from 6:42:48 to 6:42:51
Processed Order 6 by worker 1 from 6:42:50 to 6:42:51
Processed Order 5 by worker 2 from 6:42:50 to 6:42:52
Processed Order 8 by worker 1 from 6:42:51 to 6:42:52
Processed Order 7 by worker 3 from 6:42:51 to 6:42:54
Processed Order 9 by worker 2 from 6:42:52 to 6:42:54

As we can see from the results above, worker 1 processed 4 messages whereas worker 3 only processed 2 messages.

Warning: The message order is no longer guaranteed.

The order can sometimes be important. For example, a reservation system may processes new bookings, updates and cancellations. Let's say the traveller creates a booking and decides to cancel it immediately. When both requests are in a queue processed by multiple workers then there is a potential that the cancellation will be processed before the create.

Summary

A practical example was provided to process messages from a single queue using multiple workers by applying the competing consumers pattern.

The drawback of potentially processing messages out of order was briefly covered. With enough interest, I may create a follow up post to address the order issue using various techniques such as a sequential message convoy, applying sessions and using correlation IDs to preserve delivery order.

The next post will provide an illustration for adding instrumentation to monitor queue processing throughput. The performance of the competing consumers solution is compared with the single worker solution.

Broadcasting messages using the composite pattern

Mon, 21 Dec 2015 23:59:00 +1300

The previous post focused on reading messages from a queue and pushing the messages to a single command for processing.

The problem is how can we notify multiple commands to process the same request? For example, when an order is received, we may want a command to 1) send an email, 2) add an audit record and 3) call an external API.

The solution is to use the composite pattern, whereby a group of commands are treated in a similar way as a single command to fan out the messages.

Download Source Code

Setup

The core components such as the Queue Monitor were covered in the Message Queue Delivery Strategies post.

Sequential Processing

The CompositeSequentialBroadcastCommand class below implements ICommand to indicate that it exhibits the behaviour of a command. The responsibility of the composite class is to loop though a collection of commands and call each command with the same request.

public interface ICommand<in T>
{
void Execute(T message);
}
public class OrderProcessorCommand : ICommand<OrderModel>
{
private readonly int _id;
public OrderProcessorCommand(int id)
{
_id = id;
}
public void Execute(OrderModel message)
{
if (message == null) throw new ArgumentNullException("message");
var start = DateTime.Now;
Thread.Sleep(TimeSpan.FromSeconds(2));
Console.WriteLine("Processed {0} by worker {1} from {2} to {3}",
message.Name, _id, start.ToString("h:mm:ss"),
DateTime.Now.ToString("h:mm:ss"));
}
}
public class CompositeSequentialBroadcastCommand<T> : ICommand<T>
{
private readonly IEnumerable<ICommand<T>> _commands;
public CompositeSequentialBroadcastCommand(IEnumerable<ICommand<T>> commands)
{
if (commands == null) throw new ArgumentNullException("commands");
_commands = commands.ToList();
}
public void Execute(T message)
{
foreach (var command in _commands)
{
command.Execute(message);
}
}
}

Let's see what happens when 3 sequential workers are configured to process 2 messages.

using (var queue = new MessageQueue(@".\Private$\Orders"))
{
queue.Formatter = new BinaryMessageFormatter();
var workers = new List<ICommand<OrderModel>>();
for (var workerId = 1; workerId <= 3; workerId++)
{
workers.Add(new OrderProcessorCommand(workerId));
}
var command = new CompositeSequentialBroadcastCommand<OrderModel>(workers);
var queueMonitor = new QueueMonitor<OrderModel>(queue, command);
queueMonitor.Start();
}

Processed Order 1 by worker 1 from 6:26:34 to 6:26:36
Processed Order 1 by worker 2 from 6:26:36 to 6:26:38
Processed Order 1 by worker 3 from 6:26:38 to 6:26:40
Processed Order 2 by worker 1 from 6:26:40 to 6:26:42
Processed Order 2 by worker 2 from 6:26:42 to 6:26:44
Processed Order 2 by worker 3 from 6:26:44 to 6:26:46

The output above shows that each worker processed the same message in sequence. If worker 2 failed then worker 3 would not be executed. This can be a great pattern when the sequence is important. For example, only send the welcome email if the order has been created successfully.

Parallel Processing

The CompositeParallelBroadcastCommand class is similar to the previous example. The only difference is that this class will call the commands in parallel.

class CompositeParallelBroadcastCommand<T> : ICommand<T>
{
private readonly IEnumerable<ICommand<T>> _commands;
public CompositeParallelBroadcastCommand(IEnumerable<ICommand<T>> commands)
{
if (commands == null) throw new ArgumentNullException("commands");
_commands = commands.ToList();
}
public void Execute(T message)
{
Parallel.ForEach(_commands, c => c.Execute(message));
}
}

Let's see what happens when parallel 3 workers are configured to process 2 messages.

using (var queue = new MessageQueue(@".\Private$\Orders"))
{
queue.Formatter = new BinaryMessageFormatter();
var workers = new List<ICommand<OrderModel>>();
for (var workerId = 1; workerId <= 3; workerId++)
{
workers.Add(new OrderProcessorCommand(workerId));
}
var command = new CompositeParallelBroadcastCommand<OrderModel>(workers);
var queueMonitor = new QueueMonitor<OrderModel>(queue, command);
queueMonitor.Start();
}

Processed Order 1 by worker 2 from 6:08:36 to 6:08:38
Processed Order 1 by worker 1 from 6:08:36 to 6:08:38
Processed Order 1 by worker 3 from 6:08:36 to 6:08:38
Processed Order 2 by worker 3 from 6:08:38 to 6:08:40
Processed Order 2 by worker 2 from 6:08:38 to 6:08:40
Processed Order 2 by worker 1 from 6:08:38 to 6:08:40

The output above shows that each worker processed the same message concurrently. This can be a great pattern when sequencing doesn�t matters. For example, send a welcome email and add an audit record at the same time.

Summary

This post focused on broadcasting a single message to multiple receivers. The composite pattern allows the behaviour of the Queue Monitor push to change without modifying the code.

Coming soon: The next post will focus on processing a single request by dispatching messaging to a pool of competing consumer processors.