Commit 76fcf389 authored by Yang Chi's avatar Yang Chi Committed by facebook-github-bot-1

Remove per-write buffer callback from AsyncSocket

Summary: Remove per-write buffer callback from AsyncSocket

Reviewed By: afrind

Differential Revision: D2733142

fb-gh-sync-id: 91214a8c833bbd479bf049c2bb72d660e0c30f50
parent 34de70e2
...@@ -148,8 +148,7 @@ void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf, ...@@ -148,8 +148,7 @@ void AsyncPipeWriter::write(unique_ptr<folly::IOBuf> buf,
void AsyncPipeWriter::writeChain(folly::AsyncWriter::WriteCallback* callback, void AsyncPipeWriter::writeChain(folly::AsyncWriter::WriteCallback* callback,
std::unique_ptr<folly::IOBuf>&& buf, std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags, WriteFlags) {
BufferCallback*) {
write(std::move(buf), callback); write(std::move(buf), callback);
} }
......
...@@ -147,20 +147,21 @@ class AsyncPipeWriter : public EventHandler, ...@@ -147,20 +147,21 @@ class AsyncPipeWriter : public EventHandler,
} }
// AsyncWriter methods // AsyncWriter methods
void write(folly::AsyncWriter::WriteCallback* callback, const void* buf, void write(folly::AsyncWriter::WriteCallback* callback,
size_t bytes, WriteFlags flags = WriteFlags::NONE, const void* buf,
BufferCallback* bufCallback = nullptr) override { size_t bytes,
writeChain(callback, IOBuf::wrapBuffer(buf, bytes), flags, bufCallback); WriteFlags flags = WriteFlags::NONE) override {
writeChain(callback, IOBuf::wrapBuffer(buf, bytes), flags);
} }
void writev(folly::AsyncWriter::WriteCallback*, const iovec*, void writev(folly::AsyncWriter::WriteCallback*,
size_t, WriteFlags = WriteFlags::NONE, const iovec*,
BufferCallback* = nullptr) 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(folly::AsyncWriter::WriteCallback* callback,
std::unique_ptr<folly::IOBuf>&& buf, std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override;
BufferCallback* bufCallback = nullptr) override;
private: private:
void handlerReady(uint16_t events) noexcept override; void handlerReady(uint16_t events) noexcept override;
......
...@@ -63,16 +63,14 @@ const AsyncSocketException socketShutdownForWritesEx( ...@@ -63,16 +63,14 @@ const AsyncSocketException socketShutdownForWritesEx(
*/ */
class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest { class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest {
public: public:
static BytesWriteRequest* newRequest( static BytesWriteRequest* newRequest(AsyncSocket* socket,
AsyncSocket* socket,
WriteCallback* callback, WriteCallback* callback,
const iovec* ops, const iovec* ops,
uint32_t opCount, uint32_t opCount,
uint32_t partialWritten, uint32_t partialWritten,
uint32_t bytesWritten, uint32_t bytesWritten,
unique_ptr<IOBuf>&& ioBuf, unique_ptr<IOBuf>&& ioBuf,
WriteFlags flags, WriteFlags flags) {
BufferCallback* bufferCallback = nullptr) {
assert(opCount > 0); assert(opCount > 0);
// Since we put a variable size iovec array at the end // Since we put a variable size iovec array at the end
// of each BytesWriteRequest, we have to manually allocate the memory. // of each BytesWriteRequest, we have to manually allocate the memory.
...@@ -84,7 +82,7 @@ class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest { ...@@ -84,7 +82,7 @@ class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest {
return new(buf) BytesWriteRequest(socket, callback, ops, opCount, return new(buf) BytesWriteRequest(socket, callback, ops, opCount,
partialWritten, bytesWritten, partialWritten, bytesWritten,
std::move(ioBuf), flags, bufferCallback); std::move(ioBuf), flags);
} }
void destroy() override { void destroy() override {
...@@ -138,9 +136,8 @@ class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest { ...@@ -138,9 +136,8 @@ class AsyncSocket::BytesWriteRequest : public AsyncSocket::WriteRequest {
uint32_t partialBytes, uint32_t partialBytes,
uint32_t bytesWritten, uint32_t bytesWritten,
unique_ptr<IOBuf>&& ioBuf, unique_ptr<IOBuf>&& ioBuf,
WriteFlags flags, WriteFlags flags)
BufferCallback* bufferCallback = nullptr) : AsyncSocket::WriteRequest(socket, callback)
: AsyncSocket::WriteRequest(socket, callback, bufferCallback)
, opCount_(opCount) , opCount_(opCount)
, opIndex_(0) , opIndex_(0)
, flags_(flags) , flags_(flags)
...@@ -611,46 +608,43 @@ AsyncSocket::ReadCallback* AsyncSocket::getReadCallback() const { ...@@ -611,46 +608,43 @@ AsyncSocket::ReadCallback* AsyncSocket::getReadCallback() const {
} }
void AsyncSocket::write(WriteCallback* callback, void AsyncSocket::write(WriteCallback* callback,
const void* buf, size_t bytes, WriteFlags flags, const void* buf, size_t bytes, WriteFlags flags) {
BufferCallback* bufCallback) {
iovec op; iovec op;
op.iov_base = const_cast<void*>(buf); op.iov_base = const_cast<void*>(buf);
op.iov_len = bytes; op.iov_len = bytes;
writeImpl(callback, &op, 1, unique_ptr<IOBuf>(), flags, bufCallback); writeImpl(callback, &op, 1, unique_ptr<IOBuf>(), flags);
} }
void AsyncSocket::writev(WriteCallback* callback, void AsyncSocket::writev(WriteCallback* callback,
const iovec* vec, const iovec* vec,
size_t count, size_t count,
WriteFlags flags, WriteFlags flags) {
BufferCallback* bufCallback) { writeImpl(callback, vec, count, unique_ptr<IOBuf>(), flags);
writeImpl(callback, vec, count, unique_ptr<IOBuf>(), flags, bufCallback);
} }
void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf, void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf,
WriteFlags flags, BufferCallback* bufCallback) { WriteFlags flags) {
constexpr size_t kSmallSizeMax = 64; constexpr size_t kSmallSizeMax = 64;
size_t count = buf->countChainElements(); size_t count = buf->countChainElements();
if (count <= kSmallSizeMax) { if (count <= kSmallSizeMax) {
iovec vec[BOOST_PP_IF(FOLLY_HAVE_VLA, count, kSmallSizeMax)]; iovec vec[BOOST_PP_IF(FOLLY_HAVE_VLA, count, kSmallSizeMax)];
writeChainImpl(callback, vec, count, std::move(buf), flags, bufCallback); writeChainImpl(callback, vec, count, std::move(buf), flags);
} else { } else {
iovec* vec = new iovec[count]; iovec* vec = new iovec[count];
writeChainImpl(callback, vec, count, std::move(buf), flags, bufCallback); writeChainImpl(callback, vec, count, std::move(buf), flags);
delete[] vec; delete[] vec;
} }
} }
void AsyncSocket::writeChainImpl(WriteCallback* callback, iovec* vec, void AsyncSocket::writeChainImpl(WriteCallback* callback, iovec* vec,
size_t count, unique_ptr<IOBuf>&& buf, WriteFlags flags, size_t count, unique_ptr<IOBuf>&& buf, WriteFlags flags) {
BufferCallback* bufCallback) {
size_t veclen = buf->fillIov(vec, count); size_t veclen = buf->fillIov(vec, count);
writeImpl(callback, vec, veclen, std::move(buf), flags, bufCallback); writeImpl(callback, vec, veclen, std::move(buf), flags);
} }
void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec, void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec,
size_t count, unique_ptr<IOBuf>&& buf, size_t count, unique_ptr<IOBuf>&& buf,
WriteFlags flags, BufferCallback* bufCallback) { WriteFlags flags) {
VLOG(6) << "AsyncSocket::writev() this=" << this << ", fd=" << fd_ VLOG(6) << "AsyncSocket::writev() this=" << this << ", fd=" << fd_
<< ", callback=" << callback << ", count=" << count << ", callback=" << callback << ", count=" << count
<< ", state=" << state_; << ", state=" << state_;
...@@ -694,11 +688,7 @@ void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec, ...@@ -694,11 +688,7 @@ void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec,
callback->writeSuccess(); callback->writeSuccess();
} }
return; return;
} else { // continue writing the next writeReq } // else { continue writing the next writeReq }
if (bufCallback) {
bufCallback->onEgressBuffered();
}
}
mustRegister = true; mustRegister = true;
} }
} else if (!connecting()) { } else if (!connecting()) {
...@@ -711,8 +701,7 @@ void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec, ...@@ -711,8 +701,7 @@ void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec,
try { try {
req = BytesWriteRequest::newRequest(this, callback, vec + countWritten, req = BytesWriteRequest::newRequest(this, callback, vec + countWritten,
count - countWritten, partialWritten, count - countWritten, partialWritten,
bytesWritten, std::move(ioBuf), flags, bytesWritten, std::move(ioBuf), flags);
bufCallback);
} catch (const std::exception& ex) { } catch (const std::exception& ex) {
// we mainly expect to catch std::bad_alloc here // we mainly expect to catch std::bad_alloc here
AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR, AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
...@@ -1515,11 +1504,6 @@ void AsyncSocket::handleWrite() noexcept { ...@@ -1515,11 +1504,6 @@ void AsyncSocket::handleWrite() noexcept {
} }
// We'll continue around the loop, trying to write another request // We'll continue around the loop, trying to write another request
} else { } else {
// Notify BufferCallback:
BufferCallback* bufferCallback = writeReqHead_->getBufferCallback();
if (bufferCallback) {
bufferCallback->onEgressBuffered();
}
// Partial write. // Partial write.
writeReqHead_->consume(); writeReqHead_->consume();
// Stop after a partial write; it's highly likely that a subsequent write // Stop after a partial write; it's highly likely that a subsequent write
......
...@@ -328,15 +328,12 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -328,15 +328,12 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
ReadCallback* getReadCallback() const override; ReadCallback* getReadCallback() const override;
void write(WriteCallback* callback, const void* buf, size_t bytes, void write(WriteCallback* callback, const void* buf, size_t bytes,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override;
BufferCallback* bufCallback = nullptr) override;
void writev(WriteCallback* callback, const iovec* vec, size_t count, void writev(WriteCallback* callback, const iovec* vec, size_t count,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override;
BufferCallback* bufCallback = nullptr) override;
void writeChain(WriteCallback* callback, void writeChain(WriteCallback* callback,
std::unique_ptr<folly::IOBuf>&& buf, std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override;
BufferCallback* bufCallback = nullptr) override;
class WriteRequest; class WriteRequest;
virtual void writeRequest(WriteRequest* req); virtual void writeRequest(WriteRequest* req);
...@@ -518,11 +515,8 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -518,11 +515,8 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
*/ */
class WriteRequest { class WriteRequest {
public: public:
WriteRequest( WriteRequest(AsyncSocket* socket, WriteCallback* callback) :
AsyncSocket* socket, socket_(socket), callback_(callback) {}
WriteCallback* callback,
BufferCallback* bufferCallback = nullptr) :
socket_(socket), callback_(callback), bufferCallback_(bufferCallback) {}
virtual void start() {}; virtual void start() {};
...@@ -560,10 +554,6 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -560,10 +554,6 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
socket_->appBytesWritten_ += count; socket_->appBytesWritten_ += count;
} }
BufferCallback* getBufferCallback() const {
return bufferCallback_;
}
protected: protected:
// protected destructor, to ensure callers use destroy() // protected destructor, to ensure callers use destroy()
virtual ~WriteRequest() {} virtual ~WriteRequest() {}
...@@ -572,7 +562,6 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -572,7 +562,6 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
WriteRequest* next_{nullptr}; ///< pointer to next WriteRequest WriteRequest* next_{nullptr}; ///< pointer to next WriteRequest
WriteCallback* callback_; ///< completion callback WriteCallback* callback_; ///< completion callback
uint32_t totalBytesWritten_{0}; ///< total bytes written uint32_t totalBytesWritten_{0}; ///< total bytes written
BufferCallback* bufferCallback_{nullptr};
}; };
protected: protected:
...@@ -701,11 +690,10 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -701,11 +690,10 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
* @param count Number of IOBufs to write, beginning at start of buf. * @param count Number of IOBufs to write, beginning at start of buf.
* @param buf Chain of iovecs. * @param buf Chain of iovecs.
* @param flags set of flags for the underlying write calls, like cork * @param flags set of flags for the underlying write calls, like cork
* @param bufCallback Callback when egress data begins to buffer
*/ */
void writeChainImpl(WriteCallback* callback, iovec* vec, void writeChainImpl(WriteCallback* callback, iovec* vec,
size_t count, std::unique_ptr<folly::IOBuf>&& buf, size_t count, std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags flags, BufferCallback* bufCallback = nullptr); WriteFlags flags);
/** /**
* Write as much data as possible to the socket without blocking, * Write as much data as possible to the socket without blocking,
...@@ -723,12 +711,10 @@ class AsyncSocket : virtual public AsyncTransportWrapper { ...@@ -723,12 +711,10 @@ class AsyncSocket : virtual public AsyncTransportWrapper {
* the IOBuf is transferred here; upon completion of * the IOBuf is transferred here; upon completion of
* the write, the AsyncSocket deletes the IOBuf. * the write, the AsyncSocket deletes the IOBuf.
* @param flags Set of write flags. * @param flags Set of write flags.
* @param bufCallback Callback when egress data buffers up
*/ */
void writeImpl(WriteCallback* callback, const iovec* vec, size_t count, void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
std::unique_ptr<folly::IOBuf>&& buf, std::unique_ptr<folly::IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE);
BufferCallback* bufCallback = nullptr);
/** /**
* Attempt to write to the socket. * Attempt to write to the socket.
......
...@@ -464,12 +464,6 @@ class AsyncReader { ...@@ -464,12 +464,6 @@ class AsyncReader {
class AsyncWriter { class AsyncWriter {
public: public:
class BufferCallback {
public:
virtual ~BufferCallback() {}
virtual void onEgressBuffered() = 0;
};
class WriteCallback { class WriteCallback {
public: public:
virtual ~WriteCallback() = default; virtual ~WriteCallback() = default;
...@@ -499,15 +493,12 @@ class AsyncWriter { ...@@ -499,15 +493,12 @@ class AsyncWriter {
// 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(WriteCallback* callback, const void* buf, size_t bytes,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) = 0;
BufferCallback* bufCallback = nullptr) = 0;
virtual void writev(WriteCallback* callback, const iovec* vec, size_t count, virtual void writev(WriteCallback* callback, const iovec* vec, size_t count,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) = 0;
BufferCallback* bufCallback = nullptr) = 0;
virtual void writeChain(WriteCallback* callback, virtual void writeChain(WriteCallback* callback,
std::unique_ptr<IOBuf>&& buf, std::unique_ptr<IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) = 0;
BufferCallback* bufCallback = nullptr) = 0;
protected: protected:
virtual ~AsyncWriter() = default; virtual ~AsyncWriter() = default;
...@@ -525,19 +516,15 @@ class AsyncTransportWrapper : virtual public AsyncTransport, ...@@ -525,19 +516,15 @@ class AsyncTransportWrapper : virtual public AsyncTransport,
// to keep compatibility. // to keep compatibility.
using ReadCallback = AsyncReader::ReadCallback; using ReadCallback = AsyncReader::ReadCallback;
using WriteCallback = AsyncWriter::WriteCallback; using WriteCallback = AsyncWriter::WriteCallback;
using BufferCallback = AsyncWriter::BufferCallback;
virtual void setReadCB(ReadCallback* callback) override = 0; virtual void setReadCB(ReadCallback* callback) override = 0;
virtual ReadCallback* getReadCallback() const override = 0; virtual ReadCallback* getReadCallback() const override = 0;
virtual void write(WriteCallback* callback, const void* buf, size_t bytes, virtual void write(WriteCallback* callback, const void* buf, size_t bytes,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override = 0;
BufferCallback* bufCallback = nullptr) override = 0;
virtual void writev(WriteCallback* callback, const iovec* vec, size_t count, virtual void writev(WriteCallback* callback, const iovec* vec, size_t count,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override = 0;
BufferCallback* bufCallback = nullptr) override = 0;
virtual void writeChain(WriteCallback* callback, virtual void writeChain(WriteCallback* callback,
std::unique_ptr<IOBuf>&& buf, std::unique_ptr<IOBuf>&& buf,
WriteFlags flags = WriteFlags::NONE, WriteFlags flags = WriteFlags::NONE) override = 0;
BufferCallback* bufCallback = nullptr) override = 0;
/** /**
* The transport wrapper may wrap another transport. This returns the * The transport wrapper may wrap another transport. This returns the
* transport that is wrapped. It returns nullptr if there is no wrapped * transport that is wrapped. It returns nullptr if there is no wrapped
......
...@@ -60,23 +60,6 @@ class ConnCallback : public AsyncSocket::ConnectCallback { ...@@ -60,23 +60,6 @@ class ConnCallback : public AsyncSocket::ConnectCallback {
VoidCallback errorCallback; VoidCallback errorCallback;
}; };
class BufferCallback : public AsyncTransportWrapper::BufferCallback {
public:
BufferCallback()
: buffered_(false) {}
void onEgressBuffered() override {
buffered_ = true;
}
bool hasBuffered() const {
return buffered_;
}
private:
bool buffered_{false};
};
class WriteCallback : public AsyncTransportWrapper::WriteCallback { class WriteCallback : public AsyncTransportWrapper::WriteCallback {
public: public:
WriteCallback() WriteCallback()
......
...@@ -2238,32 +2238,3 @@ TEST(AsyncSocketTest, NumPendingMessagesInQueue) { ...@@ -2238,32 +2238,3 @@ TEST(AsyncSocketTest, NumPendingMessagesInQueue) {
eventBase.loop(); eventBase.loop();
} }
TEST(AsyncSocketTest, BufferTest) {
TestServer server;
EventBase evb;
AsyncSocket::OptionMap option{{{SOL_SOCKET, SO_SNDBUF}, 128}};
std::shared_ptr<AsyncSocket> socket = AsyncSocket::newSocket(&evb);
ConnCallback ccb;
socket->connect(&ccb, server.getAddress(), 30, option);
char buf[100 * 1024];
memset(buf, 'c', sizeof(buf));
WriteCallback wcb;
BufferCallback bcb;
socket->write(&wcb, buf, sizeof(buf), WriteFlags::NONE, &bcb);
evb.loop();
CHECK_EQ(ccb.state, STATE_SUCCEEDED);
CHECK_EQ(wcb.state, STATE_SUCCEEDED);
ASSERT_TRUE(bcb.hasBuffered());
socket->close();
server.verifyConnection(buf, sizeof(buf));
ASSERT_TRUE(socket->isClosedBySelf());
ASSERT_FALSE(socket->isClosedByPeer());
}
...@@ -27,31 +27,15 @@ class MockAsyncTransport: public AsyncTransportWrapper { ...@@ -27,31 +27,15 @@ class MockAsyncTransport: public AsyncTransportWrapper {
MOCK_METHOD1(setReadCB, void(ReadCallback*)); MOCK_METHOD1(setReadCB, void(ReadCallback*));
MOCK_CONST_METHOD0(getReadCallback, ReadCallback*()); MOCK_CONST_METHOD0(getReadCallback, ReadCallback*());
MOCK_CONST_METHOD0(getReadCB, ReadCallback*()); MOCK_CONST_METHOD0(getReadCB, ReadCallback*());
MOCK_METHOD5(write, void(WriteCallback*, MOCK_METHOD4(write, void(WriteCallback*, const void*, size_t, WriteFlags));
const void*, size_t, MOCK_METHOD4(writev, void(WriteCallback*, const iovec*, size_t, WriteFlags));
WriteFlags, MOCK_METHOD3(writeChain,
BufferCallback*)); void(WriteCallback*, std::shared_ptr<folly::IOBuf>, WriteFlags));
MOCK_METHOD5(writev, void(WriteCallback*,
const iovec*, size_t,
WriteFlags,
BufferCallback*));
MOCK_METHOD4(writeChain,
void(WriteCallback*,
std::shared_ptr<folly::IOBuf>,
WriteFlags,
BufferCallback*));
void writeChain(WriteCallback* callback, void writeChain(WriteCallback* callback,
std::unique_ptr<folly::IOBuf>&& iob, std::unique_ptr<folly::IOBuf>&& iob,
WriteFlags flags = WriteFlags flags = WriteFlags::NONE) override {
WriteFlags::NONE, writeChain(callback, std::shared_ptr<folly::IOBuf>(iob.release()), flags);
BufferCallback* bufCB = nullptr) override {
writeChain(
callback,
std::shared_ptr<folly::IOBuf>(iob.release()),
flags,
bufCB);
} }
MOCK_METHOD0(close, void()); MOCK_METHOD0(close, void());
......
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