Fix the long-blocking read for Valkey RDMA.

During the valkey benchmark, when we set the data size as 16KB, the
benchmark will be blocking on GET. The reason behind is that RDMA
event is edge triggered and every incoming data has to be read totally
instead of read partially. We modify the valkeyBufferRead to enable a
total read over the RDMA buffer when the conneciton type is RDMA. To
realize this logic we need to keep read the buffer until the read
result data size is smaller than the attempt read size.

In addition, we need to deal with a corner case for the logic above.
If the message happened to be 16KB and the first read finish all the
available data, the second read will be triggered and block there until
return an error. To solve this problem, we make read non-blocking
except for the first read.

Signed-off-by: Ruihong Wang <ruihong@google.com>

Author: ruihong123

src/rdma.c

@@ -511,7 +511,12 @@ static ssize_t valkeyRdmaRead(valkeyContext *c, char *buf, size_t bufcap) {
         return VALKEY_ERR;
     }
 
-    end = vk_msec_now() + timed;
+    if (timed > 0) {
+        end = vk_msec_now() + timed;
+    } else {
+        // end = -1 marks that we want a non-blocking read.
+        end = -1;
+    }
 
 pollcq:
     /* try to poll a CQ first */
@@ -531,6 +536,9 @@ static ssize_t valkeyRdmaRead(valkeyContext *c, char *buf, size_t bufcap) {
         }
 
         return toread;
+    } else if (ctx->recv_offset == ctx->rx_offset && end == -1) {
+        // non-blocking read over an empty buffer shall directly return here.
+        return 0;
     }
 
     if (valkeyRdmaPollCqCm(c, end) == VALKEY_OK) {

src/valkey.c

@@ -1009,13 +1009,41 @@ int valkeyBufferRead(valkeyContext *c) {
     if (c->err)
         return VALKEY_ERR;
 
-    nread = c->funcs->read(c, buf, sizeof(buf));
-    if (nread < 0) {
-        return VALKEY_ERR;
-    }
-    if (nread > 0 && valkeyReaderFeed(c->reader, buf, nread) != VALKEY_OK) {
-        valkeySetError(c, c->reader->err, c->reader->errstr);
-        return VALKEY_ERR;
+    if (c->connection_type != VALKEY_CONN_RDMA) {
+        nread = c->funcs->read(c, buf, sizeof(buf));
+        if (nread < 0) {
+            return VALKEY_ERR;
+        }
+        if (nread > 0 && valkeyReaderFeed(c->reader, buf, nread) != VALKEY_OK) {
+            valkeySetError(c, c->reader->err, c->reader->errstr);
+            return VALKEY_ERR;
+        }
+    } else {
+        /* For Valkey over RDMA, as the ready-to-read signal only exists per arrived RDMA write.
+         * This funciton has to copy all the data from communication buffer to client buffer.*/
+        struct timeval *recorded_timeout = c->command_timeout;
+        struct timeval zero_timeout;
+        while (1) {
+            nread = c->funcs->read(c, buf, sizeof(buf));
+            if (nread < 0) {
+                return VALKEY_ERR;
+            }
+            if (nread > 0 && valkeyReaderFeed(c->reader, buf, nread) != VALKEY_OK) {
+                valkeySetError(c, c->reader->err, c->reader->errstr);
+                return VALKEY_ERR;
+            }
+            if ((uint32_t)nread < sizeof(buf))
+                break;
+            /* Only the first read is required to be blocking, the rest of reads should be non-blocking
+             * and keep read until there is no data in the communicaiton buffer.*/
+            zero_timeout.tv_sec = 0;
+            zero_timeout.tv_usec = 0;
+            c->command_timeout = &zero_timeout;
+        }
+        /*Recover the timeout back*/
+        if (c->command_timeout != recorded_timeout) {
+            c->command_timeout = recorded_timeout;
+        }
     }
     return VALKEY_OK;
 }