Multi-Database Support

.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', 'Operations.SortedSetOperations', 'Cluster.ClusterMigrate', 'Cluster.ClusterOperations', 'Lua.LuaScripts', 'Lua.LuaScriptCacheOperations','Lua.LuaRunnerOperations','Operations.CustomOperations', 'Operations.RawStringOperations', 'Operations.ScriptOperations', 'Operations.ModuleOperations', 'Operations.PubSubOperations', 'Network.BasicOperations', 'Network.RawStringOperations' ]
+        test: [ 'Operations.BasicOperations', 'Operations.ObjectOperations', 'Operations.HashObjectOperations', 'Operations.SortedSetOperations', 'Cluster.ClusterMigrate', 'Cluster.ClusterOperations', 'Lua.LuaScripts', 'Lua.LuaScriptCacheOperations','Lua.LuaRunnerOperations','Operations.CustomOperations', 'Operations.RawStringOperations', 'Operations.ScriptOperations', 'Operations.ModuleOperations', 'Operations.PubSubOperations', 'Operations.ServerOperations', 'Network.BasicOperations', 'Network.RawStringOperations' ]
     steps:
       - name: Check out code
         uses: actions/checkout@v4

benchmark/BDN.benchmark/Operations/ServerOperations.cs

@@ -0,0 +1,52 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using BenchmarkDotNet.Attributes;
+using Embedded.server;
+
+namespace BDN.benchmark.Operations
+{
+    /// <summary>
+    /// Benchmark for ServerOperations
+    /// </summary>
+    [MemoryDiagnoser]
+    public unsafe class ServerOperations : OperationsBase
+    {
+        static ReadOnlySpan<byte> SELECTUNSELECT => "*2\r\n$6\r\nSELECT\r\n$1\r\n1\r\n*2\r\n$6\r\nSELECT\r\n$1\r\n0\r\n"u8;
+        Request selectUnselect;
+
+        static ReadOnlySpan<byte> SWAPDB => "*3\r\n$6\r\nSWAPDB\r\n$1\r\n1\r\n$1\r\n0\r\n"u8;
+        Request swapDb;
+
+        public override void GlobalSetup()
+        {
+            base.GlobalSetup();
+
+            SetupOperation(ref selectUnselect, SELECTUNSELECT);
+            SetupOperation(ref swapDb, SWAPDB);
+
+            // Pre-populate data in DB0
+            SlowConsumeMessage("*3\r\n$3\r\nSET\r\n$1\r\na\r\n$1\r\na\r\n"u8);
+            SlowConsumeMessage("*3\r\n$5\r\nLPUSH\r\n$1\r\nd\r\n$1\r\nf\r\n"u8);
+            SlowConsumeMessage("*3\r\n$4\r\nSADD\r\n$1\r\ne\r\n$1\r\nb\r\n"u8);
+
+            // Pre-populate data in DB1
+            SlowConsumeMessage("*2\r\n$6\r\nSELECT\r\n$1\r\n1\r\n"u8);
+            SlowConsumeMessage("*3\r\n$3\r\nSET\r\n$1\r\nb\r\n$1\r\nb\r\n"u8);
+            SlowConsumeMessage("*3\r\n$5\r\nLPUSH\r\n$1\r\nf\r\n$1\r\nh\r\n"u8);
+            SlowConsumeMessage("*3\r\n$4\r\nSADD\r\n$1\r\ng\r\n$1\r\ni\r\n"u8);
+        }
+
+        [Benchmark]
+        public void SelectUnselect()
+        {
+            Send(selectUnselect);
+        }
+
+        [Benchmark]
+        public void SwapDb()
+        {
+            Send(swapDb);
+        }
+    }
+}

libs/cluster/Server/Migration/MigrateSessionKeys.cs

@@ -182,14 +182,14 @@ public bool MigrateKeys()
                     return false;
 
                 // Migrate main store keys
-                _gcs.InitializeIterationBuffer(clusterProvider.storeWrapper.loggingFrequncy);
+                _gcs.InitializeIterationBuffer(clusterProvider.storeWrapper.loggingFrequency);
                 if (!MigrateKeysFromMainStore())
                     return false;
 
                 // Migrate object store keys
                 if (!clusterProvider.serverOptions.DisableObjects)
                 {
-                    _gcs.InitializeIterationBuffer(clusterProvider.storeWrapper.loggingFrequncy);
+                    _gcs.InitializeIterationBuffer(clusterProvider.storeWrapper.loggingFrequency);
                     if (!MigrateKeysFromObjectStore())
                         return false;
                 }

