chore: use BackedArguments for implementing a pipeline message

Remove duplicate code for collapsing multiple blobs, similarly to #6171

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

Author: romange

src/common/backed_args.h

@@ -33,6 +33,16 @@ class BackedArguments {
     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();
   }

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->HeapMemory();
     }
     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()->HeapMemory();
     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->HeapMemory();
   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)->HeapMemory();
       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.