Commit 928ed23a authored by Nathan Bronson's avatar Nathan Bronson Committed by Facebook Github Bot

clang-format folly/io subdir

Summary: Automated reformat of folly/io subdir in preparation for other changes

Reviewed By: yfeldblum

Differential Revision: D8559473

fbshipit-source-id: c94d9c05ee77e25b6a61ee7a47b472ccea1f18f3
parent 38589135
...@@ -41,8 +41,8 @@ void Appender::vprintf(const char* fmt, va_list ap) { ...@@ -41,8 +41,8 @@ void Appender::vprintf(const char* fmt, va_list ap) {
}; };
// First try writing into our available data space. // First try writing into our available data space.
int ret = vsnprintf(reinterpret_cast<char*>(writableData()), length(), int ret =
fmt, ap); vsnprintf(reinterpret_cast<char*>(writableData()), length(), fmt, ap);
if (ret < 0) { if (ret < 0) {
throw std::runtime_error("error formatting printf() data"); throw std::runtime_error("error formatting printf() data");
} }
...@@ -58,16 +58,17 @@ void Appender::vprintf(const char* fmt, va_list ap) { ...@@ -58,16 +58,17 @@ void Appender::vprintf(const char* fmt, va_list ap) {
// There wasn't enough room for the data. // There wasn't enough room for the data.
// Allocate more room, and then retry. // Allocate more room, and then retry.
ensure(len + 1); ensure(len + 1);
ret = vsnprintf(reinterpret_cast<char*>(writableData()), length(), ret =
fmt, apCopy); vsnprintf(reinterpret_cast<char*>(writableData()), length(), fmt, apCopy);
if (ret < 0) { if (ret < 0) {
throw std::runtime_error("error formatting printf() data"); throw std::runtime_error("error formatting printf() data");
} }
len = size_t(ret); len = size_t(ret);
if (len >= length()) { if (len >= length()) {
// This shouldn't ever happen. // This shouldn't ever happen.
throw std::runtime_error("unexpectedly out of buffer space on second " throw std::runtime_error(
"vsnprintf() attmept"); "unexpectedly out of buffer space on second "
"vsnprintf() attmept");
} }
append(len); append(len);
} }
......
...@@ -55,7 +55,9 @@ namespace detail { ...@@ -55,7 +55,9 @@ namespace detail {
template <class Derived, class BufType> template <class Derived, class BufType>
class CursorBase { class CursorBase {
// Make all the templated classes friends for copy constructor. // Make all the templated classes friends for copy constructor.
template <class D, typename B> friend class CursorBase; template <class D, typename B>
friend class CursorBase;
public: public:
explicit CursorBase(BufType* buf) : crtBuf_(buf), buffer_(buf) { explicit CursorBase(BufType* buf) : crtBuf_(buf), buffer_(buf) {
if (crtBuf_) { if (crtBuf_) {
...@@ -166,7 +168,8 @@ class CursorBase { ...@@ -166,7 +168,8 @@ class CursorBase {
// We are at the end of a buffer, but it isn't the last buffer. // We are at the end of a buffer, but it isn't the last buffer.
// We might still be at the end if the remaining buffers in the chain are // We might still be at the end if the remaining buffers in the chain are
// empty. // empty.
const IOBuf* buf = crtBuf_->next();; const IOBuf* buf = crtBuf_->next();
;
while (buf != buffer_) { while (buf != buffer_) {
if (buf->length() > 0) { if (buf->length() > 0) {
return false; return false;
...@@ -516,7 +519,7 @@ class CursorBase { ...@@ -516,7 +519,7 @@ class CursorBase {
* Return the distance between two cursors. * Return the distance between two cursors.
*/ */
size_t operator-(const CursorBase& other) const { size_t operator-(const CursorBase& other) const {
BufType *otherBuf = other.crtBuf_; BufType* otherBuf = other.crtBuf_;
size_t len = 0; size_t len = 0;
if (otherBuf != crtBuf_) { if (otherBuf != crtBuf_) {
...@@ -572,7 +575,7 @@ class CursorBase { ...@@ -572,7 +575,7 @@ class CursorBase {
(uint64_t)(crtEnd_ - crtBegin_) == crtBuf_->length()); (uint64_t)(crtEnd_ - crtBegin_) == crtBuf_->length());
} }
~CursorBase() { } ~CursorBase() {}
BufType* head() { BufType* head() {
return buffer_; return buffer_;
...@@ -629,7 +632,7 @@ class CursorBase { ...@@ -629,7 +632,7 @@ class CursorBase {
} }
void readFixedStringSlow(std::string* str, size_t len) { void readFixedStringSlow(std::string* str, size_t len) {
for (size_t available; (available = length()) < len; ) { for (size_t available; (available = length()) < len;) {
str->append(reinterpret_cast<const char*>(data()), available); str->append(reinterpret_cast<const char*>(data()), available);
if (UNLIKELY(!tryAdvanceBuffer())) { if (UNLIKELY(!tryAdvanceBuffer())) {
throw_exception<std::out_of_range>("string underflow"); throw_exception<std::out_of_range>("string underflow");
...@@ -650,7 +653,7 @@ class CursorBase { ...@@ -650,7 +653,7 @@ class CursorBase {
} }
uint8_t* p = reinterpret_cast<uint8_t*>(buf); uint8_t* p = reinterpret_cast<uint8_t*>(buf);
size_t copied = 0; size_t copied = 0;
for (size_t available; (available = length()) < len; ) { for (size_t available; (available = length()) < len;) {
memcpy(p, data(), available); memcpy(p, data(), available);
copied += available; copied += available;
if (UNLIKELY(!tryAdvanceBuffer())) { if (UNLIKELY(!tryAdvanceBuffer())) {
...@@ -673,7 +676,7 @@ class CursorBase { ...@@ -673,7 +676,7 @@ class CursorBase {
size_t skipAtMostSlow(size_t len) { size_t skipAtMostSlow(size_t len) {
size_t skipped = 0; size_t skipped = 0;
for (size_t available; (available = length()) < len; ) { for (size_t available; (available = length()) < len;) {
skipped += available; skipped += available;
if (UNLIKELY(!tryAdvanceBuffer())) { if (UNLIKELY(!tryAdvanceBuffer())) {
return skipped; return skipped;
...@@ -710,8 +713,7 @@ class CursorBase { ...@@ -710,8 +713,7 @@ class CursorBase {
} }
} }
void advanceDone() { void advanceDone() {}
}
}; };
} // namespace detail } // namespace detail
...@@ -719,11 +721,11 @@ class CursorBase { ...@@ -719,11 +721,11 @@ class CursorBase {
class Cursor : public detail::CursorBase<Cursor, const IOBuf> { class Cursor : public detail::CursorBase<Cursor, const IOBuf> {
public: public:
explicit Cursor(const IOBuf* buf) explicit Cursor(const IOBuf* buf)
: detail::CursorBase<Cursor, const IOBuf>(buf) {} : detail::CursorBase<Cursor, const IOBuf>(buf) {}
template <class OtherDerived, class OtherBuf> template <class OtherDerived, class OtherBuf>
explicit Cursor(const detail::CursorBase<OtherDerived, OtherBuf>& cursor) explicit Cursor(const detail::CursorBase<OtherDerived, OtherBuf>& cursor)
: detail::CursorBase<Cursor, const IOBuf>(cursor) {} : detail::CursorBase<Cursor, const IOBuf>(cursor) {}
}; };
namespace detail { namespace detail {
...@@ -732,8 +734,7 @@ template <class Derived> ...@@ -732,8 +734,7 @@ template <class Derived>
class Writable { class Writable {
public: public:
template <class T> template <class T>
typename std::enable_if<std::is_arithmetic<T>::value>::type typename std::enable_if<std::is_arithmetic<T>::value>::type write(T value) {
write(T value) {
const uint8_t* u8 = reinterpret_cast<const uint8_t*>(&value); const uint8_t* u8 = reinterpret_cast<const uint8_t*>(&value);
Derived* d = static_cast<Derived*>(this); Derived* d = static_cast<Derived*>(this);
d->push(u8, sizeof(T)); d->push(u8, sizeof(T));
...@@ -782,7 +783,7 @@ class Writable { ...@@ -782,7 +783,7 @@ class Writable {
size_t pushAtMost(Cursor cursor, size_t len) { size_t pushAtMost(Cursor cursor, size_t len) {
size_t written = 0; size_t written = 0;
for(;;) { for (;;) {
auto currentBuffer = cursor.peekBytes(); auto currentBuffer = cursor.peekBytes();
const uint8_t* crtData = currentBuffer.data(); const uint8_t* crtData = currentBuffer.data();
size_t available = currentBuffer.size(); size_t available = currentBuffer.size();
...@@ -808,25 +809,21 @@ class Writable { ...@@ -808,25 +809,21 @@ class Writable {
} // namespace detail } // namespace detail
enum class CursorAccess { enum class CursorAccess { PRIVATE, UNSHARE };
PRIVATE,
UNSHARE
};
template <CursorAccess access> template <CursorAccess access>
class RWCursor class RWCursor : public detail::CursorBase<RWCursor<access>, IOBuf>,
: public detail::CursorBase<RWCursor<access>, IOBuf>, public detail::Writable<RWCursor<access>> {
public detail::Writable<RWCursor<access>> {
friend class detail::CursorBase<RWCursor<access>, IOBuf>; friend class detail::CursorBase<RWCursor<access>, IOBuf>;
public: public:
explicit RWCursor(IOBuf* buf) explicit RWCursor(IOBuf* buf)
: detail::CursorBase<RWCursor<access>, IOBuf>(buf), : detail::CursorBase<RWCursor<access>, IOBuf>(buf), maybeShared_(true) {}
maybeShared_(true) {}
template <class OtherDerived, class OtherBuf> template <class OtherDerived, class OtherBuf>
explicit RWCursor(const detail::CursorBase<OtherDerived, OtherBuf>& cursor) explicit RWCursor(const detail::CursorBase<OtherDerived, OtherBuf>& cursor)
: detail::CursorBase<RWCursor<access>, IOBuf>(cursor), : detail::CursorBase<RWCursor<access>, IOBuf>(cursor),
maybeShared_(true) {} maybeShared_(true) {}
/** /**
* Gather at least n bytes contiguously into the current buffer, * Gather at least n bytes contiguously into the current buffer,
* by coalescing subsequent buffers from the chain as necessary. * by coalescing subsequent buffers from the chain as necessary.
...@@ -972,10 +969,7 @@ typedef RWCursor<CursorAccess::UNSHARE> RWUnshareCursor; ...@@ -972,10 +969,7 @@ typedef RWCursor<CursorAccess::UNSHARE> RWUnshareCursor;
class Appender : public detail::Writable<Appender> { class Appender : public detail::Writable<Appender> {
public: public:
Appender(IOBuf* buf, uint64_t growth) Appender(IOBuf* buf, uint64_t growth)
: buffer_(buf), : buffer_(buf), crtBuf_(buf->prev()), growth_(growth) {}
crtBuf_(buf->prev()),
growth_(growth) {
}
uint8_t* writableData() { uint8_t* writableData() {
return crtBuf_->writableTail(); return crtBuf_->writableTail();
...@@ -1074,7 +1068,7 @@ class Appender : public detail::Writable<Appender> { ...@@ -1074,7 +1068,7 @@ class Appender : public detail::Writable<Appender> {
* This method may throw exceptions on error. * This method may throw exceptions on error.
*/ */
void printf(FOLLY_PRINTF_FORMAT const char* fmt, ...) void printf(FOLLY_PRINTF_FORMAT const char* fmt, ...)
FOLLY_PRINTF_FORMAT_ATTR(2, 3); FOLLY_PRINTF_FORMAT_ATTR(2, 3);
void vprintf(const char* fmt, va_list ap); void vprintf(const char* fmt, va_list ap);
......
This diff is collapsed.
This diff is collapsed.
...@@ -35,8 +35,7 @@ const size_t MAX_PACK_COPY = 4096; ...@@ -35,8 +35,7 @@ const size_t MAX_PACK_COPY = 4096;
/** /**
* Convenience function to append chain src to chain dst. * Convenience function to append chain src to chain dst.
*/ */
void void appendToChain(unique_ptr<IOBuf>& dst, unique_ptr<IOBuf>&& src, bool pack) {
appendToChain(unique_ptr<IOBuf>& dst, unique_ptr<IOBuf>&& src, bool pack) {
if (dst == nullptr) { if (dst == nullptr) {
dst = std::move(src); dst = std::move(src);
} else { } else {
...@@ -109,8 +108,7 @@ IOBufQueue& IOBufQueue::operator=(IOBufQueue&& other) { ...@@ -109,8 +108,7 @@ IOBufQueue& IOBufQueue::operator=(IOBufQueue&& other) {
return *this; return *this;
} }
std::pair<void*, uint64_t> std::pair<void*, uint64_t> IOBufQueue::headroom() {
IOBufQueue::headroom() {
// Note, headroom is independent from the tail, so we don't need to flush the // Note, headroom is independent from the tail, so we don't need to flush the
// cache. // cache.
if (head_) { if (head_) {
...@@ -120,8 +118,7 @@ IOBufQueue::headroom() { ...@@ -120,8 +118,7 @@ IOBufQueue::headroom() {
} }
} }
void void IOBufQueue::markPrepended(uint64_t n) {
IOBufQueue::markPrepended(uint64_t n) {
if (n == 0) { if (n == 0) {
return; return;
} }
...@@ -132,8 +129,7 @@ IOBufQueue::markPrepended(uint64_t n) { ...@@ -132,8 +129,7 @@ IOBufQueue::markPrepended(uint64_t n) {
chainLength_ += n; chainLength_ += n;
} }
void void IOBufQueue::prepend(const void* buf, uint64_t n) {
IOBufQueue::prepend(const void* buf, uint64_t n) {
// We're not touching the tail, so we don't need to flush the cache. // We're not touching the tail, so we don't need to flush the cache.
auto hroom = head_->headroom(); auto hroom = head_->headroom();
if (!head_ || hroom < n) { if (!head_ || hroom < n) {
...@@ -144,8 +140,7 @@ IOBufQueue::prepend(const void* buf, uint64_t n) { ...@@ -144,8 +140,7 @@ IOBufQueue::prepend(const void* buf, uint64_t n) {
chainLength_ += n; chainLength_ += n;
} }
void void IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) {
IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) {
if (!buf) { if (!buf) {
return; return;
} }
...@@ -156,8 +151,7 @@ IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) { ...@@ -156,8 +151,7 @@ IOBufQueue::append(unique_ptr<IOBuf>&& buf, bool pack) {
appendToChain(head_, std::move(buf), pack); appendToChain(head_, std::move(buf), pack);
} }
void void IOBufQueue::append(IOBufQueue& other, bool pack) {
IOBufQueue::append(IOBufQueue& other, bool pack) {
if (!other.head_) { if (!other.head_) {
return; return;
} }
...@@ -175,16 +169,16 @@ IOBufQueue::append(IOBufQueue& other, bool pack) { ...@@ -175,16 +169,16 @@ IOBufQueue::append(IOBufQueue& other, bool pack) {
other.chainLength_ = 0; other.chainLength_ = 0;
} }
void void IOBufQueue::append(const void* buf, size_t len) {
IOBufQueue::append(const void* buf, size_t len) {
auto guard = updateGuard(); auto guard = updateGuard();
auto src = static_cast<const uint8_t*>(buf); auto src = static_cast<const uint8_t*>(buf);
while (len != 0) { while (len != 0) {
if ((head_ == nullptr) || head_->prev()->isSharedOne() || if ((head_ == nullptr) || head_->prev()->isSharedOne() ||
(head_->prev()->tailroom() == 0)) { (head_->prev()->tailroom() == 0)) {
appendToChain(head_, appendToChain(
IOBuf::create(std::max(MIN_ALLOC_SIZE, head_,
std::min(len, MAX_ALLOC_SIZE))), IOBuf::create(
std::max(MIN_ALLOC_SIZE, std::min(len, MAX_ALLOC_SIZE))),
false); false);
} }
IOBuf* last = head_->prev(); IOBuf* last = head_->prev();
...@@ -197,8 +191,7 @@ IOBufQueue::append(const void* buf, size_t len) { ...@@ -197,8 +191,7 @@ IOBufQueue::append(const void* buf, size_t len) {
} }
} }
void void IOBufQueue::wrapBuffer(const void* buf, size_t len, uint64_t blockSize) {
IOBufQueue::wrapBuffer(const void* buf, size_t len, uint64_t blockSize) {
auto src = static_cast<const uint8_t*>(buf); auto src = static_cast<const uint8_t*>(buf);
while (len != 0) { while (len != 0) {
size_t n = std::min(len, size_t(blockSize)); size_t n = std::min(len, size_t(blockSize));
...@@ -208,9 +201,10 @@ IOBufQueue::wrapBuffer(const void* buf, size_t len, uint64_t blockSize) { ...@@ -208,9 +201,10 @@ IOBufQueue::wrapBuffer(const void* buf, size_t len, uint64_t blockSize) {
} }
} }
pair<void*,uint64_t> pair<void*, uint64_t> IOBufQueue::preallocateSlow(
IOBufQueue::preallocateSlow(uint64_t min, uint64_t newAllocationSize, uint64_t min,
uint64_t max) { uint64_t newAllocationSize,
uint64_t max) {
// Avoid grabbing update guard, since we're manually setting the cache ptrs. // Avoid grabbing update guard, since we're manually setting the cache ptrs.
flushCache(); flushCache();
// Allocate a new buffer of the requested max size. // Allocate a new buffer of the requested max size.
......
...@@ -291,16 +291,15 @@ class IOBufQueue { ...@@ -291,16 +291,15 @@ class IOBufQueue {
* releasing the buffers), if possible. If pack is false, we leave * releasing the buffers), if possible. If pack is false, we leave
* the chain topology unchanged. * the chain topology unchanged.
*/ */
void append(std::unique_ptr<folly::IOBuf>&& buf, void append(std::unique_ptr<folly::IOBuf>&& buf, bool pack = false);
bool pack=false);
/** /**
* Add a queue to the end of this queue. The queue takes ownership of * Add a queue to the end of this queue. The queue takes ownership of
* all buffers from the other queue. * all buffers from the other queue.
*/ */
void append(IOBufQueue& other, bool pack=false); void append(IOBufQueue& other, bool pack = false);
void append(IOBufQueue&& other, bool pack=false) { void append(IOBufQueue&& other, bool pack = false) {
append(other, pack); // call lvalue reference overload, above append(other, pack); // call lvalue reference overload, above
} }
/** /**
...@@ -329,8 +328,10 @@ class IOBufQueue { ...@@ -329,8 +328,10 @@ class IOBufQueue {
* Every buffer except for the last will wrap exactly blockSize bytes. * Every buffer except for the last will wrap exactly blockSize bytes.
* Importantly, this method may be used to wrap buffers larger than 4GB. * Importantly, this method may be used to wrap buffers larger than 4GB.
*/ */
void wrapBuffer(const void* buf, size_t len, void wrapBuffer(
uint64_t blockSize=(1U << 31)); // default block size: 2GB const void* buf,
size_t len,
uint64_t blockSize = (1U << 31)); // default block size: 2GB
/** /**
* Obtain a writable block of contiguous bytes at the end of this * Obtain a writable block of contiguous bytes at the end of this
...@@ -352,9 +353,10 @@ class IOBufQueue { ...@@ -352,9 +353,10 @@ class IOBufQueue {
* callback, tell the application how much of the buffer they've * callback, tell the application how much of the buffer they've
* filled with data. * filled with data.
*/ */
std::pair<void*,uint64_t> preallocate( std::pair<void*, uint64_t> preallocate(
uint64_t min, uint64_t newAllocationSize, uint64_t min,
uint64_t max = std::numeric_limits<uint64_t>::max()) { uint64_t newAllocationSize,
uint64_t max = std::numeric_limits<uint64_t>::max()) {
dcheckCacheIntegrity(); dcheckCacheIntegrity();
if (LIKELY(writableTail() != nullptr && tailroom() >= min)) { if (LIKELY(writableTail() != nullptr && tailroom() >= min)) {
......
...@@ -25,16 +25,21 @@ ...@@ -25,16 +25,21 @@
namespace folly { namespace folly {
class RecordIOReader::Iterator : public boost::iterator_facade< class RecordIOReader::Iterator : public boost::iterator_facade<
RecordIOReader::Iterator, RecordIOReader::Iterator,
const std::pair<ByteRange, off_t>, const std::pair<ByteRange, off_t>,
boost::forward_traversal_tag> { boost::forward_traversal_tag> {
friend class boost::iterator_core_access; friend class boost::iterator_core_access;
friend class RecordIOReader; friend class RecordIOReader;
private: private:
Iterator(ByteRange range, uint32_t fileId, off_t pos); Iterator(ByteRange range, uint32_t fileId, off_t pos);
reference dereference() const { return recordAndPos_; } reference dereference() const {
bool equal(const Iterator& other) const { return range_ == other.range_; } return recordAndPos_;
}
bool equal(const Iterator& other) const {
return range_ == other.range_;
}
void increment() { void increment() {
size_t skip = recordio_helpers::headerSize() + recordAndPos_.first.size(); size_t skip = recordio_helpers::headerSize() + recordAndPos_.first.size();
recordAndPos_.second += off_t(skip); recordAndPos_.second += off_t(skip);
...@@ -49,10 +54,18 @@ class RecordIOReader::Iterator : public boost::iterator_facade< ...@@ -49,10 +54,18 @@ class RecordIOReader::Iterator : public boost::iterator_facade<
std::pair<ByteRange, off_t> recordAndPos_; std::pair<ByteRange, off_t> recordAndPos_;
}; };
inline auto RecordIOReader::cbegin() const -> Iterator { return seek(0); } inline auto RecordIOReader::cbegin() const -> Iterator {
inline auto RecordIOReader::begin() const -> Iterator { return cbegin(); } return seek(0);
inline auto RecordIOReader::cend() const -> Iterator { return seek(off_t(-1)); } }
inline auto RecordIOReader::end() const -> Iterator { return cend(); } inline auto RecordIOReader::begin() const -> Iterator {
return cbegin();
}
inline auto RecordIOReader::cend() const -> Iterator {
return seek(off_t(-1));
}
inline auto RecordIOReader::end() const -> Iterator {
return cend();
}
inline auto RecordIOReader::seek(off_t pos) const -> Iterator { inline auto RecordIOReader::seek(off_t pos) const -> Iterator {
return Iterator(map_.range(), fileId_, pos); return Iterator(map_.range(), fileId_, pos);
} }
...@@ -69,22 +82,25 @@ struct Header { ...@@ -69,22 +82,25 @@ struct Header {
// occurrence must start at least 4 bytes later) // occurrence must start at least 4 bytes later)
static constexpr uint32_t kMagic = 0xeac313a1; static constexpr uint32_t kMagic = 0xeac313a1;
uint32_t magic; uint32_t magic;
uint8_t version; // backwards incompatible version, currently 0 uint8_t version; // backwards incompatible version, currently 0
uint8_t hashFunction; // 0 = SpookyHashV2 uint8_t hashFunction; // 0 = SpookyHashV2
uint16_t flags; // reserved (must be 0) uint16_t flags; // reserved (must be 0)
uint32_t fileId; // unique file ID uint32_t fileId; // unique file ID
uint32_t dataLength; uint32_t dataLength;
uint64_t dataHash; uint64_t dataHash;
uint32_t headerHash; // must be last uint32_t headerHash; // must be last
} FOLLY_PACK_ATTR; } FOLLY_PACK_ATTR;
FOLLY_PACK_POP FOLLY_PACK_POP
static_assert(offsetof(Header, headerHash) + sizeof(Header::headerHash) == static_assert(
sizeof(Header), "invalid header layout"); offsetof(Header, headerHash) + sizeof(Header::headerHash) == sizeof(Header),
"invalid header layout");
} // namespace recordio_detail } // namespace recordio_detail
constexpr size_t headerSize() { return sizeof(recordio_detail::Header); } constexpr size_t headerSize() {
return sizeof(recordio_detail::Header);
}
inline RecordInfo findRecord(ByteRange range, uint32_t fileId) { inline RecordInfo findRecord(ByteRange range, uint32_t fileId) {
return findRecord(range, range, fileId); return findRecord(range, range, fileId);
......
...@@ -31,10 +31,10 @@ namespace folly { ...@@ -31,10 +31,10 @@ namespace folly {
using namespace recordio_helpers; using namespace recordio_helpers;
RecordIOWriter::RecordIOWriter(File file, uint32_t fileId) RecordIOWriter::RecordIOWriter(File file, uint32_t fileId)
: file_(std::move(file)), : file_(std::move(file)),
fileId_(fileId), fileId_(fileId),
writeLock_(file_, std::defer_lock), writeLock_(file_, std::defer_lock),
filePos_(0) { filePos_(0) {
if (!writeLock_.try_lock()) { if (!writeLock_.try_lock()) {
throw std::runtime_error("RecordIOWriter: file locked by another process"); throw std::runtime_error("RecordIOWriter: file locked by another process");
} }
...@@ -48,7 +48,7 @@ RecordIOWriter::RecordIOWriter(File file, uint32_t fileId) ...@@ -48,7 +48,7 @@ RecordIOWriter::RecordIOWriter(File file, uint32_t fileId)
void RecordIOWriter::write(std::unique_ptr<IOBuf> buf) { void RecordIOWriter::write(std::unique_ptr<IOBuf> buf) {
size_t totalLength = prependHeader(buf, fileId_); size_t totalLength = prependHeader(buf, fileId_);
if (totalLength == 0) { if (totalLength == 0) {
return; // nothing to do return; // nothing to do
} }
DCHECK_EQ(buf->computeChainDataLength(), totalLength); DCHECK_EQ(buf->computeChainDataLength(), totalLength);
...@@ -70,14 +70,10 @@ void RecordIOWriter::write(std::unique_ptr<IOBuf> buf) { ...@@ -70,14 +70,10 @@ void RecordIOWriter::write(std::unique_ptr<IOBuf> buf) {
} }
RecordIOReader::RecordIOReader(File file, uint32_t fileId) RecordIOReader::RecordIOReader(File file, uint32_t fileId)
: map_(std::move(file)), : map_(std::move(file)), fileId_(fileId) {}
fileId_(fileId) {
}
RecordIOReader::Iterator::Iterator(ByteRange range, uint32_t fileId, off_t pos) RecordIOReader::Iterator::Iterator(ByteRange range, uint32_t fileId, off_t pos)
: range_(range), : range_(range), fileId_(fileId), recordAndPos_(ByteRange(), 0) {
fileId_(fileId),
recordAndPos_(ByteRange(), 0) {
if (size_t(pos) >= range_.size()) { if (size_t(pos) >= range_.size()) {
// Note that this branch can execute if pos is negative as well. // Note that this branch can execute if pos is negative as well.
recordAndPos_.second = off_t(-1); recordAndPos_.second = off_t(-1);
...@@ -93,7 +89,7 @@ void RecordIOReader::Iterator::advanceToValid() { ...@@ -93,7 +89,7 @@ void RecordIOReader::Iterator::advanceToValid() {
ByteRange record = findRecord(range_, fileId_).record; ByteRange record = findRecord(range_, fileId_).record;
if (record.empty()) { if (record.empty()) {
recordAndPos_ = std::make_pair(ByteRange(), off_t(-1)); recordAndPos_ = std::make_pair(ByteRange(), off_t(-1));
range_.clear(); // at end range_.clear(); // at end
} else { } else {
size_t skipped = size_t(record.begin() - range_.begin()); size_t skipped = size_t(record.begin() - range_.begin());
DCHECK_GE(skipped, headerSize()); DCHECK_GE(skipped, headerSize());
...@@ -110,11 +106,11 @@ using recordio_detail::Header; ...@@ -110,11 +106,11 @@ using recordio_detail::Header;
namespace { namespace {
constexpr uint32_t kHashSeed = 0xdeadbeef; // for mcurtiss constexpr uint32_t kHashSeed = 0xdeadbeef; // for mcurtiss
uint32_t headerHash(const Header& header) { uint32_t headerHash(const Header& header) {
return hash::SpookyHashV2::Hash32(&header, offsetof(Header, headerHash), return hash::SpookyHashV2::Hash32(
kHashSeed); &header, offsetof(Header, headerHash), kHashSeed);
} }
std::pair<size_t, uint64_t> dataLengthAndHash(const IOBuf* buf) { std::pair<size_t, uint64_t> dataLengthAndHash(const IOBuf* buf) {
...@@ -146,7 +142,7 @@ size_t prependHeader(std::unique_ptr<IOBuf>& buf, uint32_t fileId) { ...@@ -146,7 +142,7 @@ size_t prependHeader(std::unique_ptr<IOBuf>& buf, uint32_t fileId) {
} }
auto lengthAndHash = dataLengthAndHash(buf.get()); auto lengthAndHash = dataLengthAndHash(buf.get());
if (lengthAndHash.first == 0) { if (lengthAndHash.first == 0) {
return 0; // empty, nothing to do, no zero-length records return 0; // empty, nothing to do, no zero-length records
} }
// Prepend to the first buffer in the chain if we have room, otherwise // Prepend to the first buffer in the chain if we have room, otherwise
...@@ -172,15 +168,13 @@ size_t prependHeader(std::unique_ptr<IOBuf>& buf, uint32_t fileId) { ...@@ -172,15 +168,13 @@ size_t prependHeader(std::unique_ptr<IOBuf>& buf, uint32_t fileId) {
} }
RecordInfo validateRecord(ByteRange range, uint32_t fileId) { RecordInfo validateRecord(ByteRange range, uint32_t fileId) {
if (range.size() <= headerSize()) { // records may not be empty if (range.size() <= headerSize()) { // records may not be empty
return {0, {}}; return {0, {}};
} }
const Header* header = reinterpret_cast<const Header*>(range.begin()); const Header* header = reinterpret_cast<const Header*>(range.begin());
range.advance(sizeof(Header)); range.advance(sizeof(Header));
if (header->magic != Header::kMagic || if (header->magic != Header::kMagic || header->version != 0 ||
header->version != 0 || header->hashFunction != 0 || header->flags != 0 ||
header->hashFunction != 0 ||
header->flags != 0 ||
(fileId != 0 && header->fileId != fileId) || (fileId != 0 && header->fileId != fileId) ||
header->dataLength > range.size()) { header->dataLength > range.size()) {
return {0, {}}; return {0, {}};
...@@ -195,19 +189,18 @@ RecordInfo validateRecord(ByteRange range, uint32_t fileId) { ...@@ -195,19 +189,18 @@ RecordInfo validateRecord(ByteRange range, uint32_t fileId) {
return {header->fileId, range}; return {header->fileId, range};
} }
RecordInfo findRecord(ByteRange searchRange, RecordInfo
ByteRange wholeRange, findRecord(ByteRange searchRange, ByteRange wholeRange, uint32_t fileId) {
uint32_t fileId) {
static const uint32_t magic = Header::kMagic; static const uint32_t magic = Header::kMagic;
static const ByteRange magicRange(reinterpret_cast<const uint8_t*>(&magic), static const ByteRange magicRange(
sizeof(magic)); reinterpret_cast<const uint8_t*>(&magic), sizeof(magic));
DCHECK_GE(searchRange.begin(), wholeRange.begin()); DCHECK_GE(searchRange.begin(), wholeRange.begin());
DCHECK_LE(searchRange.end(), wholeRange.end()); DCHECK_LE(searchRange.end(), wholeRange.end());
const uint8_t* start = searchRange.begin(); const uint8_t* start = searchRange.begin();
const uint8_t* end = std::min(searchRange.end(), const uint8_t* end =
wholeRange.end() - sizeof(Header)); std::min(searchRange.end(), wholeRange.end() - sizeof(Header));
// end-1: the last place where a Header could start // end-1: the last place where a Header could start
while (start < end) { while (start < end) {
auto p = ByteRange(start, end + sizeof(magic)).find(magicRange); auto p = ByteRange(start, end + sizeof(magic)).find(magicRange);
......
...@@ -69,7 +69,9 @@ class RecordIOWriter { ...@@ -69,7 +69,9 @@ class RecordIOWriter {
* Return the position in the file where the next byte will be written. * Return the position in the file where the next byte will be written.
* Conservative, as stuff can be written at any time from another thread. * Conservative, as stuff can be written at any time from another thread.
*/ */
off_t filePos() const { return filePos_; } off_t filePos() const {
return filePos_;
}
private: private:
File file_; File file_;
...@@ -129,7 +131,7 @@ namespace recordio_helpers { ...@@ -129,7 +131,7 @@ namespace recordio_helpers {
/** /**
* Header size. * Header size.
*/ */
constexpr size_t headerSize(); // defined in RecordIO-inl.h constexpr size_t headerSize(); // defined in RecordIO-inl.h
/** /**
* Write a header in the buffer. We will prepend the header to the front * Write a header in the buffer. We will prepend the header to the front
...@@ -157,9 +159,8 @@ struct RecordInfo { ...@@ -157,9 +159,8 @@ struct RecordInfo {
uint32_t fileId; uint32_t fileId;
ByteRange record; ByteRange record;
}; };
RecordInfo findRecord(ByteRange searchRange, RecordInfo
ByteRange wholeRange, findRecord(ByteRange searchRange, ByteRange wholeRange, uint32_t fileId);
uint32_t fileId);
/** /**
* Search for the first valid record in range. * Search for the first valid record in range.
......
...@@ -122,9 +122,9 @@ void ShutdownSocketSet::shutdown(int fd, bool abortive) { ...@@ -122,9 +122,9 @@ void ShutdownSocketSet::shutdown(int fd, bool abortive) {
} }
CHECK_EQ(prevState, MUST_CLOSE) CHECK_EQ(prevState, MUST_CLOSE)
<< "Invalid prev state for fd " << fd << ": " << int(prevState); << "Invalid prev state for fd " << fd << ": " << int(prevState);
folly::closeNoInt(fd); // ignore errors, nothing to do folly::closeNoInt(fd); // ignore errors, nothing to do
CHECK( CHECK(
sref.compare_exchange_strong(prevState, FREE, std::memory_order_relaxed)) sref.compare_exchange_strong(prevState, FREE, std::memory_order_relaxed))
......
...@@ -70,7 +70,7 @@ class ShutdownSocketSet : private boost::noncopyable { ...@@ -70,7 +70,7 @@ class ShutdownSocketSet : private boost::noncopyable {
* read() and write() operations to the socket will fail. During normal * read() and write() operations to the socket will fail. During normal
* operation, just call ::shutdown() on the socket. * operation, just call ::shutdown() on the socket.
*/ */
void shutdown(int fd, bool abortive=false); void shutdown(int fd, bool abortive = false);
/** /**
* Immediate shutdown of all connections. This is a hard-hitting hammer; * Immediate shutdown of all connections. This is a hard-hitting hammer;
...@@ -92,7 +92,7 @@ class ShutdownSocketSet : private boost::noncopyable { ...@@ -92,7 +92,7 @@ class ShutdownSocketSet : private boost::noncopyable {
* *
* This is async-signal-safe and ignores errors. * This is async-signal-safe and ignores errors.
*/ */
void shutdownAll(bool abortive=false); void shutdownAll(bool abortive = false);
private: private:
void doShutdown(int fd, bool abortive); void doShutdown(int fd, bool abortive);
......
...@@ -38,6 +38,7 @@ namespace folly { ...@@ -38,6 +38,7 @@ namespace folly {
template <class T> template <class T>
class TypedIOBuf { class TypedIOBuf {
static_assert(std::is_standard_layout<T>::value, "must be standard layout"); static_assert(std::is_standard_layout<T>::value, "must be standard layout");
public: public:
typedef T value_type; typedef T value_type;
typedef value_type& reference; typedef value_type& reference;
...@@ -46,7 +47,7 @@ class TypedIOBuf { ...@@ -46,7 +47,7 @@ class TypedIOBuf {
typedef value_type* iterator; typedef value_type* iterator;
typedef const value_type* const_iterator; typedef const value_type* const_iterator;
explicit TypedIOBuf(IOBuf* buf) : buf_(buf) { } explicit TypedIOBuf(IOBuf* buf) : buf_(buf) {}
IOBuf* ioBuf() { IOBuf* ioBuf() {
return buf_; return buf_;
...@@ -73,7 +74,9 @@ class TypedIOBuf { ...@@ -73,7 +74,9 @@ class TypedIOBuf {
uint32_t length() const { uint32_t length() const {
return sdiv(buf_->length()); return sdiv(buf_->length());
} }
uint32_t size() const { return length(); } uint32_t size() const {
return length();
}
uint32_t headroom() const { uint32_t headroom() const {
return sdiv(buf_->headroom()); return sdiv(buf_->headroom());
...@@ -117,14 +120,28 @@ class TypedIOBuf { ...@@ -117,14 +120,28 @@ class TypedIOBuf {
void reserve(uint32_t minHeadroom, uint32_t minTailroom) { void reserve(uint32_t minHeadroom, uint32_t minTailroom) {
buf_->reserve(smul(minHeadroom), smul(minTailroom)); buf_->reserve(smul(minHeadroom), smul(minTailroom));
} }
void reserve(uint32_t minTailroom) { reserve(0, minTailroom); } void reserve(uint32_t minTailroom) {
reserve(0, minTailroom);
}
const T* cbegin() const { return data(); } const T* cbegin() const {
const T* cend() const { return tail(); } return data();
const T* begin() const { return cbegin(); } }
const T* end() const { return cend(); } const T* cend() const {
T* begin() { return writableData(); } return tail();
T* end() { return writableTail(); } }
const T* begin() const {
return cbegin();
}
const T* end() const {
return cend();
}
T* begin() {
return writableData();
}
T* end() {
return writableTail();
}
const T& front() const { const T& front() const {
assert(!empty()); assert(!empty());
...@@ -163,7 +180,9 @@ class TypedIOBuf { ...@@ -163,7 +180,9 @@ class TypedIOBuf {
void push(const T& data) { void push(const T& data) {
push(&data, &data + 1); push(&data, &data + 1);
} }
void push_back(const T& data) { push(data); } void push_back(const T& data) {
push(data);
}
/** /**
* Append multiple elements in a sequence; will call distance(). * Append multiple elements in a sequence; will call distance().
......
...@@ -18,10 +18,10 @@ ...@@ -18,10 +18,10 @@
#include <folly/FileUtil.h> #include <folly/FileUtil.h>
#include <folly/io/async/AsyncSocketException.h> #include <folly/io/async/AsyncSocketException.h>
using std::string;
using std::unique_ptr;
using folly::IOBuf; using folly::IOBuf;
using folly::IOBufQueue; using folly::IOBufQueue;
using std::string;
using std::unique_ptr;
namespace folly { namespace folly {
...@@ -30,8 +30,8 @@ AsyncPipeReader::~AsyncPipeReader() { ...@@ -30,8 +30,8 @@ AsyncPipeReader::~AsyncPipeReader() {
} }
void AsyncPipeReader::failRead(const AsyncSocketException& ex) { void AsyncPipeReader::failRead(const AsyncSocketException& ex) {
VLOG(5) << "AsyncPipeReader(this=" << this << ", fd=" << fd_ << VLOG(5) << "AsyncPipeReader(this=" << this << ", fd=" << fd_
"): failed while reading: " << ex.what(); << "): failed while reading: " << ex.what();
DCHECK(readCallback_ != nullptr); DCHECK(readCallback_ != nullptr);
AsyncReader::ReadCallback* callback = readCallback_; AsyncReader::ReadCallback* callback = readCallback_;
...@@ -124,8 +124,8 @@ void AsyncPipeReader::handlerReady(uint16_t events) noexcept { ...@@ -124,8 +124,8 @@ void AsyncPipeReader::handlerReady(uint16_t events) noexcept {
// No more data to read right now. // No more data to read right now.
return; return;
} else if (bytesRead < 0) { } else if (bytesRead < 0) {
AsyncSocketException ex(AsyncSocketException::INVALID_STATE, AsyncSocketException ex(
"read failed", errno); AsyncSocketException::INVALID_STATE, "read failed", errno);
failRead(ex); failRead(ex);
return; return;
} else { } else {
...@@ -142,13 +142,13 @@ void AsyncPipeReader::handlerReady(uint16_t events) noexcept { ...@@ -142,13 +142,13 @@ void AsyncPipeReader::handlerReady(uint16_t events) noexcept {
} }
} }
void AsyncPipeWriter::write(
void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf, unique_ptr<folly::IOBuf> buf,
AsyncWriter::WriteCallback* callback) { AsyncWriter::WriteCallback* callback) {
if (closed()) { if (closed()) {
if (callback) { if (callback) {
AsyncSocketException ex(AsyncSocketException::NOT_OPEN, AsyncSocketException ex(
"attempt to write to closed pipe"); AsyncSocketException::NOT_OPEN, "attempt to write to closed pipe");
callback->writeErr(0, ex); callback->writeErr(0, ex);
} }
return; return;
...@@ -157,9 +157,9 @@ void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf, ...@@ -157,9 +157,9 @@ void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf,
folly::IOBufQueue iobq; folly::IOBufQueue iobq;
iobq.append(std::move(buf)); iobq.append(std::move(buf));
std::pair<folly::IOBufQueue, AsyncWriter::WriteCallback*> p( std::pair<folly::IOBufQueue, AsyncWriter::WriteCallback*> p(
std::move(iobq), callback); std::move(iobq), callback);
queue_.emplace_back(std::move(p)); queue_.emplace_back(std::move(p));
if (wasEmpty) { if (wasEmpty) {
handleWrite(); handleWrite();
} else { } else {
CHECK(!queue_.empty()); CHECK(!queue_.empty());
...@@ -167,9 +167,10 @@ void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf, ...@@ -167,9 +167,10 @@ void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf,
} }
} }
void AsyncPipeWriter::writeChain(folly::AsyncWriter::WriteCallback* callback, void AsyncPipeWriter::writeChain(
std::unique_ptr<folly::IOBuf>&& buf, folly::AsyncWriter::WriteCallback* callback,
WriteFlags) { std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags) {
write(std::move(buf), callback); write(std::move(buf), callback);
} }
...@@ -186,8 +187,8 @@ void AsyncPipeWriter::closeOnEmpty() { ...@@ -186,8 +187,8 @@ void AsyncPipeWriter::closeOnEmpty() {
void AsyncPipeWriter::closeNow() { void AsyncPipeWriter::closeNow() {
VLOG(5) << "close now"; VLOG(5) << "close now";
if (!queue_.empty()) { if (!queue_.empty()) {
failAllWrites(AsyncSocketException(AsyncSocketException::NOT_OPEN, failAllWrites(AsyncSocketException(
"closed with pending writes")); AsyncSocketException::NOT_OPEN, "closed with pending writes"));
} }
if (fd_ >= 0) { if (fd_ >= 0) {
unregisterHandler(); unregisterHandler();
...@@ -213,7 +214,6 @@ void AsyncPipeWriter::failAllWrites(const AsyncSocketException& ex) { ...@@ -213,7 +214,6 @@ void AsyncPipeWriter::failAllWrites(const AsyncSocketException& ex) {
} }
} }
void AsyncPipeWriter::handlerReady(uint16_t events) noexcept { void AsyncPipeWriter::handlerReady(uint16_t events) noexcept {
CHECK(events & EventHandler::WRITE); CHECK(events & EventHandler::WRITE);
...@@ -225,7 +225,7 @@ void AsyncPipeWriter::handleWrite() { ...@@ -225,7 +225,7 @@ void AsyncPipeWriter::handleWrite() {
assert(!queue_.empty()); assert(!queue_.empty());
do { do {
auto& front = queue_.front(); auto& front = queue_.front();
folly::IOBufQueue &curQueue = front.first; folly::IOBufQueue& curQueue = front.first;
DCHECK(!curQueue.empty()); DCHECK(!curQueue.empty());
// someday, support writev. The logic for partial writes is a bit complex // someday, support writev. The logic for partial writes is a bit complex
const IOBuf* head = curQueue.front(); const IOBuf* head = curQueue.front();
...@@ -238,8 +238,8 @@ void AsyncPipeWriter::handleWrite() { ...@@ -238,8 +238,8 @@ void AsyncPipeWriter::handleWrite() {
registerHandler(EventHandler::WRITE); registerHandler(EventHandler::WRITE);
return; return;
} else { } else {
failAllWrites(AsyncSocketException(AsyncSocketException::INTERNAL_ERROR, failAllWrites(AsyncSocketException(
"write failed", errno)); AsyncSocketException::INTERNAL_ERROR, "write failed", errno));
closeNow(); closeNow();
return; return;
} }
......
...@@ -35,8 +35,9 @@ class AsyncPipeReader : public EventHandler, ...@@ -35,8 +35,9 @@ class AsyncPipeReader : public EventHandler,
public AsyncReader, public AsyncReader,
public DelayedDestruction { public DelayedDestruction {
public: public:
typedef std::unique_ptr<AsyncPipeReader, typedef std::
folly::DelayedDestruction::Destructor> UniquePtr; unique_ptr<AsyncPipeReader, folly::DelayedDestruction::Destructor>
UniquePtr;
template <typename... Args> template <typename... Args>
static UniquePtr newReader(Args&&... args) { static UniquePtr newReader(Args&&... args) {
...@@ -44,8 +45,7 @@ class AsyncPipeReader : public EventHandler, ...@@ -44,8 +45,7 @@ class AsyncPipeReader : public EventHandler,
} }
AsyncPipeReader(folly::EventBase* eventBase, int pipeFd) AsyncPipeReader(folly::EventBase* eventBase, int pipeFd)
: EventHandler(eventBase, pipeFd), : EventHandler(eventBase, pipeFd), fd_(pipeFd) {}
fd_(pipeFd) {}
/** /**
* Set the read callback and automatically install/uninstall the handler * Set the read callback and automatically install/uninstall the handler
...@@ -96,8 +96,9 @@ class AsyncPipeWriter : public EventHandler, ...@@ -96,8 +96,9 @@ class AsyncPipeWriter : public EventHandler,
public AsyncWriter, public AsyncWriter,
public DelayedDestruction { public DelayedDestruction {
public: public:
typedef std::unique_ptr<AsyncPipeWriter, typedef std::
folly::DelayedDestruction::Destructor> UniquePtr; unique_ptr<AsyncPipeWriter, folly::DelayedDestruction::Destructor>
UniquePtr;
template <typename... Args> template <typename... Args>
static UniquePtr newWriter(Args&&... args) { static UniquePtr newWriter(Args&&... args) {
...@@ -105,15 +106,15 @@ class AsyncPipeWriter : public EventHandler, ...@@ -105,15 +106,15 @@ class AsyncPipeWriter : public EventHandler,
} }
AsyncPipeWriter(folly::EventBase* eventBase, int pipeFd) AsyncPipeWriter(folly::EventBase* eventBase, int pipeFd)
: EventHandler(eventBase, pipeFd), : EventHandler(eventBase, pipeFd), fd_(pipeFd) {}
fd_(pipeFd) {}
/** /**
* Asynchronously write the given iobuf to this pipe, and invoke the callback * Asynchronously write the given iobuf to this pipe, and invoke the callback
* on success/error. * on success/error.
*/ */
void write(std::unique_ptr<folly::IOBuf> iob, void write(
AsyncWriter::WriteCallback* wcb = nullptr); std::unique_ptr<folly::IOBuf> iob,
AsyncWriter::WriteCallback* wcb = nullptr);
/** /**
* Set a special hook to close the socket (otherwise, will call close()) * Set a special hook to close the socket (otherwise, will call close())
...@@ -148,21 +149,24 @@ class AsyncPipeWriter : public EventHandler, ...@@ -148,21 +149,24 @@ class AsyncPipeWriter : public EventHandler,
} }
// AsyncWriter methods // AsyncWriter methods
void write(folly::AsyncWriter::WriteCallback* callback, void write(
const void* buf, folly::AsyncWriter::WriteCallback* callback,
size_t bytes, const void* buf,
WriteFlags flags = WriteFlags::NONE) override { size_t bytes,
WriteFlags flags = WriteFlags::NONE) override {
writeChain(callback, IOBuf::wrapBuffer(buf, bytes), flags); writeChain(callback, IOBuf::wrapBuffer(buf, bytes), flags);
} }
void writev(folly::AsyncWriter::WriteCallback*, void writev(
const iovec*, folly::AsyncWriter::WriteCallback*,
size_t, const iovec*,
WriteFlags = WriteFlags::NONE) override { size_t,
WriteFlags = WriteFlags::NONE) override {
throw std::runtime_error("writev is not supported. Please use writeChain."); throw std::runtime_error("writev is not supported. Please use writeChain.");
} }
void writeChain(folly::AsyncWriter::WriteCallback* callback, void writeChain(
std::unique_ptr<folly::IOBuf>&& buf, folly::AsyncWriter::WriteCallback* callback,
WriteFlags flags = WriteFlags::NONE) override; std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE) override;
private: private:
void handlerReady(uint16_t events) noexcept override; void handlerReady(uint16_t events) noexcept override;
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
...@@ -26,8 +26,7 @@ using std::string; ...@@ -26,8 +26,7 @@ using std::string;
namespace folly { namespace folly {
AsyncSignalHandler::AsyncSignalHandler(EventBase* eventBase) AsyncSignalHandler::AsyncSignalHandler(EventBase* eventBase)
: eventBase_(eventBase) { : eventBase_(eventBase) {}
}
AsyncSignalHandler::~AsyncSignalHandler() { AsyncSignalHandler::~AsyncSignalHandler() {
// Unregister any outstanding events // Unregister any outstanding events
...@@ -52,27 +51,24 @@ void AsyncSignalHandler::detachEventBase() { ...@@ -52,27 +51,24 @@ void AsyncSignalHandler::detachEventBase() {
void AsyncSignalHandler::registerSignalHandler(int signum) { void AsyncSignalHandler::registerSignalHandler(int signum) {
pair<SignalEventMap::iterator, bool> ret = pair<SignalEventMap::iterator, bool> ret =
signalEvents_.insert(make_pair(signum, event())); signalEvents_.insert(make_pair(signum, event()));
if (!ret.second) { if (!ret.second) {
// This signal has already been registered // This signal has already been registered
throw std::runtime_error(folly::to<string>( throw std::runtime_error(
"handler already registered for signal ", folly::to<string>("handler already registered for signal ", signum));
signum));
} }
struct event* ev = &(ret.first->second); struct event* ev = &(ret.first->second);
try { try {
signal_set(ev, signum, libeventCallback, this); signal_set(ev, signum, libeventCallback, this);
if (event_base_set(eventBase_->getLibeventBase(), ev) != 0 ) { if (event_base_set(eventBase_->getLibeventBase(), ev) != 0) {
throw std::runtime_error(folly::to<string>( throw std::runtime_error(folly::to<string>(
"error initializing event handler for signal ", "error initializing event handler for signal ", signum));
signum));
} }
if (event_add(ev, nullptr) != 0) { if (event_add(ev, nullptr) != 0) {
throw std::runtime_error(folly::to<string>( throw std::runtime_error(
"error adding event handler for signal ", folly::to<string>("error adding event handler for signal ", signum));
signum));
} }
} catch (...) { } catch (...) {
signalEvents_.erase(ret.first); signalEvents_.erase(ret.first);
...@@ -84,17 +80,19 @@ void AsyncSignalHandler::unregisterSignalHandler(int signum) { ...@@ -84,17 +80,19 @@ void AsyncSignalHandler::unregisterSignalHandler(int signum) {
SignalEventMap::iterator it = signalEvents_.find(signum); SignalEventMap::iterator it = signalEvents_.find(signum);
if (it == signalEvents_.end()) { if (it == signalEvents_.end()) {
throw std::runtime_error(folly::to<string>( throw std::runtime_error(folly::to<string>(
"unable to unregister handler for signal ", "unable to unregister handler for signal ",
signum, ": signal not registered")); signum,
": signal not registered"));
} }
event_del(&it->second); event_del(&it->second);
signalEvents_.erase(it); signalEvents_.erase(it);
} }
void AsyncSignalHandler::libeventCallback(libevent_fd_t signum, void AsyncSignalHandler::libeventCallback(
short /* events */, libevent_fd_t signum,
void* arg) { short /* events */,
void* arg) {
AsyncSignalHandler* handler = static_cast<AsyncSignalHandler*>(arg); AsyncSignalHandler* handler = static_cast<AsyncSignalHandler*>(arg);
handler->signalReceived(int(signum)); handler->signalReceived(int(signum));
} }
......
...@@ -105,8 +105,8 @@ class AsyncSignalHandler { ...@@ -105,8 +105,8 @@ class AsyncSignalHandler {
typedef std::map<int, struct event> SignalEventMap; typedef std::map<int, struct event> SignalEventMap;
// Forbidden copy constructor and assignment operator // Forbidden copy constructor and assignment operator
AsyncSignalHandler(AsyncSignalHandler const &); AsyncSignalHandler(AsyncSignalHandler const&);
AsyncSignalHandler& operator=(AsyncSignalHandler const &); AsyncSignalHandler& operator=(AsyncSignalHandler const&);
static void libeventCallback(libevent_fd_t signum, short events, void* arg); static void libeventCallback(libevent_fd_t signum, short events, void* arg);
......
This diff is collapsed.
This diff is collapsed.
...@@ -26,32 +26,27 @@ namespace folly { ...@@ -26,32 +26,27 @@ namespace folly {
AsyncTimeout::AsyncTimeout(TimeoutManager* timeoutManager) AsyncTimeout::AsyncTimeout(TimeoutManager* timeoutManager)
: timeoutManager_(timeoutManager) { : timeoutManager_(timeoutManager) {
folly_event_set( folly_event_set(
&event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this); &event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this);
event_.ev_base = nullptr; event_.ev_base = nullptr;
timeoutManager_->attachTimeoutManager( timeoutManager_->attachTimeoutManager(
this, this, TimeoutManager::InternalEnum::NORMAL);
TimeoutManager::InternalEnum::NORMAL);
} }
AsyncTimeout::AsyncTimeout(EventBase* eventBase) AsyncTimeout::AsyncTimeout(EventBase* eventBase) : timeoutManager_(eventBase) {
: timeoutManager_(eventBase) {
folly_event_set( folly_event_set(
&event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this); &event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this);
event_.ev_base = nullptr; event_.ev_base = nullptr;
if (eventBase) { if (eventBase) {
timeoutManager_->attachTimeoutManager( timeoutManager_->attachTimeoutManager(
this, this, TimeoutManager::InternalEnum::NORMAL);
TimeoutManager::InternalEnum::NORMAL);
} }
} }
AsyncTimeout::AsyncTimeout(TimeoutManager* timeoutManager, AsyncTimeout::AsyncTimeout(
InternalEnum internal) TimeoutManager* timeoutManager,
InternalEnum internal)
: timeoutManager_(timeoutManager) { : timeoutManager_(timeoutManager) {
folly_event_set( folly_event_set(
&event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this); &event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this);
event_.ev_base = nullptr; event_.ev_base = nullptr;
...@@ -60,14 +55,13 @@ AsyncTimeout::AsyncTimeout(TimeoutManager* timeoutManager, ...@@ -60,14 +55,13 @@ AsyncTimeout::AsyncTimeout(TimeoutManager* timeoutManager,
AsyncTimeout::AsyncTimeout(EventBase* eventBase, InternalEnum internal) AsyncTimeout::AsyncTimeout(EventBase* eventBase, InternalEnum internal)
: timeoutManager_(eventBase) { : timeoutManager_(eventBase) {
folly_event_set( folly_event_set(
&event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this); &event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this);
event_.ev_base = nullptr; event_.ev_base = nullptr;
timeoutManager_->attachTimeoutManager(this, internal); timeoutManager_->attachTimeoutManager(this, internal);
} }
AsyncTimeout::AsyncTimeout(): timeoutManager_(nullptr) { AsyncTimeout::AsyncTimeout() : timeoutManager_(nullptr) {
folly_event_set( folly_event_set(
&event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this); &event_, -1, EV_TIMEOUT, &AsyncTimeout::libeventCallback, this);
event_.ev_base = nullptr; event_.ev_base = nullptr;
......
...@@ -121,10 +121,12 @@ class AsyncTimeout : private boost::noncopyable { ...@@ -121,10 +121,12 @@ class AsyncTimeout : private boost::noncopyable {
* internal event. TimeoutManager::loop() will return when there are no more * internal event. TimeoutManager::loop() will return when there are no more
* non-internal events remaining. * non-internal events remaining.
*/ */
void attachTimeoutManager(TimeoutManager* timeoutManager, void attachTimeoutManager(
InternalEnum internal = InternalEnum::NORMAL); TimeoutManager* timeoutManager,
void attachEventBase(EventBase* eventBase, InternalEnum internal = InternalEnum::NORMAL);
InternalEnum internal = InternalEnum::NORMAL); void attachEventBase(
EventBase* eventBase,
InternalEnum internal = InternalEnum::NORMAL);
/** /**
* Detach the timeout from its TimeoutManager. * Detach the timeout from its TimeoutManager.
...@@ -175,9 +177,8 @@ class AsyncTimeout : private boost::noncopyable { ...@@ -175,9 +177,8 @@ class AsyncTimeout : private boost::noncopyable {
*/ */
template <typename TCallback> template <typename TCallback>
static std::unique_ptr<AsyncTimeout> make( static std::unique_ptr<AsyncTimeout> make(
TimeoutManager &manager, TimeoutManager& manager,
TCallback &&callback TCallback&& callback);
);
/** /**
* Convenience function that wraps a function object as * Convenience function that wraps a function object as
...@@ -210,10 +211,9 @@ class AsyncTimeout : private boost::noncopyable { ...@@ -210,10 +211,9 @@ class AsyncTimeout : private boost::noncopyable {
*/ */
template <typename TCallback> template <typename TCallback>
static std::unique_ptr<AsyncTimeout> schedule( static std::unique_ptr<AsyncTimeout> schedule(
TimeoutManager::timeout_type timeout, TimeoutManager::timeout_type timeout,
TimeoutManager &manager, TimeoutManager& manager,
TCallback &&callback TCallback&& callback);
);
private: private:
static void libeventCallback(libevent_fd_t fd, short events, void* arg); static void libeventCallback(libevent_fd_t fd, short events, void* arg);
...@@ -239,20 +239,15 @@ namespace detail { ...@@ -239,20 +239,15 @@ namespace detail {
* @author: Marcelo Juchem <marcelo@fb.com> * @author: Marcelo Juchem <marcelo@fb.com>
*/ */
template <typename TCallback> template <typename TCallback>
struct async_timeout_wrapper: struct async_timeout_wrapper : public AsyncTimeout {
public AsyncTimeout
{
template <typename UCallback> template <typename UCallback>
async_timeout_wrapper(TimeoutManager *manager, UCallback &&callback): async_timeout_wrapper(TimeoutManager* manager, UCallback&& callback)
AsyncTimeout(manager), : AsyncTimeout(manager), callback_(std::forward<UCallback>(callback)) {}
callback_(std::forward<UCallback>(callback))
{}
void timeoutExpired() noexcept override { void timeoutExpired() noexcept override {
static_assert( static_assert(
noexcept(std::declval<TCallback>()()), noexcept(std::declval<TCallback>()()),
"callback must be declared noexcept, e.g.: `[]() noexcept {}`" "callback must be declared noexcept, e.g.: `[]() noexcept {}`");
);
callback_(); callback_();
} }
...@@ -264,23 +259,18 @@ struct async_timeout_wrapper: ...@@ -264,23 +259,18 @@ struct async_timeout_wrapper:
template <typename TCallback> template <typename TCallback>
std::unique_ptr<AsyncTimeout> AsyncTimeout::make( std::unique_ptr<AsyncTimeout> AsyncTimeout::make(
TimeoutManager &manager, TimeoutManager& manager,
TCallback &&callback TCallback&& callback) {
) {
return std::unique_ptr<AsyncTimeout>( return std::unique_ptr<AsyncTimeout>(
new detail::async_timeout_wrapper<typename std::decay<TCallback>::type>( new detail::async_timeout_wrapper<typename std::decay<TCallback>::type>(
std::addressof(manager), std::addressof(manager), std::forward<TCallback>(callback)));
std::forward<TCallback>(callback)
)
);
} }
template <typename TCallback> template <typename TCallback>
std::unique_ptr<AsyncTimeout> AsyncTimeout::schedule( std::unique_ptr<AsyncTimeout> AsyncTimeout::schedule(
TimeoutManager::timeout_type timeout, TimeoutManager::timeout_type timeout,
TimeoutManager &manager, TimeoutManager& manager,
TCallback &&callback TCallback&& callback) {
) {
auto wrapper = AsyncTimeout::make(manager, std::forward<TCallback>(callback)); auto wrapper = AsyncTimeout::make(manager, std::forward<TCallback>(callback));
wrapper->scheduleTimeout(timeout); wrapper->scheduleTimeout(timeout);
return wrapper; return wrapper;
......
...@@ -29,11 +29,11 @@ ...@@ -29,11 +29,11 @@
constexpr bool kOpenSslModeMoveBufferOwnership = constexpr bool kOpenSslModeMoveBufferOwnership =
#ifdef SSL_MODE_MOVE_BUFFER_OWNERSHIP #ifdef SSL_MODE_MOVE_BUFFER_OWNERSHIP
true true
#else #else
false false
#endif #endif
; ;
namespace folly { namespace folly {
...@@ -72,7 +72,7 @@ enum class WriteFlags : uint32_t { ...@@ -72,7 +72,7 @@ enum class WriteFlags : uint32_t {
*/ */
inline WriteFlags operator|(WriteFlags a, WriteFlags b) { inline WriteFlags operator|(WriteFlags a, WriteFlags b) {
return static_cast<WriteFlags>( return static_cast<WriteFlags>(
static_cast<uint32_t>(a) | static_cast<uint32_t>(b)); static_cast<uint32_t>(a) | static_cast<uint32_t>(b));
} }
/* /*
...@@ -88,7 +88,7 @@ inline WriteFlags& operator|=(WriteFlags& a, WriteFlags b) { ...@@ -88,7 +88,7 @@ inline WriteFlags& operator|=(WriteFlags& a, WriteFlags b) {
*/ */
inline WriteFlags operator&(WriteFlags a, WriteFlags b) { inline WriteFlags operator&(WriteFlags a, WriteFlags b) {
return static_cast<WriteFlags>( return static_cast<WriteFlags>(
static_cast<uint32_t>(a) & static_cast<uint32_t>(b)); static_cast<uint32_t>(a) & static_cast<uint32_t>(b));
} }
/* /*
...@@ -120,7 +120,6 @@ inline bool isSet(WriteFlags a, WriteFlags b) { ...@@ -120,7 +120,6 @@ inline bool isSet(WriteFlags a, WriteFlags b) {
return (a & b) == b; return (a & b) == b;
} }
/** /**
* AsyncTransport defines an asynchronous API for streaming I/O. * AsyncTransport defines an asynchronous API for streaming I/O.
* *
...@@ -382,7 +381,9 @@ class AsyncTransport : public DelayedDestruction, public AsyncSocketBase { ...@@ -382,7 +381,9 @@ class AsyncTransport : public DelayedDestruction, public AsyncSocketBase {
/** /**
* Get the certificate used to authenticate the peer. * Get the certificate used to authenticate the peer.
*/ */
virtual ssl::X509UniquePtr getPeerCert() const { return nullptr; } virtual ssl::X509UniquePtr getPeerCert() const {
return nullptr;
}
/** /**
* The local certificate used for this connection. May be null * The local certificate used for this connection. May be null
...@@ -459,7 +460,9 @@ class AsyncTransport : public DelayedDestruction, public AsyncSocketBase { ...@@ -459,7 +460,9 @@ class AsyncTransport : public DelayedDestruction, public AsyncSocketBase {
* False if the transport does not have replay protection, but will in the * False if the transport does not have replay protection, but will in the
* future. * future.
*/ */
virtual bool isReplaySafe() const { return true; } virtual bool isReplaySafe() const {
return true;
}
/** /**
* Set the ReplaySafeCallback on this transport. * Set the ReplaySafeCallback on this transport.
...@@ -580,8 +583,8 @@ class AsyncReader { ...@@ -580,8 +583,8 @@ class AsyncReader {
* @param readBuf The unique pointer of read buffer. * @param readBuf The unique pointer of read buffer.
*/ */
virtual void readBufferAvailable(std::unique_ptr<IOBuf> /*readBuf*/) virtual void readBufferAvailable(
noexcept {} std::unique_ptr<IOBuf> /*readBuf*/) noexcept {}
/** /**
* readEOF() will be invoked when the transport is closed. * readEOF() will be invoked when the transport is closed.
...@@ -636,18 +639,26 @@ class AsyncWriter { ...@@ -636,18 +639,26 @@ class AsyncWriter {
* @param bytesWritten The number of bytes that were successfull * @param bytesWritten The number of bytes that were successfull
* @param ex An exception describing the error that occurred. * @param ex An exception describing the error that occurred.
*/ */
virtual void writeErr(size_t bytesWritten, virtual void writeErr(
const AsyncSocketException& ex) noexcept = 0; size_t bytesWritten,
const AsyncSocketException& ex) noexcept = 0;
}; };
// Write methods that aren't part of AsyncTransport // Write methods that aren't part of AsyncTransport
virtual void write(WriteCallback* callback, const void* buf, size_t bytes, virtual void write(
WriteFlags flags = WriteFlags::NONE) = 0; WriteCallback* callback,
virtual void writev(WriteCallback* callback, const iovec* vec, size_t count, const void* buf,
WriteFlags flags = WriteFlags::NONE) = 0; size_t bytes,
virtual void writeChain(WriteCallback* callback, WriteFlags flags = WriteFlags::NONE) = 0;
std::unique_ptr<IOBuf>&& buf, virtual void writev(
WriteFlags flags = WriteFlags::NONE) = 0; WriteCallback* callback,
const iovec* vec,
size_t count,
WriteFlags flags = WriteFlags::NONE) = 0;
virtual void writeChain(
WriteCallback* callback,
std::unique_ptr<IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE) = 0;
protected: protected:
virtual ~AsyncWriter() = default; virtual ~AsyncWriter() = default;
...@@ -663,8 +674,8 @@ class AsyncTransportWrapper : virtual public AsyncTransport, ...@@ -663,8 +674,8 @@ class AsyncTransportWrapper : virtual public AsyncTransport,
// Alias for inherited members from AsyncReader and AsyncWriter // Alias for inherited members from AsyncReader and AsyncWriter
// to keep compatibility. // to keep compatibility.
using ReadCallback = AsyncReader::ReadCallback; using ReadCallback = AsyncReader::ReadCallback;
using WriteCallback = AsyncWriter::WriteCallback; using WriteCallback = AsyncWriter::WriteCallback;
void setReadCB(ReadCallback* callback) override = 0; void setReadCB(ReadCallback* callback) override = 0;
ReadCallback* getReadCallback() const override = 0; ReadCallback* getReadCallback() const override = 0;
void write( void write(
...@@ -711,7 +722,7 @@ class AsyncTransportWrapper : virtual public AsyncTransport, ...@@ -711,7 +722,7 @@ class AsyncTransportWrapper : virtual public AsyncTransport,
template <class T> template <class T>
T* getUnderlyingTransport() { T* getUnderlyingTransport() {
return const_cast<T*>(static_cast<const AsyncTransportWrapper*>(this) return const_cast<T*>(static_cast<const AsyncTransportWrapper*>(this)
->getUnderlyingTransport<T>()); ->getUnderlyingTransport<T>());
} }
}; };
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment