Distributed transactions

.gitignore

@@ -197,7 +197,9 @@ nativebin/
 
 # JetBrains
 cs/.idea/
-cs/remote/.idea
-cs/libdpr/.idea
+cs/remote/.idea/
+cs/research/libdpr/.idea/
+cs/libdpr/.idea/
+cs/research/darq/.idea/
 
 cs/**/BenchmarkDotNet.Artifacts/

Dockerfile

@@ -0,0 +1,28 @@
+FROM mcr.microsoft.com/dotnet/sdk:7.0 AS build-env
+WORKDIR /app
+
+# Build Travel Reservation
+COPY ./cs .
+WORKDIR /app/research/darq/TravelReservation
+RUN dotnet restore
+RUN dotnet build -c Release -o out
+
+# Build EventProcessing
+WORKDIR /app/research/darq/EventProcessing
+RUN dotnet restore
+RUN dotnet publish -c Release -o out
+
+# Build CoordinatorMicrobench
+WORKDIR /app/research/darq/CoordinatorMicrobench
+RUN dotnet restore
+RUN dotnet publish -c Release -o out
+
+FROM mcr.microsoft.com/dotnet/aspnet:7.0
+WORKDIR /app
+COPY --from=build-env /app/TravelReservation-latency ./TravelReservation-latency
+COPY --from=build-env /app/TravelReservation-thr ./TravelReservation-thr
+COPY --from=build-env /app/EventProcessing ./EventProcessing-latency
+COPY --from=build-env /app/research/darq/TravelReservation/out ./TravelReservation
+COPY --from=build-env /app/research/darq/EventProcessing/out ./EventProcessing
+COPY --from=build-env /app/research/darq/CoordinatorMicrobench/out ./CoordinatorMicrobench
+EXPOSE 4022

cs/playground/FasterLogMLSDTest/Program.cs

@@ -4,7 +4,9 @@
 using System;
 using System.Diagnostics;
 using System.IO;
+using System.Linq;
 using System.Threading;
+using System.Threading.Tasks;
 using FASTER.core;
 
 namespace FasterLogStress
@@ -16,27 +18,55 @@ public class Program
         static readonly byte[] entry = new byte[100];
         private static string commitPath;
 
-        public static void Main()
+        private static byte[] buffer;
+
+        public static async Task Main()
         {
-            commitPath = "FasterLogStress/";
+            using var settings = new FasterLogSettings("./Test", deleteDirOnDispose: true)
+            {
+                AutoRefreshSafeTailAddress = true
+            };
+            log = new FasterLog(settings);
 
-            // Clean up log files from previous test runs in case they weren't cleaned up
-            // We loop to ensure clean-up as deleteOnClose does not always work for MLSD
-            while (Directory.Exists(commitPath))
-                Directory.Delete(commitPath, true);
+            buffer = new byte[2048];
+            Random.Shared.NextBytes(buffer);
 
-            // Create devices \ log for test
-            device = new ManagedLocalStorageDevice(commitPath + "ManagedLocalStore.log", deleteOnClose: true);
-            log = new FasterLog(new FasterLogSettings { LogDevice = device, PageSizeBits = 12, MemorySizeBits = 14 });
+            await Task.WhenAll(EnqueueThread(), ScanThread());
+            log.Dispose();
+        }
+
+        static async Task EnqueueThread()
+        {
+            for (int count = 0; count < 5; ++count)
+            {
+                await log.EnqueueAsync(buffer);
+                await Task.Delay(1000);
+            }
 
-            ManagedLocalStoreBasicTest();
+            log.CompleteLog();
+            Console.WriteLine("Enqueue complete");
+        }
 
-            log.Dispose();
-            device.Dispose();
+        static async Task ScanThread()
+        {
+            using var iterator = log.Scan(log.BeginAddress, long.MaxValue, scanUncommitted: true);
+            while (true)
+            {
+                byte[] result;
+                while (!iterator.GetNext(out result, out _, out _))
+                {
+                    if (iterator.Ended)
+                    {
+                        Console.WriteLine("Scan complete");
+                        return;
+                    }
 
-            // Clean up log files
-            if (Directory.Exists(commitPath))
-                Directory.Delete(commitPath, true);
+                    await iterator.WaitAsync();
+                }
+
+                Console.WriteLine("Received buffer");
+                Debug.Assert(result.SequenceEqual(buffer));
+            }
         }
 
 

cs/remote/src/FASTER.common/ElasticCircularBuffer.cs

