chore: introduce BackedArguments

In order to support async Dispatch we must use arguments allocated on heap.
This PR introduces a basic ParsedArgument class that will be used by the protocol parser.

Also, we introduce a new directory that will contain low-level common utilities and classes that
can be used by any other components: facade, core and server. We move StringOrView there.

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

Author: romange

src/common/backed_args.h

@@ -0,0 +1,90 @@
+// Copyright 2025, DragonflyDB authors.  All rights reserved.
+// See LICENSE for licensing terms.
+//
+
+#pragma once
+
+#include <absl/container/inlined_vector.h>
+
+#include <string_view>
+
+namespace cmn {
+
+#ifdef ABSL_HAVE_ADDRESS_SANITIZER
+constexpr size_t kReqStorageSize = 88;
+#else
+constexpr size_t kReqStorageSize = 88;
+#endif
+
+class BackedArguments {
+ public:
+  BackedArguments() {
+  }
+
+  template <typename I> BackedArguments(I begin, I end);
+
+  size_t UsedMemory() const {
+    return lens_.capacity() * sizeof(uint32_t) + storage_.capacity();
+  }
+
+  // The capacity is chosen so that we allocate a fully utilized (128 bytes) block.
+  using StorageType = absl::InlinedVector<char, kReqStorageSize>;
+
+  std::string_view Front() const {
+    return std::string_view{storage_.data(), lens_[0]};
+  }
+
+  /*auto MakeArgs() const {
+    absl::InlinedVector<std::string_view, 5> args(lens.size());
+    const char* ptr = storage.data();
+    for (size_t i = 0; i < lens.size(); ++i) {
+      args[i] = std::string_view{ptr, lens[i]};
+      ptr += lens[i] + 1;  // +1 for '\0'
+    }
+    return args;
+  }*/
+
+  size_t size() const {
+    return lens_.size();
+  }
+
+  size_t elem_len(size_t i) const {
+    return lens_[i];
+  }
+
+  size_t elem_capacity(size_t i) const {
+    return elem_len(i) + 1;
+  }
+
+  std::string_view at(uint32_t index, uint32_t offset) const {
+    const char* ptr = storage_.data() + offset;
+    return std::string_view{ptr + offset, lens_[index]};
+  }
+
+ protected:
+  absl::InlinedVector<uint32_t, 5> lens_;
+  StorageType storage_;
+};
+
+static_assert(sizeof(BackedArguments) == 128);
+
+template <typename I> BackedArguments::BackedArguments(I begin, I end) {
+  lens_.reserve(end - begin);
+  size_t total_size = 0;
+  for (auto it = begin; it != end; ++it) {
+    lens_.push_back(it->size());
+    total_size += it->size() + 1;  // +1 for '\0'
+  }
+  storage_.resize(total_size);
+
+  char* next = storage_.data();
+  for (auto it = begin; it != end; ++it) {
+    if (!it->empty()) {
+      memcpy(next, it->data(), it->size());
+    }
+    next[it->size()] = '\0';
+    next += it->size() + 1;
+  }
+}
+
+}  // namespace cmn

src/facade/dragonfly_connection.cc

