From 0e90456e3a71527b85e92371e66f337da2e6c601 Mon Sep 17 00:00:00 2001
From: Caleb Lloyd <caleb@synadia.com>
Date: Tue, 30 Jan 2024 20:59:53 -0500
Subject: [PATCH 1/2] MicroBenchmark for PublishParallel

Signed-off-by: Caleb Lloyd <caleb@synadia.com>
---
 .../MicroBenchmark/PublishParallelBench.cs    | 89 +++++++++++++++++++
 ...nBuffersBench.cs => PublishSerialBench.cs} | 12 ++-
 2 files changed, 97 insertions(+), 4 deletions(-)
 create mode 100644 sandbox/MicroBenchmark/PublishParallelBench.cs
 rename sandbox/MicroBenchmark/{SerializationBuffersBench.cs => PublishSerialBench.cs} (72%)

diff --git a/sandbox/MicroBenchmark/PublishParallelBench.cs b/sandbox/MicroBenchmark/PublishParallelBench.cs
new file mode 100644
index 000000000..835a339ca
--- /dev/null
+++ b/sandbox/MicroBenchmark/PublishParallelBench.cs
@@ -0,0 +1,89 @@
+using System.Text.Json.Serialization;
+using BenchmarkDotNet.Attributes;
+using NATS.Client.Core;
+
+#pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.
+
+namespace MicroBenchmark;
+
+[MemoryDiagnoser]
+[ShortRunJob]
+[PlainExporter]
+public class PublishParallelBench
+{
+    public const int TotalMsgs = 100_000;
+
+    private static readonly PublishParallelObj Data = new PublishParallelObj
+    {
+        String = new('a', 32),
+        Bool = true,
+        Int = 42,
+        Dictionary = new Dictionary<string, string>
+        {
+            { new('a', 32), new('a', 32) },
+            { new('b', 32), new('b', 32) },
+            { new('c', 32), new('c', 32) },
+        },
+        List = new List<string>
+        {
+            new('a', 32),
+            new('b', 32),
+            new('c', 32),
+        },
+    };
+
+    private NatsConnection _nats;
+
+    [Params(1, 2, 4)]
+    public int Concurrency { get; set; }
+
+    [GlobalSetup]
+    public async Task Setup()
+    {
+        var registry = new NatsJsonContextSerializerRegistry(PublishParallelJsonContext.Default);
+        _nats = new NatsConnection(NatsOpts.Default with { SerializerRegistry = registry });
+        await _nats.ConnectAsync();
+    }
+
+    [Benchmark]
+    public async Task PublishParallelAsync()
+    {
+        var tasks = new List<Task>();
+        for (var i = 0; i < Concurrency; i++)
+        {
+            tasks.Add(Task.Run(async () =>
+            {
+                for (var j = 0; j < TotalMsgs / Concurrency; j++)
+                {
+                    await _nats.PublishAsync("test", Data);
+                }
+            }));
+        }
+
+        await Task.WhenAll(tasks);
+        await _nats.PingAsync();
+    }
+}
+
+internal record PublishParallelObj
+{
+    [JsonPropertyName("string")]
+    public string String { get; set; }
+
+    [JsonPropertyName("bool")]
+    public bool Bool { get; set; }
+
+    [JsonPropertyName("int")]
+    public int Int { get; set; }
+
+    [JsonPropertyName("dictionary")]
+    public Dictionary<string, string> Dictionary { get; set; }
+
+    [JsonPropertyName("list")]
+    public List<string> List { get; set; }
+}
+
+[JsonSerializable(typeof(PublishParallelObj))]
+internal partial class PublishParallelJsonContext : JsonSerializerContext
+{
+}
diff --git a/sandbox/MicroBenchmark/SerializationBuffersBench.cs b/sandbox/MicroBenchmark/PublishSerialBench.cs
similarity index 72%
rename from sandbox/MicroBenchmark/SerializationBuffersBench.cs
rename to sandbox/MicroBenchmark/PublishSerialBench.cs
index 8e15c07be..52e5c1242 100644
--- a/sandbox/MicroBenchmark/SerializationBuffersBench.cs
+++ b/sandbox/MicroBenchmark/PublishSerialBench.cs
@@ -7,7 +7,7 @@ namespace MicroBenchmark;
 [MemoryDiagnoser]
 [ShortRunJob]
 [PlainExporter]
-public class SerializationBuffersBench
+public class PublishSerialBench
 {
     private static readonly string Data = new('0', 126);
 
@@ -17,16 +17,20 @@ public class SerializationBuffersBench
     public int Iter { get; set; }
 
     [GlobalSetup]
-    public void Setup() => _nats = new NatsConnection();
+    public async Task SetupAsync()
+    {
+        _nats = new NatsConnection();
+        await _nats.ConnectAsync();
+    }
 
     [Benchmark]
-    public async ValueTask<TimeSpan> PublishAsync()
+    public async Task PublishAsync()
     {
         for (var i = 0; i < Iter; i++)
         {
             await _nats.PublishAsync("foo", Data);
         }
 
-        return await _nats.PingAsync();
+        await _nats.PingAsync();
     }
 }

From a13caa9b446ffa79354453af72aad27504858d39 Mon Sep 17 00:00:00 2001
From: Caleb Lloyd <caleb@synadia.com>
Date: Tue, 30 Jan 2024 21:13:06 -0500
Subject: [PATCH 2/2] increase to 1M msgs to reduce error

Signed-off-by: Caleb Lloyd <caleb@synadia.com>
---
 sandbox/MicroBenchmark/PublishParallelBench.cs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sandbox/MicroBenchmark/PublishParallelBench.cs b/sandbox/MicroBenchmark/PublishParallelBench.cs
index 835a339ca..27f4b3310 100644
--- a/sandbox/MicroBenchmark/PublishParallelBench.cs
+++ b/sandbox/MicroBenchmark/PublishParallelBench.cs
@@ -11,7 +11,7 @@ namespace MicroBenchmark;
 [PlainExporter]
 public class PublishParallelBench
 {
-    public const int TotalMsgs = 100_000;
+    public const int TotalMsgs = 1_000_000;
 
     private static readonly PublishParallelObj Data = new PublishParallelObj
     {