Support async RESP responses to pending GET calls (due to disk read)

libs/common/Networking/GarnetTcpNetworkSender.cs

@@ -50,6 +50,11 @@ public class GarnetTcpNetworkSender : NetworkSenderBase
 
         readonly LimitedFixedBufferPool networkPool;
 
+        /// <summary>
+        /// NOTE: This variable should not be marked as readonly as it is a mutable struct
+        /// </summary>
+        SpinLock spinLock;
+
         /// <summary>
         /// 
         /// </summary>
@@ -67,6 +72,7 @@ public GarnetTcpNetworkSender(
             this.saeaStack = new(2 * ThrottleMax);
             this.responseObject = null;
             this.ThrottleMax = throttleMax;
+            this.spinLock = new();
 
             var endpoint = socket.RemoteEndPoint as IPEndPoint;
             if (endpoint != null)
@@ -79,6 +85,49 @@ public GarnetTcpNetworkSender(
         /// <inheritdoc />
         public override string RemoteEndpointName => remoteEndpoint;
 
+        /// <inheritdoc />
+        public override void Enter()
+        {
+            var lockTaken = false;
+            spinLock.Enter(ref lockTaken);
+            Debug.Assert(lockTaken);
+        }
+
+        /// <inheritdoc />
+        public override unsafe void EnterAndGetResponseObject(out byte* head, out byte* tail)
+        {
+            var lockTaken = false;
+            spinLock.Enter(ref lockTaken);
+            Debug.Assert(lockTaken);
+            Debug.Assert(responseObject == null);
+            if (!saeaStack.TryPop(out responseObject, out bool disposed))
+            {
+                if (disposed)
+                    ThrowDisposed();
+                responseObject = new GarnetSaeaBuffer(SeaaBuffer_Completed, networkPool);
+            }
+            head = responseObject.buffer.entryPtr;
+            tail = responseObject.buffer.entryPtr + responseObject.buffer.entry.Length;
+        }
+
+        /// <inheritdoc />
+        public override void Exit()
+        {
+            spinLock.Exit();
+        }
+
+        /// <inheritdoc />
+        public override unsafe void ExitAndReturnResponseObject()
+        {
+            if (responseObject != null)
+            {
+                ReturnBuffer(responseObject);
+                responseObject = null;
+            }
+            spinLock.Exit();
+        }
+
+
         /// <inheritdoc />
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         public override void GetResponseObject()

libs/common/Networking/INetworkSender.cs

@@ -21,6 +21,28 @@ public interface INetworkSender : IDisposable
         /// </summary>
         string RemoteEndpointName { get; }
 
+        /// <summary>
+        /// Enter exclusive use of network sender.
+        /// </summary>
+        void Enter();
+
+        /// <summary>
+        /// Enter exclusive use of network sender. Allocate and get response object pointers.
+        /// </summary>
+        /// <param name="head"></param>
+        /// <param name="tail"></param>
+        unsafe void EnterAndGetResponseObject(out byte* head, out byte* tail);
+
+        /// <summary>
+        /// Exit exclusive use of network sender.
+        /// </summary>
+        void Exit();
+
+        /// <summary>
+        /// Exit exclusive use of network sender. Free response object.
+        /// </summary>
+        void ExitAndReturnResponseObject();
+
         /// <summary>
         /// Allocate a new response object
         /// </summary>

libs/common/Networking/NetworkHandler.cs

@@ -530,6 +530,28 @@ unsafe void ShiftTransportReceiveBuffer()
             }
         }
 
+        /// <inheritdoc />
+        public override void Enter()
+            => networkSender.Enter();
+
+        /// <inheritdoc />
+        public override unsafe void EnterAndGetResponseObject(out byte* head, out byte* tail)
+        {
+            networkSender.Enter();
+            head = transportSendBufferPtr;
+            tail = transportSendBufferPtr + transportSendBuffer.Length;
+        }
+
+        /// <inheritdoc />
+        public override void Exit()
+            => networkSender.Exit();
+
+        /// <inheritdoc />
+        public override void ExitAndReturnResponseObject()
+        {
+            networkSender.Exit();
+        }
+
         /// <inheritdoc />
         public override void GetResponseObject() { }
 

libs/common/Networking/NetworkSenderBase.cs

@@ -46,6 +46,18 @@ public NetworkSenderBase(int serverBufferSize)
         /// <inheritdoc />
         public abstract string RemoteEndpointName { get; }
 
+        /// <inheritdoc />
+        public abstract void Enter();
+
+        /// <inheritdoc />
+        public abstract unsafe void EnterAndGetResponseObject(out byte* head, out byte* tail);
+
+        /// <inheritdoc />
+        public abstract void Exit();
+
+        /// <inheritdoc />
+        public abstract void ExitAndReturnResponseObject();
+
         /// <inheritdoc />
         public abstract void GetResponseObject();
 