@@ -377,15 +377,18 @@ class PipelineCacheSizeTracker {
 
 thread_local PipelineCacheSizeTracker tl_pipe_cache_sz_tracker;
 
-void Connection::PipelineMessage::SetArgs(const RespVec& args) {
-  auto* next = storage.data();
-  for (size_t i = 0; i < args.size(); ++i) {
-    RespExpr::Buffer buf = args[i].GetBuf();
-    size_t s = buf.size();
+void Connection::PipelineMessage::SetArgs(const RespVec& src, size_t backed_size) {
+  storage_.resize(backed_size);
+  lens_.resize(src.size());
+
+  auto* next = storage_.data();
+  for (size_t i = 0; i < src.size(); ++i) {
+    RespExpr::Buffer buf = src[i].GetBuf();
+    uint32_t s = buf.size();
     if (s)
       memcpy(next, buf.data(), s);
     next[s] = '\0';
-    this->args[i] = MutableSlice(next, s);
+    lens_[i] = s;
     next += (s + 1);
   }
 }
@@ -425,23 +428,13 @@ Connection::MCPipelineMessage::MCPipelineMessage(MemcacheParser::Command cmd_in,
   }
 }
 
-void Connection::PipelineMessage::Reset(size_t nargs, size_t capacity) {
-  storage.resize(capacity);
-  args.resize(nargs);
-}
-
-size_t Connection::PipelineMessage::StorageCapacity() const {
-  return storage.capacity() + args.capacity();
-}
-
 size_t Connection::MessageHandle::UsedMemory() const {
   struct MessageSize {
     size_t operator()(const PubMessagePtr& msg) {
       return sizeof(PubMessage) + (msg->channel.size() + msg->message.size());
     }
     size_t operator()(const PipelineMessagePtr& msg) {
-      return sizeof(PipelineMessage) + msg->args.capacity() * sizeof(MutableSlice) +
-             msg->storage.capacity();
+      return sizeof(PipelineMessage) + msg->UsedMemory();
     }
     size_t operator()(const MonitorMessage& msg) {
       return msg.capacity();
@@ -555,11 +548,10 @@ void Connection::AsyncOperations::operator()(const PubMessage& pub_msg) {
 }
 
 void Connection::AsyncOperations::operator()(Connection::PipelineMessage& msg) {
-  DVLOG(2) << "Dispatching pipeline: " << ToSV(msg.args.front());
+  DVLOG(2) << "Dispatching pipeline: " << msg.Front();
 
   ++self->local_stats_.cmds;
-  self->service_->DispatchCommand(ParsedArgs{msg.args}, self->reply_builder_.get(),
-                                  self->cc_.get());
+  self->service_->DispatchCommand(ParsedArgs{msg}, self->reply_builder_.get(), self->cc_.get());
 
   self->last_interaction_ = time(nullptr);
   self->skip_next_squashing_ = false;
@@ -644,7 +636,7 @@ Connection::Connection(Protocol protocol, util::HttpListenerBase* http_listener,
   static atomic_uint32_t next_id{1};
 
   constexpr size_t kReqSz = sizeof(Connection::PipelineMessage);
-  static_assert(kReqSz <= 256 && kReqSz >= 200);
+  static_assert(kReqSz <= 256);
 
   switch (protocol) {
     case Protocol::REDIS:
@@ -1537,16 +1529,17 @@ void Connection::SquashPipeline() {
   DCHECK_EQ(reply_builder_->GetProtocol(), Protocol::REDIS);  // Only Redis is supported.
 
   vector<ParsedArgs> squash_cmds;
-  squash_cmds.reserve(dispatch_q_.size());
+
+  squash_cmds.reserve(std::min<uint32_t>(dispatch_q_.size(), pipeline_squash_limit_cached));
 
   uint64_t start = CycleClock::Now();
 
   for (const auto& msg : dispatch_q_) {
     CHECK(holds_alternative<PipelineMessagePtr>(msg.handle))
         << msg.handle.index() << " on " << DebugInfo();
 
-    auto& pmsg = get<PipelineMessagePtr>(msg.handle);
-    squash_cmds.emplace_back(ParsedArgs(pmsg->args));
+    const auto& pmsg = get<PipelineMessagePtr>(msg.handle);
+    squash_cmds.emplace_back(*pmsg);
     if (squash_cmds.size() >= pipeline_squash_limit_cached) {
       // We reached the limit of commands to squash, so we dispatch them.
       break;
@@ -1766,14 +1759,12 @@ Connection::PipelineMessagePtr Connection::FromArgs(const RespVec& args) {
   static_assert(alignof(PipelineMessage) == 8);
 
   PipelineMessagePtr ptr;
-  if (ptr = GetFromPipelinePool(); ptr) {
-    ptr->Reset(args.size(), backed_sz);
-  } else {
+  if (ptr = GetFromPipelinePool(); !ptr) {
     // We must construct in place here, since there is a slice that uses memory locations
-    ptr = make_unique<PipelineMessage>(args.size(), backed_sz);
+    ptr = make_unique<PipelineMessage>();
   }
 
-  ptr->SetArgs(args);
+  ptr->SetArgs(args, backed_sz);
   return ptr;
 }
 
@@ -1782,7 +1773,7 @@ void Connection::ShrinkPipelinePool() {
     return;
 
   if (tl_pipe_cache_sz_tracker.CheckAndUpdateWatermark(pipeline_req_pool_.size())) {
-    stats_->pipeline_cmd_cache_bytes -= pipeline_req_pool_.back()->StorageCapacity();
+    stats_->pipeline_cmd_cache_bytes -= pipeline_req_pool_.back()->UsedMemory();
     pipeline_req_pool_.pop_back();
   }
 }
@@ -1792,7 +1783,7 @@ Connection::PipelineMessagePtr Connection::GetFromPipelinePool() {
     return nullptr;
 
   auto ptr = std::move(pipeline_req_pool_.back());
-  stats_->pipeline_cmd_cache_bytes -= ptr->StorageCapacity();
+  stats_->pipeline_cmd_cache_bytes -= ptr->UsedMemory();
   pipeline_req_pool_.pop_back();
   return ptr;
 }
@@ -1968,7 +1959,7 @@ void Connection::RecycleMessage(MessageHandle msg) {
     pending_pipeline_cmd_cnt_--;
     pending_pipeline_bytes_ -= used_mem;
     if (stats_->pipeline_cmd_cache_bytes < qbp.pipeline_cache_limit) {
-      stats_->pipeline_cmd_cache_bytes += (*pipe)->StorageCapacity();
+      stats_->pipeline_cmd_cache_bytes += (*pipe)->UsedMemory();
       pipeline_req_pool_.push_back(std::move(*pipe));
     }
   }

src/facade/dragonfly_connection.h

@@ -13,6 +13,7 @@
 #include <utility>
 #include <variant>
 
+#include "common/backed_args.h"
 #include "facade/acl_commands_def.h"
 #include "facade/facade_types.h"
 #include "facade/memcache_parser.h"
@@ -80,22 +81,8 @@ class Connection : public util::Connection {
   };
 
   // Pipeline message, accumulated Redis command to be executed.
-  struct PipelineMessage {
-    PipelineMessage(size_t nargs, size_t capacity) : args(nargs), storage(capacity) {
-    }
-
-    void Reset(size_t nargs, size_t capacity);
-
-    void SetArgs(const RespVec& args);
-
-    size_t StorageCapacity() const;
-
-    // mi_stl_allocator uses mi heap internally.
-    // The capacity is chosen so that we allocate a fully utilized (256 bytes) block.
-    using StorageType = absl::InlinedVector<char, kReqStorageSize>;
-
-    absl::InlinedVector<std::string_view, 6> args;
-    StorageType storage;
+  struct PipelineMessage : public cmn::BackedArguments {
+    void SetArgs(const RespVec& args, size_t backed_size);
   };
 
   // Pipeline message, accumulated Memcached command to be executed.

src/facade/facade_types.h

@@ -13,6 +13,7 @@
 #include <variant>
 
 #include "base/iterator.h"
+#include "common/backed_args.h"
 #include "facade/op_status.h"
 
 namespace facade {
@@ -45,31 +46,102 @@ class ParsedArgs {
  public:
   ParsedArgs() = default;
 
-  explicit ParsedArgs(ArgSlice args) : args_(args) {  // NOLINT google-explicit-constructor
+  ParsedArgs(const cmn::BackedArguments& bargs)  // NOLINT google-explicit-constructor
+      : args_(&bargs) {
   }
 
+  // TODO: to remove this constructor in favor of BackedArguments.
+  // The only reason to keep it is because our RedisParser returns ArgSlice with memory
+  // held elsewhere. We will have to use BackedArguments everywhere to support async invocations.
+  ParsedArgs(ArgSlice slice)  // NOLINT google-explicit-constructor
+      : args_(slice) {
+  }
+
+  ParsedArgs(const ParsedArgs& other) = default;
+  ParsedArgs& operator=(const ParsedArgs& bargs) = default;
+
   size_t size() const {
-    return args_.size();
+    return std::visit([](const auto& args) { return args.size(); }, args_);
   }
 
   bool empty() const {
-    return args_.empty();
+    return size() == 0;
   }
 
   ParsedArgs Tail() const {
-    return ParsedArgs{args_.subspan(1)};
+    return std::visit([](const auto& args) { return args.Tail(); }, args_);
   }
 
   std::string_view Front() const {
-    return args_.front();
+    return std::visit([](const auto& args) { return args.front(); }, args_);
   }
 
-  ArgSlice ToSlice() const {
-    return args_;
+  ArgSlice ToSlice(CmdArgVec* scratch) const {
+    return std::visit([scratch](const auto& args) { return args.ToSlice(scratch); }, args_);
+  }
+
+  size_t UsedMemory() const {
+    return std::visit([](const auto& args) { return args.UsedMemory(); }, args_);
   }
 
  private:
-  absl::Span<const std::string_view> args_;
+  struct WrapperBacked {
+    WrapperBacked(const cmn::BackedArguments* args) : args_(args) {  // NOLINT
+    }
+
+    const cmn::BackedArguments* args_;
+    uint32_t index_ = 0;
+    uint32_t ofs_bytes_ = 0;
+
+    ParsedArgs Tail() const {
+      ParsedArgs res(*args_);
+      WrapperBacked* wb = std::get_if<WrapperBacked>(&res.args_);
+      wb->index_ = index_ + 1;
+      wb->ofs_bytes_ = ofs_bytes_ + args_->elem_capacity(index_);
+      return res;
+    };
+
+    size_t size() const {
+      return args_->size() - index_;
+    }
+
+    std::string_view front() const {
+      return args_->at(index_, ofs_bytes_);
+    }
+
+    ArgSlice ToSlice(CmdArgVec* scratch) const {
+      scratch->resize(size());
+      size_t offset = ofs_bytes_;
+      for (size_t i = 0; i < scratch->size(); ++i) {
+        (*scratch)[i] = args_->at(index_ + i, offset);
+        offset += args_->elem_capacity(index_ + i);
+      }
+      return ArgSlice{scratch->data(), scratch->size()};
+    }
+
+    size_t UsedMemory() const {
+      return args_->UsedMemory();
+    }
+  };
+
+  struct Slice : public ArgSlice {
+    using ArgSlice::ArgSlice;
+    Slice(ArgSlice other) : ArgSlice(other) {  // NOLINT
+    }
+
+    ParsedArgs Tail() const {
+      return ParsedArgs{subspan(1)};
+    }
+
+    ArgSlice ToSlice(std::vector<std::string_view>* /*scratch*/) const {
+      return *this;
+    }
+
+    constexpr size_t UsedMemory() const {
+      return 0;
+    }
+  };
+  std::variant<Slice, WrapperBacked> args_;
 };
 
 inline std::string_view ToSV(std::string_view slice) {
@@ -188,7 +260,8 @@ struct ReplyStats {
   absl::flat_hash_map<std::string, uint64_t> err_count;
   size_t script_error_count = 0;
 
-  // This variable can be updated directly from shard threads when they allocate memory for replies.
+  // This variable can be updated directly from shard threads when they allocate memory for
+  // replies.
   std::atomic<size_t> squashing_current_reply_size{0};
 
   ReplyStats() = default;