folly/io/IOBufQueue.cpp

   1 /*
   2  * Copyright 2017 Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include <folly/io/IOBufQueue.h>
  18
  19 #include <string.h>
  20
  21 #include <stdexcept>
  22
  23 using std::make_pair;
  24 using std::pair;
  25 using std::unique_ptr;
  26
  27 namespace {
  28
  29 using folly::IOBuf;
  30
  31 const size_t MIN_ALLOC_SIZE = 2000;
  32 const size_t MAX_ALLOC_SIZE = 8000;
  33 const size_t MAX_PACK_COPY = 4096;
  34
  35 /**
  36  * Convenience function to append chain src to chain dst.
  37  */
  38 void
  39 appendToChain(unique_ptr<IOBuf>& dst, unique_ptr<IOBuf>&& src, bool pack) {
  40   if (dst == nullptr) {
  41     dst = std::move(src);
  42   } else {
  43     IOBuf* tail = dst->prev();
  44     if (pack) {
  45       // Copy up to MAX_PACK_COPY bytes if we can free buffers; this helps
  46       // reduce wastage (the tail's tailroom and the head's headroom) when
  47       // joining two IOBufQueues together.
  48       size_t copyRemaining = MAX_PACK_COPY;
  49       uint64_t n;
  50       while (src &&
  51              (n = src->length()) < copyRemaining &&
  52              n < tail->tailroom()) {
  53         memcpy(tail->writableTail(), src->data(), n);
  54         tail->append(n);
  55         copyRemaining -= n;
  56         src = src->pop();
  57       }
  58     }
  59     if (src) {
  60       tail->appendChain(std::move(src));
  61     }
  62   }
  63 }
  64
  65 } // anonymous namespace
  66
  67 namespace folly {
  68
  69 IOBufQueue::IOBufQueue(const Options& options)
  70   : options_(options),
  71     chainLength_(0) {
  72 }
  73
  74 IOBufQueue::IOBufQueue(IOBufQueue&& other) noexcept
  75   : options_(other.options_),
  76     chainLength_(other.chainLength_),
  77     head_(std::move(other.head_)) {
  78   other.chainLength_ = 0;
  79 }
  80
  81 IOBufQueue& IOBufQueue::operator=(IOBufQueue&& other) {
  82   if (&other != this) {
  83     options_ = other.options_;
  84     chainLength_ = other.chainLength_;
  85     head_ = std::move(other.head_);
  86     other.chainLength_ = 0;
  87   }
  88   return *this;
  89 }
  90
  91 std::pair<void*, uint64_t>
  92 IOBufQueue::headroom() {
  93   if (head_) {
  94     return std::make_pair(head_->writableBuffer(), head_->headroom());
  95   } else {
  96     return std::make_pair(nullptr, 0);
  97   }
  98 }
  99
 100 void
 101 IOBufQueue::markPrepended(uint64_t n) {
 102   if (n == 0) {
 103     return;
 104   }
 105   assert(head_);
 106   head_->prepend(n);
 107   chainLength_ += n;
 108 }
 109
 110 void
 111 IOBufQueue::prepend(const void* buf, uint64_t n) {
 112   auto p = headroom();
 113   if (n > p.second) {
 114     throw std::overflow_error("Not enough room to prepend");
 115   }
 116   memcpy(static_cast<char*>(p.first) + p.second - n, buf, n);
 117   markPrepended(n);
 118 }
 119
 120 void
 121 IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) {
 122   if (!buf) {
 123     return;
 124   }
 125   if (options_.cacheChainLength) {
 126     chainLength_ += buf->computeChainDataLength();
 127   }
 128   appendToChain(head_, std::move(buf), pack);
 129 }
 130
 131 void
 132 IOBufQueue::append(IOBufQueue& other, bool pack) {
 133   if (!other.head_) {
 134     return;
 135   }
 136   if (options_.cacheChainLength) {
 137     if (other.options_.cacheChainLength) {
 138       chainLength_ += other.chainLength_;
 139     } else {
 140       chainLength_ += other.head_->computeChainDataLength();
 141     }
 142   }
 143   appendToChain(head_, std::move(other.head_), pack);
 144   other.chainLength_ = 0;
 145 }
 146
 147 void
 148 IOBufQueue::append(const void* buf, size_t len) {
 149   auto src = static_cast<const uint8_t*>(buf);
 150   while (len != 0) {
 151     if ((head_ == nullptr) || head_->prev()->isSharedOne() ||
 152         (head_->prev()->tailroom() == 0)) {
 153       appendToChain(head_,
 154           IOBuf::create(std::max(MIN_ALLOC_SIZE,
 155               std::min(len, MAX_ALLOC_SIZE))),
 156           false);
 157     }
 158     IOBuf* last = head_->prev();
 159     uint64_t copyLen = std::min(len, (size_t)last->tailroom());
 160     memcpy(last->writableTail(), src, copyLen);
 161     src += copyLen;
 162     last->append(copyLen);
 163     chainLength_ += copyLen;
 164     len -= copyLen;
 165   }
 166 }
 167
 168 void
 169 IOBufQueue::wrapBuffer(const void* buf, size_t len, uint64_t blockSize) {
 170   auto src = static_cast<const uint8_t*>(buf);
 171   while (len != 0) {
 172     size_t n = std::min(len, size_t(blockSize));
 173     append(IOBuf::wrapBuffer(src, n));
 174     src += n;
 175     len -= n;
 176   }
 177 }
 178
 179 pair<void*,uint64_t>
 180 IOBufQueue::preallocateSlow(uint64_t min, uint64_t newAllocationSize,
 181                             uint64_t max) {
 182   // Allocate a new buffer of the requested max size.
 183   unique_ptr<IOBuf> newBuf(IOBuf::create(std::max(min, newAllocationSize)));
 184   appendToChain(head_, std::move(newBuf), false);
 185   IOBuf* last = head_->prev();
 186   return make_pair(last->writableTail(),
 187                    std::min(max, last->tailroom()));
 188 }
 189
 190 unique_ptr<IOBuf> IOBufQueue::split(size_t n, bool throwOnUnderflow) {
 191   unique_ptr<IOBuf> result;
 192   while (n != 0) {
 193     if (head_ == nullptr) {
 194       if (throwOnUnderflow) {
 195         throw std::underflow_error(
 196             "Attempt to remove more bytes than are present in IOBufQueue");
 197       } else {
 198         break;
 199       }
 200     } else if (head_->length() <= n) {
 201       n -= head_->length();
 202       chainLength_ -= head_->length();
 203       unique_ptr<IOBuf> remainder = head_->pop();
 204       appendToChain(result, std::move(head_), false);
 205       head_ = std::move(remainder);
 206     } else {
 207       unique_ptr<IOBuf> clone = head_->cloneOne();
 208       clone->trimEnd(clone->length() - n);
 209       appendToChain(result, std::move(clone), false);
 210       head_->trimStart(n);
 211       chainLength_ -= n;
 212       break;
 213     }
 214   }
 215   if (UNLIKELY(result == nullptr)) {
 216     return IOBuf::create(0);
 217   }
 218   return result;
 219 }
 220
 221 void IOBufQueue::trimStart(size_t amount) {
 222   auto trimmed = trimStartAtMost(amount);
 223   if (trimmed != amount) {
 224     throw std::underflow_error(
 225         "Attempt to trim more bytes than are present in IOBufQueue");
 226   }
 227 }
 228
 229 size_t IOBufQueue::trimStartAtMost(size_t amount) {
 230   auto original = amount;
 231   while (amount > 0) {
 232     if (!head_) {
 233       break;
 234     }
 235     if (head_->length() > amount) {
 236       head_->trimStart(amount);
 237       chainLength_ -= amount;
 238       amount = 0;
 239       break;
 240     }
 241     amount -= head_->length();
 242     chainLength_ -= head_->length();
 243     head_ = head_->pop();
 244   }
 245   return original - amount;
 246 }
 247
 248 void IOBufQueue::trimEnd(size_t amount) {
 249   auto trimmed = trimEndAtMost(amount);
 250   if (trimmed != amount) {
 251     throw std::underflow_error(
 252         "Attempt to trim more bytes than are present in IOBufQueue");
 253   }
 254 }
 255
 256 size_t IOBufQueue::trimEndAtMost(size_t amount) {
 257   auto original = amount;
 258   while (amount > 0) {
 259     if (!head_) {
 260       break;
 261     }
 262     if (head_->prev()->length() > amount) {
 263       head_->prev()->trimEnd(amount);
 264       chainLength_ -= amount;
 265       amount = 0;
 266       break;
 267     }
 268     amount -= head_->prev()->length();
 269     chainLength_ -= head_->prev()->length();
 270
 271     if (head_->isChained()) {
 272       head_->prev()->unlink();
 273     } else {
 274       head_.reset();
 275     }
 276   }
 277   return original - amount;
 278 }
 279
 280 std::unique_ptr<folly::IOBuf> IOBufQueue::pop_front() {
 281   if (!head_) {
 282     return nullptr;
 283   }
 284   chainLength_ -= head_->length();
 285   std::unique_ptr<folly::IOBuf> retBuf = std::move(head_);
 286   head_ = retBuf->pop();
 287   return retBuf;
 288 }
 289
 290 void IOBufQueue::clear() {
 291   if (!head_) {
 292     return;
 293   }
 294   IOBuf* buf = head_.get();
 295   do {
 296     buf->clear();
 297     buf = buf->next();
 298   } while (buf != head_.get());
 299   chainLength_ = 0;
 300 }
 301
 302 void IOBufQueue::appendToString(std::string& out) const {
 303   if (!head_) {
 304     return;
 305   }
 306   auto len =
 307     options_.cacheChainLength ? chainLength_ : head_->computeChainDataLength();
 308   out.reserve(out.size() + len);
 309
 310   for (auto range : *head_) {
 311     out.append(reinterpret_cast<const char*>(range.data()), range.size());
 312   }
 313 }
 314
 315 void IOBufQueue::gather(uint64_t maxLength) {
 316   if (head_ != nullptr) {
 317     head_->gather(maxLength);
 318   }
 319 }
 320
 321 } // folly