libs/cluster/Session/RespClusterMigrateCommands.cs

@@ -26,7 +26,7 @@ private void TrackImportProgress(int keyCount, bool isMainStore, bool completed
         {
             totalKeyCount += keyCount;
             var duration = TimeSpan.FromTicks(Stopwatch.GetTimestamp() - lastLog);
-            if (completed || lastLog == 0 || duration >= clusterProvider.storeWrapper.loggingFrequncy)
+            if (completed || lastLog == 0 || duration >= clusterProvider.storeWrapper.loggingFrequency)
             {
                 logger?.LogTrace("[{op}]: isMainStore:({storeType}) totalKeyCount:({totalKeyCount})", completed ? "COMPLETED" : "IMPORTING", isMainStore, totalKeyCount.ToString("N0"));
                 lastLog = Stopwatch.GetTimestamp();

libs/host/Configuration/Options.cs

@@ -569,6 +569,10 @@ internal sealed class Options
         [Option("unixsocketperm", Required = false, HelpText = "Unix socket permissions in octal (Unix platforms only)")]
         public int UnixSocketPermission { get; set; }
 
+        [IntRangeValidation(1, 256, isRequired: false)]
+        [Option("max-databases", Required = false, HelpText = "Max number of logical databases allowed in a single Garnet server instance")]
+        public int MaxDatabases { get; set; }
+
         /// <summary>
         /// This property contains all arguments that were not parsed by the command line argument parser
         /// </summary>
@@ -813,7 +817,8 @@ public GarnetServerOptions GetServerOptions(ILogger logger = null)
                 SkipRDBRestoreChecksumValidation = SkipRDBRestoreChecksumValidation.GetValueOrDefault(),
                 LuaOptions = EnableLua.GetValueOrDefault() ? new LuaOptions(LuaMemoryManagementMode, LuaScriptMemoryLimit, LuaScriptTimeoutMs == 0 ? Timeout.InfiniteTimeSpan : TimeSpan.FromMilliseconds(LuaScriptTimeoutMs), LuaLoggingMode, logger) : null,
                 UnixSocketPath = UnixSocketPath,
-                UnixSocketPermission = unixSocketPermissions
+                UnixSocketPermission = unixSocketPermissions,
+                MaxDatabases = MaxDatabases,
             };
         }
 

libs/host/Configuration/Redis/RedisOptions.cs

@@ -97,6 +97,9 @@ internal class RedisOptions
 
         [RedisOption("slowlog-max-len", nameof(Options.SlowLogMaxEntries))]
         public Option<int> SlowLogMaxLen { get; set; }
+
+        [RedisOption("databases", nameof(Options.MaxDatabases))]
+        public Option<int> Databases { get; set; }
     }
 
     /// <summary>

libs/host/GarnetServer.cs

@@ -46,10 +46,6 @@ static string GetVersion()
 
         private readonly GarnetServerOptions opts;
         private IGarnetServer server;
-        private TsavoriteKV<SpanByte, SpanByte, MainStoreFunctions, MainStoreAllocator> store;
-        private TsavoriteKV<byte[], IGarnetObject, ObjectStoreFunctions, ObjectStoreAllocator> objectStore;
-        private IDevice aofDevice;
-        private TsavoriteLog appendOnlyFile;
         private SubscribeBroker subscribeBroker;
         private KVSettings<SpanByte, SpanByte> kvSettings;
         private KVSettings<byte[], IGarnetObject> objKvSettings;
@@ -227,35 +223,44 @@ private void InitializeServer()
             if (!setMax && !ThreadPool.SetMaxThreads(maxThreads, maxCPThreads))
                 throw new Exception($"Unable to call ThreadPool.SetMaxThreads with {maxThreads}, {maxCPThreads}");
 
-            CreateMainStore(clusterFactory, out var checkpointDir);
-            CreateObjectStore(clusterFactory, customCommandManager, checkpointDir, out var objectStoreSizeTracker);
+            StoreWrapper.DatabaseCreatorDelegate createDatabaseDelegate = (int dbId) =>
+                CreateDatabase(dbId, opts, clusterFactory, customCommandManager);
 
             if (!opts.DisablePubSub)
                 subscribeBroker = new SubscribeBroker(null, opts.PubSubPageSizeBytes(), opts.SubscriberRefreshFrequencyMs, true, logger);
 
