folly/io/IOBufQueue.cpp

   1 /*
   2  * Copyright 2013 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; // Must fit within a uint32_t
  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       uint32_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)
  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*, uint32_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(uint32_t n) {
 102   if (n == 0) {
 103     return;
 104   }
 105   assert(head_);
 106   head_->prepend(n);
 107   if (options_.cacheChainLength) {
 108     chainLength_ += n;
 109   }
 110 }
 111
 112 void
 113 IOBufQueue::prepend(const void* buf, uint32_t n) {
 114   auto p = headroom();
 115   if (n > p.second) {
 116     throw std::overflow_error("Not enough room to prepend");
 117   }
 118   memcpy(static_cast<char*>(p.first) + p.second - n, buf, n);
 119   markPrepended(n);
 120 }
 121
 122 void
 123 IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) {
 124   if (!buf) {
 125     return;
 126   }
 127   if (options_.cacheChainLength) {
 128     chainLength_ += buf->computeChainDataLength();
 129   }
 130   appendToChain(head_, std::move(buf), pack);
 131 }
 132
 133 void
 134 IOBufQueue::append(IOBufQueue& other, bool pack) {
 135   if (!other.head_) {
 136     return;
 137   }
 138   if (options_.cacheChainLength) {
 139     if (other.options_.cacheChainLength) {
 140       chainLength_ += other.chainLength_;
 141     } else {
 142       chainLength_ += other.head_->computeChainDataLength();
 143     }
 144   }
 145   appendToChain(head_, std::move(other.head_), pack);
 146   other.chainLength_ = 0;
 147 }
 148
 149 void
 150 IOBufQueue::append(const void* buf, size_t len) {
 151   auto src = static_cast<const uint8_t*>(buf);
 152   while (len != 0) {
 153     if ((head_ == nullptr) || head_->prev()->isSharedOne() ||
 154         (head_->prev()->tailroom() == 0)) {
 155       appendToChain(head_, std::move(
 156           IOBuf::create(std::max(MIN_ALLOC_SIZE,
 157               std::min(len, MAX_ALLOC_SIZE)))),
 158           false);
 159     }
 160     IOBuf* last = head_->prev();
 161     uint32_t copyLen = std::min(len, (size_t)last->tailroom());
 162     memcpy(last->writableTail(), src, copyLen);
 163     src += copyLen;
 164     last->append(copyLen);
 165     if (options_.cacheChainLength) {
 166       chainLength_ += copyLen;
 167     }
 168     len -= copyLen;
 169   }
 170 }
 171
 172 void
 173 IOBufQueue::wrapBuffer(const void* buf, size_t len, uint32_t blockSize) {
 174   auto src = static_cast<const uint8_t*>(buf);
 175   while (len != 0) {
 176     size_t n = std::min(len, size_t(blockSize));
 177     append(IOBuf::wrapBuffer(src, n));
 178     src += n;
 179     len -= n;
 180   }
 181 }
 182
 183 pair<void*,uint32_t>
 184 IOBufQueue::preallocate(uint32_t min, uint32_t newAllocationSize,
 185                         uint32_t max) {
 186   if (head_ != nullptr) {
 187     // If there's enough space left over at the end of the queue, use that.
 188     IOBuf* last = head_->prev();
 189     if (!last->isSharedOne()) {
 190       uint32_t avail = last->tailroom();
 191       if (avail >= min) {
 192         return make_pair(last->writableTail(), std::min(max, avail));
 193       }
 194     }
 195   }
 196   // Allocate a new buffer of the requested max size.
 197   unique_ptr<IOBuf> newBuf(IOBuf::create(std::max(min, newAllocationSize)));
 198   appendToChain(head_, std::move(newBuf), false);
 199   IOBuf* last = head_->prev();
 200   return make_pair(last->writableTail(),
 201                    std::min(max, last->tailroom()));
 202 }
 203
 204 void
 205 IOBufQueue::postallocate(uint32_t n) {
 206   head_->prev()->append(n);
 207   if (options_.cacheChainLength) {
 208     chainLength_ += n;
 209   }
 210 }
 211
 212 unique_ptr<IOBuf>
 213 IOBufQueue::split(size_t n) {
 214   unique_ptr<IOBuf> result;
 215   while (n != 0) {
 216     if (head_ == nullptr) {
 217       throw std::underflow_error(
 218           "Attempt to remove more bytes than are present in IOBufQueue");
 219     } else if (head_->length() <= n) {
 220       n -= head_->length();
 221       if (options_.cacheChainLength) {
 222         chainLength_ -= head_->length();
 223       }
 224       unique_ptr<IOBuf> remainder = head_->pop();
 225       appendToChain(result, std::move(head_), false);
 226       head_ = std::move(remainder);
 227     } else {
 228       unique_ptr<IOBuf> clone = head_->cloneOne();
 229       clone->trimEnd(clone->length() - n);
 230       appendToChain(result, std::move(clone), false);
 231       head_->trimStart(n);
 232       if (options_.cacheChainLength) {
 233         chainLength_ -= n;
 234       }
 235       break;
 236     }
 237   }
 238   return std::move(result);
 239 }
 240
 241 void IOBufQueue::trimStart(size_t amount) {
 242   while (amount > 0) {
 243     if (!head_) {
 244       throw std::underflow_error(
 245         "Attempt to trim more bytes than are present in IOBufQueue");
 246     }
 247     if (head_->length() > amount) {
 248       head_->trimStart(amount);
 249       if (options_.cacheChainLength) {
 250         chainLength_ -= amount;
 251       }
 252       break;
 253     }
 254     amount -= head_->length();
 255     if (options_.cacheChainLength) {
 256       chainLength_ -= head_->length();
 257     }
 258     head_ = head_->pop();
 259   }
 260 }
 261
 262 void IOBufQueue::trimEnd(size_t amount) {
 263   while (amount > 0) {
 264     if (!head_) {
 265       throw std::underflow_error(
 266         "Attempt to trim more bytes than are present in IOBufQueue");
 267     }
 268     if (head_->prev()->length() > amount) {
 269       head_->prev()->trimEnd(amount);
 270       if (options_.cacheChainLength) {
 271         chainLength_ -= amount;
 272       }
 273       break;
 274     }
 275     amount -= head_->prev()->length();
 276     if (options_.cacheChainLength) {
 277       chainLength_ -= head_->prev()->length();
 278     }
 279
 280     if (head_->isChained()) {
 281       head_->prev()->unlink();
 282     } else {
 283       head_.reset();
 284     }
 285   }
 286 }
 287
 288 std::unique_ptr<folly::IOBuf> IOBufQueue::pop_front() {
 289   if (!head_) {
 290     return nullptr;
 291   }
 292   if (options_.cacheChainLength) {
 293     chainLength_ -= head_->length();
 294   }
 295   std::unique_ptr<folly::IOBuf> retBuf = std::move(head_);
 296   head_ = retBuf->pop();
 297   return retBuf;
 298 }
 299
 300 void IOBufQueue::clear() {
 301   if (!head_) {
 302     return;
 303   }
 304   IOBuf* buf = head_.get();
 305   do {
 306     buf->clear();
 307     buf = buf->next();
 308   } while (buf != head_.get());
 309   if (options_.cacheChainLength) {
 310     chainLength_ = 0;
 311   }
 312 }
 313
 314 } // folly