[C#] Misc remote cleanup (#569)

* Misc remote cleanup

* nit

cs/remote/samples/FixedLenServer/Types.cs

@@ -53,15 +53,15 @@ public struct Output
     }
 
 
-    public struct Functions : IFunctions<Key, Value, Input, Output, long>
+    public struct Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
     {
         // No locking needed for atomic types such as Value
         public bool SupportsLocking => false;
 
         // Callbacks
         public void RMWCompletionCallback(ref Key key, ref Input input, ref Output output, long ctx, Status status) { }
 
-        public void ReadCompletionCallback(ref Key key, ref Input input, ref Output output, long ctx, Status status) { }
+        public void ReadCompletionCallback(ref Key key, ref Input input, ref Output output, long ctx, Status status, RecordInfo recordInfo) { }
 
         public void UpsertCompletionCallback(ref Key key, ref Value value, long ctx) { }
 
@@ -72,17 +72,23 @@ public void CheckpointCompletionCallback(string sessionId, CommitPoint commitPoi
 
         // Read functions
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public void SingleReader(ref Key key, ref Input input, ref Value value, ref Output dst) => dst.value = value;
+        public void SingleReader(ref Key key, ref Input input, ref Value value, ref Output dst, long address)
+        {
+            dst.value = value;
+        }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public void ConcurrentReader(ref Key key, ref Input input, ref Value value, ref Output dst) => dst.value = value;
+        public void ConcurrentReader(ref Key key, ref Input input, ref Value value, ref Output dst, ref RecordInfo recordInfo, long address)
+        {
+            dst.value = value;
+        }
 
         // Upsert functions
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public void SingleWriter(ref Key key, ref Value src, ref Value dst) => dst = src;
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst)
+        public bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst, ref RecordInfo recordInfo, long address)
         {
             dst = src;
             return true;
@@ -97,7 +103,7 @@ public void InitialUpdater(ref Key key, ref Input input, ref Value value, ref Ou
         }
 
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
-        public bool InPlaceUpdater(ref Key key, ref Input input, ref Value value, ref Output output)
+        public bool InPlaceUpdater(ref Key key, ref Input input, ref Value value, ref Output output, ref RecordInfo recordInfo, long address)
         {
             Interlocked.Add(ref value.value, input.value);
             output.value = value;
@@ -114,5 +120,7 @@ public void CopyUpdater(ref Key key, ref Input input, ref Value oldValue, ref Va
         public void Lock(ref RecordInfo recordInfo, ref Key key, ref Value value, LockType lockType, ref long lockContext) { }
 
         public bool Unlock(ref RecordInfo recordInfo, ref Key key, ref Value value, LockType lockType, long lockContext) => true;
+
+        public void ConcurrentDeleter(ref Key key, ref Value value, ref RecordInfo recordInfo, long address) { }
     }
 }

cs/remote/src/FASTER.server/BinaryServerSession.cs

@@ -11,7 +11,7 @@ namespace FASTER.server
 {
     internal unsafe sealed class BinaryServerSession<Key, Value, Input, Output, Functions, ParameterSerializer>
         : FasterKVServerSessionBase<Key, Value, Input, Output, Functions, ParameterSerializer>
-        where Functions : IFunctions<Key, Value, Input, Output, long>
+        where Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
         where ParameterSerializer : IServerSerializer<Key, Value, Input, Output>
     {
         readonly HeaderReaderWriter hrw;

cs/remote/src/FASTER.server/FasterKVProvider.cs

@@ -1,6 +1,4 @@
 using System;
-using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.Net.Sockets;
 using FASTER.common;
 using FASTER.core;
@@ -17,7 +15,7 @@ namespace FASTER.server
     /// <typeparam name="Functions"></typeparam>
     /// <typeparam name="ParameterSerializer"></typeparam>
     public sealed class FasterKVProvider<Key, Value, Input, Output, Functions, ParameterSerializer> : ISessionProvider
-            where Functions : IFunctions<Key, Value, Input, Output, long>
+            where Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
             where ParameterSerializer : IServerSerializer<Key, Value, Input, Output>
     {
         readonly FasterKV<Key, Value> store;

cs/remote/src/FASTER.server/FasterKVServerSessionBase.cs

@@ -5,10 +5,10 @@
 namespace FASTER.server
 {
     internal abstract class FasterKVServerSessionBase<Key, Value, Input, Output, Functions, ParameterSerializer> : FasterKVServerSessionBase<Output>
-        where Functions : IFunctions<Key, Value, Input, Output, long>
+        where Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
         where ParameterSerializer : IServerSerializer<Key, Value, Input, Output>
     {
-        protected readonly ClientSession<Key, Value, Input, Output, long, ServerKVFunctions<Key, Value, Input, Output, Functions>> session;
+        protected readonly AdvancedClientSession<Key, Value, Input, Output, long, ServerKVFunctions<Key, Value, Input, Output, Functions>> session;
         protected readonly ParameterSerializer serializer;
 
         public FasterKVServerSessionBase(Socket socket, FasterKV<Key, Value> store, Functions functions,

cs/remote/src/FASTER.server/FasterServer.cs

@@ -43,7 +43,7 @@ public FasterServer(string address, int port, int networkBufferSize = default)
             if (networkBufferSize == default)
                 this.networkBufferSize = BufferSizeUtils.ClientBufferSize(new MaxSizeSettings());
 
-            var ip = IPAddress.Parse(address);
+            var ip = address == null ? IPAddress.Any : IPAddress.Parse(address);
             var endPoint = new IPEndPoint(ip, port);
             servSocket = new Socket(ip.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
             servSocket.Bind(endPoint);
@@ -98,7 +98,7 @@ private bool HandleNewConnection(SocketAsyncEventArgs e)
                 return false;
             }
 
-            // Ok to create new event args on accept because we assume a connection to be long-running
+            // Ok to create new event args on accept because we assume a connection to be long-running            
             var receiveEventArgs = new SocketAsyncEventArgs();
             var buffer = new byte[networkBufferSize];
             receiveEventArgs.SetBuffer(buffer, 0, networkBufferSize);
@@ -247,7 +247,15 @@ private unsafe bool CreateSession(SocketAsyncEventArgs e)
 
             connArgs.session.AddBytesRead(connArgs.bytesRead);
             var _newHead = connArgs.session.TryConsumeMessages(e.Buffer);
-            e.SetBuffer(_newHead, e.Buffer.Length - _newHead);
+            if (_newHead == e.Buffer.Length)
+            {
+                // Need to grow input buffer
+                var newBuffer = new byte[e.Buffer.Length * 2];
+                Array.Copy(e.Buffer, newBuffer, e.Buffer.Length);
+                e.SetBuffer(newBuffer, _newHead, newBuffer.Length - _newHead);
+            }
+            else
+                e.SetBuffer(_newHead, e.Buffer.Length - _newHead);
             return true;
         }
 

cs/remote/src/FASTER.server/SpanByteFasterKVProvider.cs → cs/remote/src/FASTER.server/Providers/SpanByteFasterKVProvider.cs

@@ -10,7 +10,7 @@ namespace FASTER.server
     /// </summary>
     public sealed class SpanByteFasterKVProvider : ISessionProvider
     {
-        readonly StoreWrapper<SpanByte, SpanByte> storeWrapper;
+        readonly FasterKV<SpanByte, SpanByte> store;
         readonly SpanByteServerSerializer serializer;
         readonly SubscribeKVBroker<SpanByte, SpanByte, SpanByte, IKeyInputSerializer<SpanByte, SpanByte>> kvBroker;
         readonly SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>> broker;
@@ -22,11 +22,20 @@ public sealed class SpanByteFasterKVProvider : ISessionProvider
         /// <param name="store"></param>
         /// <param name="kvBroker"></param>
         /// <param name="broker"></param>
-        /// <param name="tryRecover"></param>
+        /// <param name="serverOptions"></param>
         /// <param name="maxSizeSettings"></param>
-        public SpanByteFasterKVProvider(FasterKV<SpanByte, SpanByte> store, SubscribeKVBroker<SpanByte, SpanByte, SpanByte, IKeyInputSerializer<SpanByte, SpanByte>> kvBroker = null, SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>> broker = null, bool tryRecover = true, MaxSizeSettings maxSizeSettings = default)
+        public SpanByteFasterKVProvider(FasterKV<SpanByte, SpanByte> store, SubscribeKVBroker<SpanByte, SpanByte, SpanByte, IKeyInputSerializer<SpanByte, SpanByte>> kvBroker = null, SubscribeBroker<SpanByte, SpanByte, IKeySerializer<SpanByte>> broker = null, ServerOptions serverOptions = null, MaxSizeSettings maxSizeSettings = default)
         {
-            this.storeWrapper = new StoreWrapper<SpanByte, SpanByte>(store, tryRecover);
+            this.store = store;
+            if ((serverOptions ?? new ServerOptions()).Recover)
+            {
+                try
+                {
+                    store.Recover();
+                }
+                catch
+                { }
+            }
             this.kvBroker = kvBroker;
             this.broker = broker;
             this.serializer = new SpanByteServerSerializer();
@@ -40,10 +49,10 @@ public IServerSession GetSession(WireFormat wireFormat, Socket socket)
             {
                 case WireFormat.WebSocket:
                     return new WebsocketServerSession<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, SpanByteFunctionsForServer<long>, SpanByteServerSerializer>
-                        (socket, storeWrapper.store, new SpanByteFunctionsForServer<long>(), serializer, maxSizeSettings, kvBroker, broker);
+                        (socket, store, new SpanByteFunctionsForServer<long>(), serializer, maxSizeSettings, kvBroker, broker);
                 default:
                     return new BinaryServerSession<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, SpanByteFunctionsForServer<long>, SpanByteServerSerializer>
-                        (socket, storeWrapper.store, new SpanByteFunctionsForServer<long>(), serializer, maxSizeSettings, kvBroker, broker);
+                        (socket, store, new SpanByteFunctionsForServer<long>(), serializer, maxSizeSettings, kvBroker, broker);
             }
         }
     }

cs/remote/src/FASTER.server/PubSub/SubscribeBroker.cs

@@ -28,6 +28,7 @@ public sealed class SubscribeBroker<Key, Value, KeyValueSerializer> : IDisposabl
         private AsyncQueue<(byte[], byte[])> publishQueue;
         readonly IKeySerializer<Key> keySerializer;
         readonly FasterLog log;
+        readonly IDevice device;
         readonly CancellationTokenSource cts = new();
         readonly ManualResetEvent done = new(true);
         bool disposed = false;
@@ -41,7 +42,7 @@ public sealed class SubscribeBroker<Key, Value, KeyValueSerializer> : IDisposabl
         public SubscribeBroker(IKeySerializer<Key> keySerializer, string logDir, bool startFresh = true)
         {
             this.keySerializer = keySerializer;
-            var device = logDir == null ? new NullDevice() : Devices.CreateLogDevice(logDir + "/pubsubkv", preallocateFile: false);
+            device = logDir == null ? new NullDevice() : Devices.CreateLogDevice(logDir + "/pubsubkv", preallocateFile: false);
             device.Initialize((long)(1 << 30) * 64);
             log = new FasterLog(new FasterLogSettings { LogDevice = device });
             if (startFresh)
@@ -139,8 +140,8 @@ private async Task Start(CancellationToken cancellationToken = default)
                     if (disposed)
                         break;
 
-                    var iter = log.Scan(log.BeginAddress, long.MaxValue, scanUncommitted: true);
-                    await iter.WaitAsync(cts.Token);
+                    using var iter = log.Scan(log.BeginAddress, long.MaxValue, scanUncommitted: true);
+                    await iter.WaitAsync(cancellationToken);
                     while (iter.GetNext(out byte[] subscriptionKey, out _, out _, out _))
                     {
                         if (!iter.GetNext(out byte[] subscriptionValue, out _, out long currentAddress, out long nextAddress))
@@ -376,6 +377,7 @@ public void Dispose()
             subscriptions?.Clear();
             prefixSubscriptions?.Clear();
             log.Dispose();
+            device.Dispose();
         }
     }
 }

cs/remote/src/FASTER.server/PubSub/SubscribeKVBroker.cs

@@ -28,6 +28,7 @@ public sealed class SubscribeKVBroker<Key, Value, Input, KeyInputSerializer> : I
         private AsyncQueue<byte[]> publishQueue;
         readonly IKeyInputSerializer<Key, Input> keyInputSerializer;
         readonly FasterLog log;
+        readonly IDevice device;
         readonly CancellationTokenSource cts = new();
         readonly ManualResetEvent done = new(true);
         bool disposed = false;
@@ -41,7 +42,7 @@ public sealed class SubscribeKVBroker<Key, Value, Input, KeyInputSerializer> : I
         public SubscribeKVBroker(IKeyInputSerializer<Key, Input> keyInputSerializer, string logDir, bool startFresh = true)
         {
             this.keyInputSerializer = keyInputSerializer;
-            var device = logDir == null ? new NullDevice() : Devices.CreateLogDevice(logDir + "/pubsubkv", preallocateFile: false);
+            device = logDir == null ? new NullDevice() : Devices.CreateLogDevice(logDir + "/pubsubkv", preallocateFile: false);
             device.Initialize((long)(1 << 30) * 64);
             log = new FasterLog(new FasterLogSettings { LogDevice = device });
             if (startFresh)
@@ -101,7 +102,7 @@ internal async Task Start(CancellationToken cancellationToken = default)
                     if (disposed)
                         break;
 
-                    var iter = log.Scan(log.BeginAddress, long.MaxValue, scanUncommitted: true);
+                    using var iter = log.Scan(log.BeginAddress, long.MaxValue, scanUncommitted: true);
                     await iter.WaitAsync(cancellationToken);
                     while (iter.GetNext(out byte[] subscriptionKey, out int entryLength, out long currentAddress, out long nextAddress))
                     {
@@ -255,6 +256,7 @@ public void Dispose()
             subscriptions?.Clear();
             prefixSubscriptions?.Clear();
             log.Dispose();
+            device.Dispose();
         }
     }
 }

cs/remote/src/FASTER.server/ServerKVFunctions.cs

@@ -5,8 +5,8 @@
 
 namespace FASTER.server
 {
-    internal struct ServerKVFunctions<Key, Value, Input, Output, Functions> : IFunctions<Key, Value, Input, Output, long>
-        where Functions : IFunctions<Key, Value, Input, Output, long>
+    internal struct ServerKVFunctions<Key, Value, Input, Output, Functions> : IAdvancedFunctions<Key, Value, Input, Output, long>
+        where Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
     {
         private readonly Functions functions;
         private readonly FasterKVServerSessionBase<Output> serverNetworkSession;
@@ -22,11 +22,14 @@ public ServerKVFunctions(Functions functions, FasterKVServerSessionBase<Output>
         public void CheckpointCompletionCallback(string sessionId, CommitPoint commitPoint)
             => functions.CheckpointCompletionCallback(sessionId, commitPoint);
 
-        public void ConcurrentReader(ref Key key, ref Input input, ref Value value, ref Output dst)
-            => functions.ConcurrentReader(ref key, ref input, ref value, ref dst);
+        public void ConcurrentDeleter(ref Key key, ref Value value, ref RecordInfo recordInfo, long address)
+            => functions.ConcurrentDeleter(ref key, ref value, ref recordInfo, address);
 
-        public bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst)
-            => functions.ConcurrentWriter(ref key, ref src, ref dst);
+        public void ConcurrentReader(ref Key key, ref Input input, ref Value value, ref Output dst, ref RecordInfo recordInfo, long address)
+            => functions.ConcurrentReader(ref key, ref input, ref value, ref dst, ref recordInfo, address);
+
+        public bool ConcurrentWriter(ref Key key, ref Value src, ref Value dst, ref RecordInfo recordInfo, long address)
+            => functions.ConcurrentWriter(ref key, ref src, ref dst, ref recordInfo, address);
 
         public bool NeedCopyUpdate(ref Key key, ref Input input, ref Value oldValue, ref Output output)
             => functions.NeedCopyUpdate(ref key, ref input, ref oldValue, ref output);
@@ -40,13 +43,13 @@ public void DeleteCompletionCallback(ref Key key, long ctx)
         public void InitialUpdater(ref Key key, ref Input input, ref Value value, ref Output output)
             => functions.InitialUpdater(ref key, ref input, ref value, ref output);
 
-        public bool InPlaceUpdater(ref Key key, ref Input input, ref Value value, ref Output output)
-            => functions.InPlaceUpdater(ref key, ref input, ref value, ref output);
+        public bool InPlaceUpdater(ref Key key, ref Input input, ref Value value, ref Output output, ref RecordInfo recordInfo, long address)
+            => functions.InPlaceUpdater(ref key, ref input, ref value, ref output, ref recordInfo, address);
 
-        public void ReadCompletionCallback(ref Key key, ref Input input, ref Output output, long ctx, Status status)
+        public void ReadCompletionCallback(ref Key key, ref Input input, ref Output output, long ctx, Status status, RecordInfo recordInfo)
         {
             serverNetworkSession.CompleteRead(ref output, ctx, status);
-            functions.ReadCompletionCallback(ref key, ref input, ref output, ctx, status);
+            functions.ReadCompletionCallback(ref key, ref input, ref output, ctx, status, recordInfo);
         }
 
         public void RMWCompletionCallback(ref Key key, ref Input input, ref Output output, long ctx, Status status)
@@ -55,8 +58,8 @@ public void RMWCompletionCallback(ref Key key, ref Input input, ref Output outpu
             functions.RMWCompletionCallback(ref key, ref input, ref output, ctx, status);
         }
 
-        public void SingleReader(ref Key key, ref Input input, ref Value value, ref Output dst)
-            => functions.SingleReader(ref key, ref input, ref value, ref dst);
+        public void SingleReader(ref Key key, ref Input input, ref Value value, ref Output dst, long address)
+            => functions.SingleReader(ref key, ref input, ref value, ref dst, address);
 
         public void SingleWriter(ref Key key, ref Value src, ref Value dst)
             => functions.SingleWriter(ref key, ref src, ref dst);

cs/remote/src/FASTER.server/Servers/FixedLenServer.cs

@@ -15,7 +15,7 @@ public sealed class FixedLenServer<Key, Value, Input, Output, Functions> : Gener
         where Value : unmanaged
         where Input : unmanaged
         where Output : unmanaged
-        where Functions : IFunctions<Key, Value, Input, Output, long>
+        where Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
     {
         /// <summary>
         /// Create server instance; use Start to start the server.

cs/remote/src/FASTER.server/Servers/GenericServer.cs

@@ -12,7 +12,7 @@ namespace FASTER.server
     /// FASTER server for generic types
     /// </summary>
     public class GenericServer<Key, Value, Input, Output, Functions, ParameterSerializer> : IDisposable
-            where Functions : IFunctions<Key, Value, Input, Output, long>
+            where Functions : IAdvancedFunctions<Key, Value, Input, Output, long>
             where ParameterSerializer : IServerSerializer<Key, Value, Input, Output>
     {
         readonly ServerOptions opts;