First implementation of new SI API

cs/benchmark/FasterSpanByteYcsbBenchmark.cs

@@ -300,7 +300,7 @@ public unsafe void Run()
             long startTailAddress = store.Log.TailAddress;
             Console.WriteLine("Start tail address = " + startTailAddress);
 
-            if (!storeWasRecovered && this.backupMode.HasFlag(BackupMode.Backukp) && kPeriodicCheckpointMilliseconds <= 0)
+            if (!storeWasRecovered && this.backupMode.HasFlag(BackupMode.Backup) && kPeriodicCheckpointMilliseconds <= 0)
             {
                 Console.WriteLine("Checkpointing FasterKV for fast restart");
                 store.TakeFullCheckpoint(out _);

cs/benchmark/FasterYcsbBenchmark.cs

@@ -82,7 +82,7 @@ public enum Op : ulong
 
         volatile bool done = false;
 
-        public FASTER_YcsbBenchmark(int threadCount_, int numaStyle_, string distribution_, int readPercent_, int backupOptions_)
+        public FASTER_YcsbBenchmark(int threadCount_, int numaStyle_, string distribution_, int readPercent_, int backupOptions_, int indexType_)
         {
             // Pin loading thread if it is not used for checkpointing
             if (kPeriodicCheckpointMilliseconds <= 0)
@@ -110,13 +110,26 @@ public FASTER_YcsbBenchmark(int threadCount_, int numaStyle_, string distributio
 
             var path = "D:\\data\\FasterYcsbBenchmark\\";
             device = Devices.CreateLogDevice(path + "hlog", preallocateFile: true);
+            var secondaryIndexType = (SecondaryIndexType)indexType_;
 
             if (kSmallMemoryLog)
                 store = new FasterKV<Key, Value>
-                    (kMaxKey / 2, new LogSettings { LogDevice = device, PreallocateLog = true, PageSizeBits = 22, SegmentSizeBits = 26, MemorySizeBits = 26 }, new CheckpointSettings { CheckPointType = CheckpointType.FoldOver, CheckpointDir = path });
+                    (kMaxKey / 2, new LogSettings { LogDevice = device, PreallocateLog = true, PageSizeBits = 22, SegmentSizeBits = 26, MemorySizeBits = 26 },
+                    new CheckpointSettings { CheckPointType = CheckpointType.FoldOver, CheckpointDir = path },
+                    supportsMutableIndexes: secondaryIndexType != SecondaryIndexType.None);
             else
                 store = new FasterKV<Key, Value>
-                    (kMaxKey / 2, new LogSettings { LogDevice = device, PreallocateLog = true }, new CheckpointSettings { CheckPointType = CheckpointType.FoldOver, CheckpointDir = path });
+                    (kMaxKey / 2, new LogSettings { LogDevice = device, PreallocateLog = true },
+                    new CheckpointSettings { CheckPointType = CheckpointType.FoldOver, CheckpointDir = path },
+                    supportsMutableIndexes: secondaryIndexType != SecondaryIndexType.None);
+
+            if (secondaryIndexType != SecondaryIndexType.None)
+            {
+                if (secondaryIndexType.HasFlag(SecondaryIndexType.Key))
+                    store.SecondaryIndexBroker.AddIndex(new NullKeyIndex<Key>());
+                if (secondaryIndexType.HasFlag(SecondaryIndexType.Value))
+                    store.SecondaryIndexBroker.AddIndex(new NullValueIndex<Value>());
+            }
         }
 
         private void RunYcsb(int thread_idx)
@@ -295,7 +308,7 @@ public unsafe void Run()
             long startTailAddress = store.Log.TailAddress;
             Console.WriteLine("Start tail address = " + startTailAddress);
 
-            if (!storeWasRecovered && this.backupMode.HasFlag(BackupMode.Backukp) && kPeriodicCheckpointMilliseconds <= 0)
+            if (!storeWasRecovered && this.backupMode.HasFlag(BackupMode.Backup) && kPeriodicCheckpointMilliseconds <= 0)
             {
                 Console.WriteLine("Checkpointing FasterKV for fast restart");
                 store.TakeFullCheckpoint(out _);

cs/benchmark/Program.cs

@@ -28,6 +28,14 @@ class Options
                         "\n    3 = Both (Recover FasterKV if the Checkpoint is available, else populate FasterKV from data and Checkpoint it so it can be Restored in a subsequent run)")]
         public int Backup { get; set; }
 
+        [Option('i', "index", Required = false, Default = 0,
+             HelpText = "Secondary index type(s); these implement a no-op index to test the overhead on FasterKV operations:" +
+                        "\n    0 = None (default)" +
+                        "\n    1 = Key-based index" +
+                        "\n    2 = Value-based index" +
+                        "\n    3 = Both index types")]
+        public int SecondaryIndexType { get; set; }
+
         [Option('r', "read_percent", Required = false, Default = 50,
          HelpText = "Percentage of reads (-1 for 100% read-modify-write")]
         public int ReadPercent { get; set; }
@@ -44,7 +52,13 @@ enum BenchmarkType : int
 
     [Flags] enum BackupMode : int
     {
-        None, Restore, Backukp, Both
+        None, Restore, Backup, Both
+    };
+
+    [Flags]
+    enum SecondaryIndexType : int
+    {
+        None, Key, Value, Both
     };
 
     public class Program
@@ -62,7 +76,7 @@ public static void Main(string[] args)
 
             if (b == BenchmarkType.Ycsb)
             {
-                var test = new FASTER_YcsbBenchmark(options.ThreadCount, options.NumaStyle, options.Distribution, options.ReadPercent, options.Backup);
+                var test = new FASTER_YcsbBenchmark(options.ThreadCount, options.NumaStyle, options.Distribution, options.ReadPercent, options.Backup, options.SecondaryIndexType);
                 test.Run();
             }
             else if (b == BenchmarkType.SpanByte)

cs/benchmark/SecondaryIndexes.cs

@@ -0,0 +1,33 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT license.
+
+using FASTER.core;
+
+namespace FASTER.benchmark
+{
+    class NullKeyIndex<Key> : ISecondaryKeyIndex<Key>
+    {
+        public string Name => "KeyIndex";
+
+        public bool IsMutable => true;
+
+        public void Delete(ref Key key) { }
+
+        public void Insert(ref Key key) { }
+
+        public void Upsert(ref Key key, bool isMutable) { }
+    }
+
+    class NullValueIndex<Value> : ISecondaryValueIndex<Value>
+    {
+        public string Name => "ValueIndex";
+
+        public bool IsMutable => true;
+
+        public void Delete(ref Value value, long recordId) { }
+
+        public void Insert(ref Value value, long recordId) { }
+
+        public void Upsert(ref Value value, long recordId, bool isMutable) { }
+    }
+}

cs/src/core/Allocator/GenericAllocator.cs

@@ -922,7 +922,7 @@ protected override bool RetrievedFullRecord(byte* record, ref AsyncIOContext<Key
         /// <returns></returns>
         public override bool KeyHasObjects()
         {
-            return SerializerSettings.keySerializer != null;
+            return SerializerSettings?.keySerializer != null;
         }
 
         /// <summary>
@@ -931,7 +931,7 @@ public override bool KeyHasObjects()
         /// <returns></returns>
         public override bool ValueHasObjects()
         {
-            return SerializerSettings.valueSerializer != null;
+            return SerializerSettings?.valueSerializer != null;
         }
         #endregion
 

cs/src/core/ClientSession/AdvancedClientSession.cs

@@ -737,15 +737,54 @@ public void ConcurrentReader(ref Key key, ref Input input, ref Value value, ref
 
             public bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst, long address)
             {
-                return _clientSession.functions.ConcurrentWriter(ref key, ref src, ref dst, address);
+                if (!_clientSession.fht.SupportsMutableIndexes)
+                    return _clientSession.functions.ConcurrentWriter(ref key, ref src, ref dst, address);
+
+                ref RecordInfo recordInfo = ref _clientSession.fht.RecordAccessor.SpinLockRecordInfo(address);
+                try
+                {
+                    if (!recordInfo.Tombstone && _clientSession.functions.ConcurrentWriter(ref key, ref src, ref dst, address))
+                    {
+                        // KeyIndexes do not need notification of in-place updates because the key does not change.
+                        if (_clientSession.fht.SecondaryIndexBroker.MutableValueIndexCount > 0)
+                            _clientSession.fht.SecondaryIndexBroker.Upsert(ref dst, address);
+                        return true;
+                    }
+                }
+                finally
+                {
+                    recordInfo.Unlock();
+                }
+                return false;
+            }
+
+            public bool ConcurrentDeleter(ref Key key, ref Value value, long address)
+            {
+                if (!_clientSession.fht.SupportsMutableIndexes)
+                    return _clientSession.functions.ConcurrentDeleter(ref key, ref value, address);
+
+                ref RecordInfo recordInfo = ref _clientSession.fht.RecordAccessor.SpinLockRecordInfo(address);
+                try
+                {
+                    if (_clientSession.fht.SecondaryIndexBroker.MutableKeyIndexCount > 0)
+                        _clientSession.fht.SecondaryIndexBroker.Delete(ref key);
+                    if (_clientSession.fht.SecondaryIndexBroker.MutableValueIndexCount > 0)
+                        _clientSession.fht.SecondaryIndexBroker.Delete(ref value, address);
+                    return _clientSession.functions.ConcurrentDeleter(ref key, ref value, address);
+                }
+                finally
+                {
+                    recordInfo.Unlock();
+                }
+                return true;
             }
 
             public bool NeedCopyUpdate(ref Key key, ref Input input, ref Value oldValue)
                 => _clientSession.functions.NeedCopyUpdate(ref key, ref input, ref oldValue);
 
-            public void CopyUpdater(ref Key key, ref Input input, ref Value oldValue, ref Value newValue, long oldAddress, long newAddress)
+            public void CopyUpdater(ref Key key, ref Input input, ref Value oldValue, ref Value newValue, long address)
             {
-                _clientSession.functions.CopyUpdater(ref key, ref input, ref oldValue, ref newValue, oldAddress, newAddress);
+                _clientSession.functions.CopyUpdater(ref key, ref input, ref oldValue, ref newValue, address);
             }
 
             public void DeleteCompletionCallback(ref Key key, Context ctx)
@@ -770,7 +809,25 @@ public void InitialUpdater(ref Key key, ref Input input, ref Value value, long a
 
             public bool InPlaceUpdater(ref Key key, ref Input input, ref Value value, long address)
             {
-                return _clientSession.functions.InPlaceUpdater(ref key, ref input, ref value, address);
+                if (!_clientSession.fht.SupportsMutableIndexes)
+                    return _clientSession.functions.InPlaceUpdater(ref key, ref input, ref value, address);
+
+                ref RecordInfo recordInfo = ref _clientSession.fht.RecordAccessor.SpinLockRecordInfo(address);
+                try
+                {
+                    if (!recordInfo.Tombstone && _clientSession.functions.InPlaceUpdater(ref key, ref input, ref value, address))
+                    {
+                        // KeyIndexes do not need notification of in-place updates because the key does not change.
+                        if (_clientSession.fht.SecondaryIndexBroker.MutableValueIndexCount > 0)
+                            _clientSession.fht.SecondaryIndexBroker.Upsert(ref value, address);
+                        return true;
+                    }
+                }
+                finally
+                {
+                    recordInfo.Unlock();
+                }
+                return false;
             }
 
             public void ReadCompletionCallback(ref Key key, ref Input input, ref Output output, Context ctx, Status status, RecordInfo recordInfo)

cs/src/core/ClientSession/ClientSession.cs

@@ -747,13 +747,53 @@ public void ConcurrentReader(ref Key key, ref Input input, ref Value value, ref
 
             public bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst, long address)
             {
-                return _clientSession.functions.ConcurrentWriter(ref key, ref src, ref dst);
+                if (!_clientSession.fht.SupportsMutableIndexes)
+                    return _clientSession.functions.ConcurrentWriter(ref key, ref src, ref dst);
+
+                ref RecordInfo recordInfo = ref _clientSession.fht.RecordAccessor.SpinLockRecordInfo(address);
+                try
+                {
+                    if (!recordInfo.Tombstone && _clientSession.functions.ConcurrentWriter(ref key, ref src, ref dst))
+                    {
+                        // KeyIndexes do not need notification of in-place updates because the key does not change.
+                        if (_clientSession.fht.SecondaryIndexBroker.MutableValueIndexCount > 0)
+                            _clientSession.fht.SecondaryIndexBroker.Upsert(ref dst, address);
+                        return true;
+                    }
+                }
+                finally
+                {
+                    recordInfo.Unlock();
+                }
+                return false;
+            }
+
+            public bool ConcurrentDeleter(ref Key key, ref Value value, long address)
+            {
+                // Non-Advanced IFunctions has no ConcurrentDeleter
+                if (!_clientSession.fht.SupportsMutableIndexes)
+                    return false;
+
+                ref RecordInfo recordInfo = ref _clientSession.fht.RecordAccessor.SpinLockRecordInfo(address);
+                try
+                {
+                    if (_clientSession.fht.SecondaryIndexBroker.MutableKeyIndexCount > 0)
+                        _clientSession.fht.SecondaryIndexBroker.Delete(ref key);
+                    if (_clientSession.fht.SecondaryIndexBroker.MutableValueIndexCount > 0)
+                        _clientSession.fht.SecondaryIndexBroker.Delete(ref value, address);
+                    _clientSession.fht.SetRecordDeleted(ref recordInfo, ref value);
+                }
+                finally
+                {
+                    recordInfo.Unlock();
+                }
+                return true;
             }
 
             public bool NeedCopyUpdate(ref Key key, ref Input input, ref Value oldValue)
                 => _clientSession.functions.NeedCopyUpdate(ref key, ref input, ref oldValue);
 
-            public void CopyUpdater(ref Key key, ref Input input, ref Value oldValue, ref Value newValue, long oldAddress, long newAddress)
+            public void CopyUpdater(ref Key key, ref Input input, ref Value oldValue, ref Value newValue, long address)
             {
                 _clientSession.functions.CopyUpdater(ref key, ref input, ref oldValue, ref newValue);
             }
@@ -780,7 +820,25 @@ public void InitialUpdater(ref Key key, ref Input input, ref Value value, long a
 
             public bool InPlaceUpdater(ref Key key, ref Input input, ref Value value, long address)
             {
-                return _clientSession.functions.InPlaceUpdater(ref key, ref input, ref value);
+                if (!_clientSession.fht.SupportsMutableIndexes)
+                    return _clientSession.functions.InPlaceUpdater(ref key, ref input, ref value);
+
+                ref RecordInfo recordInfo = ref _clientSession.fht.RecordAccessor.SpinLockRecordInfo(address);
+                try
+                {
+                    if (!recordInfo.Tombstone && _clientSession.functions.InPlaceUpdater(ref key, ref input, ref value))
+                    {
+                        // KeyIndexes do not need notification of in-place updates because the key does not change.
+                        if (_clientSession.fht.SecondaryIndexBroker.MutableValueIndexCount > 0)
+                            _clientSession.fht.SecondaryIndexBroker.Upsert(ref value, address);
+                        return true;
+                    }
+                }
+                finally
+                {
+                    recordInfo.Unlock();
+                }
+                return false;
             }
 
             public void ReadCompletionCallback(ref Key key, ref Input input, ref Output output, Context ctx, Status status, RecordInfo recordInfo)

cs/src/core/Index/Common/Contexts.cs

@@ -89,6 +89,7 @@ internal struct PendingContext<Input, Output, Context>
             internal const byte kSkipReadCache = 0x01;
             internal const byte kNoKey = 0x02;
             internal const byte kSkipCopyReadsToTail = 0x04;
+            internal const byte kIsNewRecord = 0x08;
 
             internal static byte GetOperationFlags(ReadFlags readFlags, bool noKey = false)
             {
@@ -119,6 +120,12 @@ internal bool SkipCopyReadsToTail
                 set => operationFlags = value ? (byte)(operationFlags | kSkipCopyReadsToTail) : (byte)(operationFlags & ~kSkipCopyReadsToTail);
             }
 
+            internal bool IsNewRecord
+            {
+                get => (operationFlags & kIsNewRecord) != 0;
+                set => operationFlags = value ? (byte)(operationFlags | kIsNewRecord) : (byte)(operationFlags & ~kIsNewRecord);
+            }
+
             public void Dispose()
             {
                 key?.Dispose();