-            CreateAOF();
-
             logger?.LogTrace("TLS is {tlsEnabled}", opts.TlsOptions == null ? "disabled" : "enabled");
 
-            if (logger != null)
-            {
-                var configMemoryLimit = (store.IndexSize * 64) + store.Log.MaxMemorySizeBytes + (store.ReadCache?.MaxMemorySizeBytes ?? 0) + (appendOnlyFile?.MaxMemorySizeBytes ?? 0);
-                if (objectStore != null)
-                    configMemoryLimit += objectStore.IndexSize * 64 + objectStore.Log.MaxMemorySizeBytes + (objectStore.ReadCache?.MaxMemorySizeBytes ?? 0) + (objectStoreSizeTracker?.TargetSize ?? 0) + (objectStoreSizeTracker?.ReadCacheTargetSize ?? 0);
-                logger.LogInformation("Total configured memory limit: {configMemoryLimit}", configMemoryLimit);
-            }
-
             if (opts.EndPoint is UnixDomainSocketEndPoint)
             {
                 // Delete existing unix socket file, if it exists.
                 File.Delete(opts.UnixSocketPath);
             }
 
             // Create Garnet TCP server if none was provided.
-            this.server ??= new GarnetServerTcp(opts.EndPoint, 0, opts.TlsOptions, opts.NetworkSendThrottleMax, opts.NetworkConnectionLimit, opts.UnixSocketPath, opts.UnixSocketPermission, logger);
+            this.server ??= new GarnetServerTcp(opts.EndPoint, 0, opts.TlsOptions, opts.NetworkSendThrottleMax,
+                opts.NetworkConnectionLimit, opts.UnixSocketPath, opts.UnixSocketPermission, logger);
 
-            storeWrapper = new StoreWrapper(version, RedisProtocolVersion, server, store, objectStore, objectStoreSizeTracker,
-                    customCommandManager, appendOnlyFile, opts, subscribeBroker, clusterFactory: clusterFactory, loggerFactory: loggerFactory);
+            storeWrapper = new StoreWrapper(version, RedisProtocolVersion, server,
+                customCommandManager, opts, subscribeBroker,
+                createDatabaseDelegate: createDatabaseDelegate,
+                clusterFactory: clusterFactory,
+                loggerFactory: loggerFactory);
+
+            if (logger != null)
+            {
+                var configMemoryLimit = (storeWrapper.store.IndexSize * 64) +
+                                        storeWrapper.store.Log.MaxMemorySizeBytes +
+                                        (storeWrapper.store.ReadCache?.MaxMemorySizeBytes ?? 0) +
+                                        (storeWrapper.appendOnlyFile?.MaxMemorySizeBytes ?? 0);
+                if (storeWrapper.objectStore != null)
+                    configMemoryLimit += storeWrapper.objectStore.IndexSize * 64 +
+                                         storeWrapper.objectStore.Log.MaxMemorySizeBytes +
+                                         (storeWrapper.objectStore.ReadCache?.MaxMemorySizeBytes ?? 0) +
+                                         (storeWrapper.objectStoreSizeTracker?.TargetSize ?? 0) +
+                                         (storeWrapper.objectStoreSizeTracker?.ReadCacheTargetSize ?? 0);
+                logger.LogInformation("Total configured memory limit: {configMemoryLimit}", configMemoryLimit);
+            }
 
             // Create session provider for Garnet
             Provider = new GarnetProvider(storeWrapper, subscribeBroker);
@@ -270,6 +275,17 @@ private void InitializeServer()
             LoadModules(customCommandManager);
         }
 
+        private GarnetDatabase CreateDatabase(int dbId, GarnetServerOptions serverOptions, ClusterFactory clusterFactory,
+            CustomCommandManager customCommandManager)
+        {
+            var store = CreateMainStore(dbId, clusterFactory);
+            var objectStore = CreateObjectStore(dbId, clusterFactory, customCommandManager, out var objectStoreSizeTracker);
+            var (aofDevice, aof) = CreateAOF(dbId);
+            return new GarnetDatabase(dbId, store, objectStore, objectStoreSizeTracker, aofDevice, aof,
+                serverOptions.AdjustedIndexMaxCacheLines == 0,
+                serverOptions.AdjustedObjectStoreIndexMaxCacheLines == 0);
+        }
+
         private void LoadModules(CustomCommandManager customCommandManager)
         {
             if (opts.LoadModuleCS == null)
@@ -294,82 +310,77 @@ private void LoadModules(CustomCommandManager customCommandManager)
             }
         }
 