@@ -5,6 +5,7 @@
 using System.Collections.Generic;
 using System.Runtime.CompilerServices;
 using System.Runtime.Serialization;
+using System.Threading;
 
 namespace FASTER.common
 {
@@ -92,7 +93,8 @@ public sealed class ElasticCircularBuffer<T> : IEnumerable<T>
         private readonly LinkedList<CircularBuffer<T>> buffers;
         private LinkedListNode<CircularBuffer<T>> head;
         private LinkedListNode<CircularBuffer<T>> tail;
-
+        private int count;
+
         /// <summary>
         /// Constructor
         /// </summary>
@@ -103,6 +105,9 @@ public ElasticCircularBuffer()
             buffers.AddFirst(node);
             tail = head = node;
         }
+
+        public int ApproxCount => count;
+
 
         /// <summary>
         /// Enqueue
@@ -111,6 +116,8 @@ public ElasticCircularBuffer()
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public void Enqueue(ref T value)
         {
+            Interlocked.Increment(ref count);
+
             if (tail.Value.IsFull())
             {
                 tail.Value.Sealed = true;
@@ -158,6 +165,7 @@ public T Dequeue()
                 if (head == null) head = buffers.First;
                 temp.Value.Sealed = false;
             }
+            Interlocked.Decrement(ref count);
             return head.Value.Dequeue();
         }
 

cs/remote/src/FASTER.common/HeaderReaderWriter.cs

@@ -34,7 +34,22 @@ public unsafe int ReadPendingSeqNo(ref byte* dst)
         public unsafe bool Write(MessageType s, ref byte* dst, int length)
         {
             if (length < 1) return false;
-            *dst++ = (byte)s;
+            *dst++ = (byte) s;
+            return true;
+        }
+
+        /// <summary>
+        /// Write message type to memory
+        /// </summary>
+        /// <param name="s">Message type</param>
+        /// <param name="dst">Destination memory</param>
+        /// <param name="length">Length of destination</param>
+        /// <returns>Whether write succeeded</returns>
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        public unsafe bool Write(byte s, ref byte* dst, int length)
+        {
+            if (length < 1) return false;
+            *dst++ = s;
             return true;
         }
 

cs/remote/src/FASTER.common/INetworkSender.cs

@@ -54,12 +54,7 @@ public interface INetworkSender : IDisposable
         /// <summary>
         /// Send response (caller owns buffer space)
         /// </summary>
-        void SendResponse(byte[] buffer, int offset, int count, object context);
-
-        /// <summary>
-        /// Send response (caller owns buffer space)
-        /// </summary>
-        void SendCallback(object context);
+        void SendResponse(byte[] buffer, int offset, int count, Action sendCallback);
 
         /// <summary>
         /// Dispose, optionally waiting for ongoing outgoing calls to complete

cs/remote/src/FASTER.common/NetworkSenderBase.cs

