Add Latency Benchmark (#120)

ArtemisNetCoreClient.sln

@@ -18,6 +18,12 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Throughput_NMS.AMQP", "benc
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Throughput_ArtemisNetCoreClient", "benchmark\Throughput_ArtemisNetCoreClient\Throughput_ArtemisNetCoreClient.csproj", "{204727DC-1368-4BD9-A918-7788AE474782}"
 EndProject
+Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Latency", "Latency", "{74042889-6E96-4D1F-9C80-9C364E93F754}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Latency_NMS.AMQP", "benchmark\Latency_NMS.AMQP\Latency_NMS.AMQP.csproj", "{F9FEEE09-32B1-4810-92F4-DCC051CA88DB}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Latency_ArtemisNetCoreClient", "benchmark\Latency_ArtemisNetCoreClient\Latency_ArtemisNetCoreClient.csproj", "{21B4D046-1177-485B-BD5E-9C90383A3A6D}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -48,6 +54,14 @@ Global
 		{204727DC-1368-4BD9-A918-7788AE474782}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{204727DC-1368-4BD9-A918-7788AE474782}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{204727DC-1368-4BD9-A918-7788AE474782}.Release|Any CPU.Build.0 = Release|Any CPU
+		{F9FEEE09-32B1-4810-92F4-DCC051CA88DB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{F9FEEE09-32B1-4810-92F4-DCC051CA88DB}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{F9FEEE09-32B1-4810-92F4-DCC051CA88DB}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{F9FEEE09-32B1-4810-92F4-DCC051CA88DB}.Release|Any CPU.Build.0 = Release|Any CPU
+		{21B4D046-1177-485B-BD5E-9C90383A3A6D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{21B4D046-1177-485B-BD5E-9C90383A3A6D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{21B4D046-1177-485B-BD5E-9C90383A3A6D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{21B4D046-1177-485B-BD5E-9C90383A3A6D}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(NestedProjects) = preSolution
 		{12B2FE3E-96BE-4AC2-8D93-9371BB3B1A14} = {F1AC22BC-1EC4-46A7-96B6-6C117EF29DB9}
@@ -56,5 +70,8 @@ Global
 		{A2F7F999-AD5C-4E82-8D01-E05B9FAFE806} = {F1AC22BC-1EC4-46A7-96B6-6C117EF29DB9}
 		{7E352FBC-72E8-4D8B-B35F-5EB7ECC234B3} = {A2F7F999-AD5C-4E82-8D01-E05B9FAFE806}
 		{204727DC-1368-4BD9-A918-7788AE474782} = {A2F7F999-AD5C-4E82-8D01-E05B9FAFE806}
+		{74042889-6E96-4D1F-9C80-9C364E93F754} = {F1AC22BC-1EC4-46A7-96B6-6C117EF29DB9}
+		{F9FEEE09-32B1-4810-92F4-DCC051CA88DB} = {74042889-6E96-4D1F-9C80-9C364E93F754}
+		{21B4D046-1177-485B-BD5E-9C90383A3A6D} = {74042889-6E96-4D1F-9C80-9C364E93F754}
 	EndGlobalSection
 EndGlobal

README.md

@@ -151,6 +151,22 @@ The performance is assessed by measuring the round-trip time and calculating the
   <img src="./readme/PingPong_Benchmark.svg" alt="Benchmark Results Diagram"/>
 </div>
 
+### Latency Benchmark
+
+This benchmark measures the latency of message delivery between a producer and a consumer.
+
+- **`Producer`**: Initiates the sending of 10,000 messages, each embedded with a high-precision timestamp immediately before dispatch.
+- **`Consumer`**: Receives messages and calculates the latency for each by comparing the embedded timestamp with the time of receipt.
+
+| Library                | Average Latency (µs)   |
+|------------------------|------------------------|
+| ArtemisNetCoreClient   | 524.31                 |
+| NMS.AMQP               | 552.49                 |
+
+<div align="center">
+  <img src="./readme/Latency_Benchmark.svg" alt="Benchmark Results Diagram"/>
+</div>
+
 ## Running the tests
 
 To run the tests, you need an Apache ActiveMQ Artemis server. The server can be hosted in a Docker container.

benchmark/Latency_ArtemisNetCoreClient/Consumer.cs

@@ -0,0 +1,57 @@
+using System.Diagnostics;
+using ActiveMQ.Artemis.Core.Client;
+
+namespace Latency_ArtemisNetCoreClient;
+
+public class Consumer : IAsyncDisposable
+{
+    private readonly IConnection _connection;
+    private readonly ISession _session;
+    private readonly IConsumer _consumer;
+
+    private Consumer(IConnection connection, ISession session, IConsumer consumer)
+    {
+        _connection = connection;
+        _session = session;
+        _consumer = consumer;
+    }
+
+    public static async Task<Consumer> CreateAsync(Endpoint endpoint)
+    {
+        var connectionFactory = new ConnectionFactory();
+        var connection = await connectionFactory.CreateAsync(endpoint);
+        var session = await connection.CreateSessionAsync();
+
+        var consumer = await session.CreateConsumerAsync(new ConsumerConfiguration { QueueName = "latency" });
+        return new Consumer(connection, session, consumer);
+    }
+
+    public Task<double[]> StartConsumingAsync(int messages)
+    {
+        return Task.Run(async () =>
+        {
+            var latencies = new double[messages];
+
+            for (var i = 0; i < messages; i++)
+            {
+                var message = await _consumer.ReceiveMessageAsync();
+                var receiveTimestamp = Stopwatch.GetTimestamp();
+                var sendTimestamp = (long) message.Properties["Timestamp"]!;
+                latencies[i] = (receiveTimestamp - sendTimestamp) * 1_000_000.0 / Stopwatch.Frequency;
+                
+                // AMQP doesn't support waiting for the confirmation from the broker for message acknowledgment.
+                // So if we want to compare apples to apples we need to use the fire-and-forget method of acknowledgment.
+                _consumer.Acknowledge(message.MessageDelivery);
+            }
+
+            return latencies;
+        });
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _consumer.DisposeAsync();
+        await _session.DisposeAsync();
+        await _connection.DisposeAsync();
+    }
+}

benchmark/Latency_ArtemisNetCoreClient/Latency_ArtemisNetCoreClient.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <ProjectReference Include="..\..\src\ArtemisNetCoreClient\ArtemisNetCoreClient.csproj" />
+    </ItemGroup>
+
+</Project>

benchmark/Latency_ArtemisNetCoreClient/Producer.cs

@@ -0,0 +1,65 @@
+using System.Diagnostics;
+using ActiveMQ.Artemis.Core.Client;
+
+namespace Latency_ArtemisNetCoreClient;
+
+public class Producer : IAsyncDisposable
+{
+    private readonly IConnection _connection;
+    private readonly ISession _session;
+    private readonly IProducer _producer;
+    private readonly Random _random;
+
+    private Producer(IConnection connection, ISession session, IProducer producer)
+    {
+        _connection = connection;
+        _session = session;
+        _producer = producer;
+        _random = new Random();
+    }
+
+    public static async Task<Producer> CreateAsync(Endpoint endpoint)
+    {
+        var connectionFactory = new ConnectionFactory();
+        var connection = await connectionFactory.CreateAsync(endpoint);
+        var session = await connection.CreateSessionAsync();
+
+        var producer = await session.CreateProducerAsync(new ProducerConfiguration { Address = "latency" });
+        return new Producer(connection, session, producer);
+    }
+
+    public Task SendMessagesAsync(int messages, int payloadSize)
+    {
+        return Task.Run(async () =>
+        {
+            for (var i = 0; i < messages; i++)
+            {
+                var timestamp = Stopwatch.GetTimestamp(); // Get high precision timestamp
+                var message = new Message
+                {
+                    Body = GenerateRandomData(payloadSize),
+                    Properties = new Dictionary<string, object?>
+                    {
+                        ["Timestamp"] = timestamp
+                    },
+                    Durable = true
+                };
+                await _producer.SendMessageAsync(message);
+            }
+        });
+    }
+
+    private byte[] GenerateRandomData(int size)
+    {
+        var data = new byte[size];
+        _random.NextBytes(data);
+        return data;
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        await _producer.DisposeAsync();
+        await _session.DisposeAsync();
+        await _connection.DisposeAsync();
+    }
+}

benchmark/Latency_ArtemisNetCoreClient/Program.cs

@@ -0,0 +1,31 @@
+using ActiveMQ.Artemis.Core.Client;
+
+namespace Latency_ArtemisNetCoreClient;
+
+public static class Program
+{
+    static async Task Main(string[] args)
+    {
+        var endpoint = new Endpoint
+        {
+            Host = "localhost",
+            Port = 61616,
+            User = "artemis",
+            Password = "artemis"
+        };
+
+        var messages = 10_000;
+
+        for (int i = 0; i < 10; i++)
+        {
+            await using var consumer = await Consumer.CreateAsync(endpoint);
+            var startConsumingTask = consumer.StartConsumingAsync(messages: messages);
+
+            await using var producer = await Producer.CreateAsync(endpoint);
+            await producer.SendMessagesAsync(messages: messages, payloadSize: 1024);
+
+            var latencies = await startConsumingTask;
+            Console.WriteLine($"Latency: avg:{latencies.Average():F2}µs, min:{latencies.Min():F2}µs, max:{latencies.Max():F2}µs");
+        }
+    }
+}

benchmark/Latency_NMS.AMQP/Consumer.cs

@@ -0,0 +1,56 @@
+using System.Diagnostics;
+using Apache.NMS;
+using Apache.NMS.AMQP;
+
+namespace Latency_NMS.AMQP;
+
+public class Consumer : IDisposable
+{
+    private readonly IConnection _connection;
+    private readonly ISession _session;
+    private readonly IMessageConsumer _consumer;
+
+    private Consumer(IConnection connection, ISession session, IMessageConsumer consumer)
+    {
+        _connection = connection;
+        _session = session;
+        _consumer = consumer;
+    }
+
+    public static async Task<Consumer> CreateAsync(NmsConnectionFactory connectionFactory)
+    {
+        var connection = await connectionFactory.CreateConnectionAsync();
+        var session = await connection.CreateSessionAsync();
+        await connection.StartAsync();
+
+        var consumer = await session.CreateConsumerAsync(await session.GetQueueAsync("latency"));
+        return new Consumer(connection, session, consumer);
+    }
+
+    public Task<double[]> StartConsumingAsync(int messages)
+    {
+        return Task.Run(async () =>
+        {
+            var latencies = new double[messages];
+
+            for (var i = 0; i < messages; i++)
+            {
+                var message = await _consumer.ReceiveAsync();
+                var receiveTimestamp = Stopwatch.GetTimestamp();
+                var sendTimestamp = (long) message.Properties["Timestamp"];
+                latencies[i] = (receiveTimestamp - sendTimestamp) * 1_000_000.0 / Stopwatch.Frequency;
+                // ReSharper disable once MethodHasAsyncOverload
+                message.Acknowledge();
+            }
+
+            return latencies;
+        });
+    }
+
+    public void Dispose()
+    {
+        _consumer.Dispose();
+        _session.Dispose();
+        _connection.Dispose();
+    }
+}

benchmark/Latency_NMS.AMQP/Latency_NMS.AMQP.csproj

@@ -0,0 +1,14 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <OutputType>Exe</OutputType>
+        <TargetFramework>net8.0</TargetFramework>
+        <ImplicitUsings>enable</ImplicitUsings>
+        <Nullable>enable</Nullable>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <PackageReference Include="Apache.NMS.AMQP" Version="2.2.0" />
+    </ItemGroup>
+
+</Project>

benchmark/Latency_NMS.AMQP/Producer.cs

@@ -0,0 +1,61 @@
+using System.Diagnostics;
+using Apache.NMS;
+using Apache.NMS.AMQP;
+
+namespace Latency_NMS.AMQP;
+
+public class Producer : IDisposable
+{
+    private readonly IConnection _connection;
+    private readonly ISession _session;
+    private readonly IMessageProducer _producer;
+    private readonly Random _random;
+
+    private Producer(IConnection connection, ISession session, IMessageProducer producer)
+    {
+        _connection = connection;
+        _session = session;
+        _producer = producer;
+        _random = new Random();
+    }
+
+    public static async Task<Producer> CreateAsync(NmsConnectionFactory connectionFactory)
+    {
+        var connection = await connectionFactory.CreateConnectionAsync();
+        var session = await connection.CreateSessionAsync();
+
+        var producer = await session.CreateProducerAsync(await session.GetQueueAsync("latency"));
+        return new Producer(connection, session, producer);
+    }
+
+    public Task SendMessagesAsync(int messages, int payloadSize)
+    {
+        return Task.Run(async () =>
+        {
+            for (var i = 0; i < messages; i++)
+            {
+                var timestamp = Stopwatch.GetTimestamp(); // Get high precision timestamp
+                var pingMessage = await _producer.CreateBytesMessageAsync(GenerateRandomData(payloadSize));
+                // pingMessage.NMSDeliveryMode = MsgDeliveryMode.Persistent;
+                pingMessage.Properties["Timestamp"] = timestamp;
+                
+                // ReSharper disable once MethodHasAsyncOverload
+                _producer.Send(pingMessage);
+            }
+        });
+    }
+
+    private byte[] GenerateRandomData(int size)
+    {
+        byte[] data = new byte[size];
+        _random.NextBytes(data);
+        return data;
+    }
+
+    public void Dispose()
+    {
+        _producer.Dispose();
+        _session.Dispose();
+        _connection.Dispose();
+    }
+}