STREAMS API in Garnet

Garnet.sln

@@ -118,6 +118,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ETag", "samples\ETag\ETag.c
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Garnet.fuzz", "test\Garnet.fuzz\Garnet.fuzz.csproj", "{7A42F7AA-EE93-49B1-8711-A1D6D948F5FC}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Btree", "playground\BTree\Btree.csproj", "{CE12831B-2805-469E-8208-759DC4B4862C}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -358,6 +360,14 @@ Global
 		{7A42F7AA-EE93-49B1-8711-A1D6D948F5FC}.Release|Any CPU.Build.0 = Release|Any CPU
 		{7A42F7AA-EE93-49B1-8711-A1D6D948F5FC}.Release|x64.ActiveCfg = Release|Any CPU
 		{7A42F7AA-EE93-49B1-8711-A1D6D948F5FC}.Release|x64.Build.0 = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|x64.ActiveCfg = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Debug|x64.Build.0 = Debug|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|Any CPU.Build.0 = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|x64.ActiveCfg = Release|Any CPU
+		{CE12831B-2805-469E-8208-759DC4B4862C}.Release|x64.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -395,6 +405,7 @@ Global
 		{4FBA1587-BAFC-49F8-803A-D1CF431A26F5} = {7068BB97-1958-4060-B5F1-859464592E56}
 		{7A42F7AA-EE93-49B1-8711-A1D6D948F5FC} = {9A03717A-4E0B-49CA-8579-A02A4C1D003F}
 		{1A5DF817-D0DD-4F0B-AE3A-C9CD0E03C9D5} = {D8A9CE6E-91B9-4B84-B44A-2BCF1161A793}
+		{CE12831B-2805-469E-8208-759DC4B4862C} = {69A71E2C-00E3-42F3-854E-BE157A24834E}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {2C02C405-4798-41CA-AF98-61EDFEF6772E}

libs/common/RespReadUtils.cs

@@ -630,6 +630,16 @@ public static bool TryReadInt64WithLengthHeader(out long number, ref byte* ptr,
             return true;
         }
 
+         /// <summary>
+        /// Tries to read a Ulong from the given ASCII-encoded RESP string. 
+        /// Note: this does not check for any length headers and is simply an accessor to TryReadUlong.
+        /// </summary>
+        /// <param name="number">If parsing was successful, contains the parsed ulong value.</param>
+        /// <param name="ptr">The starting position in the RESP string. Will be advanced if parsing is successful.</param>
+        /// <param name="end">The current end of the RESP string.</param>
+        /// <returns>True if a ulong was successfully parsed.</returns>
+        public static bool ReadUlong(out ulong number, ref byte* ptr, byte* end) => TryReadUInt64(ref ptr, end, out number, out _);
+
         /// <summary>
         /// Read long with length header
         /// </summary>

libs/host/Configuration/Options.cs

@@ -615,6 +615,10 @@ public IEnumerable<string> LuaAllowedFunctions
         [Option("max-databases", Required = false, HelpText = "Max number of logical databases allowed in a single Garnet server instance")]
         public int MaxDatabases { get; set; }
 
+        [OptionValidation]
+        [Option("streams", Required = false, HelpText = "Enable streams on server.")]
+        public bool? EnableStreams { get; set; }
+
         /// <summary>
         /// This property contains all arguments that were not parsed by the command line argument parser
         /// </summary>
@@ -883,6 +887,7 @@ public GarnetServerOptions GetServerOptions(ILogger logger = null)
                 UnixSocketPath = UnixSocketPath,
                 UnixSocketPermission = unixSocketPermissions,
                 MaxDatabases = MaxDatabases,
+                EnableStreams = EnableStreams.GetValueOrDefault(),
             };
         }
 

libs/host/defaults.conf

@@ -405,5 +405,8 @@
 	"UnixSocketPermission": 0,
 
 	/* Max number of logical databases allowed in a single Garnet server instance */
-	"MaxDatabases": 16
+	"MaxDatabases": 16,
+
+	/* Enable use of streams inside Garnet */
+	"EnableStreams": false,
 }

