Multi-Database Support (#1005)

* Added MaxDatabases option

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* wip

* aof

* wip

* wip

* wip

* wip

* wip

* bugfixes

* wip

* tests

* format

* small fix

* wip

* test

* simplify dispose logic for gcs at AofSyncTask

* fix

* format

* fix

* wip

* test

* readding multithreading tests

* test

* test

* wip

* test

* format

* wip

* Added FLUSHALL + tests

* Ignore LC MT test

* format

* Added DB ID to client info

* SAVE, BGSAVE, LASTSAVE with ID - not tested yet

* ensure waitForSync before dispose

* wip

* wip

* wip

* Adding HCOLLECT info & docs to GarnetCommandDocs/Info & updating RespCommandDocs/Info

* Updated command info & docs for SAVE, BGSAVE, LASTSAVE & COMMITAOF

* format

* bugfixes

* Adding SPUBLISH, SSUBSCRIBE TO GarnetCommandsInfo.json

* wip

* wip

* wip

* IDatabaseManager refactor - wip (broken)

* wip (broken)

* wip

* wip

* quit code change

* wip

* wip

* Fixing SWAPDB

* wip

* wip

* add explicit fail on fail to dispose

* replayStoreWrapper should not create database

* wip

* wip - tests passing

* wip

* format

* format

* some merge fixes

* format

* bugfix

* fix

* Added ServerOperations to BDN + allocation improvements

* Added website docs

* Fixing ACL test

* Initialize databaseManager as single database unless multi explicitly needed

* merge fix + format

* merging from main + removing unnecessary usings

* bugfix

* Test fix + consolidation of getting directory paths for checkpointing and AOF

* Updating DB IDs after swap

* Some cleanup

* Some cleanup

* more cleanup + format

* small fixes

* Fixed MaxDatabases validation

* more cleanup

* format + small fix

* Calling select & swapdb from lua script

* Disabling a flaky assert

* Fix

* Comment added

* format

* fix

* Addressing comments + adding some tests

* format

* Added some more tests

* fix

* Avoiding allocations while acquiring databases lock

* couple fixes

* Added some website docs for multi-db

* Switching GarnetDatabase & GarnetDatabaseSession from scruct to class

* format

* removed unnecessary benchmarks

* Making properties in GD & GDS readonly

* cleanup

* Switching info to show current-db data, addressing some comments

* format

* Fix tests

* fix

* fix

* disabling select & swapdb inside a txn for now, added tests.

---------

Co-authored-by: Vasileios Zois <vazois@microsoft.com>

Author: TalZaccai

libs/cluster/Server/ClusterProvider.cs

@@ -440,7 +440,7 @@ internal ReplicationLogCheckpointManager GetReplicationLogCheckpointManager(Stor
         internal bool BumpAndWaitForEpochTransition()
         {
             BumpCurrentEpoch();
-            foreach (var server in storeWrapper.TcpServer)
+            foreach (var server in storeWrapper.Servers)
             {
                 while (true)
                 {

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/cluster/Session/RespClusterReplicationCommands.cs

@@ -505,7 +505,7 @@ private bool NetworkClusterFlushAll(out bool invalidParameters)
             }
 
             // Flush all keys
-            clusterProvider.storeWrapper.ExecuteFlushDb(RespCommand.FLUSHALL, false, 0);
+            clusterProvider.storeWrapper.FlushAllDatabases(unsafeTruncateLog: false);
 
             while (!RespWriteUtils.TryWriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
                 SendAndReset();

libs/host/Configuration/Options.cs

@@ -607,6 +607,10 @@ public IEnumerable<string> LuaAllowedFunctions
         [Option("unixsocketperm", Required = false, HelpText = "Unix socket permissions in octal (Unix platforms only)")]
         public int UnixSocketPermission { get; set; }
 
+        [IntRangeValidation(1, 256, isRequired: true)]
+        [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>
@@ -872,7 +876,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, LuaAllowedFunctions, 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

@@ -47,10 +47,6 @@ static string GetVersion()
 
         private readonly GarnetServerOptions opts;
         private IGarnetServer[] servers;
-        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;
@@ -233,24 +229,14 @@ private void InitializeServer()
                 throw new Exception($"Unable to call ThreadPool.SetMaxThreads with {maxThreads}, {maxCPThreads}");
 
             opts.Initialize(loggerFactory);
-            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);
-            }
-
             // Create Garnet TCP server if none was provided.
             if (servers == null)
             {
@@ -268,8 +254,28 @@ private void InitializeServer()
                 }
             }
 
-            storeWrapper = new StoreWrapper(version, RedisProtocolVersion, servers, store, objectStore, objectStoreSizeTracker,
-                    customCommandManager, appendOnlyFile, opts, subscribeBroker, clusterFactory: clusterFactory, loggerFactory: loggerFactory);
+            storeWrapper = new StoreWrapper(version, RedisProtocolVersion, servers, 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);
+            }
+
+            var maxDatabases = opts.EnableCluster ? 1 : opts.MaxDatabases;
+            logger?.LogInformation("Max number of logical databases allowed on server: {maxDatabases}", maxDatabases);
 
             // Create session provider for Garnet
             Provider = new GarnetProvider(storeWrapper, subscribeBroker);
@@ -285,6 +291,17 @@ private void InitializeServer()
             LoadModules(customCommandManager);
         }
 
+        private GarnetDatabase CreateDatabase(int dbId, GarnetServerOptions serverOptions, ClusterFactory clusterFactory,
+            CustomCommandManager customCommandManager)
+        {
+            var store = CreateMainStore(dbId, clusterFactory, out var epoch, out var stateMachineDriver);
+            var objectStore = CreateObjectStore(dbId, clusterFactory, customCommandManager, epoch, stateMachineDriver, out var objectStoreSizeTracker);
+            var (aofDevice, aof) = CreateAOF(dbId);
+            return new GarnetDatabase(dbId, store, objectStore, epoch, stateMachineDriver, objectStoreSizeTracker,
+                aofDevice, aof, serverOptions.AdjustedIndexMaxCacheLines == 0,
+                serverOptions.AdjustedObjectStoreIndexMaxCacheLines == 0);
+        }
+
         private void LoadModules(CustomCommandManager customCommandManager)
         {
             if (opts.LoadModuleCS == null)
@@ -309,82 +326,82 @@ private void LoadModules(CustomCommandManager customCommandManager)
             }
         }
 
