Merge branch 'main' into talzacc/bdn_module

.github/workflows/ci-bdnbenchmark.yml

@@ -42,7 +42,7 @@ jobs:
         os: [ ubuntu-latest, windows-latest ]
         framework: [ 'net8.0' ]
         configuration: [ 'Release' ]
-        test: [ 'Operations.BasicOperations', 'Operations.ObjectOperations', 'Operations.HashObjectOperations', 'Cluster.ClusterMigrate', 'Cluster.ClusterOperations', 'Lua.LuaScripts', 'Operations.CustomOperations', 'Operations.RawStringOperations', 'Operations.ScriptOperations','Network.BasicOperations', 'Network.RawStringOperations' ]
+        test: [ 'Operations.BasicOperations', 'Operations.ObjectOperations', 'Operations.HashObjectOperations', 'Cluster.ClusterMigrate', 'Cluster.ClusterOperations', 'Lua.LuaScripts', 'Lua.LuaScriptCacheOperations','Lua.LuaRunnerOperations','Operations.CustomOperations', 'Operations.RawStringOperations', 'Operations.ScriptOperations','Network.BasicOperations', 'Network.RawStringOperations' ]
     steps:
       - name: Check out code
         uses: actions/checkout@v4

Version.props

@@ -1,6 +1,6 @@
 <Project>
 	<!-- Versioning property for builds and packages -->
 	<PropertyGroup>
-		<VersionPrefix>1.0.50</VersionPrefix>
+		<VersionPrefix>1.0.51</VersionPrefix>
 	</PropertyGroup>
 </Project>

benchmark/BDN.benchmark/Embedded/EmbeddedRespServer.cs

@@ -2,8 +2,10 @@
 // Licensed under the MIT license.
 
 using Garnet;
+using Garnet.common;
 using Garnet.server;
 using Microsoft.Extensions.Logging;
