folly/experimental/io/IOBuf.cpp

   1 /*
   2  * Copyright 2012 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 #define __STDC_LIMIT_MACROS
  18
  19 #include "folly/experimental/io/IOBuf.h"
  20
  21 #include "folly/Malloc.h"
  22 #include "folly/Likely.h"
  23
  24 #include <stdexcept>
  25 #include <assert.h>
  26 #include <stdint.h>
  27 #include <stdlib.h>
  28
  29 using std::unique_ptr;
  30
  31 namespace folly {
  32
  33 const uint32_t IOBuf::kMaxIOBufSize;
  34 // Note: Applying offsetof() to an IOBuf is legal according to C++11, since
  35 // IOBuf is a standard-layout class.  However, this isn't legal with earlier
  36 // C++ standards, which require that offsetof() only be used with POD types.
  37 //
  38 // This code compiles with g++ 4.6, but not with g++ 4.4 or earlier versions.
  39 const uint32_t IOBuf::kMaxInternalDataSize =
  40   kMaxIOBufSize - offsetof(folly::IOBuf, int_.buf);
  41
  42 IOBuf::SharedInfo::SharedInfo()
  43   : freeFn(NULL),
  44     userData(NULL) {
  45   // Use relaxed memory ordering here.  Since we are creating a new SharedInfo,
  46   // no other threads should be referring to it yet.
  47   refcount.store(1, std::memory_order_relaxed);
  48 }
  49
  50 IOBuf::SharedInfo::SharedInfo(FreeFunction fn, void* arg)
  51   : freeFn(fn),
  52     userData(arg) {
  53   // Use relaxed memory ordering here.  Since we are creating a new SharedInfo,
  54   // no other threads should be referring to it yet.
  55   refcount.store(1, std::memory_order_relaxed);
  56 }
  57
  58 void* IOBuf::operator new(size_t size) {
  59   // Since IOBuf::create() manually allocates space for some IOBuf objects
  60   // using malloc(), override operator new so that all IOBuf objects are
  61   // always allocated using malloc().  This way operator delete can always know
  62   // that free() is the correct way to deallocate the memory.
  63   void* ptr = malloc(size);
  64
  65   // operator new is not allowed to return NULL
  66   if (UNLIKELY(ptr == NULL)) {
  67     throw std::bad_alloc();
  68   }
  69
  70   return ptr;
  71 }
  72
  73 void* IOBuf::operator new(size_t size, void* ptr) {
  74   assert(size <= kMaxIOBufSize);
  75   return ptr;
  76 }
  77
  78 void IOBuf::operator delete(void* ptr) {
  79   // For small buffers, IOBuf::create() manually allocates the space for the
  80   // IOBuf object using malloc().  Therefore we override delete to ensure that
  81   // the IOBuf space is freed using free() rather than a normal delete.
  82   free(ptr);
  83 }
  84
  85 unique_ptr<IOBuf> IOBuf::create(uint32_t capacity) {
  86   // If the desired capacity is less than kMaxInternalDataSize,
  87   // just allocate a single region large enough for both the IOBuf header and
  88   // the data.
  89   if (capacity <= kMaxInternalDataSize) {
  90     void* buf = malloc(kMaxIOBufSize);
  91     if (UNLIKELY(buf == NULL)) {
  92       throw std::bad_alloc();
  93     }
  94
  95     uint8_t* bufEnd = static_cast<uint8_t*>(buf) +kMaxIOBufSize;
  96     unique_ptr<IOBuf> iobuf(new(buf) IOBuf(bufEnd));
  97     assert(iobuf->capacity() >= capacity);
  98     return iobuf;
  99   }
 100
 101   // Allocate an external buffer
 102   uint8_t* buf;
 103   SharedInfo* sharedInfo;
 104   uint32_t actualCapacity;
 105   allocExtBuffer(capacity, &buf, &sharedInfo, &actualCapacity);
 106
 107   // Allocate the IOBuf header
 108   try {
 109     return unique_ptr<IOBuf>(new IOBuf(kExtAllocated, 0,
 110                                        buf, actualCapacity,
 111                                        buf, 0,
 112                                        sharedInfo));
 113   } catch (...) {
 114     free(buf);
 115     throw;
 116   }
 117 }
 118
 119 unique_ptr<IOBuf> IOBuf::takeOwnership(void* buf, uint32_t capacity,
 120                                        FreeFunction freeFn,
 121                                        void* userData,
 122                                        bool freeOnError) {
 123   SharedInfo* sharedInfo = NULL;
 124   try {
 125     sharedInfo = new SharedInfo(freeFn, userData);
 126
 127     uint8_t* bufPtr = static_cast<uint8_t*>(buf);
 128     return unique_ptr<IOBuf>(new IOBuf(kExtUserSupplied, kFlagFreeSharedInfo,
 129                                        bufPtr, capacity,
 130                                        bufPtr, capacity,
 131                                        sharedInfo));
 132   } catch (...) {
 133     delete sharedInfo;
 134     if (freeOnError) {
 135       if (freeFn) {
 136         try {
 137           freeFn(buf, userData);
 138         } catch (...) {
 139           // The user's free function is not allowed to throw.
 140           abort();
 141         }
 142       } else {
 143         free(buf);
 144       }
 145     }
 146     throw;
 147   }
 148 }
 149
 150 unique_ptr<IOBuf> IOBuf::wrapBuffer(const void* buf, uint32_t capacity) {
 151   // We cast away the const-ness of the buffer here.
 152   // This is okay since IOBuf users must use unshare() to create a copy of
 153   // this buffer before writing to the buffer.
 154   uint8_t* bufPtr = static_cast<uint8_t*>(const_cast<void*>(buf));
 155   return unique_ptr<IOBuf>(new IOBuf(kExtUserSupplied, kFlagUserOwned,
 156                                      bufPtr, capacity,
 157                                      bufPtr, capacity,
 158                                      NULL));
 159 }
 160
 161 IOBuf::IOBuf(uint8_t* end)
 162   : next_(this),
 163     prev_(this),
 164     data_(int_.buf),
 165     length_(0),
 166     flags_(0) {
 167   int_.capacity = end - int_.buf;
 168   assert(int_.capacity <= kMaxInternalDataSize);
 169 }
 170
 171 IOBuf::IOBuf(ExtBufTypeEnum type,
 172              uint32_t flags,
 173              uint8_t* buf,
 174              uint32_t capacity,
 175              uint8_t* data,
 176              uint32_t length,
 177              SharedInfo* sharedInfo)
 178   : next_(this),
 179     prev_(this),
 180     data_(data),
 181     length_(length),
 182     flags_(kFlagExt | flags) {
 183   ext_.capacity = capacity;
 184   ext_.type = type;
 185   ext_.buf = buf;
 186   ext_.sharedInfo = sharedInfo;
 187
 188   assert(data >= buf);
 189   assert(data + length <= buf + capacity);
 190   assert(static_cast<bool>(flags & kFlagUserOwned) ==
 191          (sharedInfo == NULL));
 192 }
 193
 194 IOBuf::~IOBuf() {
 195   // Destroying an IOBuf destroys the entire chain.
 196   // Users of IOBuf should only explicitly delete the head of any chain.
 197   // The other elements in the chain will be automatically destroyed.
 198   while (next_ != this) {
 199     // Since unlink() returns unique_ptr() and we don't store it,
 200     // it will automatically delete the unlinked element.
 201     (void)next_->unlink();
 202   }
 203
 204   if (flags_ & kFlagExt) {
 205     decrementRefcount();
 206   }
 207 }
 208
 209 bool IOBuf::empty() const {
 210   const IOBuf* current = this;
 211   do {
 212     if (current->length() != 0) {
 213       return false;
 214     }
 215     current = current->next_;
 216   } while (current != this);
 217   return true;
 218 }
 219
 220 uint32_t IOBuf::countChainElements() const {
 221   uint32_t numElements = 1;
 222   for (IOBuf* current = next_; current != this; current = current->next_) {
 223     ++numElements;
 224   }
 225   return numElements;
 226 }
 227
 228 uint64_t IOBuf::computeChainDataLength() const {
 229   uint64_t fullLength = length_;
 230   for (IOBuf* current = next_; current != this; current = current->next_) {
 231     fullLength += current->length_;
 232   }
 233   return fullLength;
 234 }
 235
 236 void IOBuf::prependChain(unique_ptr<IOBuf>&& iobuf) {
 237   // Take ownership of the specified IOBuf
 238   IOBuf* other = iobuf.release();
 239
 240   // Remember the pointer to the tail of the other chain
 241   IOBuf* otherTail = other->prev_;
 242
 243   // Hook up prev_->next_ to point at the start of the other chain,
 244   // and other->prev_ to point at prev_
 245   prev_->next_ = other;
 246   other->prev_ = prev_;
 247
 248   // Hook up otherTail->next_ to point at us,
 249   // and prev_ to point back at otherTail,
 250   otherTail->next_ = this;
 251   prev_ = otherTail;
 252 }
 253
 254 unique_ptr<IOBuf> IOBuf::clone() const {
 255   unique_ptr<IOBuf> newHead(cloneOne());
 256
 257   for (IOBuf* current = next_; current != this; current = current->next_) {
 258     newHead->prependChain(current->cloneOne());
 259   }
 260
 261   return newHead;
 262 }
 263
 264 unique_ptr<IOBuf> IOBuf::cloneOne() const {
 265   if (flags_ & kFlagExt) {
 266     unique_ptr<IOBuf> iobuf(new IOBuf(static_cast<ExtBufTypeEnum>(ext_.type),
 267                                       flags_, ext_.buf, ext_.capacity,
 268                                       data_, length_,
 269                                       ext_.sharedInfo));
 270     if (ext_.sharedInfo) {
 271       ext_.sharedInfo->refcount.fetch_add(1, std::memory_order_acq_rel);
 272     }
 273     return iobuf;
 274   } else {
 275     // We have an internal data buffer that cannot be shared
 276     // Allocate a new IOBuf and copy the data into it.
 277     unique_ptr<IOBuf> iobuf(IOBuf::create(int_.capacity));
 278     assert((iobuf->flags_ & kFlagExt) == 0);
 279     iobuf->data_ += headroom();
 280     memcpy(iobuf->data_, data_, length_);
 281     iobuf->length_ = length_;
 282     return iobuf;
 283   }
 284 }
 285
 286 void IOBuf::unshareOneSlow() {
 287   // Internal buffers are always unshared, so unshareOneSlow() can only be
 288   // called for external buffers
 289   assert(flags_ & kFlagExt);
 290
 291   // Allocate a new buffer for the data
 292   uint8_t* buf;
 293   SharedInfo* sharedInfo;
 294   uint32_t actualCapacity;
 295   allocExtBuffer(ext_.capacity, &buf, &sharedInfo, &actualCapacity);
 296
 297   // Copy the data
 298   // Maintain the same amount of headroom.  Since we maintained the same
 299   // minimum capacity we also maintain at least the same amount of tailroom.
 300   uint32_t headlen = headroom();
 301   memcpy(buf + headlen, data_, length_);
 302
 303   // Release our reference on the old buffer
 304   decrementRefcount();
 305   // Make sure kFlagExt is set, and kFlagUserOwned and kFlagFreeSharedInfo
 306   // are not set.
 307   flags_ = kFlagExt;
 308
 309   // Update the buffer pointers to point to the new buffer
 310   data_ = buf + headlen;
 311   ext_.buf = buf;
 312   ext_.sharedInfo = sharedInfo;
 313 }
 314
 315 void IOBuf::unshareChained() {
 316   // unshareChained() should only be called if we are part of a chain of
 317   // multiple IOBufs.  The caller should have already verified this.
 318   assert(isChained());
 319
 320   IOBuf* current = this;
 321   while (true) {
 322     if (current->isSharedOne()) {
 323       // we have to unshare
 324       break;
 325     }
 326
 327     current = current->next_;
 328     if (current == this) {
 329       // None of the IOBufs in the chain are shared,
 330       // so return without doing anything
 331       return;
 332     }
 333   }
 334
 335   // We have to unshare.  Let coalesceSlow() do the work.
 336   coalesceSlow();
 337 }
 338
 339 void IOBuf::coalesceSlow(size_t maxLength) {
 340   // coalesceSlow() should only be called if we are part of a chain of multiple
 341   // IOBufs.  The caller should have already verified this.
 342   assert(isChained());
 343   assert(length_ < maxLength);
 344
 345   // Compute the length of the entire chain
 346   uint64_t newLength = 0;
 347   IOBuf* end = this;
 348   do {
 349     newLength += end->length_;
 350     end = end->next_;
 351   } while (newLength < maxLength && end != this);
 352
 353   uint64_t newHeadroom = headroom();
 354   uint64_t newTailroom = end->prev_->tailroom();
 355   uint64_t newCapacity = newLength + newHeadroom + newTailroom;
 356   if (newCapacity > UINT32_MAX) {
 357     throw std::overflow_error("IOBuf chain too large to coalesce");
 358   }
 359
 360   // Allocate space for the coalesced buffer.
 361   // We always convert to an external buffer, even if we happened to be an
 362   // internal buffer before.
 363   uint8_t* newBuf;
 364   SharedInfo* newInfo;
 365   uint32_t actualCapacity;
 366   allocExtBuffer(newCapacity, &newBuf, &newInfo, &actualCapacity);
 367
 368   // Copy the data into the new buffer
 369   uint8_t* newData = newBuf + newHeadroom;
 370   uint8_t* p = newData;
 371   IOBuf* current = this;
 372   do {
 373     memcpy(p, current->data_, current->length_);
 374     p += current->length_;
 375     current = current->next_;
 376   } while (current != end);
 377
 378   // Point at the new buffer
 379   if (flags_ & kFlagExt) {
 380     decrementRefcount();
 381   }
 382
 383   // Make sure kFlagExt is set, and kFlagUserOwned and kFlagFreeSharedInfo
 384   // are not set.
 385   flags_ = kFlagExt;
 386
 387   ext_.capacity = actualCapacity;
 388   ext_.type = kExtAllocated;
 389   ext_.buf = newBuf;
 390   ext_.sharedInfo = newInfo;
 391   data_ = newData;
 392   length_ = newLength;
 393
 394   // Separate from the rest of our chain.
 395   // Since we don't store the unique_ptr returned by separateChain(),
 396   // this will immediately delete the returned subchain.
 397   (void)separateChain(next_, end->prev_);
 398
 399   // We should be only element left in the chain now
 400   assert(length_ >= maxLength || !isChained());
 401 }
 402
 403 void IOBuf::decrementRefcount() {
 404   assert(flags_ & kFlagExt);
 405
 406   // Externally owned buffers don't have a SharedInfo object and aren't managed
 407   // by the reference count
 408   if (flags_ & kFlagUserOwned) {
 409     assert(ext_.sharedInfo == NULL);
 410     return;
 411   }
 412
 413   // Decrement the refcount
 414   uint32_t newcnt = ext_.sharedInfo->refcount.fetch_sub(
 415       1, std::memory_order_acq_rel);
 416   // Note that fetch_sub() returns the value before we decremented.
 417   // If it is 1, we were the only remaining user; if it is greater there are
 418   // still other users.
 419   if (newcnt > 1) {
 420     return;
 421   }
 422
 423   // We were the last user.  Free the buffer
 424   if (ext_.sharedInfo->freeFn != NULL) {
 425     try {
 426       ext_.sharedInfo->freeFn(ext_.buf, ext_.sharedInfo->userData);
 427     } catch (...) {
 428       // The user's free function should never throw.  Otherwise we might
 429       // throw from the IOBuf destructor.  Other code paths like coalesce()
 430       // also assume that decrementRefcount() cannot throw.
 431       abort();
 432     }
 433   } else {
 434     free(ext_.buf);
 435   }
 436
 437   // Free the SharedInfo if it was allocated separately.
 438   //
 439   // This is only used by takeOwnership().
 440   //
 441   // To avoid this special case handling in decrementRefcount(), we could have
 442   // takeOwnership() set a custom freeFn() that calls the user's free function
 443   // then frees the SharedInfo object.  (This would require that
 444   // takeOwnership() store the user's free function with its allocated
 445   // SharedInfo object.)  However, handling this specially with a flag seems
 446   // like it shouldn't be problematic.
 447   if (flags_ & kFlagFreeSharedInfo) {
 448     delete ext_.sharedInfo;
 449   }
 450 }
 451
 452 void IOBuf::reserveSlow(uint32_t minHeadroom, uint32_t minTailroom) {
 453   size_t newCapacity = (size_t)length_ + minHeadroom + minTailroom;
 454   CHECK_LT(newCapacity, UINT32_MAX);
 455
 456   // We'll need to reallocate the buffer.
 457   // There are a few options.
 458   // - If we have enough total room, move the data around in the buffer
 459   //   and adjust the data_ pointer.
 460   // - If we're using an internal buffer, we'll switch to an external
 461   //   buffer with enough headroom and tailroom.
 462   // - If we have enough headroom (headroom() >= minHeadroom) but not too much
 463   //   (so we don't waste memory), we can try one of two things, depending on
 464   //   whether we use jemalloc or not:
 465   //   - If using jemalloc, we can try to expand in place, avoiding a memcpy()
 466   //   - If not using jemalloc and we don't have too much to copy,
 467   //     we'll use realloc() (note that realloc might have to copy
 468   //     headroom + data + tailroom, see smartRealloc in folly/Malloc.h)
 469   // - Otherwise, bite the bullet and reallocate.
 470   if (headroom() + tailroom() >= minHeadroom + minTailroom) {
 471     uint8_t* newData = writableBuffer() + minHeadroom;
 472     memmove(newData, data_, length_);
 473     data_ = newData;
 474     return;
 475   }
 476
 477   size_t newAllocatedCapacity = goodExtBufferSize(newCapacity);
 478   uint8_t* newBuffer = nullptr;
 479   uint32_t newHeadroom = 0;
 480   uint32_t oldHeadroom = headroom();
 481
 482   if ((flags_ & kFlagExt) && length_ != 0 && oldHeadroom >= minHeadroom) {
 483     if (usingJEMalloc()) {
 484       size_t headSlack = oldHeadroom - minHeadroom;
 485       // We assume that tailroom is more useful and more important than
 486       // tailroom (not least because realloc / rallocm allow us to grow the
 487       // buffer at the tail, but not at the head)  So, if we have more headroom
 488       // than we need, we consider that "wasted".  We arbitrarily define "too
 489       // much" headroom to be 25% of the capacity.
 490       if (headSlack * 4 <= newCapacity) {
 491         size_t allocatedCapacity = capacity() + sizeof(SharedInfo);
 492         void* p = ext_.buf;
 493         if (allocatedCapacity >= jemallocMinInPlaceExpandable) {
 494           int r = rallocm(&p, &newAllocatedCapacity, newAllocatedCapacity,
 495                           0, ALLOCM_NO_MOVE);
 496           if (r == ALLOCM_SUCCESS) {
 497             newBuffer = static_cast<uint8_t*>(p);
 498             newHeadroom = oldHeadroom;
 499           } else if (r == ALLOCM_ERR_OOM) {
 500             // shouldn't happen as we don't actually allocate new memory
 501             // (due to ALLOCM_NO_MOVE)
 502             throw std::bad_alloc();
 503           }
 504           // if ALLOCM_ERR_NOT_MOVED, do nothing, fall back to
 505           // malloc/memcpy/free
 506         }
 507       }
 508     } else {  // Not using jemalloc
 509       size_t copySlack = capacity() - length_;
 510       if (copySlack * 2 <= length_) {
 511         void* p = realloc(ext_.buf, newAllocatedCapacity);
 512         if (UNLIKELY(p == nullptr)) {
 513           throw std::bad_alloc();
 514         }
 515         newBuffer = static_cast<uint8_t*>(p);
 516         newHeadroom = oldHeadroom;
 517       }
 518     }
 519   }
 520
 521   // None of the previous reallocation strategies worked (or we're using
 522   // an internal buffer).  malloc/copy/free.
 523   if (newBuffer == nullptr) {
 524     void* p = malloc(newAllocatedCapacity);
 525     if (UNLIKELY(p == nullptr)) {
 526       throw std::bad_alloc();
 527     }
 528     newBuffer = static_cast<uint8_t*>(p);
 529     memcpy(newBuffer + minHeadroom, data_, length_);
 530     if (flags_ & kFlagExt) {
 531       free(ext_.buf);
 532     }
 533     newHeadroom = minHeadroom;
 534   }
 535
 536   SharedInfo* info;
 537   uint32_t cap;
 538   initExtBuffer(newBuffer, newAllocatedCapacity, &info, &cap);
 539
 540   flags_ = kFlagExt;
 541
 542   ext_.capacity = cap;
 543   ext_.type = kExtAllocated;
 544   ext_.buf = newBuffer;
 545   ext_.sharedInfo = info;
 546   data_ = newBuffer + newHeadroom;
 547   // length_ is unchanged
 548 }
 549
 550 void IOBuf::allocExtBuffer(uint32_t minCapacity,
 551                            uint8_t** bufReturn,
 552                            SharedInfo** infoReturn,
 553                            uint32_t* capacityReturn) {
 554   size_t mallocSize = goodExtBufferSize(minCapacity);
 555   uint8_t* buf = static_cast<uint8_t*>(malloc(mallocSize));
 556   if (UNLIKELY(buf == NULL)) {
 557     throw std::bad_alloc();
 558   }
 559   initExtBuffer(buf, mallocSize, infoReturn, capacityReturn);
 560   *bufReturn = buf;
 561 }
 562
 563 size_t IOBuf::goodExtBufferSize(uint32_t minCapacity) {
 564   // Determine how much space we should allocate.  We'll store the SharedInfo
 565   // for the external buffer just after the buffer itself.  (We store it just
 566   // after the buffer rather than just before so that the code can still just
 567   // use free(ext_.buf) to free the buffer.)
 568   size_t minSize = static_cast<size_t>(minCapacity) + sizeof(SharedInfo);
 569   // Add room for padding so that the SharedInfo will be aligned on an 8-byte
 570   // boundary.
 571   minSize = (minSize + 7) & ~7;
 572
 573   // Use goodMallocSize() to bump up the capacity to a decent size to request
 574   // from malloc, so we can use all of the space that malloc will probably give
 575   // us anyway.
 576   return goodMallocSize(minSize);
 577 }
 578
 579 void IOBuf::initExtBuffer(uint8_t* buf, size_t mallocSize,
 580                           SharedInfo** infoReturn,
 581                           uint32_t* capacityReturn) {
 582   // Find the SharedInfo storage at the end of the buffer
 583   // and construct the SharedInfo.
 584   uint8_t* infoStart = (buf + mallocSize) - sizeof(SharedInfo);
 585   SharedInfo* sharedInfo = new(infoStart) SharedInfo;
 586
 587   size_t actualCapacity = infoStart - buf;
 588   // On the unlikely possibility that the actual capacity is larger than can
 589   // fit in a uint32_t after adding room for the refcount and calling
 590   // goodMallocSize(), truncate downwards if necessary.
 591   if (actualCapacity >= UINT32_MAX) {
 592     *capacityReturn = UINT32_MAX;
 593   } else {
 594     *capacityReturn = actualCapacity;
 595   }
 596
 597   *infoReturn = sharedInfo;
 598 }
 599
 600 } // folly