Commit 1185881d authored by James Sedgwick's avatar James Sedgwick Committed by Viswanath Sivakumar

transportActive and transportInactive

Summary:
These are equivalents to Netty's channelActive and channelInactive, but we've been calling channels transports so I'm staying consistent.
I skipped integrating this into TAsyncTransportHandler because thrift still does manual CB attachment/detachment and it's unclear how that fits into this model
If my suspicions are correct, it *should* be possible to make attachReadCallback and detachReadCallback private in AsyncSocketHandler, right? And perhaps get rid of the event base modifier methods? What's our use case for those?

Test Plan: unit, employ in telnet server

Reviewed By: davejwatson@fb.com

Subscribers: fugalh, alandau, bmatheny, folly-diffs@, jsedgwick, yfeldblum, chalfant

FB internal diff: D2044520

Signature: t1:2044520:1431551998:af1de358b5dbefcca148814015d8e9f63f458d5d
parent f639b454
...@@ -36,6 +36,9 @@ class ClientBootstrap { ...@@ -36,6 +36,9 @@ class ClientBootstrap {
, bootstrap_(bootstrap) {} , bootstrap_(bootstrap) {}
void connectSuccess() noexcept override { void connectSuccess() noexcept override {
if (bootstrap_->getPipeline()) {
bootstrap_->getPipeline()->transportActive();
}
promise_.setValue(bootstrap_->getPipeline()); promise_.setValue(bootstrap_->getPipeline());
delete this; delete this;
} }
...@@ -77,9 +80,6 @@ class ClientBootstrap { ...@@ -77,9 +80,6 @@ class ClientBootstrap {
socket->connect( socket->connect(
new ConnectCallback(std::move(promise), this), address); new ConnectCallback(std::move(promise), this), address);
pipeline_ = pipelineFactory_->newPipeline(socket); pipeline_ = pipelineFactory_->newPipeline(socket);
if (pipeline_) {
pipeline_->attachTransport(socket);
}
}); });
return retval; return retval;
} }
......
...@@ -89,6 +89,7 @@ class ServerAcceptor ...@@ -89,6 +89,7 @@ class ServerAcceptor
std::shared_ptr<AsyncSocket>( std::shared_ptr<AsyncSocket>(
transport.release(), transport.release(),
folly::DelayedDestruction::Destructor()))); folly::DelayedDestruction::Destructor())));
pipeline->transportActive();
auto connection = new ServerConnection(std::move(pipeline)); auto connection = new ServerConnection(std::move(pipeline));
Acceptor::addConnection(connection); Acceptor::addConnection(connection);
} }
......
...@@ -37,9 +37,7 @@ class AsyncSocketHandler ...@@ -37,9 +37,7 @@ class AsyncSocketHandler
AsyncSocketHandler(AsyncSocketHandler&&) = default; AsyncSocketHandler(AsyncSocketHandler&&) = default;
~AsyncSocketHandler() { ~AsyncSocketHandler() {
if (socket_) { detachReadCallback();
detachReadCallback();
}
} }
void attachReadCallback() { void attachReadCallback() {
...@@ -47,9 +45,14 @@ class AsyncSocketHandler ...@@ -47,9 +45,14 @@ class AsyncSocketHandler
} }
void detachReadCallback() { void detachReadCallback() {
if (socket_->getReadCallback() == this) { if (socket_ && socket_->getReadCallback() == this) {
socket_->setReadCB(nullptr); socket_->setReadCB(nullptr);
} }
auto ctx = getContext();
if (ctx && !firedInactive_) {
firedInactive_ = true;
ctx->fireTransportInactive();
}
} }
void attachEventBase(folly::EventBase* eventBase) { void attachEventBase(folly::EventBase* eventBase) {
...@@ -65,8 +68,14 @@ class AsyncSocketHandler ...@@ -65,8 +68,14 @@ class AsyncSocketHandler
} }
} }
void attachPipeline(Context* ctx) override { void transportActive(Context* ctx) override {
ctx->getPipeline()->setTransport(socket_);
attachReadCallback(); attachReadCallback();
ctx->fireTransportActive();
}
void detachPipeline(Context* ctx) override {
detachReadCallback();
} }
folly::Future<void> write( folly::Future<void> write(
...@@ -149,6 +158,7 @@ class AsyncSocketHandler ...@@ -149,6 +158,7 @@ class AsyncSocketHandler
folly::IOBufQueue bufQueue_{folly::IOBufQueue::cacheChainLength()}; folly::IOBufQueue bufQueue_{folly::IOBufQueue::cacheChainLength()};
std::shared_ptr<AsyncSocket> socket_{nullptr}; std::shared_ptr<AsyncSocket> socket_{nullptr};
bool firedInactive_{false};
}; };
}} }}
...@@ -29,10 +29,7 @@ class HandlerBase { ...@@ -29,10 +29,7 @@ class HandlerBase {
virtual ~HandlerBase() {} virtual ~HandlerBase() {}
virtual void attachPipeline(Context* ctx) {} virtual void attachPipeline(Context* ctx) {}
virtual void attachTransport(Context* ctx) {}
virtual void detachPipeline(Context* ctx) {} virtual void detachPipeline(Context* ctx) {}
virtual void detachTransport(Context* ctx) {}
Context* getContext() { Context* getContext() {
if (attachCount_ != 1) { if (attachCount_ != 1) {
...@@ -67,6 +64,12 @@ class Handler : public HandlerBase<HandlerContext<Rout, Wout>> { ...@@ -67,6 +64,12 @@ class Handler : public HandlerBase<HandlerContext<Rout, Wout>> {
virtual void readException(Context* ctx, exception_wrapper e) { virtual void readException(Context* ctx, exception_wrapper e) {
ctx->fireReadException(std::move(e)); ctx->fireReadException(std::move(e));
} }
virtual void transportActive(Context* ctx) {
ctx->fireTransportActive();
}
virtual void transportInactive(Context* ctx) {
ctx->fireTransportInactive();
}
virtual Future<void> write(Context* ctx, Win msg) = 0; virtual Future<void> write(Context* ctx, Win msg) = 0;
virtual Future<void> close(Context* ctx) { virtual Future<void> close(Context* ctx) {
...@@ -81,8 +84,6 @@ class Handler : public HandlerBase<HandlerContext<Rout, Wout>> { ...@@ -81,8 +84,6 @@ class Handler : public HandlerBase<HandlerContext<Rout, Wout>> {
exception_wrapper e) {} exception_wrapper e) {}
virtual void channelRegistered(HandlerContext* ctx) {} virtual void channelRegistered(HandlerContext* ctx) {}
virtual void channelUnregistered(HandlerContext* ctx) {} virtual void channelUnregistered(HandlerContext* ctx) {}
virtual void channelActive(HandlerContext* ctx) {}
virtual void channelInactive(HandlerContext* ctx) {}
virtual void channelReadComplete(HandlerContext* ctx) {} virtual void channelReadComplete(HandlerContext* ctx) {}
virtual void userEventTriggered(HandlerContext* ctx, void* evt) {} virtual void userEventTriggered(HandlerContext* ctx, void* evt) {}
virtual void channelWritabilityChanged(HandlerContext* ctx) {} virtual void channelWritabilityChanged(HandlerContext* ctx) {}
...@@ -120,6 +121,12 @@ class InboundHandler : public HandlerBase<InboundHandlerContext<Rout>> { ...@@ -120,6 +121,12 @@ class InboundHandler : public HandlerBase<InboundHandlerContext<Rout>> {
virtual void readException(Context* ctx, exception_wrapper e) { virtual void readException(Context* ctx, exception_wrapper e) {
ctx->fireReadException(std::move(e)); ctx->fireReadException(std::move(e));
} }
virtual void transportActive(Context* ctx) {
ctx->fireTransportActive();
}
virtual void transportInactive(Context* ctx) {
ctx->fireTransportInactive();
}
}; };
template <class Win, class Wout = Win> template <class Win, class Wout = Win>
......
...@@ -25,9 +25,6 @@ class PipelineContext { ...@@ -25,9 +25,6 @@ class PipelineContext {
virtual void attachPipeline() = 0; virtual void attachPipeline() = 0;
virtual void detachPipeline() = 0; virtual void detachPipeline() = 0;
virtual void attachTransport() = 0;
virtual void detachTransport() = 0;
template <class H, class HandlerContext> template <class H, class HandlerContext>
void attachContext(H* handler, HandlerContext* ctx) { void attachContext(H* handler, HandlerContext* ctx) {
if (++handler->attachCount_ == 1) { if (++handler->attachCount_ == 1) {
...@@ -48,6 +45,8 @@ class InboundLink { ...@@ -48,6 +45,8 @@ class InboundLink {
virtual void read(In msg) = 0; virtual void read(In msg) = 0;
virtual void readEOF() = 0; virtual void readEOF() = 0;
virtual void readException(exception_wrapper e) = 0; virtual void readException(exception_wrapper e) = 0;
virtual void transportActive() = 0;
virtual void transportInactive() = 0;
}; };
template <class Out> template <class Out>
...@@ -86,16 +85,6 @@ class ContextImplBase : public PipelineContext { ...@@ -86,16 +85,6 @@ class ContextImplBase : public PipelineContext {
attached_ = false; attached_ = false;
} }
void attachTransport() override {
DestructorGuard dg(pipeline_);
handler_->attachTransport(impl_);
}
void detachTransport() override {
DestructorGuard dg(pipeline_);
handler_->detachTransport(impl_);
}
void setNextIn(PipelineContext* ctx) override { void setNextIn(PipelineContext* ctx) override {
auto nextIn = dynamic_cast<InboundLink<typename H::rout>*>(ctx); auto nextIn = dynamic_cast<InboundLink<typename H::rout>*>(ctx);
if (nextIn) { if (nextIn) {
...@@ -181,6 +170,20 @@ class ContextImpl ...@@ -181,6 +170,20 @@ class ContextImpl
} }
} }
void fireTransportActive() override {
DestructorGuard dg(this->pipeline_);
if (this->nextIn_) {
this->nextIn_->transportActive();
}
}
void fireTransportInactive() override {
DestructorGuard dg(this->pipeline_);
if (this->nextIn_) {
this->nextIn_->transportInactive();
}
}
Future<void> fireWrite(Wout msg) override { Future<void> fireWrite(Wout msg) override {
DestructorGuard dg(this->pipeline_); DestructorGuard dg(this->pipeline_);
if (this->nextOut_) { if (this->nextOut_) {
...@@ -205,10 +208,6 @@ class ContextImpl ...@@ -205,10 +208,6 @@ class ContextImpl
return this->pipeline_; return this->pipeline_;
} }
std::shared_ptr<AsyncTransport> getTransport() override {
return this->pipeline_->getTransport();
}
void setWriteFlags(WriteFlags flags) override { void setWriteFlags(WriteFlags flags) override {
this->pipeline_->setWriteFlags(flags); this->pipeline_->setWriteFlags(flags);
} }
...@@ -243,6 +242,16 @@ class ContextImpl ...@@ -243,6 +242,16 @@ class ContextImpl
this->handler_->readException(this, std::move(e)); this->handler_->readException(this, std::move(e));
} }
void transportActive() override {
DestructorGuard dg(this->pipeline_);
this->handler_->transportActive(this);
}
void transportInactive() override {
DestructorGuard dg(this->pipeline_);
this->handler_->transportInactive(this);
}
// OutboundLink overrides // OutboundLink overrides
Future<void> write(Win msg) override { Future<void> write(Win msg) override {
DestructorGuard dg(this->pipeline_); DestructorGuard dg(this->pipeline_);
...@@ -310,12 +319,22 @@ class InboundContextImpl ...@@ -310,12 +319,22 @@ class InboundContextImpl
} }
} }
PipelineBase* getPipeline() override { void fireTransportActive() override {
return this->pipeline_; DestructorGuard dg(this->pipeline_);
if (this->nextIn_) {
this->nextIn_->transportActive();
}
} }
std::shared_ptr<AsyncTransport> getTransport() override { void fireTransportInactive() override {
return this->pipeline_->getTransport(); DestructorGuard dg(this->pipeline_);
if (this->nextIn_) {
this->nextIn_->transportInactive();
}
}
PipelineBase* getPipeline() override {
return this->pipeline_;
} }
// InboundLink overrides // InboundLink overrides
...@@ -334,6 +353,16 @@ class InboundContextImpl ...@@ -334,6 +353,16 @@ class InboundContextImpl
this->handler_->readException(this, std::move(e)); this->handler_->readException(this, std::move(e));
} }
void transportActive() override {
DestructorGuard dg(this->pipeline_);
this->handler_->transportActive(this);
}
void transportInactive() override {
DestructorGuard dg(this->pipeline_);
this->handler_->transportInactive(this);
}
private: private:
using DestructorGuard = typename P::DestructorGuard; using DestructorGuard = typename P::DestructorGuard;
}; };
...@@ -387,10 +416,6 @@ class OutboundContextImpl ...@@ -387,10 +416,6 @@ class OutboundContextImpl
return this->pipeline_; return this->pipeline_;
} }
std::shared_ptr<AsyncTransport> getTransport() override {
return this->pipeline_->getTransport();
}
// OutboundLink overrides // OutboundLink overrides
Future<void> write(Win msg) override { Future<void> write(Win msg) override {
DestructorGuard dg(this->pipeline_); DestructorGuard dg(this->pipeline_);
......
...@@ -32,13 +32,16 @@ class HandlerContext { ...@@ -32,13 +32,16 @@ class HandlerContext {
virtual void fireRead(In msg) = 0; virtual void fireRead(In msg) = 0;
virtual void fireReadEOF() = 0; virtual void fireReadEOF() = 0;
virtual void fireReadException(exception_wrapper e) = 0; virtual void fireReadException(exception_wrapper e) = 0;
virtual void fireTransportActive() = 0;
virtual void fireTransportInactive() = 0;
virtual Future<void> fireWrite(Out msg) = 0; virtual Future<void> fireWrite(Out msg) = 0;
virtual Future<void> fireClose() = 0; virtual Future<void> fireClose() = 0;
virtual PipelineBase* getPipeline() = 0; virtual PipelineBase* getPipeline() = 0;
std::shared_ptr<AsyncTransport> getTransport() {
virtual std::shared_ptr<AsyncTransport> getTransport() = 0; return getPipeline()->getTransport();
}
virtual void setWriteFlags(WriteFlags flags) = 0; virtual void setWriteFlags(WriteFlags flags) = 0;
virtual WriteFlags getWriteFlags() = 0; virtual WriteFlags getWriteFlags() = 0;
...@@ -67,10 +70,13 @@ class InboundHandlerContext { ...@@ -67,10 +70,13 @@ class InboundHandlerContext {
virtual void fireRead(In msg) = 0; virtual void fireRead(In msg) = 0;
virtual void fireReadEOF() = 0; virtual void fireReadEOF() = 0;
virtual void fireReadException(exception_wrapper e) = 0; virtual void fireReadException(exception_wrapper e) = 0;
virtual void fireTransportActive() = 0;
virtual void fireTransportInactive() = 0;
virtual PipelineBase* getPipeline() = 0; virtual PipelineBase* getPipeline() = 0;
std::shared_ptr<AsyncTransport> getTransport() {
virtual std::shared_ptr<AsyncTransport> getTransport() = 0; return getPipeline()->getTransport();
}
// TODO Need get/set writeFlags, readBufferSettings? Probably not. // TODO Need get/set writeFlags, readBufferSettings? Probably not.
// Do we even really need them stored in the pipeline at all? // Do we even really need them stored in the pipeline at all?
...@@ -86,8 +92,9 @@ class OutboundHandlerContext { ...@@ -86,8 +92,9 @@ class OutboundHandlerContext {
virtual Future<void> fireClose() = 0; virtual Future<void> fireClose() = 0;
virtual PipelineBase* getPipeline() = 0; virtual PipelineBase* getPipeline() = 0;
std::shared_ptr<AsyncTransport> getTransport() {
virtual std::shared_ptr<AsyncTransport> getTransport() = 0; return getPipeline()->getTransport();
}
}; };
enum class HandlerDir { enum class HandlerDir {
......
...@@ -35,11 +35,6 @@ Pipeline<R, W>::~Pipeline() { ...@@ -35,11 +35,6 @@ Pipeline<R, W>::~Pipeline() {
} }
} }
template <class R, class W>
std::shared_ptr<AsyncTransport> Pipeline<R, W>::getTransport() {
return transport_;
}
template <class R, class W> template <class R, class W>
void Pipeline<R, W>::setWriteFlags(WriteFlags flags) { void Pipeline<R, W>::setWriteFlags(WriteFlags flags) {
writeFlags_ = flags; writeFlags_ = flags;
...@@ -82,6 +77,24 @@ Pipeline<R, W>::readEOF() { ...@@ -82,6 +77,24 @@ Pipeline<R, W>::readEOF() {
front_->readEOF(); front_->readEOF();
} }
template <class R, class W>
template <class T>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
Pipeline<R, W>::transportActive() {
if (front_) {
front_->transportActive();
}
}
template <class R, class W>
template <class T>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
Pipeline<R, W>::transportInactive() {
if (front_) {
front_->transportInactive();
}
}
template <class R, class W> template <class R, class W>
template <class T> template <class T>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type typename std::enable_if<!std::is_same<T, Nothing>::value>::type
...@@ -221,23 +234,6 @@ bool Pipeline<R, W>::setOwner(H* handler) { ...@@ -221,23 +234,6 @@ bool Pipeline<R, W>::setOwner(H* handler) {
return false; return false;
} }
template <class R, class W>
void Pipeline<R, W>::attachTransport(
std::shared_ptr<AsyncTransport> transport) {
transport_ = std::move(transport);
for (auto& ctx : ctxs_) {
ctx->attachTransport();
}
}
template <class R, class W>
void Pipeline<R, W>::detachTransport() {
transport_ = nullptr;
for (auto& ctx : ctxs_) {
ctx->detachTransport();
}
}
template <class R, class W> template <class R, class W>
template <class Context> template <class Context>
void Pipeline<R, W>::addContextFront(Context* ctx) { void Pipeline<R, W>::addContextFront(Context* ctx) {
......
...@@ -43,8 +43,17 @@ class PipelineBase { ...@@ -43,8 +43,17 @@ class PipelineBase {
} }
} }
void setTransport(std::shared_ptr<AsyncTransport> transport) {
transport_ = transport;
}
std::shared_ptr<AsyncTransport> getTransport() {
return transport_;
}
private: private:
PipelineManager* manager_{nullptr}; PipelineManager* manager_{nullptr};
std::shared_ptr<AsyncTransport> transport_;
}; };
struct Nothing{}; struct Nothing{};
...@@ -63,8 +72,6 @@ class Pipeline : public PipelineBase, public DelayedDestruction { ...@@ -63,8 +72,6 @@ class Pipeline : public PipelineBase, public DelayedDestruction {
Pipeline(); Pipeline();
~Pipeline(); ~Pipeline();
std::shared_ptr<AsyncTransport> getTransport();
void setWriteFlags(WriteFlags flags); void setWriteFlags(WriteFlags flags);
WriteFlags getWriteFlags(); WriteFlags getWriteFlags();
...@@ -83,6 +90,14 @@ class Pipeline : public PipelineBase, public DelayedDestruction { ...@@ -83,6 +90,14 @@ class Pipeline : public PipelineBase, public DelayedDestruction {
typename std::enable_if<!std::is_same<T, Nothing>::value>::type typename std::enable_if<!std::is_same<T, Nothing>::value>::type
readException(exception_wrapper e); readException(exception_wrapper e);
template <class T = R>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
transportActive();
template <class T = R>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
transportInactive();
template <class T = W> template <class T = W>
typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
write(W msg); write(W msg);
...@@ -121,10 +136,6 @@ class Pipeline : public PipelineBase, public DelayedDestruction { ...@@ -121,10 +136,6 @@ class Pipeline : public PipelineBase, public DelayedDestruction {
template <class H> template <class H>
bool setOwner(H* handler); bool setOwner(H* handler);
void attachTransport(std::shared_ptr<AsyncTransport> transport);
void detachTransport();
protected: protected:
explicit Pipeline(bool isStatic); explicit Pipeline(bool isStatic);
...@@ -137,7 +148,6 @@ class Pipeline : public PipelineBase, public DelayedDestruction { ...@@ -137,7 +148,6 @@ class Pipeline : public PipelineBase, public DelayedDestruction {
template <class Context> template <class Context>
Pipeline& addHelper(std::shared_ptr<Context>&& ctx, bool front); Pipeline& addHelper(std::shared_ptr<Context>&& ctx, bool front);
std::shared_ptr<AsyncTransport> transport_;
WriteFlags writeFlags_{WriteFlags::NONE}; WriteFlags writeFlags_{WriteFlags::NONE};
std::pair<uint64_t, uint64_t> readBufferSettings_{2048, 2048}; std::pair<uint64_t, uint64_t> readBufferSettings_{2048, 2048};
......
...@@ -42,8 +42,7 @@ TEST(OutputBufferingHandlerTest, Basic) { ...@@ -42,8 +42,7 @@ TEST(OutputBufferingHandlerTest, Basic) {
EventBase eb; EventBase eb;
auto socket = AsyncSocket::newSocket(&eb); auto socket = AsyncSocket::newSocket(&eb);
EXPECT_CALL(mockHandler, attachTransport(_)); pipeline.setTransport(socket);
pipeline.attachTransport(socket);
// Buffering should prevent writes until the EB loops, and the writes should // Buffering should prevent writes until the EB loops, and the writes should
// be batched into one write call. // be batched into one write call.
......
...@@ -299,21 +299,3 @@ TEST(Pipeline, DynamicConstruction) { ...@@ -299,21 +299,3 @@ TEST(Pipeline, DynamicConstruction) {
.finalize()); .finalize());
} }
} }
TEST(Pipeline, AttachTransport) {
IntHandler handler;
EXPECT_CALL(handler, attachPipeline(_));
StaticPipeline<int, int, IntHandler>
pipeline(&handler);
EventBase eb;
auto socket = AsyncSocket::newSocket(&eb);
EXPECT_CALL(handler, attachTransport(_));
pipeline.attachTransport(socket);
EXPECT_CALL(handler, detachTransport(_));
pipeline.detachTransport();
EXPECT_CALL(handler, detachPipeline(_));
}
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