+using Tsavorite.core;
 
 namespace Embedded.server
 {
@@ -13,6 +15,7 @@ namespace Embedded.server
     internal sealed class EmbeddedRespServer : GarnetServer
     {
         readonly GarnetServerEmbedded garnetServerEmbedded;
+        readonly SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>> subscribeBroker;
 
         /// <summary>
         /// Creates an EmbeddedRespServer instance
@@ -23,6 +26,13 @@ internal sealed class EmbeddedRespServer : GarnetServer
         public EmbeddedRespServer(GarnetServerOptions opts, ILoggerFactory loggerFactory = null, GarnetServerEmbedded server = null) : base(opts, loggerFactory, server)
         {
             this.garnetServerEmbedded = server;
+            this.subscribeBroker = opts.DisablePubSub ? null :
+                new SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>>(
+                    new SpanByteKeySerializer(),
+                    null,
+                    opts.PubSubPageSizeBytes(),
+                    opts.SubscriberRefreshFrequencyMs,
+                    true);
         }
 
         /// <summary>
@@ -38,7 +48,7 @@ public EmbeddedRespServer(GarnetServerOptions opts, ILoggerFactory loggerFactory
         /// <returns>A new RESP server session</returns>
         internal RespServerSession GetRespSession()
         {
-            return new RespServerSession(0, new EmbeddedNetworkSender(), storeWrapper, null, null, true);
+            return new RespServerSession(0, new EmbeddedNetworkSender(), storeWrapper, subscribeBroker: subscribeBroker, null, true);
         }
 
         internal EmbeddedNetworkHandler GetNetworkHandler()

benchmark/BDN.benchmark/Lua/LuaParams.cs

@@ -1,26 +1,38 @@
 // Copyright (c) Microsoft Corporation.
 // Licensed under the MIT license.
 
+using BenchmarkDotNet.Code;
+using Garnet.server;
+
 namespace BDN.benchmark.Lua
 {
     /// <summary>
-    /// Cluster parameters
+    /// Lua parameters
     /// </summary>
-    public struct LuaParams
+    public readonly struct LuaParams
     {
+        public readonly LuaMemoryManagementMode Mode { get; }
+        public readonly bool MemoryLimit { get; }
+
         /// <summary>
         /// Constructor
         /// </summary>
-        public LuaParams()
+        public LuaParams(LuaMemoryManagementMode mode, bool memoryLimit)
         {
+            Mode = mode;
+            MemoryLimit = memoryLimit;
         }
 
+        /// <summary>
+        /// Get the equivalent <see cref="LuaOptions"/>.
+        /// </summary>
+        public LuaOptions CreateOptions()
+        => new(Mode, MemoryLimit ? "2m" : "");
+
         /// <summary>
         /// String representation
         /// </summary>
         public override string ToString()
-        {
-            return "None";
-        }
+        => $"{Mode},{(MemoryLimit ? "Limit" : "None")}";
     }
 }

benchmark/BDN.benchmark/Lua/LuaRunnerOperations.cs

@@ -139,9 +139,13 @@ public unsafe class LuaRunnerOperations
         /// Lua parameters provider
         /// </summary>
         public IEnumerable<LuaParams> LuaParamsProvider()
-        {
-            yield return new();
-        }
+        => [
+            new(LuaMemoryManagementMode.Native, false),
+            new(LuaMemoryManagementMode.Tracked, false),
+            new(LuaMemoryManagementMode.Tracked, true),
+            new(LuaMemoryManagementMode.Managed, false),
+            new(LuaMemoryManagementMode.Managed, true),
+        ];
 
         private EmbeddedRespServer server;
         private RespServerSession session;
@@ -151,16 +155,21 @@ public IEnumerable<LuaParams> LuaParamsProvider()
         private LuaRunner smallCompileRunner;
         private LuaRunner largeCompileRunner;
 
+        private LuaOptions opts;
+
         [GlobalSetup]
         public void GlobalSetup()
         {
-            server = new EmbeddedRespServer(new GarnetServerOptions() { EnableLua = true, QuietMode = true });
+            opts = Params.CreateOptions();
+
+            server = new EmbeddedRespServer(new GarnetServerOptions() { EnableLua = true, QuietMode = true, LuaOptions = opts });
 
             session = server.GetRespSession();
-            paramsRunner = new LuaRunner("return nil");
 
-            smallCompileRunner = new LuaRunner(SmallScript);
-            largeCompileRunner = new LuaRunner(LargeScript);
+            paramsRunner = new LuaRunner(opts, "return nil");
+
+            smallCompileRunner = new LuaRunner(opts, SmallScript);
+            largeCompileRunner = new LuaRunner(opts, LargeScript);
         }
 
         [GlobalCleanup]
@@ -171,6 +180,18 @@ public void GlobalCleanup()
             paramsRunner.Dispose();
         }
 
+        [IterationSetup]
+        public void IterationSetup()
+        {
+            session.EnterAndGetResponseObject();
+        }
+
+        [IterationCleanup]
+        public void IterationCleanup()
+        {
+            session.ExitAndReturnResponseObject();
+        }
+
         [Benchmark]
         public void ResetParametersSmall()
         {
@@ -194,13 +215,13 @@ public void ResetParametersLarge()
         [Benchmark]
         public void ConstructSmall()
         {
-            using var runner = new LuaRunner(SmallScript);
+            using var runner = new LuaRunner(opts, SmallScript);
         }
 
         [Benchmark]
         public void ConstructLarge()
         {
-            using var runner = new LuaRunner(LargeScript);
+            using var runner = new LuaRunner(opts, LargeScript);
         }
 
         [Benchmark]

benchmark/BDN.benchmark/Lua/LuaScriptCacheOperations.cs

@@ -22,9 +22,13 @@ public class LuaScriptCacheOperations
         /// Lua parameters provider
         /// </summary>
         public IEnumerable<LuaParams> LuaParamsProvider()
-        {
-            yield return new();
-        }
+        => [
+            new(LuaMemoryManagementMode.Native, false),
+            new(LuaMemoryManagementMode.Tracked, false),
+            new(LuaMemoryManagementMode.Tracked, true),
+            new(LuaMemoryManagementMode.Managed, false),
+            new(LuaMemoryManagementMode.Managed, true),
+        ];
 
         private EmbeddedRespServer server;
         private StoreWrapper storeWrapper;
@@ -38,7 +42,9 @@ public IEnumerable<LuaParams> LuaParamsProvider()
         [GlobalSetup]
         public void GlobalSetup()
         {
-            server = new EmbeddedRespServer(new GarnetServerOptions() { EnableLua = true, QuietMode = true });
+            var options = Params.CreateOptions();
+
+            server = new EmbeddedRespServer(new GarnetServerOptions() { EnableLua = true, QuietMode = true, LuaOptions = options });
             storeWrapper = server.StoreWrapper;
             sessionScriptCache = new SessionScriptCache(storeWrapper, new GarnetNoAuthAuthenticator());
             session = server.GetRespSession();

benchmark/BDN.benchmark/Lua/LuaScripts.cs

@@ -22,23 +22,29 @@ public unsafe class LuaScripts
         /// Lua parameters provider
         /// </summary>
         public IEnumerable<LuaParams> LuaParamsProvider()
-        {
-            yield return new();
-        }
+        => [
+            new(LuaMemoryManagementMode.Native, false),
+            new(LuaMemoryManagementMode.Tracked, false),
+            new(LuaMemoryManagementMode.Tracked, true),
+            new(LuaMemoryManagementMode.Managed, false),
+            new(LuaMemoryManagementMode.Managed, true),
+        ];
 
         LuaRunner r1, r2, r3, r4;
         readonly string[] keys = ["key1"];
 
         [GlobalSetup]
         public void GlobalSetup()
         {
-            r1 = new LuaRunner("return");
+            var options = Params.CreateOptions();
+
+            r1 = new LuaRunner(options, "return");
             r1.CompileForRunner();
-            r2 = new LuaRunner("return 1 + 1");
+            r2 = new LuaRunner(options, "return 1 + 1");
             r2.CompileForRunner();
-            r3 = new LuaRunner("return KEYS[1]");
+            r3 = new LuaRunner(options, "return KEYS[1]");
             r3.CompileForRunner();
-            r4 = new LuaRunner("return redis.call(KEYS[1])");
+            r4 = new LuaRunner(options, "return redis.call(KEYS[1])");
             r4.CompileForRunner();
         }
 

benchmark/BDN.benchmark/Operations/OperationsBase.cs

@@ -43,6 +43,7 @@ public IEnumerable<OperationParams> OperationParamsProvider()
         internal const int batchSize = 100;
         internal EmbeddedRespServer server;
         internal RespServerSession session;
+        internal RespServerSession subscribeSession;
 
         /// <summary>
         /// Setup
@@ -55,7 +56,9 @@ public virtual void GlobalSetup()
                 QuietMode = true,
                 EnableLua = true,
                 DisablePubSub = true,
+                LuaOptions = new(LuaMemoryManagementMode.Native, ""),
             };
+
             if (Params.useAof)
             {
                 opts.EnableAOF = true;
@@ -72,7 +75,7 @@ public virtual void GlobalSetup()
                 if (Params.useACLs)
                 {
                     aclFile = Path.GetTempFileName();
-                    File.WriteAllText(aclFile, @"user default on nopass -@all +ping +set +get +setex +incr +decr +incrby +decrby +zadd +zrem +lpush +lpop +sadd +srem +hset +hdel +@custom");
+                    File.WriteAllText(aclFile, @"user default on nopass -@all +ping +set +get +setex +incr +decr +incrby +decrby +zadd +zrem +lpush +lpop +sadd +srem +hset +hdel +publish +subscribe +@custom");
                     opts.AuthSettings = new AclAuthenticationPasswordSettings(aclFile);
                 }
 
@@ -84,8 +87,6 @@ public virtual void GlobalSetup()
                 if (aclFile != null)
                     File.Delete(aclFile);
             }
-
-            session = server.GetRespSession();
         }
 
         /// <summary>
@@ -95,6 +96,7 @@ public virtual void GlobalSetup()
         public virtual void GlobalCleanup()
         {
             session.Dispose();
+            subscribeSession?.Dispose();
             server.Dispose();
         }
 

benchmark/BDN.benchmark/Operations/PubSubOperations.cs

@@ -0,0 +1,55 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using BenchmarkDotNet.Attributes;
+using Embedded.server;
+using Garnet.server;
+
+namespace BDN.benchmark.Operations
+{
+    /// <summary>
+    /// Benchmark for PubSubOperations
+    /// </summary>
+    [MemoryDiagnoser]
+    public unsafe class PubSubOperations : OperationsBase
+    {
+        static ReadOnlySpan<byte> PUBLISH => "*3\r\n$7\r\nPUBLISH\r\n$7\r\nchannel\r\n$7\r\nmessage\r\n"u8;
+        Request publish;
+
+        static ReadOnlySpan<byte> SUBSCRIBE => "*2\r\n$9\r\nSUBSCRIBE\r\n$7\r\nchannel\r\n"u8;
+        Request subscribe;
+
+        public override void GlobalSetup()
+        {
+            var opts = new GarnetServerOptions
+            {
+                QuietMode = true,
+                DisablePubSub = false,
+            };
+
+            server = new EmbeddedRespServer(opts, null, new GarnetServerEmbedded());
+            session = server.GetRespSession();
+            subscribeSession = server.GetRespSession();
+
+            SetupOperation(ref publish, PUBLISH);
+            SetupOperation(ref subscribe, SUBSCRIBE);
+
+            // Subscribe to secondary session
+            _ = subscribeSession.TryConsumeMessages(subscribe.bufferPtr, subscribe.buffer.Length);
+
+            // Warm up
+            SlowConsumeMessage(new Span<byte>(publish.bufferPtr, publish.buffer.Length));
+            SlowConsumeMessage(new Span<byte>(publish.bufferPtr, publish.buffer.Length));
+            SlowConsumeMessage(new Span<byte>(publish.bufferPtr, publish.buffer.Length));
+            SlowConsumeMessage(new Span<byte>(publish.bufferPtr, publish.buffer.Length));
+            SlowConsumeMessage(new Span<byte>(publish.bufferPtr, publish.buffer.Length));
+            SlowConsumeMessage(new Span<byte>(publish.bufferPtr, publish.buffer.Length));
+        }
+
+        [Benchmark]
+        public void Publish()
+        {
+            Send(publish);
+        }
+    }
+}