folly/io/Cursor.h

   1 /*
   2  * Copyright 2014 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 #ifndef FOLLY_CURSOR_H
  18 #define FOLLY_CURSOR_H
  19
  20 #include <assert.h>
  21 #include <cstdarg>
  22 #include <stdexcept>
  23 #include <string.h>
  24 #include <type_traits>
  25 #include <memory>
  26
  27 #include <folly/Bits.h>
  28 #include <folly/io/IOBuf.h>
  29 #include <folly/io/IOBufQueue.h>
  30 #include <folly/Likely.h>
  31 #include <folly/Memory.h>
  32 #include <folly/Portability.h>
  33 #include <folly/Range.h>
  34
  35 /**
  36  * Cursor class for fast iteration over IOBuf chains.
  37  *
  38  * Cursor - Read-only access
  39  *
  40  * RWPrivateCursor - Read-write access, assumes private access to IOBuf chain
  41  * RWUnshareCursor - Read-write access, calls unshare on write (COW)
  42  * Appender        - Write access, assumes private access to IOBuf chian
  43  *
  44  * Note that RW cursors write in the preallocated part of buffers (that is,
  45  * between the buffer's data() and tail()), while Appenders append to the end
  46  * of the buffer (between the buffer's tail() and bufferEnd()).  Appenders
  47  * automatically adjust the buffer pointers, so you may only use one
  48  * Appender with a buffer chain; for this reason, Appenders assume private
  49  * access to the buffer (you need to call unshare() yourself if necessary).
  50  **/
  51 namespace folly { namespace io {
  52 namespace detail {
  53
  54 template <class Derived, typename BufType>
  55 class CursorBase {
  56  public:
  57   const uint8_t* data() const {
  58     return crtBuf_->data() + offset_;
  59   }
  60
  61   /*
  62    * Return the remaining space available in the current IOBuf.
  63    *
  64    * May return 0 if the cursor is at the end of an IOBuf.  Use peek() instead
  65    * if you want to avoid this.  peek() will advance to the next non-empty
  66    * IOBuf (up to the end of the chain) if the cursor is currently pointing at
  67    * the end of a buffer.
  68    */
  69   size_t length() const {
  70     return crtBuf_->length() - offset_;
  71   }
  72
  73   /*
  74    * Return the space available until the end of the entire IOBuf chain.
  75    */
  76   size_t totalLength() const {
  77     if (crtBuf_ == buffer_) {
  78       return crtBuf_->computeChainDataLength() - offset_;
  79     }
  80     CursorBase end(buffer_->prev());
  81     end.offset_ = end.buffer_->length();
  82     return end - *this;
  83   }
  84
  85   Derived& operator+=(size_t offset) {
  86     Derived* p = static_cast<Derived*>(this);
  87     p->skip(offset);
  88     return *p;
  89   }
  90   Derived operator+(size_t offset) const {
  91     Derived other(*this);
  92     other.skip(offset);
  93     return other;
  94   }
  95
  96   /**
  97    * Compare cursors for equality/inequality.
  98    *
  99    * Two cursors are equal if they are pointing to the same location in the
 100    * same IOBuf chain.
 101    */
 102   bool operator==(const Derived& other) const {
 103     return (offset_ == other.offset_) && (crtBuf_ == other.crtBuf_);
 104   }
 105   bool operator!=(const Derived& other) const {
 106     return !operator==(other);
 107   }
 108
 109   template <class T>
 110   typename std::enable_if<std::is_arithmetic<T>::value, T>::type
 111   read() {
 112     T val;
 113     pull(&val, sizeof(T));
 114     return val;
 115   }
 116
 117   template <class T>
 118   T readBE() {
 119     return Endian::big(read<T>());
 120   }
 121
 122   template <class T>
 123   T readLE() {
 124     return Endian::little(read<T>());
 125   }
 126
 127   /**
 128    * Read a fixed-length string.
 129    *
 130    * The std::string-based APIs should probably be avoided unless you
 131    * ultimately want the data to live in an std::string. You're better off
 132    * using the pull() APIs to copy into a raw buffer otherwise.
 133    */
 134   std::string readFixedString(size_t len) {
 135     std::string str;
 136
 137     str.reserve(len);
 138     for (;;) {
 139       // Fast path: it all fits in one buffer.
 140       size_t available = length();
 141       if (LIKELY(available >= len)) {
 142         str.append(reinterpret_cast<const char*>(data()), len);
 143         offset_ += len;
 144         return str;
 145       }
 146
 147       str.append(reinterpret_cast<const char*>(data()), available);
 148       if (UNLIKELY(!tryAdvanceBuffer())) {
 149         throw std::out_of_range("string underflow");
 150       }
 151       len -= available;
 152     }
 153   }
 154
 155   /**
 156    * Read a string consisting of bytes until the given terminator character is
 157    * seen. Raises an std::length_error if maxLength bytes have been processed
 158    * before the terminator is seen.
 159    *
 160    * See comments in readFixedString() about when it's appropriate to use this
 161    * vs. using pull().
 162    */
 163   std::string readTerminatedString(
 164     char termChar = '\0',
 165     size_t maxLength = std::numeric_limits<size_t>::max()) {
 166     std::string str;
 167
 168     for (;;) {
 169       const uint8_t* buf = data();
 170       size_t buflen = length();
 171
 172       size_t i = 0;
 173       while (i < buflen && buf[i] != termChar) {
 174         ++i;
 175
 176         // Do this check after incrementing 'i', as even though we start at the
 177         // 0 byte, it still represents a single character
 178         if (str.length() + i >= maxLength) {
 179           throw std::length_error("string overflow");
 180         }
 181       }
 182
 183       str.append(reinterpret_cast<const char*>(buf), i);
 184       if (i < buflen) {
 185         skip(i + 1);
 186         return str;
 187       }
 188
 189       skip(i);
 190
 191       if (UNLIKELY(!tryAdvanceBuffer())) {
 192         throw std::out_of_range("string underflow");
 193       }
 194     }
 195   }
 196
 197   explicit CursorBase(BufType* buf)
 198     : crtBuf_(buf)
 199     , offset_(0)
 200     , buffer_(buf) {}
 201
 202   // Make all the templated classes friends for copy constructor.
 203   template <class D, typename B> friend class CursorBase;
 204
 205   /*
 206    * Copy constructor.
 207    *
 208    * This also allows constructing a CursorBase from other derived types.
 209    * For instance, this allows constructing a Cursor from an RWPrivateCursor.
 210    */
 211   template <class OtherDerived, class OtherBuf>
 212   explicit CursorBase(const CursorBase<OtherDerived, OtherBuf>& cursor)
 213     : crtBuf_(cursor.crtBuf_),
 214       offset_(cursor.offset_),
 215       buffer_(cursor.buffer_) {}
 216
 217   // reset cursor to point to a new buffer.
 218   void reset(BufType* buf) {
 219     crtBuf_ = buf;
 220     buffer_ = buf;
 221     offset_ = 0;
 222   }
 223
 224   /**
 225    * Return the available data in the current buffer.
 226    * If you want to gather more data from the chain into a contiguous region
 227    * (for hopefully zero-copy access), use gather() before peek().
 228    */
 229   std::pair<const uint8_t*, size_t> peek() {
 230     // Ensure that we're pointing to valid data
 231     size_t available = length();
 232     while (UNLIKELY(available == 0 && tryAdvanceBuffer())) {
 233       available = length();
 234     }
 235
 236     return std::make_pair(data(), available);
 237   }
 238
 239   void pull(void* buf, size_t len) {
 240     if (UNLIKELY(pullAtMost(buf, len) != len)) {
 241       throw std::out_of_range("underflow");
 242     }
 243   }
 244
 245   void clone(std::unique_ptr<folly::IOBuf>& buf, size_t len) {
 246     if (UNLIKELY(cloneAtMost(buf, len) != len)) {
 247       throw std::out_of_range("underflow");
 248     }
 249   }
 250
 251   void clone(folly::IOBuf& buf, size_t len) {
 252     if (UNLIKELY(cloneAtMost(buf, len) != len)) {
 253       throw std::out_of_range("underflow");
 254     }
 255   }
 256
 257   void skip(size_t len) {
 258     if (UNLIKELY(skipAtMost(len) != len)) {
 259       throw std::out_of_range("underflow");
 260     }
 261   }
 262
 263   size_t pullAtMost(void* buf, size_t len) {
 264     uint8_t* p = reinterpret_cast<uint8_t*>(buf);
 265     size_t copied = 0;
 266     for (;;) {
 267       // Fast path: it all fits in one buffer.
 268       size_t available = length();
 269       if (LIKELY(available >= len)) {
 270         memcpy(p, data(), len);
 271         offset_ += len;
 272         return copied + len;
 273       }
 274
 275       memcpy(p, data(), available);
 276       copied += available;
 277       if (UNLIKELY(!tryAdvanceBuffer())) {
 278         return copied;
 279       }
 280       p += available;
 281       len -= available;
 282     }
 283   }
 284
 285   size_t cloneAtMost(folly::IOBuf& buf, size_t len) {
 286     buf = folly::IOBuf();
 287
 288     std::unique_ptr<folly::IOBuf> tmp;
 289     size_t copied = 0;
 290     for (int loopCount = 0; true; ++loopCount) {
 291       // Fast path: it all fits in one buffer.
 292       size_t available = length();
 293       if (LIKELY(available >= len)) {
 294         if (loopCount == 0) {
 295           crtBuf_->cloneOneInto(buf);
 296           buf.trimStart(offset_);
 297           buf.trimEnd(buf.length() - len);
 298         } else {
 299           tmp = crtBuf_->cloneOne();
 300           tmp->trimStart(offset_);
 301           tmp->trimEnd(tmp->length() - len);
 302           buf.prependChain(std::move(tmp));
 303         }
 304
 305         offset_ += len;
 306         return copied + len;
 307       }
 308
 309
 310       if (loopCount == 0) {
 311         crtBuf_->cloneOneInto(buf);
 312         buf.trimStart(offset_);
 313       } else {
 314         tmp = crtBuf_->cloneOne();
 315         tmp->trimStart(offset_);
 316         buf.prependChain(std::move(tmp));
 317       }
 318
 319       copied += available;
 320       if (UNLIKELY(!tryAdvanceBuffer())) {
 321         return copied;
 322       }
 323       len -= available;
 324     }
 325   }
 326
 327   size_t cloneAtMost(std::unique_ptr<folly::IOBuf>& buf, size_t len) {
 328     if (!buf) {
 329       buf = make_unique<folly::IOBuf>();
 330     }
 331
 332     return cloneAtMost(*buf, len);
 333   }
 334
 335   size_t skipAtMost(size_t len) {
 336     size_t skipped = 0;
 337     for (;;) {
 338       // Fast path: it all fits in one buffer.
 339       size_t available = length();
 340       if (LIKELY(available >= len)) {
 341         offset_ += len;
 342         return skipped + len;
 343       }
 344
 345       skipped += available;
 346       if (UNLIKELY(!tryAdvanceBuffer())) {
 347         return skipped;
 348       }
 349       len -= available;
 350     }
 351   }
 352
 353   /**
 354    * Return the distance between two cursors.
 355    */
 356   size_t operator-(const CursorBase& other) const {
 357     BufType *otherBuf = other.crtBuf_;
 358     size_t len = 0;
 359
 360     if (otherBuf != crtBuf_) {
 361       len += otherBuf->length() - other.offset_;
 362
 363       for (otherBuf = otherBuf->next();
 364            otherBuf != crtBuf_ && otherBuf != other.buffer_;
 365            otherBuf = otherBuf->next()) {
 366         len += otherBuf->length();
 367       }
 368
 369       if (otherBuf == other.buffer_) {
 370         throw std::out_of_range("wrap-around");
 371       }
 372
 373       len += offset_;
 374     } else {
 375       if (offset_ < other.offset_) {
 376         throw std::out_of_range("underflow");
 377       }
 378
 379       len += offset_ - other.offset_;
 380     }
 381
 382     return len;
 383   }
 384
 385   /**
 386    * Return the distance from the given IOBuf to the this cursor.
 387    */
 388   size_t operator-(const BufType* buf) const {
 389     size_t len = 0;
 390
 391     BufType *curBuf = buf;
 392     while (curBuf != crtBuf_) {
 393       len += curBuf->length();
 394       curBuf = curBuf->next();
 395       if (curBuf == buf || curBuf == buffer_) {
 396         throw std::out_of_range("wrap-around");
 397       }
 398     }
 399
 400     len += offset_;
 401     return len;
 402   }
 403
 404  protected:
 405   BufType* crtBuf_;
 406   size_t offset_;
 407
 408   ~CursorBase(){}
 409
 410   BufType* head() {
 411     return buffer_;
 412   }
 413
 414   bool tryAdvanceBuffer() {
 415     BufType* nextBuf = crtBuf_->next();
 416     if (UNLIKELY(nextBuf == buffer_)) {
 417       offset_ = crtBuf_->length();
 418       return false;
 419     }
 420
 421     offset_ = 0;
 422     crtBuf_ = nextBuf;
 423     static_cast<Derived*>(this)->advanceDone();
 424     return true;
 425   }
 426
 427  private:
 428   void advanceDone() {
 429   }
 430
 431   BufType* buffer_;
 432 };
 433
 434 } //namespace detail
 435
 436 class Cursor : public detail::CursorBase<Cursor, const IOBuf> {
 437  public:
 438   explicit Cursor(const IOBuf* buf)
 439     : detail::CursorBase<Cursor, const IOBuf>(buf) {}
 440
 441   template <class OtherDerived, class OtherBuf>
 442   explicit Cursor(const detail::CursorBase<OtherDerived, OtherBuf>& cursor)
 443     : detail::CursorBase<Cursor, const IOBuf>(cursor) {}
 444 };
 445
 446 namespace detail {
 447
 448 template <class Derived>
 449 class Writable {
 450  public:
 451   template <class T>
 452   typename std::enable_if<std::is_arithmetic<T>::value>::type
 453   write(T value) {
 454     const uint8_t* u8 = reinterpret_cast<const uint8_t*>(&value);
 455     Derived* d = static_cast<Derived*>(this);
 456     d->push(u8, sizeof(T));
 457   }
 458
 459   template <class T>
 460   void writeBE(T value) {
 461     Derived* d = static_cast<Derived*>(this);
 462     d->write(Endian::big(value));
 463   }
 464
 465   template <class T>
 466   void writeLE(T value) {
 467     Derived* d = static_cast<Derived*>(this);
 468     d->write(Endian::little(value));
 469   }
 470
 471   void push(const uint8_t* buf, size_t len) {
 472     Derived* d = static_cast<Derived*>(this);
 473     if (d->pushAtMost(buf, len) != len) {
 474       throw std::out_of_range("overflow");
 475     }
 476   }
 477
 478   void push(ByteRange buf) {
 479     if (this->pushAtMost(buf) != buf.size()) {
 480       throw std::out_of_range("overflow");
 481     }
 482   }
 483
 484   size_t pushAtMost(ByteRange buf) {
 485     Derived* d = static_cast<Derived*>(this);
 486     return d->pushAtMost(buf.data(), buf.size());
 487   }
 488
 489   /**
 490    * push len bytes of data from input cursor, data could be in an IOBuf chain.
 491    * If input cursor contains less than len bytes, or this cursor has less than
 492    * len bytes writable space, an out_of_range exception will be thrown.
 493    */
 494   void push(Cursor cursor, size_t len) {
 495     if (this->pushAtMost(cursor, len) != len) {
 496       throw std::out_of_range("overflow");
 497     }
 498   }
 499
 500   size_t pushAtMost(Cursor cursor, size_t len) {
 501     size_t written = 0;
 502     for(;;) {
 503       auto currentBuffer = cursor.peek();
 504       const uint8_t* crtData = currentBuffer.first;
 505       size_t available = currentBuffer.second;
 506       if (available == 0) {
 507         // end of buffer chain
 508         return written;
 509       }
 510       // all data is in current buffer
 511       if (available >= len) {
 512         this->push(crtData, len);
 513         cursor.skip(len);
 514         return written + len;
 515       }
 516
 517       // write the whole current IOBuf
 518       this->push(crtData, available);
 519       cursor.skip(available);
 520       written += available;
 521       len -= available;
 522     }
 523   }
 524 };
 525
 526 } // namespace detail
 527
 528 enum class CursorAccess {
 529   PRIVATE,
 530   UNSHARE
 531 };
 532
 533 template <CursorAccess access>
 534 class RWCursor
 535   : public detail::CursorBase<RWCursor<access>, IOBuf>,
 536     public detail::Writable<RWCursor<access>> {
 537   friend class detail::CursorBase<RWCursor<access>, IOBuf>;
 538  public:
 539   explicit RWCursor(IOBuf* buf)
 540     : detail::CursorBase<RWCursor<access>, IOBuf>(buf),
 541       maybeShared_(true) {}
 542
 543   template <class OtherDerived, class OtherBuf>
 544   explicit RWCursor(const detail::CursorBase<OtherDerived, OtherBuf>& cursor)
 545     : detail::CursorBase<RWCursor<access>, IOBuf>(cursor),
 546       maybeShared_(true) {}
 547   /**
 548    * Gather at least n bytes contiguously into the current buffer,
 549    * by coalescing subsequent buffers from the chain as necessary.
 550    */
 551   void gather(size_t n) {
 552     // Forbid attempts to gather beyond the end of this IOBuf chain.
 553     // Otherwise we could try to coalesce the head of the chain and end up
 554     // accidentally freeing it, invalidating the pointer owned by external
 555     // code.
 556     //
 557     // If crtBuf_ == head() then IOBuf::gather() will perform all necessary
 558     // checking.  We only have to perform an explicit check here when calling
 559     // gather() on a non-head element.
 560     if (this->crtBuf_ != this->head() && this->totalLength() < n) {
 561       throw std::overflow_error("cannot gather() past the end of the chain");
 562     }
 563     this->crtBuf_->gather(this->offset_ + n);
 564   }
 565   void gatherAtMost(size_t n) {
 566     size_t size = std::min(n, this->totalLength());
 567     return this->crtBuf_->gather(this->offset_ + size);
 568   }
 569
 570   using detail::Writable<RWCursor<access>>::pushAtMost;
 571   size_t pushAtMost(const uint8_t* buf, size_t len) {
 572     size_t copied = 0;
 573     for (;;) {
 574       // Fast path: the current buffer is big enough.
 575       size_t available = this->length();
 576       if (LIKELY(available >= len)) {
 577         if (access == CursorAccess::UNSHARE) {
 578           maybeUnshare();
 579         }
 580         memcpy(writableData(), buf, len);
 581         this->offset_ += len;
 582         return copied + len;
 583       }
 584
 585       if (access == CursorAccess::UNSHARE) {
 586         maybeUnshare();
 587       }
 588       memcpy(writableData(), buf, available);
 589       copied += available;
 590       if (UNLIKELY(!this->tryAdvanceBuffer())) {
 591         return copied;
 592       }
 593       buf += available;
 594       len -= available;
 595     }
 596   }
 597
 598   void insert(std::unique_ptr<folly::IOBuf> buf) {
 599     folly::IOBuf* nextBuf;
 600     if (this->offset_ == 0) {
 601       // Can just prepend
 602       nextBuf = this->crtBuf_;
 603       this->crtBuf_->prependChain(std::move(buf));
 604     } else {
 605       std::unique_ptr<folly::IOBuf> remaining;
 606       if (this->crtBuf_->length() - this->offset_ > 0) {
 607         // Need to split current IOBuf in two.
 608         remaining = this->crtBuf_->cloneOne();
 609         remaining->trimStart(this->offset_);
 610         nextBuf = remaining.get();
 611         buf->prependChain(std::move(remaining));
 612       } else {
 613         // Can just append
 614         nextBuf = this->crtBuf_->next();
 615       }
 616       this->crtBuf_->trimEnd(this->length());
 617       this->crtBuf_->appendChain(std::move(buf));
 618     }
 619     // Jump past the new links
 620     this->offset_ = 0;
 621     this->crtBuf_ = nextBuf;
 622   }
 623
 624   uint8_t* writableData() {
 625     return this->crtBuf_->writableData() + this->offset_;
 626   }
 627
 628  private:
 629   void maybeUnshare() {
 630     if (UNLIKELY(maybeShared_)) {
 631       this->crtBuf_->unshareOne();
 632       maybeShared_ = false;
 633     }
 634   }
 635
 636   void advanceDone() {
 637     maybeShared_ = true;
 638   }
 639
 640   bool maybeShared_;
 641 };
 642
 643 typedef RWCursor<CursorAccess::PRIVATE> RWPrivateCursor;
 644 typedef RWCursor<CursorAccess::UNSHARE> RWUnshareCursor;
 645
 646 /**
 647  * Append to the end of a buffer chain, growing the chain (by allocating new
 648  * buffers) in increments of at least growth bytes every time.  Won't grow
 649  * (and push() and ensure() will throw) if growth == 0.
 650  *
 651  * TODO(tudorb): add a flavor of Appender that reallocates one IOBuf instead
 652  * of chaining.
 653  */
 654 class Appender : public detail::Writable<Appender> {
 655  public:
 656   Appender(IOBuf* buf, uint64_t growth)
 657     : buffer_(buf),
 658       crtBuf_(buf->prev()),
 659       growth_(growth) {
 660   }
 661
 662   uint8_t* writableData() {
 663     return crtBuf_->writableTail();
 664   }
 665
 666   size_t length() const {
 667     return crtBuf_->tailroom();
 668   }
 669
 670   /**
 671    * Mark n bytes (must be <= length()) as appended, as per the
 672    * IOBuf::append() method.
 673    */
 674   void append(size_t n) {
 675     crtBuf_->append(n);
 676   }
 677
 678   /**
 679    * Ensure at least n contiguous bytes available to write.
 680    * Postcondition: length() >= n.
 681    */
 682   void ensure(uint64_t n) {
 683     if (LIKELY(length() >= n)) {
 684       return;
 685     }
 686
 687     // Waste the rest of the current buffer and allocate a new one.
 688     // Don't make it too small, either.
 689     if (growth_ == 0) {
 690       throw std::out_of_range("can't grow buffer chain");
 691     }
 692
 693     n = std::max(n, growth_);
 694     buffer_->prependChain(IOBuf::create(n));
 695     crtBuf_ = buffer_->prev();
 696   }
 697
 698   using detail::Writable<Appender>::pushAtMost;
 699   size_t pushAtMost(const uint8_t* buf, size_t len) {
 700     size_t copied = 0;
 701     for (;;) {
 702       // Fast path: it all fits in one buffer.
 703       size_t available = length();
 704       if (LIKELY(available >= len)) {
 705         memcpy(writableData(), buf, len);
 706         append(len);
 707         return copied + len;
 708       }
 709
 710       memcpy(writableData(), buf, available);
 711       append(available);
 712       copied += available;
 713       if (UNLIKELY(!tryGrowChain())) {
 714         return copied;
 715       }
 716       buf += available;
 717       len -= available;
 718     }
 719   }
 720
 721   /*
 722    * Append to the end of this buffer, using a printf() style
 723    * format specifier.
 724    *
 725    * Note that folly/Format.h provides nicer and more type-safe mechanisms
 726    * for formatting strings, which should generally be preferred over
 727    * printf-style formatting.  Appender objects can be used directly as an
 728    * output argument for Formatter objects.  For example:
 729    *
 730    *   Appender app(&iobuf);
 731    *   format("{} {}", "hello", "world")(app);
 732    *
 733    * However, printf-style strings are still needed when dealing with existing
 734    * third-party code in some cases.
 735    *
 736    * This will always add a nul-terminating character after the end
 737    * of the output.  However, the buffer data length will only be updated to
 738    * include the data itself.  The nul terminator will be the first byte in the
 739    * buffer tailroom.
 740    *
 741    * This method may throw exceptions on error.
 742    */
 743   void printf(FOLLY_PRINTF_FORMAT const char* fmt, ...)
 744     FOLLY_PRINTF_FORMAT_ATTR(2, 3);
 745
 746   void vprintf(const char* fmt, va_list ap);
 747
 748   /*
 749    * Calling an Appender object with a StringPiece will append the string
 750    * piece.  This allows Appender objects to be used directly with
 751    * Formatter.
 752    */
 753   void operator()(StringPiece sp) {
 754     push(ByteRange(sp));
 755   }
 756
 757  private:
 758   bool tryGrowChain() {
 759     assert(crtBuf_->next() == buffer_);
 760     if (growth_ == 0) {
 761       return false;
 762     }
 763
 764     buffer_->prependChain(IOBuf::create(growth_));
 765     crtBuf_ = buffer_->prev();
 766     return true;
 767   }
 768
 769   IOBuf* buffer_;
 770   IOBuf* crtBuf_;
 771   uint64_t growth_;
 772 };
 773
 774 class QueueAppender : public detail::Writable<QueueAppender> {
 775  public:
 776   /**
 777    * Create an Appender that writes to a IOBufQueue.  When we allocate
 778    * space in the queue, we grow no more than growth bytes at once
 779    * (unless you call ensure() with a bigger value yourself).
 780    */
 781   QueueAppender(IOBufQueue* queue, uint64_t growth) {
 782     reset(queue, growth);
 783   }
 784
 785   void reset(IOBufQueue* queue, uint64_t growth) {
 786     queue_ = queue;
 787     growth_ = growth;
 788   }
 789
 790   uint8_t* writableData() {
 791     return static_cast<uint8_t*>(queue_->writableTail());
 792   }
 793
 794   size_t length() const { return queue_->tailroom(); }
 795
 796   void append(size_t n) { queue_->postallocate(n); }
 797
 798   // Ensure at least n contiguous; can go above growth_, throws if
 799   // not enough room.
 800   void ensure(uint64_t n) { queue_->preallocate(n, growth_); }
 801
 802   template <class T>
 803   typename std::enable_if<std::is_arithmetic<T>::value>::type
 804   write(T value) {
 805     // We can't fail.
 806     auto p = queue_->preallocate(sizeof(T), growth_);
 807     storeUnaligned(p.first, value);
 808     queue_->postallocate(sizeof(T));
 809   }
 810
 811   using detail::Writable<QueueAppender>::pushAtMost;
 812   size_t pushAtMost(const uint8_t* buf, size_t len) {
 813     size_t remaining = len;
 814     while (remaining != 0) {
 815       auto p = queue_->preallocate(std::min(remaining, growth_),
 816                                    growth_,
 817                                    remaining);
 818       memcpy(p.first, buf, p.second);
 819       queue_->postallocate(p.second);
 820       buf += p.second;
 821       remaining -= p.second;
 822     }
 823
 824     return len;
 825   }
 826
 827   void insert(std::unique_ptr<folly::IOBuf> buf) {
 828     if (buf) {
 829       queue_->append(std::move(buf), true);
 830     }
 831   }
 832
 833  private:
 834   folly::IOBufQueue* queue_;
 835   size_t growth_;
 836 };
 837
 838 }}  // folly::io
 839
 840 #endif // FOLLY_CURSOR_H