Functions in session (#270)

* Moving function callbacks to sessions (#230)
* Use special interface to interact with state machine(s).
* Mark synchronization listener to be struct for jit to be able to inline calls.
* Introduced IFasterSession instead of ISynchronizationListener. It combines both the listener and functions. Both legacy kv and client session provide struct based implementation to minimize impact of having interface calls.
* Move out Input, Output, Context from FasterKV into methods.
* Added simple test for multiple sessions
* Added `custom compaction functions (#272).
Fixed a bug where compaction would use stale value if multiple records with different lengths (for varlen structs) where encountered during compaction.
* Provide `ref Input input` for calculating size of new record when allocating new record during RMW.
* Removed sync methods of state machine.

cs/playground/FasterKVDiskReadBenchmark/Program.cs

@@ -117,7 +117,7 @@ static async Task AsyncReadOperator(int id)
                 Input input = default;
                 int i = 0;
 
-                var tasks = new (long, ValueTask<FasterKV<Key, Value, Input, Output, Empty, MyFuncs>.ReadAsyncResult>)[batchSize];
+                var tasks = new (long, ValueTask<FasterKV<Key, Value>.ReadAsyncResult<Input, Output, Empty, MyFuncs>>)[batchSize];
                 while (true)
                 {
                     key = new Key(NumKeys * id + rand.Next(0, NumKeys));

cs/playground/VarLenStructSample/Types.cs

@@ -77,5 +77,10 @@ public int GetLength(ref VarLenType t)
         {
             return sizeof(int) * t.length;
         }
+
+        public int GetLength<Input>(ref VarLenType t, ref Input input)
+        {
+            return sizeof(int) * t.length;
+        }
     }
 }

cs/src/core/Allocator/AllocatorBase.cs

@@ -293,6 +293,13 @@ public unsafe abstract partial class AllocatorBase<Key, Value> : IDisposable
         /// <returns></returns>
         public abstract int GetRecordSize(long physicalAddress);
 
+        /// <summary>
+        /// Get record size
+        /// </summary>
+        /// <param name="physicalAddress"></param>
+        /// <param name="input"></param>
+        /// <returns></returns>
+        public abstract int GetRecordSize<Input>(long physicalAddress, ref Input input);
 
         /// <summary>
         /// Get number of bytes required

cs/src/core/Allocator/BlittableAllocator.cs

@@ -67,6 +67,11 @@ public override int GetRecordSize(long physicalAddress)
             return recordSize;
         }
 
+        public override int GetRecordSize<Input>(long physicalAddress, ref Input input)
+        {
+            return recordSize;
+        }
+
         public override int GetAverageRecordSize()
         {
             return recordSize;

cs/src/core/Allocator/GenericAllocator.cs

@@ -114,7 +114,6 @@ public override ref RecordInfo GetInfoFromBytePointer(byte* ptr)
             return ref Unsafe.AsRef<Record<Key, Value>>(ptr).info;
         }
 
-
         public override ref Key GetKey(long physicalAddress)
         {
             // Offset within page
@@ -142,6 +141,11 @@ public override int GetRecordSize(long physicalAddress)
             return recordSize;
         }
 
+        public override int GetRecordSize<Input>(long physicalAddress, ref Input input)
+        {
+            return recordSize;
+        }
+
         public override int GetAverageRecordSize()
         {
             return recordSize;

cs/src/core/Allocator/VarLenBlittableAllocator.cs

@@ -89,6 +89,11 @@ private int ValueSize(long physicalAddress)
             return ValueLength.GetLength(ref GetValue(physicalAddress));
         }
 
+        private int ValueSize<Input>(long physicalAddress, ref Input input)
+        {
+            return ValueLength.GetLength(ref GetValue(physicalAddress), ref input);
+        }
+
         public override int GetRecordSize(long physicalAddress)
         {
             ref var recordInfo = ref GetInfo(physicalAddress);
@@ -100,6 +105,17 @@ public override int GetRecordSize(long physicalAddress)
             return size;
         }
 
+        public override int GetRecordSize<Input>(long physicalAddress, ref Input input)
+        {
+            ref var recordInfo = ref GetInfo(physicalAddress);
+            if (recordInfo.IsNull())
+                return RecordInfo.GetLength();
+
+            var size = RecordInfo.GetLength() + KeySize(physicalAddress) + ValueSize(physicalAddress, ref input);
+            size = (size + kRecordAlignment - 1) & (~(kRecordAlignment - 1));
+            return size;
+        }
+
         public override int GetRequiredRecordSize(long physicalAddress, int availableBytes)
         {
             // We need at least [record size] + [average key size] + [average value size]