libs/resources/RespCommandsInfo.json

@@ -5015,6 +5015,124 @@
       }
     ]
   },
+  {
+    "Command": "XADD",
+    "Name": "XADD",
+    "IsInternal": false,
+    "Arity": -5,
+    "Flags": "DenyOom, Fast, Write",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Fast, Stream, Write",
+    "Tips": [
+      "nondeterministic_output"
+    ],
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": "UPDATE instead of INSERT because of the optional trimming feature",
+        "Flags": "RW, Update"
+      }
+    ],
+    "SubCommands": null
+  },
+  {
+    "Command": "XDEL",
+    "Name": "XDEL",
+    "IsInternal": false,
+    "Arity": -3,
+    "Flags": "Fast, Write",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Fast, Stream, Write",
+    "Tips": null,
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": null,
+        "Flags": "RW, Delete"
+      }
+    ],
+    "SubCommands": null
+  },
+  {
+    "Command": "XLEN",
+    "Name": "XLEN",
+    "IsInternal": false,
+    "Arity": 2,
+    "Flags": "Fast, ReadOnly",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Fast, Read, Stream",
+    "Tips": null,
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": null,
+        "Flags": "RO"
+      }
+    ],
+    "SubCommands": null
+  },
+  {
+    "Command": "XRANGE",
+    "Name": "XRANGE",
+    "IsInternal": false,
+    "Arity": -4,
+    "Flags": "ReadOnly",
+    "FirstKey": 1,
+    "LastKey": 1,
+    "Step": 1,
+    "AclCategories": "Read, Slow, Stream",
+    "Tips": null,
+    "KeySpecifications": [
+      {
+        "BeginSearch": {
+          "TypeDiscriminator": "BeginSearchIndex",
+          "Index": 1
+        },
+        "FindKeys": {
+          "TypeDiscriminator": "FindKeysRange",
+          "LastKey": 0,
+          "KeyStep": 1,
+          "Limit": 0
+        },
+        "Notes": null,
+        "Flags": "RO, Access"
+      }
+    ],
+    "SubCommands": null
+  },
   {
     "Command": "ZADD",
     "Name": "ZADD",

libs/server/BTreeIndex/BTree.cs

@@ -0,0 +1,177 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT license.
+
+using System;
+using System.Collections.Generic;
+using System.Runtime.InteropServices;
+
+namespace Garnet.server.BTreeIndex
+{
+    public unsafe partial class BTree
+    {
+        BTreeNode* root;
+        BTreeNode* head;
+        BTreeNode* tail;
+        byte* tailMinKey;
+        public static readonly int MAX_TREE_DEPTH = 10; // maximum allowed depth of the tree
+        static int DEFAULT_SPLIT_LEAF_POSITION = (BTreeNode.LEAF_CAPACITY + 1) / 2; // position at which leaf node is split
+        static int SPLIT_LEAF_POSITION = BTreeNode.LEAF_CAPACITY; // position at which leaf node is split
+        static int SPLIT_INTERNAL_POSITION = BTreeNode.INTERNAL_CAPACITY; // position at which internal node is split
+
+        BTreeNode*[] rootToTailLeaf; // array of nodes from root to tail leaf
+        public BTreeStats stats; // statistics about the tree
+
+        /// <summary>
+        /// Initializes a new instance of the <see cref="BTree"/> class.
+        /// </summary>
+        public BTree(uint sectorSize)
+        {
+            var memoryBlock = (IntPtr*)NativeMemory.AlignedAlloc((nuint)BTreeNode.PAGE_SIZE, (nuint)BTreeNode.PAGE_SIZE);
+            root = BTreeNode.Create(BTreeNodeType.Leaf, memoryBlock);
+            head = tail = root;
+            root->info->next = root->info->previous = null;
+            root->info->count = 0;
+            tailMinKey = null;
+            rootToTailLeaf = new BTreeNode*[MAX_TREE_DEPTH];
+            stats = new BTreeStats();
+            stats.depth = 1;
+            stats.numLeafNodes = 1;
+            stats.numAllocates = 1;
+        }
+
+        /// <summary>
+        /// Frees the memory allocated for a node
+        /// </summary>
+        /// <param name="node">BTreeNode to free from memory</param>
+        private void Free(ref BTreeNode* node)
+        {
+            if (node == null)
+                return;
+
+            // If this is an internal node, free all its children first
+            if (node->info->type == BTreeNodeType.Internal)
+            {
+                for (int i = 0; i <= node->info->count; i++)
+                {
+                    var child = node->data.children[i];
+                    Free(ref child);
+                    node->data.children[i] = null;
+                }
+            }
+
+            // Free the memory handle
+            if (node->memoryHandle != null)
+            {
+                NativeMemory.Free(node->memoryHandle);
+                stats.numDeallocates++;
+                node = null;
+            }
+        }
+
+        /// <summary>
+        /// Frees the memory allocated for a node
+        /// </summary>
+        /// <param name="node"></param>
+        public static void FreeNode(ref BTreeNode* node)
+        {
+            if (node == null)
+                return;
+
+            // If this is an internal node, free all its children first
+            if (node->info->type == BTreeNodeType.Internal)
+            {
+                for (int i = 0; i <= node->info->count; i++)
+                {
+                    var child = node->data.children[i];
+                    FreeNode(ref child);
+                    node->data.children[i] = null;
+                }
+            }
+
+            // Free the memory handle
+            if (node->memoryHandle != null)
+            {
+                NativeMemory.Free(node->memoryHandle);
+                node = null;
+            }
+        }
+
+        /// <summary>
+        /// Deallocates the memory allocated for the B+Tree
+        /// </summary>
+        public void Deallocate()
+        {
+            if (root == null)
+                return;
+            Free(ref root);
+            Console.WriteLine("free complete");
+            stats.printStats();
+            root = null;
+            head = null;
+            tail = null;
+        }
+
+        /// <summary>
+        /// Destructor for the B+tree
+        /// </summary>
+        ~BTree()
+        {
+            Deallocate();
+        }
+
+        public ulong FastInserts => stats.totalFastInserts;
+        public ulong LeafCount => stats.numLeafNodes;
+        public ulong InternalCount => stats.numInternalNodes;
+
+        public ulong ValidCount => StatsValidCount();
+
+        public long RootValidCount => GetValidCount(root);
+
+        public long TailValidCount => GetValidCount(tail);
+
+        public long Count()
+        {
+            return stats.numKeys;
+        }
+        public ulong StatsValidCount()
+        {
+            return stats.numValidKeys;
+        }
+
+        public long GetValidCount(BTreeNode* node)
+        {
+            return node->info->validCount;
+        }
+
+        /// <summary>
+        /// Retrieves the first entry in the B+Tree (smallest key)
+        /// </summary>
+        /// <returns>entry fetched</returns>
+        public KeyValuePair<byte[], Value> First()
+        {
+            BTreeNode* leaf = head;
+            if (leaf == null)
+            {
+                return default;
+            }
+            byte[] keyBytes = new ReadOnlySpan<byte>(leaf->GetKey(0), BTreeNode.KEY_SIZE).ToArray();
+            return new KeyValuePair<byte[], Value>(keyBytes, leaf->GetValue(0));
+        }
+
+        /// <summary>
+        /// Retrieves the last entry in the B+Tree (largest key)
+        /// </summary>
+        /// <returns>entry fetched</returns>
+        public KeyValuePair<byte[], Value> Last()
+        {
+            BTreeNode* leaf = tail;
+            if (leaf == null)
+            {
+                return default;
+            }
+            byte[] keyBytes = new ReadOnlySpan<byte>(leaf->GetKey(leaf->info->count - 1), BTreeNode.KEY_SIZE).ToArray();
+            return new KeyValuePair<byte[], Value>(keyBytes, leaf->GetValue(leaf->info->count - 1));
+        }
+
+    }
+}