Improve QueueAppender/IOBufQueue performance

[folly.git] / folly / io / IOBufQueue.cpp

diff --git a/folly/io/IOBufQueue.cpp b/folly/io/IOBufQueue.cpp
index df1518e0957a289f8d576261260d3845f1e30400..d54761e45f9acf5fcf6dd26fea2cc4bd697955ae 100644 (file)
--- a/folly/io/IOBufQueue.cpp
+++ b/folly/io/IOBufQueue.cpp
@@ -67,29 +67,53 @@ appendToChain(unique_ptr<IOBuf>& dst, unique_ptr<IOBuf>&& src, bool pack) {
 namespace folly {
 
 IOBufQueue::IOBufQueue(const Options& options)
-  : options_(options),
-    chainLength_(0) {
+    : options_(options), cachePtr_(&localCache_) {
+  localCache_.attached = true;
+}
+
+IOBufQueue::~IOBufQueue() {
+  clearWritableRangeCache();
 }
 
 IOBufQueue::IOBufQueue(IOBufQueue&& other) noexcept
-  : options_(other.options_),
-    chainLength_(other.chainLength_),
-    head_(std::move(other.head_)) {
+    : options_(other.options_), cachePtr_(&localCache_) {
+  other.clearWritableRangeCache();
+  head_ = std::move(other.head_);
+  chainLength_ = other.chainLength_;
+
+  tailStart_ = other.tailStart_;
+  localCache_.cachedRange = other.localCache_.cachedRange;
+  localCache_.attached = true;
+
   other.chainLength_ = 0;
+  other.tailStart_ = nullptr;
+  other.localCache_.cachedRange = {nullptr, nullptr};
 }
 
 IOBufQueue& IOBufQueue::operator=(IOBufQueue&& other) {
   if (&other != this) {
+    other.clearWritableRangeCache();
+    clearWritableRangeCache();
+
     options_ = other.options_;
-    chainLength_ = other.chainLength_;
     head_ = std::move(other.head_);
+    chainLength_ = other.chainLength_;
+
+    tailStart_ = other.tailStart_;
+    localCache_.cachedRange = other.localCache_.cachedRange;
+    localCache_.attached = true;
+
     other.chainLength_ = 0;
+    other.tailStart_ = nullptr;
+    other.localCache_.cachedRange = {nullptr, nullptr};
   }
   return *this;
 }
 
 std::pair<void*, uint64_t>
 IOBufQueue::headroom() {
+  // Note, headroom is independent from the tail, so we don't need to flush the
+  // cache.
   if (head_) {
     return std::make_pair(head_->writableBuffer(), head_->headroom());
   } else {
@@ -102,6 +126,8 @@ IOBufQueue::markPrepended(uint64_t n) {
   if (n == 0) {
     return;
   }
+  // Note, headroom is independent from the tail, so we don't need to flush the
+  // cache.
   assert(head_);
   head_->prepend(n);
   chainLength_ += n;
@@ -109,12 +135,14 @@ IOBufQueue::markPrepended(uint64_t n) {
 
 void
 IOBufQueue::prepend(const void* buf, uint64_t n) {
-  auto p = headroom();
-  if (n > p.second) {
+  // We're not touching the tail, so we don't need to flush the cache.
+  auto hroom = head_->headroom();
+  if (!head_ || hroom < n) {
     throw std::overflow_error("Not enough room to prepend");
   }
-  memcpy(static_cast<char*>(p.first) + p.second - n, buf, n);
-  markPrepended(n);
+  memcpy(head_->writableBuffer() + hroom - n, buf, n);
+  head_->prepend(n);
+  chainLength_ += n;
 }
 
 void
@@ -122,6 +150,7 @@ IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) {
   if (!buf) {
     return;
   }
+  auto guard = updateGuard();
   if (options_.cacheChainLength) {
     chainLength_ += buf->computeChainDataLength();
   }
@@ -133,6 +162,9 @@ IOBufQueue::append(IOBufQueue& other, bool pack) {
   if (!other.head_) {
     return;
   }
+  // We're going to chain other, thus we need to grab both guards.
+  auto otherGuard = other.updateGuard();
+  auto guard = updateGuard();
   if (options_.cacheChainLength) {
     if (other.options_.cacheChainLength) {
       chainLength_ += other.chainLength_;
@@ -146,6 +178,7 @@ IOBufQueue::append(IOBufQueue& other, bool pack) {
 
 void
 IOBufQueue::append(const void* buf, size_t len) {
+  auto guard = updateGuard();
   auto src = static_cast<const uint8_t*>(buf);
   while (len != 0) {
     if ((head_ == nullptr) || head_->prev()->isSharedOne() ||
@@ -179,15 +212,20 @@ IOBufQueue::wrapBuffer(const void* buf, size_t len, uint64_t blockSize) {
 pair<void*,uint64_t>
 IOBufQueue::preallocateSlow(uint64_t min, uint64_t newAllocationSize,
                             uint64_t max) {
+  // Avoid grabbing update guard, since we're manually setting the cache ptrs.
+  flushCache();
   // Allocate a new buffer of the requested max size.
   unique_ptr<IOBuf> newBuf(IOBuf::create(std::max(min, newAllocationSize)));
+
+  tailStart_ = newBuf->writableTail();
+  cachePtr_->cachedRange = std::pair<uint8_t*, uint8_t*>(
+      tailStart_, tailStart_ + newBuf->tailroom());
   appendToChain(head_, std::move(newBuf), false);
-  IOBuf* last = head_->prev();
-  return make_pair(last->writableTail(),
-                   std::min(max, last->tailroom()));
+  return make_pair(writableTail(), std::min<uint64_t>(max, tailroom()));
 }
 
 unique_ptr<IOBuf> IOBufQueue::split(size_t n, bool throwOnUnderflow) {
+  auto guard = updateGuard();
   unique_ptr<IOBuf> result;
   while (n != 0) {
     if (head_ == nullptr) {
@@ -227,6 +265,7 @@ void IOBufQueue::trimStart(size_t amount) {
 }
 
 size_t IOBufQueue::trimStartAtMost(size_t amount) {
+  auto guard = updateGuard();
   auto original = amount;
   while (amount > 0) {
     if (!head_) {
@@ -254,6 +293,7 @@ void IOBufQueue::trimEnd(size_t amount) {
 }
 
 size_t IOBufQueue::trimEndAtMost(size_t amount) {
+  auto guard = updateGuard();
   auto original = amount;
   while (amount > 0) {
     if (!head_) {
@@ -278,6 +318,7 @@ size_t IOBufQueue::trimEndAtMost(size_t amount) {
 }
 
 std::unique_ptr<folly::IOBuf> IOBufQueue::pop_front() {
+  auto guard = updateGuard();
   if (!head_) {
     return nullptr;
   }
@@ -291,6 +332,7 @@ void IOBufQueue::clear() {
   if (!head_) {
     return;
   }
+  auto guard = updateGuard();
   IOBuf* buf = head_.get();
   do {
     buf->clear();
@@ -303,16 +345,25 @@ void IOBufQueue::appendToString(std::string& out) const {
   if (!head_) {
     return;
   }
-  auto len =
-    options_.cacheChainLength ? chainLength_ : head_->computeChainDataLength();
+  auto len = options_.cacheChainLength
+      ? chainLength_ + (cachePtr_->cachedRange.first - tailStart_)
+      : head_->computeChainDataLength() +
+          (cachePtr_->cachedRange.first - tailStart_);
   out.reserve(out.size() + len);
 
   for (auto range : *head_) {
     out.append(reinterpret_cast<const char*>(range.data()), range.size());
   }
+
+  if (tailStart_ != cachePtr_->cachedRange.first) {
+    out.append(
+        reinterpret_cast<const char*>(tailStart_),
+        cachePtr_->cachedRange.first - tailStart_);
+  }
 }
 
 void IOBufQueue::gather(uint64_t maxLength) {
+  auto guard = updateGuard();
   if (head_ != nullptr) {
     head_->gather(maxLength);
   }