libs/common/RespReadUtils.cs

@@ -450,6 +450,31 @@ public static bool ReadULongWithLengthHeader(out ulong number, ref byte* ptr, by
             return true;
         }
 
+        /// <summary>
+        /// Skip byte array with length header
+        /// </summary>
+        public static bool SkipByteArrayWithLengthHeader(ref byte* ptr, byte* end)
+        {
+            // Parse RESP string header
+            if (!ReadLengthHeader(out var length, ref ptr, end))
+                return false;
+
+            // Advance read pointer to the end of the array (including terminator)
+            var keyPtr = ptr;
+
+            ptr += length + 2;
+
+            if (ptr > end)
+                return false;
+
+            // Ensure terminator has been received
+            if (*(ushort*)(ptr - 2) != MemoryMarshal.Read<ushort>("\r\n"u8))
+            {
+                RespParsingException.ThrowUnexpectedToken(*(ptr - 2));
+            }
+            return true;
+        }
+
         /// <summary>
         /// Read byte array with length header
         /// </summary>

libs/common/RespWriteUtils.cs

@@ -29,6 +29,21 @@ public static bool WriteMapLength(int len, ref byte* curr, byte* end)
             return true;
         }
 
+        /// <summary>
+        /// Write push type length
+        /// </summary>
+        public static bool WritePushLength(int len, ref byte* curr, byte* end)
+        {
+            int numDigits = NumUtils.NumDigits(len);
+            int totalLen = 1 + numDigits + 2;
+            if (totalLen > (int)(end - curr))
+                return false;
+            *curr++ = (byte)'>';
+            NumUtils.IntToBytes(len, numDigits, ref curr);
+            WriteNewline(ref curr);
+            return true;
+        }
+
         /// <summary>
         /// Write array length
         /// </summary>

libs/host/GarnetServer.cs

@@ -47,7 +47,7 @@ public class GarnetServer : IDisposable
         /// <summary>
         /// Resp protocol version
         /// </summary>
-        readonly string redisProtocolVersion = "6.2.11";
+        readonly string redisProtocolVersion = "7.2.5";
 
         /// <summary>
         /// Metrics API

libs/server/API/GarnetApi.cs

