chore(server): Improve the implementation of SendStringArr.

src/facade/dragonfly_connection.cc

@@ -46,9 +46,9 @@ void SendProtocolError(RedisParser::Result pres, FiberSocketBase* peer) {
     res.append("invalid multibulk length\r\n");
   }
 
-  auto size_res = peer->Send(::io::Buffer(res));
-  if (!size_res) {
-    LOG(WARNING) << "Error " << size_res.error();
+  error_code ec = peer->Write(::io::Buffer(res));
+  if (ec) {
+    LOG(WARNING) << "Error " << ec;
   }
 }
 
@@ -344,10 +344,10 @@ void Connection::ConnectionFlow(FiberSocketBase* peer) {
     }
   }
 
-  error_code ec;
+  error_code ec = cc_->reply_builder()->GetError();
 
   // Main loop.
-  if (parse_status != ERROR) {
+  if (parse_status != ERROR && !ec) {
     auto res = IoLoop(peer);
 
     if (holds_alternative<error_code>(res)) {
@@ -380,15 +380,15 @@ void Connection::ConnectionFlow(FiberSocketBase* peer) {
 
     if (redis_parser_) {
       SendProtocolError(RedisParser::Result(parser_error_), peer);
-      peer->Shutdown(SHUT_RDWR);
     } else {
       string_view sv{"CLIENT_ERROR bad command line format\r\n"};
-      auto size_res = peer->Send(::io::Buffer(sv));
-      if (!size_res) {
-        LOG(WARNING) << "Error " << size_res.error();
-        ec = size_res.error();
+      error_code ec2 = peer->Write(::io::Buffer(sv));
+      if (ec2) {
+        LOG(WARNING) << "Error " << ec2;
+        ec = ec;
       }
     }
+    peer->Shutdown(SHUT_RDWR);
   }
 
   if (ec && !FiberSocketBase::IsConnClosed(ec)) {
@@ -532,9 +532,9 @@ auto Connection::IoLoop(util::FiberSocketBase* peer) -> variant<error_code, Pars
     ++stats->io_read_cnt;
     SetPhase("process");
 
-    if (redis_parser_)
+    if (redis_parser_) {
       parse_status = ParseRedis();
-    else {
+    } else {
       DCHECK(memcache_parser_);
       parse_status = ParseMemcache();
     }

src/facade/reply_builder.cc

@@ -3,6 +3,7 @@
 //
 #include "facade/reply_builder.h"
 
+#include <absl/container/fixed_array.h>
 #include <absl/strings/numbers.h>
 #include <absl/strings/str_cat.h>
 #include <double-conversion/double-to-string.h>
@@ -89,13 +90,13 @@ void SinkReplyBuilder::SendRaw(std::string_view raw) {
 }
 
 void SinkReplyBuilder::SendRawVec(absl::Span<const std::string_view> msg_vec) {
-  iovec v[msg_vec.size()];
+  absl::FixedArray<iovec, 16> arr(msg_vec.size());
 
   for (unsigned i = 0; i < msg_vec.size(); ++i) {
-    v[i].iov_base = const_cast<char*>(msg_vec[i].data());
-    v[i].iov_len = msg_vec[i].size();
+    arr[i].iov_base = const_cast<char*>(msg_vec[i].data());
+    arr[i].iov_len = msg_vec[i].size();
   }
-  Send(v, msg_vec.size());
+  Send(arr.data(), msg_vec.size());
 }
 
 MCReplyBuilder::MCReplyBuilder(::io::Sink* sink) : SinkReplyBuilder(sink) {
@@ -305,14 +306,65 @@ void RedisReplyBuilder::SendStringArr(absl::Span<const std::string_view> arr) {
   SendRaw(res);
 }
 
+// This implementation a bit complicated because it uses vectorized
+// send to send an array. The problem with that is the OS limits vector length
+// to low numbers (around 1024). Therefore, to make it robust we send the array in batches.
+// We limit the vector length to 256 and when it fills up we flush it to the socket and continue
+// iterating.
 void RedisReplyBuilder::SendStringArr(absl::Span<const string> arr) {
-  string res = absl::StrCat("*", arr.size(), kCRLF);
+  // When vector length is too long, Send returns EMSGSIZE.
+  size_t vec_len = std::min<size_t>(256u, arr.size());
 
-  for (size_t i = 0; i < arr.size(); ++i) {
-    StrAppend(&res, "$", arr[i].size(), kCRLF);
-    res.append(arr[i]).append(kCRLF);
+  absl::FixedArray<iovec, 16> vec(vec_len * 2 + 2);
+  absl::FixedArray<char, 64> meta((vec_len + 1) * 16);
+  char* next = meta.data();
+
+  *next++ = '*';
+  next = absl::numbers_internal::FastIntToBuffer(arr.size(), next);
+  *next++ = '\r';
+  *next++ = '\n';
+  vec[0].iov_base = meta.data();
+  vec[0].iov_len = next - meta.data();
+  char* start = next;
+
+  unsigned vec_indx = 1;
+  for (unsigned i = 0; i < arr.size(); ++i) {
+    const auto& src = arr[i];
+    *next++ = '$';
+    next = absl::numbers_internal::FastIntToBuffer(src.size(), next);
+    *next++ = '\r';
+    *next++ = '\n';
+    vec[vec_indx].iov_base = start;
+    vec[vec_indx].iov_len = next - start;
+    DCHECK_GT(next - start, 0);
+
+    start = next;
+    ++vec_indx;
+
+    vec[vec_indx].iov_base = const_cast<char*>(src.data());
+    vec[vec_indx].iov_len = src.size();
+
+    *next++ = '\r';
+    *next++ = '\n';
+    ++vec_indx;
+
+    if (vec_indx + 1 >= vec.size()) {
+      if (i < arr.size() - 1 || vec_indx == vec.size()) {
+        Send(vec.data(), vec_indx);
+        if (ec_)
+          return;
+
+        vec_indx = 0;
+        start = meta.data();
+        next = start + 2;
+        start[0] = '\r';
+        start[1] = '\n';
+      }
+    }
   }
-  SendRaw(res);
+  vec[vec_indx].iov_base = start;
+  vec[vec_indx].iov_len = 2;
+  Send(vec.data(), vec_indx + 1);
 }
 
 void RedisReplyBuilder::StartArray(unsigned len) {

src/facade/reply_builder.h

@@ -7,7 +7,7 @@
 #include <string_view>
 
 #include "facade/op_status.h"
-#include "io/sync_stream_interface.h"
+#include "io/io.h"
 
 namespace facade {
 

src/redis/CMakeLists.txt

@@ -15,6 +15,8 @@ add_library(redis_lib crc64.c crcspeed.c debug.c dict.c intset.c
 
 cxx_link(redis_lib  ${ZMALLOC_DEPS})
 
+target_compile_options(redis_lib PRIVATE -Wno-maybe-uninitialized)
+
 if (REDIS_ZMALLOC_MI)
   target_compile_definitions(redis_lib PUBLIC USE_ZMALLOC_MI)
 endif()

src/server/hset_family.cc

@@ -290,10 +290,7 @@ void HSetFamily::HGetGeneric(CmdArgList args, ConnectionContext* cntx, uint8_t g
   OpResult<vector<string>> result = cntx->transaction->ScheduleSingleHopT(std::move(cb));
 
   if (result) {
-    (*cntx)->StartArray(result->size());
-    for (const auto& s : *result) {
-      (*cntx)->SendBulkString(s);
-    }
+    (*cntx)->SendStringArr(absl::Span<const string>{*result});
   } else {
     (*cntx)->SendError(result.status());
   }