chore: improve interface for redis parsing

eliminate cases where we return INPUT_PENDING but do not consume the whole string.
This should simplify buffer management for the caller, so that if they pass a string that
did not result in complete parsed request, at least the whole string is consumed and can be discarded.

Signed-off-by: Roman Gershman <[email protected]>

src/facade/redis_parser.cc

@@ -35,7 +35,7 @@ auto RedisParser::Parse(Buffer str, uint32_t* consumed, RespExpr::Vec* res) -> R
     if (str.empty()) {
       resultc.first = INPUT_PENDING;
       break;
-    }
+  }
 
     switch (state_) {
       case MAP_LEN_S:
@@ -212,19 +212,36 @@ auto RedisParser::ParseInline(Buffer str) -> ResultConsumed {
 auto RedisParser::ParseLen(Buffer str, int64_t* res) -> ResultConsumed {
   DCHECK(!str.empty());
 
-  char* s = reinterpret_cast<char*>(str.data() + 1);
-  char* pos = reinterpret_cast<char*>(memchr(s, '\n', str.size() - 1));
+  DCHECK(small_len_ > 0 || str[0] == '$' || str[0] == '*' || str[0] == '%' || str[0] == '~');
+
+  const char* s = reinterpret_cast<const char*>(str.data());
+  unsigned consumed = 0;
+  const char* pos = reinterpret_cast<const char*>(memchr(s, '\n', str.size()));
   if (!pos) {
-    Result r = str.size() < 32 ? INPUT_PENDING : BAD_ARRAYLEN;
-    return {r, 0};
+    if (str.size() + small_len_ < sizeof(small_buf_)) {
+      memcpy(small_buf_ + small_len_, str.data(), str.size());
+      small_len_ += str.size();
+      return {INPUT_PENDING, str.size()};
+    }
+    return ResultConsumed{BAD_ARRAYLEN, 0};
+  }
+
+  consumed = pos - s + 1;
+  if (small_len_ > 0) {
+    memcpy(small_buf_ + small_len_, str.data(), consumed);
+    small_len_ += consumed;
+    s = small_buf_;
+    pos = small_buf_ + small_len_ - 1;
+    small_len_ = 0;
   }
 
   if (pos[-1] != '\r') {
     return {BAD_ARRAYLEN, 0};
   }
 
-  bool success = absl::SimpleAtoi(std::string_view{s, size_t(pos - s - 1)}, res);
-  return ResultConsumed{success ? OK : BAD_ARRAYLEN, (pos - s) + 2};
+  // Skip the first character and 2 last ones (\r\n).
+  bool success = absl::SimpleAtoi(std::string_view{s + 1, size_t(pos - 1 - s)}, res);
+  return ResultConsumed{success ? OK : BAD_ARRAYLEN, consumed};
 }
 
 auto RedisParser::ConsumeArrayLen(Buffer str) -> ResultConsumed {
@@ -241,8 +258,8 @@ auto RedisParser::ConsumeArrayLen(Buffer str) -> ResultConsumed {
     len *= 2;
   }
 
-  if (len < -1 || len > max_arr_len_) {
-    LOG_IF(WARNING, len > max_arr_len_) << "Multibulk len is too large " << len;
+      if (len < -1 || len > max_arr_len_) {
+        LOG_IF(WARNING, len > max_arr_len_) << "Multibulk len is too large " << len;
 
     return {BAD_ARRAYLEN, res.second};
   }
@@ -287,7 +304,9 @@ auto RedisParser::ConsumeArrayLen(Buffer str) -> ResultConsumed {
 }
 
 auto RedisParser::ParseArg(Buffer str) -> ResultConsumed {
-  char c = str[0];
+  DCHECK(!str.empty());
+
+  char c = small_len_ > 0 ? small_buf_[0] : str[0];
 
   if (c == '$') {
     int64_t len;

src/facade/redis_parser.h

@@ -47,8 +47,6 @@ class RedisParser {
    * part of str because parser caches the intermediate state internally according to 'consumed'
    * result.
    *
-   * Note: A parser does not always guarantee progress, i.e. if a small buffer was passed it may
-   * returns INPUT_PENDING with consumed == 0.
    *
    */
 
@@ -98,8 +96,9 @@ class RedisParser {
   };
 
   State state_ = CMD_COMPLETE_S;
-  bool is_broken_token_ = false;  // whether the last inline string was broken in the middle.
+  bool is_broken_token_ = false;
   bool server_mode_ = true;
+  uint8_t small_len_ = 0;
 
   uint32_t bulk_len_ = 0;
   uint32_t last_stashed_level_ = 0, last_stashed_index_ = 0;
@@ -115,6 +114,7 @@ class RedisParser {
 
   using Blob = std::vector<uint8_t>;
   std::vector<Blob> buf_stash_;
+  char small_buf_[32];
 };
 
 }  // namespace facade

src/facade/redis_parser_test.cc

@@ -230,4 +230,11 @@ TEST_F(RedisParserTest, UsedMemory) {
   EXPECT_GT(dfly::HeapSize(stash), 30000);
 }
 
+TEST_F(RedisParserTest, Eol) {
+  ASSERT_EQ(RedisParser::INPUT_PENDING, Parse("*1\r"));
+  EXPECT_EQ(3, consumed_);
+  ASSERT_EQ(RedisParser::INPUT_PENDING, Parse("\n$5\r\n"));
+  EXPECT_EQ(5, consumed_);
+}
+
 }  // namespace facade