@@ -53,6 +53,9 @@ public GarnetStatus GET_WithPending(ref SpanByte key, ref SpanByte input, ref Sp
         public bool GET_CompletePending((GarnetStatus, SpanByteAndMemory)[] outputArr, bool wait = false)
             => storageSession.GET_CompletePending(outputArr, wait, ref context);
 
+        public bool GET_CompletePending(out CompletedOutputIterator<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, long> completedOutputs, bool wait)
+            => storageSession.GET_CompletePending(out completedOutputs, wait, ref context);
+
         /// <inheritdoc />
         public unsafe GarnetStatus GETForMemoryResult(ArgSlice key, out MemoryResult<byte> value)
             => storageSession.GET(key, out value, ref context);

libs/server/API/IGarnetAdvancedApi.cs

@@ -22,6 +22,14 @@ public interface IGarnetAdvancedApi
         /// <param name="wait"></param>
         bool GET_CompletePending((GarnetStatus, SpanByteAndMemory)[] outputArr, bool wait = false);
 
+        /// <summary>
+        /// Complete pending read operations on main store
+        /// </summary>
+        /// <param name="completedOutputs"></param>
+        /// <param name="wait"></param>
+        /// <returns></returns>
+        bool GET_CompletePending(out CompletedOutputIterator<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, long> completedOutputs, bool wait = false);
+
         /// <summary>
         /// RMW operation on main store
         /// </summary>

libs/server/InputHeader.cs

@@ -36,7 +36,7 @@ public struct RespInputHeader
         /// Size of header
         /// </summary>
         public const int Size = 2;
-        internal const byte FlagMask = (byte)RespInputFlags.Deterministic - 1;
+        internal const byte FlagMask = (byte)RespInputFlags.SetGet - 1;
 
         [FieldOffset(0)]
         internal RespCommand cmd;

libs/server/Resp/AdminCommands.cs

@@ -5,6 +5,7 @@
 using System.Diagnostics;
 using System.Linq;
 using System.Text;
+using System.Threading;
 using System.Threading.Tasks;
 using Garnet.common;
 using Garnet.server.ACL;
@@ -280,7 +281,7 @@ private bool ProcessAdminCommands<TGarnetApi>(RespCommand command, ReadOnlySpan<
             }
             else if (command == RespCommand.HELLO)
             {
-                int? respProtocolVersion = null;
+                byte? respProtocolVersion = null;
                 ReadOnlySpan<byte> authUsername = default, authPassword = default;
                 string clientName = null;
 
@@ -292,7 +293,7 @@ private bool ProcessAdminCommands<TGarnetApi>(RespCommand command, ReadOnlySpan<
                         return false;
                     readHead = (int)(ptr - recvBufferPtr);
 
-                    respProtocolVersion = localRespProtocolVersion;
+                    respProtocolVersion = (byte)localRespProtocolVersion;
                     count--;
                     while (count > 0)
                     {
@@ -542,13 +543,77 @@ private bool ProcessAdminCommands<TGarnetApi>(RespCommand command, ReadOnlySpan<
             {
                 return ProcessACLCommands(bufSpan, count);
             }
-            else if ((command == RespCommand.REGISTERCS))
+            else if (command == RespCommand.REGISTERCS)
             {
                 return NetworkREGISTERCS(count, recvBufferPtr + readHead, storeWrapper.customCommandManager);
             }
+            else if (command == RespCommand.ASYNC)
+            {
+                if (respProtocolVersion > 2 && count == 1)
+                {
+                    var param = GetCommand(bufSpan, out bool success1);
+                    if (!success1) return false;
+                    if (param.SequenceEqual(CmdStrings.ON) || param.SequenceEqual(CmdStrings.on))
+                    {
+                        useAsync = true;
+                        while (!RespWriteUtils.WriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
+                            SendAndReset();
+                    }
+                    else if (param.SequenceEqual(CmdStrings.OFF) || param.SequenceEqual(CmdStrings.off))
+                    {
+                        useAsync = false;
+                        while (!RespWriteUtils.WriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
+                            SendAndReset();
+                    }
+                    else if (param.SequenceEqual(CmdStrings.BARRIER) || param.SequenceEqual(CmdStrings.barrier))
+                    {
+                        if (asyncCompleted < asyncStarted)
+                        {
+                            asyncDone = new(0);
+                            if (dcurr > networkSender.GetResponseObjectHead())
+                                Send(networkSender.GetResponseObjectHead());
+                            try
+                            {
+                                networkSender.ExitAndReturnResponseObject();
+                                while (asyncCompleted < asyncStarted) asyncDone.Wait();
+                                asyncDone.Dispose();
+                                asyncDone = null;
+                            }
+                            finally
+                            {
+                                networkSender.EnterAndGetResponseObject(out dcurr, out dend);
+                            }
+                        }
+
+                        while (!RespWriteUtils.WriteDirect(CmdStrings.RESP_OK, ref dcurr, dend))
+                            SendAndReset();
+                    }
+                    else
+                    {
+                        if (!DrainCommands(bufSpan, count - 1))
+                            return false;
+                        errorFlag = true;
+                        errorCmd = "ASYNC";
+                    }
+                }
+                else
+                {
+                    if (!DrainCommands(bufSpan, count))
+                        return false;
+                    if (respProtocolVersion <= 2)
+                    {
+                        while (!RespWriteUtils.WriteError(CmdStrings.RESP_ERR_NOT_SUPPORTED_RESP2, ref dcurr, dend))
+                            SendAndReset();
+                    }
+                    else
+                    {
+                        errorFlag = true;
+                        errorCmd = "ASYNC";
+                    }
+                }
+            }
             else
             {
-                // Unknown RESP Command
                 if (!DrainCommands(bufSpan, count))
                     return false;
                 while (!RespWriteUtils.WriteError(CmdStrings.RESP_ERR_GENERIC_UNK_CMD, ref dcurr, dend))
@@ -564,17 +629,27 @@ private bool ProcessAdminCommands<TGarnetApi>(RespCommand command, ReadOnlySpan<
             return true;
         }
 
-        void ProcessHelloCommand(int? respProtocolVersion, ReadOnlySpan<byte> username, ReadOnlySpan<byte> password, string clientName)
+        /// <summary>
+        /// Process the HELLO command
+        /// </summary>
+        void ProcessHelloCommand(byte? respProtocolVersion, ReadOnlySpan<byte> username, ReadOnlySpan<byte> password, string clientName)
         {
             if (respProtocolVersion != null)
             {
-                if (respProtocolVersion.Value != 2)
+                if (respProtocolVersion.Value is < 2 or > 3)
                 {
                     while (!RespWriteUtils.WriteError(CmdStrings.RESP_ERR_UNSUPPORTED_PROTOCOL_VERSION, ref dcurr, dend))
                         SendAndReset();
                     return;
                 }
 
+                if (respProtocolVersion.Value != this.respProtocolVersion && asyncCompleted < asyncStarted)
+                {
+                    while (!RespWriteUtils.WriteError(CmdStrings.RESP_ERR_ASYNC_PROTOCOL_CHANGE, ref dcurr, dend))
+                        SendAndReset();
+                    return;
+                }
+
                 this.respProtocolVersion = respProtocolVersion.Value;
             }