@@ -1,4 +1,6 @@
-namespace FASTER.common
+using System;
+
+namespace FASTER.common
 {
     /// <summary>
     /// NetworkSenderBase class 
@@ -57,12 +59,12 @@ public NetworkSenderBase(int serverBufferSize)
 
         /// <inheritdoc />
         public abstract bool SendResponse(int offset, int size);
-
+        
         /// <inheritdoc />
         public abstract void SendResponse(byte[] buffer, int offset, int count, object context);
 
         /// <inheritdoc />
-        public abstract void SendCallback(object context);
+        public abstract void SendResponse(byte[] buffer, int offset, int count, Action sendCallback);
 
         /// <inheritdoc />
         public abstract void Dispose();

cs/remote/src/FASTER.common/SimpleObjectPool.cs

@@ -2,7 +2,6 @@
 // Licensed under the MIT license.
 
 using System;
-using System.Diagnostics;
 using System.Runtime.CompilerServices;
 using System.Threading;
 
@@ -12,20 +11,29 @@ namespace FASTER.common
     /// Object pool
     /// </summary>
     /// <typeparam name="T"></typeparam>
-    public class SimpleObjectPool<T> : IDisposable where T : class, IDisposable
+    public class SimpleObjectPool<T> : IDisposable where T : class
     {
         private readonly Func<T> factory;
+        private readonly Action<T> destructor;
         private readonly LightConcurrentStack<T> stack;
         private int allocatedObjects;
 
         /// <summary>
         /// Constructor
         /// </summary>
-        /// <param name="factory"></param>
-        /// <param name="maxObjects"></param>
-        public SimpleObjectPool(Func<T> factory, int maxObjects = 128)
+        /// <param name="factory"> method used to create new objects of type T </param>
+        /// <param name="maxObjects">
+        /// Max number of objects that will be retained and recycled in this object pool.
+        /// Objects exceeding this count are created and destroyed on demand
+        /// </param>
+        /// <param name="destructor"> method used to dispose retained objects when they go out of scope.
+        /// WARNING: NOT invoked on retained objects before reuse
+        /// </param>
+
+        public SimpleObjectPool(Func<T> factory, int maxObjects = 128, Action<T> destructor = null)
         {
             this.factory = factory;
+            this.destructor = destructor;
             stack = new LightConcurrentStack<T>(maxObjects);
             allocatedObjects = 0;
         }
@@ -39,7 +47,7 @@ public void Dispose()
             {
                 while (stack.TryPop(out var elem))
                 {
-                    elem.Dispose();
+                    destructor?.Invoke(elem);
                     Interlocked.Decrement(ref allocatedObjects);
                 }
                 Thread.Yield();
@@ -69,7 +77,7 @@ public void Return(T obj)
         {
             if (!stack.TryPush(obj))
             {
-                obj.Dispose();
+                destructor?.Invoke(obj);
                 Interlocked.Decrement(ref allocatedObjects);
             }
         }

cs/remote/src/FASTER.common/TcpNetworkSender.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT license.
 
+using System;
 using System.Net;
 using System.Net.Sockets;
 using System.Runtime.CompilerServices;
@@ -22,6 +23,9 @@ public class TcpNetworkSender : NetworkSenderBase
         /// Response object
         /// </summary>
         protected SeaaBuffer responseObject;
+
+        // For use when user invokes send variant with user-owned buffers
+        private SimpleObjectPool<SocketAsyncEventArgs> saeaPool = new(() => new SocketAsyncEventArgs());
 
         /// <summary>
         /// Reusable SeaaBuffer
@@ -58,7 +62,8 @@ public TcpNetworkSender(
             : base(maxSizeSettings)
         {
             this.socket = socket;
-            this.reusableSeaaBuffer = new SimpleObjectPool<SeaaBuffer>(() => new SeaaBuffer(SeaaBuffer_Completed, this.serverBufferSize));
+            this.reusableSeaaBuffer = new SimpleObjectPool<SeaaBuffer>(() => new SeaaBuffer(SeaaBuffer_Completed, 
+                this.serverBufferSize), 128, s => s.Dispose());
             this.responseObject = null;
             this.ThrottleMax = throttleMax;
 
@@ -82,7 +87,8 @@ public TcpNetworkSender(
             : base(serverBufferSize)
         {
             this.socket = socket;
-            this.reusableSeaaBuffer = new SimpleObjectPool<SeaaBuffer>(() => new SeaaBuffer(SeaaBuffer_Completed, this.serverBufferSize));
+            this.reusableSeaaBuffer = new SimpleObjectPool<SeaaBuffer>(() => new SeaaBuffer(SeaaBuffer_Completed, 
+                this.serverBufferSize), 128, s => s.Dispose());
             this.responseObject = null;
             this.ThrottleMax = throttleMax;
 
@@ -154,15 +160,32 @@ public override bool SendResponse(int offset, int size)
         }
 
         /// <inheritdoc />
-        public override void SendResponse(byte[] buffer, int offset, int count, object context)
+        public override void SendResponse(byte[] buffer, int offset, int count, Action sendCallback)
         {
-            throw new System.NotImplementedException();
+            var saea = saeaPool.Checkout();
+            saea.SetBuffer(buffer, offset, count);
+            saea.UserToken = sendCallback;
+            saea.Completed += SaeaBuffer_Completed;
+
+            if (Interlocked.Increment(ref throttleCount) > ThrottleMax)
+                throttle.Wait();
+            if (!socket.SendAsync(saea))
+                SaeaBuffer_Completed(null, saea);        
+        }
+
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
+        private void SaeaBuffer_Completed(object sender, SocketAsyncEventArgs e)
+        {
+            ((Action)e.UserToken)();
+            saeaPool.Return(e);
+            if (Interlocked.Decrement(ref throttleCount) >= ThrottleMax)
+                throttle.Release();
         }
 
         /// <inheritdoc />
-        public override void SendCallback(object context)
+        public override void SendResponse(byte[] buffer, int offset, int count, object context)
         {
-            throw new System.NotImplementedException();
+            throw new NotImplementedException();
         }
 
         /// <inheritdoc />