Update RESP3 output (#1170)

* Pass respProtocolVersion in GarnetObjectStoreOutput

* Use resp output struct to simplify code and emit correct RESP3 null.

* Make more places able to pass the Resp3 flag.

* SortedSet example. TODO: convert to use input flags.

* Convert to use RespInputFlag.

* Convert Set and (nost of) List operations to output struct.
Add resp3 flags to Scan operations.

* Update comment.
Emit the right null in more places.

* Set and pass respProtocolVersion more consistently.

* Allow passing respProtocolVersion to storageSession and use it in cases where the input isn't passed to the API.

* In practice, the protocol version doesn't matter for string store - it doesn't have Operate(). We can use that to restore SetGet flag to old value and keep AOF format.

* dotnet format

* By using separate flags for RESP3 object and RESP3 string, we can keep object invariant while still marking RESP3 for RawStringInput.

* More cases where Garnet should emit RESP3 NULL.

* Info returns RESP_EMPTY, not nill on redis on equivalent case.
Async GET should be able to return RESP3 null too.

* Add RESP3 null checks into MainStore too.

* Add test

* These sortedset functions should emit double under RESP3 for score

* Update LCS command output.

* Use functionstate instead of input flags.

* Use the functionstate constructor.

* Update more commands to use RESP3 map.

* More updates for RESP3 output.
More tests.

* Extract out RespMemoryWriter. This functionality is useful beyond the server.

* Empty commit to try running the actions again.

* Convert ListPosition. Fix return value when NOTFOUND and COUNT is used. Add test.
Move all use of Reallocate to RespMemoryWriter
Remove optimal line count ;-(

* Fix test.

* Fmt

* Rename parameters.

* Add methods to RespMemoryWriter to allow use in ToRespFormat().

* Convert COMMAND to RespMemoryWriter output, add RESP3 output.

* This is simpler

* Simplify by having the caller allocate

* fmt

* Also convert keyspecs.

* Minor optimizations.

* Move header to GarnetObjectStoreOutput and start untangling code.

* Convert rest of object functions to use header and result1 directly. Then remove GarnetObjectStoreRespOutput and convert all uses to RespMemoryWriter.

* More succinct code around functionsState.

* Fmt

* Remove unnecessary unsafes. Generalizes ObjectUtils.Scan() operation to allow this.

* Pass extra parameter for HSCAN.

* Avoid stackalloc and just write directly.

* No need for these public members in RespMemoryWriter
Convert GEOSEARCHSTORE implementation to use RespMemoryWriter.
(All sites calling TryReadErrorAsString already checked for '-' character, it's fine to return false there)

* No need for ObjectUtils, we can move remaining code into GarnetObjectBase.
Allow long cursor for object SCAN commands.

* Adjust custom path to supply cursor via parseState and not arg1, so we can have a 64bit argument there.

* Ignore case in some *SCAN option comparisons.

* Scoping the argument lets us undo the change the respreadutils.

* Undo the comment change too.

* A few more RESP3 updates

* Minor push changes.

* These null arrays should emit RESP3 nulls.
Improve RespMemoryWriter.

* This arrangement is better due to padding.

* Use consistent naming (writer for RespMemoryWriter, output for GarnetObjectStoreOutput).

* Switch to using instead of try/finally. No functional change.

* Add constructors to simplify GarnetObjectStoreOutput initialization.

* Revert "Add constructors to simplify GarnetObjectStoreOutput initialization."

This reverts commit f60d08d.

* Add constructors to simplify GarnetObjectStoreOutput initialization.

* Try to make a little faster by skipping locals init.

* Revert "Try to make a little faster by skipping locals init."

This reverts commit 8139e67.

* Try to make a little faster by skipping locals init.

* No need for static reallocate anymore.

* Address review comment

Author: prvyk

libs/client/ClientSession/GarnetClientSession.cs

@@ -63,6 +63,11 @@ public sealed partial class GarnetClientSession : IServerHook, IMessageConsumer
         /// </summary>
         public bool IsConnected => socket != null && socket.Connected && !Disposed;
 
+        /// <summary>
+        /// Get raw results from tcs completion
+        /// </summary>
+        public bool RawResult = false;
+
         /// <summary>
         /// Username to authenticate the session on the server.
         /// </summary>
@@ -107,6 +112,7 @@ public GarnetClientSession(
             string authUsername = null,
             string authPassword = null,
             int networkSendThrottleMax = 8,
+            bool rawResult = false,
             ILogger logger = null)
         {
             EndPoint = endpoint;
@@ -122,6 +128,7 @@ public GarnetClientSession(
             this.disposed = 0;
             this.authUsername = authUsername;
             this.authPassword = authPassword;
+            this.RawResult = rawResult;
         }
 
         /// <summary>
@@ -459,36 +466,51 @@ private unsafe int ProcessReplies(byte* recvBufferPtr, int bytesRead)
 
             while (readHead < bytesRead)
             {
-                switch (*ptr)
+                if (RawResult)
                 {
-                    case (byte)'+':
-                        if (!RespReadResponseUtils.TryReadSimpleString(out result, ref ptr, recvBufferPtr + bytesRead))
-                            success = false;
-                        break;
-                    case (byte)':':
-                        if (!RespReadResponseUtils.TryReadIntegerAsString(out result, ref ptr, recvBufferPtr + bytesRead))
-                            success = false;
-                        break;
-
-                    case (byte)'-':
-                        error = true;
-                        if (!RespReadResponseUtils.TryReadErrorAsString(out result, ref ptr, recvBufferPtr + bytesRead))
-                            success = false;
-                        break;
-
-                    case (byte)'$':
-                        if (!RespReadResponseUtils.TryReadStringWithLengthHeader(out result, ref ptr, recvBufferPtr + bytesRead))
-                            success = false;
-                        break;
-
-                    case (byte)'*':
-                        isArray = true;
-                        if (!RespReadResponseUtils.TryReadStringArrayWithLengthHeader(out resultArray, ref ptr, recvBufferPtr + bytesRead))
-                            success = false;
-                        break;
-
-                    default:
-                        throw new Exception("Unexpected response: " + Encoding.UTF8.GetString(new Span<byte>(recvBufferPtr, bytesRead)).Replace("\n", "|").Replace("\r", "") + "]");
+                    if (RespReadUtils.TryReadString(out var tmp, ref ptr, recvBufferPtr + bytesRead))
+                    {
+                        result += tmp;
+                        result += "\r\n";
+                    }
+                    else
+                    {
+                        success = false;
+                    }
+                }
+                else
+                {
+                    switch (*ptr)
+                    {
+                        case (byte)'+':
+                            if (!RespReadResponseUtils.TryReadSimpleString(out result, ref ptr, recvBufferPtr + bytesRead))
+                                success = false;
+                            break;
+                        case (byte)':':
+                            if (!RespReadResponseUtils.TryReadIntegerAsString(out result, ref ptr, recvBufferPtr + bytesRead))
+                                success = false;
+                            break;
+
+                        case (byte)'-':
+                            error = true;
+                            if (!RespReadResponseUtils.TryReadErrorAsString(out result, ref ptr, recvBufferPtr + bytesRead))
+                                success = false;
+                            break;
+
+                        case (byte)'$':
+                            if (!RespReadResponseUtils.TryReadStringWithLengthHeader(out result, ref ptr, recvBufferPtr + bytesRead))
+                                success = false;
+                            break;
+
+                        case (byte)'*':
+                            isArray = true;
+                            if (!RespReadResponseUtils.TryReadStringArrayWithLengthHeader(out resultArray, ref ptr, recvBufferPtr + bytesRead))
+                                success = false;
+                            break;
+
+                        default:
+                            throw new Exception("Unexpected response: " + Encoding.UTF8.GetString(new Span<byte>(recvBufferPtr, bytesRead)).Replace("\n", "|").Replace("\r", "") + "]");
+                    }
                 }
 
                 if (!success) return readHead;
@@ -500,13 +522,21 @@ private unsafe int ProcessReplies(byte* recvBufferPtr, int bytesRead)
                     var tcs = tcsArrayQueue.Dequeue();
                     tcs?.SetResult(resultArray);
                 }
-                else
+                else if (!RawResult)
                 {
                     var tcs = tcsQueue.Dequeue();
                     if (error) tcs?.SetException(new Exception(result));
                     else tcs?.SetResult(result);
                 }
             }
+
+            if (RawResult)
+            {
+                var tcs = tcsQueue.Dequeue();
+                if (error) tcs?.SetException(new Exception(result));
+                else tcs?.SetResult(result);
+            }
+
             return readHead;
         }