-        private void CreateMainStore(IClusterFactory clusterFactory, out string checkpointDir)
+        private TsavoriteKV<SpanByte, SpanByte, MainStoreFunctions, MainStoreAllocator> CreateMainStore(int dbId, IClusterFactory clusterFactory)
         {
             kvSettings = opts.GetSettings(loggerFactory, out logFactory);
 
-            checkpointDir = opts.CheckpointDir ?? opts.LogDir;
-
             // Run checkpoint on its own thread to control p99
             kvSettings.ThrottleCheckpointFlushDelayMs = opts.CheckpointThrottleFlushDelayMs;
 
-            if (opts.EnableCluster)
-            {
-                kvSettings.CheckpointManager = clusterFactory.CreateCheckpointManager(opts.DeviceFactoryCreator,
-                    new DefaultCheckpointNamingScheme(checkpointDir + "/Store/checkpoints"), isMainStore: true, logger);
-            }
-            else
-            {
-                kvSettings.CheckpointManager = new DeviceLogCommitCheckpointManager(opts.DeviceFactoryCreator,
-                    new DefaultCheckpointNamingScheme(checkpointDir + "/Store/checkpoints"), removeOutdated: true);
-            }
+            var baseName = opts.GetMainStoreCheckpointDirectory(dbId);
+            var defaultNamingScheme = new DefaultCheckpointNamingScheme(baseName);
 
-            store = new(kvSettings
+            kvSettings.CheckpointManager = opts.EnableCluster ?
+                clusterFactory.CreateCheckpointManager(opts.DeviceFactoryCreator, defaultNamingScheme, isMainStore: true, logger) :
+                new DeviceLogCommitCheckpointManager(opts.DeviceFactoryCreator, defaultNamingScheme, removeOutdated: true);
+
+            return new TsavoriteKV<SpanByte, SpanByte, MainStoreFunctions, MainStoreAllocator>(kvSettings
                 , StoreFunctions<SpanByte, SpanByte>.Create()
                 , (allocatorSettings, storeFunctions) => new(allocatorSettings, storeFunctions));
         }
 
-        private void CreateObjectStore(IClusterFactory clusterFactory, CustomCommandManager customCommandManager, string CheckpointDir, out CacheSizeTracker objectStoreSizeTracker)
+        private TsavoriteKV<byte[], IGarnetObject, ObjectStoreFunctions, ObjectStoreAllocator> CreateObjectStore(int dbId, IClusterFactory clusterFactory, CustomCommandManager customCommandManager, out CacheSizeTracker objectStoreSizeTracker)
         {
             objectStoreSizeTracker = null;
-            if (!opts.DisableObjects)
-            {
-                objKvSettings = opts.GetObjectStoreSettings(this.loggerFactory?.CreateLogger("TsavoriteKV  [obj]"),
-                    out var objHeapMemorySize, out var objReadCacheHeapMemorySize);
+            if (opts.DisableObjects)
+                return null;
 
-                // Run checkpoint on its own thread to control p99
-                objKvSettings.ThrottleCheckpointFlushDelayMs = opts.CheckpointThrottleFlushDelayMs;
+            objKvSettings = opts.GetObjectStoreSettings(this.loggerFactory?.CreateLogger("TsavoriteKV  [obj]"),
+                out var objHeapMemorySize, out var objReadCacheHeapMemorySize);
+
+            // Run checkpoint on its own thread to control p99
+            objKvSettings.ThrottleCheckpointFlushDelayMs = opts.CheckpointThrottleFlushDelayMs;
+
+            var baseName = opts.GetObjectStoreCheckpointDirectory(dbId);
+            var defaultNamingScheme = new DefaultCheckpointNamingScheme(baseName);
+
+            objKvSettings.CheckpointManager = opts.EnableCluster ?
+                clusterFactory.CreateCheckpointManager(opts.DeviceFactoryCreator, defaultNamingScheme, isMainStore: false, logger) :
+                new DeviceLogCommitCheckpointManager(opts.DeviceFactoryCreator, defaultNamingScheme, removeOutdated: true);
+
+            var objStore = new TsavoriteKV<byte[], IGarnetObject, ObjectStoreFunctions, ObjectStoreAllocator>(
+                objKvSettings,
+                StoreFunctions<byte[], IGarnetObject>.Create(new ByteArrayKeyComparer(),
+                    () => new ByteArrayBinaryObjectSerializer(),
+                    () => new GarnetObjectSerializer(customCommandManager)),
+                (allocatorSettings, storeFunctions) => new(allocatorSettings, storeFunctions));
+
+            if (objHeapMemorySize > 0 || objReadCacheHeapMemorySize > 0)
+                objectStoreSizeTracker = new CacheSizeTracker(objStore, objKvSettings, objHeapMemorySize, objReadCacheHeapMemorySize,
+                    this.loggerFactory);
+
+            return objStore;
 
-                if (opts.EnableCluster)
-                    objKvSettings.CheckpointManager = clusterFactory.CreateCheckpointManager(
-                        opts.DeviceFactoryCreator,
-                        new DefaultCheckpointNamingScheme(CheckpointDir + "/ObjectStore/checkpoints"),
-                        isMainStore: false, logger);
-                else
-                    objKvSettings.CheckpointManager = new DeviceLogCommitCheckpointManager(opts.DeviceFactoryCreator,
-                        new DefaultCheckpointNamingScheme(CheckpointDir + "/ObjectStore/checkpoints"),
-                        removeOutdated: true);
-
-                objectStore = new(objKvSettings
-                    , StoreFunctions<byte[], IGarnetObject>.Create(new ByteArrayKeyComparer(),
-                        () => new ByteArrayBinaryObjectSerializer(),
-                        () => new GarnetObjectSerializer(customCommandManager))
-                    , (allocatorSettings, storeFunctions) => new(allocatorSettings, storeFunctions));
-
-                if (objHeapMemorySize > 0 || objReadCacheHeapMemorySize > 0)
-                    objectStoreSizeTracker = new CacheSizeTracker(objectStore, objKvSettings, objHeapMemorySize, objReadCacheHeapMemorySize,
-                        this.loggerFactory);
-            }
         }
 
-        private void CreateAOF()
+        private (IDevice, TsavoriteLog) CreateAOF(int dbId)
         {
             if (opts.EnableAOF)
             {
                 if (opts.FastAofTruncate && opts.CommitFrequencyMs != -1)
                     throw new Exception("Need to set CommitFrequencyMs to -1 (manual commits) with MainMemoryReplication");
 
-                opts.GetAofSettings(out var aofSettings);
-                aofDevice = aofSettings.LogDevice;
-                appendOnlyFile = new TsavoriteLog(aofSettings, logger: this.loggerFactory?.CreateLogger("TsavoriteLog [aof]"));
-
+                opts.GetAofSettings(dbId, out var aofSettings);
+                var aofDevice = aofSettings.LogDevice;
+                var appendOnlyFile = new TsavoriteLog(aofSettings, logger: this.loggerFactory?.CreateLogger("TsavoriteLog [aof]"));
                 if (opts.CommitFrequencyMs < 0 && opts.WaitForCommit)
                     throw new Exception("Cannot use CommitWait with manual commits");
-                return;
+                return (aofDevice, appendOnlyFile);
             }
 
             if (opts.CommitFrequencyMs != 0 || opts.WaitForCommit)
                 throw new Exception("Cannot use CommitFrequencyMs or CommitWait without EnableAOF");
+            return (null, null);
         }
 
         /// <summary>
@@ -415,13 +426,9 @@ private void InternalDispose()
             Provider?.Dispose();
             server.Dispose();
             subscribeBroker?.Dispose();
-            store.Dispose();
-            appendOnlyFile?.Dispose();
-            aofDevice?.Dispose();
             kvSettings.LogDevice?.Dispose();
             if (!opts.DisableObjects)
             {
-                objectStore.Dispose();
                 objKvSettings.LogDevice?.Dispose();
                 objKvSettings.ObjectLogDevice?.Dispose();
             }

libs/host/defaults.conf

@@ -384,5 +384,8 @@
 	"UnixSocketPath": null,
 
 	/* Unix socket permissions in octal (Unix platforms only) */
-	"UnixSocketPermission": 0
+	"UnixSocketPermission": 0,
+
+	/* Max number of logical databases allowed in a single Garnet server instance */
+	"MaxDatabases": 16
 }