fix(parser): Raise parser limit for array length from 8K to 64K (#158)

Add indication of parser errors to info stats.

Signed-off-by: Roman Gershman <roman@dragonflydb.io>

src/facade/dragonfly_connection.cc

@@ -228,10 +228,13 @@ void Connection::HandleRequests() {
       if (breaker_cb_) {
         should_disarm_poller = true;
 
-        poll_id = us->PollEvent(POLLERR | POLLHUP, [&](uint32_t mask) {
-          VLOG(1) << "Got event " << mask;
+        poll_id = us->PollEvent(POLLERR | POLLHUP, [&](int32_t mask) {
           cc_->conn_closing = true;
-          breaker_cb_(mask);
+          if (mask > 0) {
+            VLOG(1) << "Got event " << mask;
+            breaker_cb_(mask);
+          }
+
           evc_.notify();  // Notify dispatch fiber.
           should_disarm_poller = false;
         });
@@ -372,10 +375,12 @@ void Connection::ConnectionFlow(FiberSocketBase* peer) {
   // We wait for dispatch_fb to finish writing the previous replies before replying to the last
   // offending request.
   if (parse_status == ERROR) {
-    VLOG(1) << "Error parser status " << parse_status;
+    VLOG(1) << "Error parser status " << parser_error_;
+    ++stats->parser_err_cnt;
 
     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));
@@ -420,6 +425,8 @@ auto Connection::ParseRedis() -> ParserStatus {
         service_->DispatchCommand(cmd_list, cc_.get());
         last_interaction_ = time(nullptr);
       } else {
+        VLOG(2) << "Dispatch async";
+
         // Dispatch via queue to speedup input reading.
         Request* req = FromArgs(std::move(parse_args_), tlh);
 

src/facade/facade.cc

@@ -21,7 +21,7 @@ constexpr size_t kSizeConnStats = sizeof(ConnectionStats);
 
 ConnectionStats& ConnectionStats::operator+=(const ConnectionStats& o) {
   // To break this code deliberately if we add/remove a field to this struct.
-  static_assert(kSizeConnStats == 160);
+  static_assert(kSizeConnStats == 168);
 
   ADD(read_buf_capacity);
   ADD(io_read_cnt);
@@ -30,6 +30,7 @@ ConnectionStats& ConnectionStats::operator+=(const ConnectionStats& o) {
   ADD(io_write_bytes);
   ADD(command_cnt);
   ADD(pipelined_cmd_cnt);
+  ADD(parser_err_cnt);
   ADD(async_writes_cnt);
 
   ADD(num_conns);

src/facade/facade_types.h

@@ -35,6 +35,7 @@ struct ConnectionStats {
   size_t io_write_bytes = 0;
   size_t command_cnt = 0;
   size_t pipelined_cmd_cnt = 0;
+  size_t parser_err_cnt = 0;
 
   // Writes count that happenned via SendRawMessageAsync call.
   size_t async_writes_cnt = 0;

src/facade/redis_parser.cc

@@ -13,7 +13,7 @@ using namespace std;
 
 namespace {
 
-constexpr int kMaxArrayLen = 8192;
+constexpr int kMaxArrayLen = 65536;
 constexpr int64_t kMaxBulkLen = 64 * (1ul << 20);  // 64MB.
 
 }  // namespace

src/server/server_family.cc

@@ -862,6 +862,7 @@ void ServerFamily::Info(CmdArgList args, ConnectionContext* cntx) {
     append("total_reads_processed", m.conn_stats.io_read_cnt);
     append("total_writes_processed", m.conn_stats.io_write_cnt);
     append("async_writes_count", m.conn_stats.async_writes_cnt);
+    append("parser_err_count", m.conn_stats.parser_err_cnt);
   }
 
   if (should_enter("TIERED", true)) {