Commit d4018226 authored by Dan Melnic's avatar Dan Melnic Committed by Facebook Github Bot

Remove the zerocopy write threshold support, add support for ENOBUFS

Summary: Remove the zerocopy write threshold support since it is a little bit confusing

Reviewed By: djwatson

Differential Revision: D6256854

fbshipit-source-id: 1c992f93d7b04c4ede2fbefebde7a7ae89de3764
parent bb8d886a
...@@ -106,7 +106,7 @@ class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest { ...@@ -106,7 +106,7 @@ class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest {
writeFlags |= WriteFlags::CORK; writeFlags |= WriteFlags::CORK;
} }
socket_->adjustZeroCopyFlags(getOps(), getOpCount(), writeFlags); socket_->adjustZeroCopyFlags(writeFlags);
auto writeResult = socket_->performWrite( auto writeResult = socket_->performWrite(
getOps(), getOpCount(), writeFlags, &opsWritten_, &partialBytes_); getOps(), getOpCount(), writeFlags, &opsWritten_, &partialBytes_);
...@@ -854,45 +854,14 @@ bool AsyncSocket::setZeroCopy(bool enable) { ...@@ -854,45 +854,14 @@ bool AsyncSocket::setZeroCopy(bool enable) {
return false; return false;
} }
void AsyncSocket::setZeroCopyWriteChainThreshold(size_t threshold) {
zeroCopyWriteChainThreshold_ = threshold;
}
bool AsyncSocket::isZeroCopyRequest(WriteFlags flags) { bool AsyncSocket::isZeroCopyRequest(WriteFlags flags) {
return (zeroCopyEnabled_ && isSet(flags, WriteFlags::WRITE_MSG_ZEROCOPY)); return (zeroCopyEnabled_ && isSet(flags, WriteFlags::WRITE_MSG_ZEROCOPY));
} }
void AsyncSocket::adjustZeroCopyFlags( void AsyncSocket::adjustZeroCopyFlags(folly::WriteFlags& flags) {
folly::IOBuf* buf, if (!zeroCopyEnabled_) {
folly::WriteFlags& flags) {
if (zeroCopyEnabled_ && zeroCopyWriteChainThreshold_ && buf &&
buf->isManaged()) {
if (buf->computeChainDataLength() >= zeroCopyWriteChainThreshold_) {
flags |= folly::WriteFlags::WRITE_MSG_ZEROCOPY;
} else {
flags = unSet(flags, folly::WriteFlags::WRITE_MSG_ZEROCOPY);
}
}
}
void AsyncSocket::adjustZeroCopyFlags(
const iovec* vec,
uint32_t count,
folly::WriteFlags& flags) {
if (zeroCopyEnabled_ && zeroCopyWriteChainThreshold_) {
count = std::min<uint32_t>(count, kIovMax);
size_t sum = 0;
for (uint32_t i = 0; i < count; ++i) {
const iovec* v = vec + i;
sum += v->iov_len;
}
if (sum >= zeroCopyWriteChainThreshold_) {
flags |= folly::WriteFlags::WRITE_MSG_ZEROCOPY;
} else {
flags = unSet(flags, folly::WriteFlags::WRITE_MSG_ZEROCOPY); flags = unSet(flags, folly::WriteFlags::WRITE_MSG_ZEROCOPY);
} }
}
} }
void AsyncSocket::addZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf) { void AsyncSocket::addZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf) {
...@@ -987,7 +956,7 @@ void AsyncSocket::writev(WriteCallback* callback, ...@@ -987,7 +956,7 @@ void AsyncSocket::writev(WriteCallback* callback,
void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf, void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf,
WriteFlags flags) { WriteFlags flags) {
adjustZeroCopyFlags(buf.get(), flags); adjustZeroCopyFlags(flags);
constexpr size_t kSmallSizeMax = 64; constexpr size_t kSmallSizeMax = 64;
size_t count = buf->countChainElements(); size_t count = buf->countChainElements();
...@@ -1758,8 +1727,7 @@ void AsyncSocket::handleErrMessages() noexcept { ...@@ -1758,8 +1727,7 @@ void AsyncSocket::handleErrMessages() noexcept {
// supporting per-socket error queues. // supporting per-socket error queues.
VLOG(5) << "AsyncSocket::handleErrMessages() this=" << this << ", fd=" << fd_ VLOG(5) << "AsyncSocket::handleErrMessages() this=" << this << ", fd=" << fd_
<< ", state=" << state_; << ", state=" << state_;
if (errMessageCallback_ == nullptr && if (errMessageCallback_ == nullptr && idZeroCopyBufPtrMap_.empty()) {
(!zeroCopyEnabled_ || idZeroCopyBufPtrMap_.empty())) {
VLOG(7) << "AsyncSocket::handleErrMessages(): " VLOG(7) << "AsyncSocket::handleErrMessages(): "
<< "no callback installed - exiting."; << "no callback installed - exiting.";
return; return;
...@@ -2358,6 +2326,14 @@ AsyncSocket::WriteResult AsyncSocket::performWrite( ...@@ -2358,6 +2326,14 @@ AsyncSocket::WriteResult AsyncSocket::performWrite(
// this bug is fixed. // this bug is fixed.
tryAgain |= (errno == ENOTCONN); tryAgain |= (errno == ENOTCONN);
#endif #endif
// workaround for running with zerocopy enabled but without a proper
// memlock value - see ulimit -l
if (zeroCopyEnabled_ && (errno == ENOBUFS)) {
tryAgain = true;
zeroCopyEnabled_ = false;
}
if (!writeResult.exception && tryAgain) { if (!writeResult.exception && tryAgain) {
// TCP buffer is full; we can't write any more data right now. // TCP buffer is full; we can't write any more data right now.
*countWritten = 0; *countWritten = 0;
......
...@@ -505,18 +505,11 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -505,18 +505,11 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
void setReadCB(ReadCallback* callback) override; void setReadCB(ReadCallback* callback) override;
ReadCallback* getReadCallback() const override; ReadCallback* getReadCallback() const override;
static const size_t kDefaultZeroCopyThreshold = 0;
bool setZeroCopy(bool enable); bool setZeroCopy(bool enable);
bool getZeroCopy() const { bool getZeroCopy() const {
return zeroCopyEnabled_; return zeroCopyEnabled_;
} }
void setZeroCopyWriteChainThreshold(size_t threshold);
size_t getZeroCopyWriteChainThreshold() const {
return zeroCopyWriteChainThreshold_;
}
uint32_t getZeroCopyBufId() const { uint32_t getZeroCopyBufId() const {
return zeroCopyBufId_; return zeroCopyBufId_;
} }
...@@ -1164,11 +1157,7 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -1164,11 +1157,7 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
uint32_t getNextZeroCopyBufId() { uint32_t getNextZeroCopyBufId() {
return zeroCopyBufId_++; return zeroCopyBufId_++;
} }
void adjustZeroCopyFlags(folly::IOBuf* buf, folly::WriteFlags& flags); void adjustZeroCopyFlags(folly::WriteFlags& flags);
void adjustZeroCopyFlags(
const iovec* vec,
uint32_t count,
folly::WriteFlags& flags);
void addZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf); void addZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf);
void addZeroCopyBuf(folly::IOBuf* ptr); void addZeroCopyBuf(folly::IOBuf* ptr);
void setZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf); void setZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf);
...@@ -1240,7 +1229,6 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -1240,7 +1229,6 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
bool trackEor_{false}; bool trackEor_{false};
bool zeroCopyEnabled_{false}; bool zeroCopyEnabled_{false};
bool zeroCopyVal_{false}; bool zeroCopyVal_{false};
size_t zeroCopyWriteChainThreshold_{kDefaultZeroCopyThreshold};
}; };
#ifdef _MSC_VER #ifdef _MSC_VER
#pragma vtordisp(pop) #pragma vtordisp(pop)
......
...@@ -27,8 +27,6 @@ ...@@ -27,8 +27,6 @@
using namespace folly; using namespace folly;
static constexpr auto const kZeroCopyThreshold = 4096;
class TestAsyncSocket { class TestAsyncSocket {
public: public:
explicit TestAsyncSocket( explicit TestAsyncSocket(
...@@ -79,9 +77,6 @@ class TestAsyncSocket { ...@@ -79,9 +77,6 @@ class TestAsyncSocket {
zeroCopy_ = enable; zeroCopy_ = enable;
if (sock_) { if (sock_) {
sock_->setZeroCopy(zeroCopy_); sock_->setZeroCopy(zeroCopy_);
if (zeroCopy_) {
sock_->setZeroCopyWriteChainThreshold(kZeroCopyThreshold);
}
} }
} }
......
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