-        private void CreateMainStore(IClusterFactory clusterFactory, out string checkpointDir)
+        private TsavoriteKV<SpanByte, SpanByte, MainStoreFunctions, MainStoreAllocator> CreateMainStore(int dbId, IClusterFactory clusterFactory,
+            out LightEpoch epoch, out StateMachineDriver stateMachineDriver)
         {
-            kvSettings = opts.GetSettings(loggerFactory, out logFactory);
+            epoch = new LightEpoch();
+            stateMachineDriver = new StateMachineDriver(epoch, loggerFactory?.CreateLogger($"StateMachineDriver"));
 
-            checkpointDir = opts.CheckpointDir ?? opts.LogDir;
+            kvSettings = opts.GetSettings(loggerFactory, epoch, stateMachineDriver, out logFactory);
 
             // 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);
+
+            kvSettings.CheckpointManager = opts.EnableCluster ?
+                clusterFactory.CreateCheckpointManager(opts.DeviceFactoryCreator, defaultNamingScheme, isMainStore: true, logger) :
+                new DeviceLogCommitCheckpointManager(opts.DeviceFactoryCreator, defaultNamingScheme, removeOutdated: true);
 
-            store = new(kvSettings
+            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,
+            LightEpoch epoch, StateMachineDriver stateMachineDriver, out CacheSizeTracker objectStoreSizeTracker)
         {
             objectStoreSizeTracker = null;
-            if (!opts.DisableObjects)
-            {
-                objKvSettings = opts.GetObjectStoreSettings(loggerFactory,
-                    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(loggerFactory, epoch, stateMachineDriver,
+                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.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]"));
-
-                if (opts.CommitFrequencyMs < 0 && opts.WaitForCommit)
-                    throw new Exception("Cannot use CommitWait with manual commits");
-                return;
+                if (opts.CommitFrequencyMs != 0 || opts.WaitForCommit)
+                    throw new Exception("Cannot use CommitFrequencyMs or CommitWait without EnableAOF");
+                return (null, null);
             }
 
-            if (opts.CommitFrequencyMs != 0 || opts.WaitForCommit)
-                throw new Exception("Cannot use CommitFrequencyMs or CommitWait without EnableAOF");
+            if (opts.FastAofTruncate && opts.CommitFrequencyMs != -1)
+                throw new Exception("Need to set CommitFrequencyMs to -1 (manual commits) with FastAofTruncate");
+
+            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 (aofDevice, appendOnlyFile);
         }
 
         /// <summary>
@@ -432,13 +449,9 @@ private void InternalDispose()
             for (var i = 0; i < servers.Length; i++)
                 servers[i]?.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

@@ -399,5 +399,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
 }