Commit c7138e7c authored by James Sedgwick's avatar James Sedgwick Committed by Praveen Kumar Ramakrishnan

unidirectional pipelines

Summary:
Cleans up bootstrap a bit at the expense of a more complex Pipeline interface
This doesn't have to go in, lmk either way as I want to move on to reorganizing this code into inl headers etc

Test Plan: unit

Reviewed By: davejwatson@fb.com

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

FB internal diff: D2034634

Signature: t1:2034634:1430414670:c91712fb26353987cb471e35a54f55c869ae7cf1
parent fdd8e84b
...@@ -267,18 +267,12 @@ TEST(Bootstrap, ExistingSocket) { ...@@ -267,18 +267,12 @@ TEST(Bootstrap, ExistingSocket) {
std::atomic<int> connections{0}; std::atomic<int> connections{0};
class TestHandlerPipeline class TestHandlerPipeline : public InboundHandler<void*> {
: public HandlerAdapter<void*,
std::exception> {
public: public:
void read(Context* ctx, void* conn) { void read(Context* ctx, void* conn) {
connections++; connections++;
return ctx->fireRead(conn); return ctx->fireRead(conn);
} }
Future<void> write(Context* ctx, std::exception e) {
return ctx->fireWrite(e);
}
}; };
template <typename HandlerPipeline> template <typename HandlerPipeline>
...@@ -316,17 +310,11 @@ TEST(Bootstrap, LoadBalanceHandler) { ...@@ -316,17 +310,11 @@ TEST(Bootstrap, LoadBalanceHandler) {
CHECK(connections == 1); CHECK(connections == 1);
} }
class TestUDPPipeline class TestUDPPipeline : public InboundHandler<void*> {
: public HandlerAdapter<void*,
std::exception> {
public: public:
void read(Context* ctx, void* conn) { void read(Context* ctx, void* conn) {
connections++; connections++;
} }
Future<void> write(Context* ctx, std::exception e) {
return ctx->fireWrite(e);
}
}; };
TEST(Bootstrap, UDP) { TEST(Bootstrap, UDP) {
......
...@@ -60,8 +60,7 @@ class ServerAcceptor ...@@ -60,8 +60,7 @@ class ServerAcceptor
public: public:
explicit ServerAcceptor( explicit ServerAcceptor(
std::shared_ptr<PipelineFactory<Pipeline>> pipelineFactory, std::shared_ptr<PipelineFactory<Pipeline>> pipelineFactory,
std::shared_ptr<folly::wangle::Pipeline< std::shared_ptr<folly::wangle::Pipeline<void*>> acceptorPipeline,
void*, std::exception>> acceptorPipeline,
EventBase* base) EventBase* base)
: Acceptor(ServerSocketConfig()) : Acceptor(ServerSocketConfig())
, base_(base) , base_(base)
...@@ -105,8 +104,7 @@ class ServerAcceptor ...@@ -105,8 +104,7 @@ class ServerAcceptor
EventBase* base_; EventBase* base_;
std::shared_ptr<PipelineFactory<Pipeline>> childPipelineFactory_; std::shared_ptr<PipelineFactory<Pipeline>> childPipelineFactory_;
std::shared_ptr<folly::wangle::Pipeline< std::shared_ptr<folly::wangle::Pipeline<void*>> acceptorPipeline_;
void*, std::exception>> acceptorPipeline_;
}; };
template <typename Pipeline> template <typename Pipeline>
...@@ -114,22 +112,19 @@ class ServerAcceptorFactory : public AcceptorFactory { ...@@ -114,22 +112,19 @@ class ServerAcceptorFactory : public AcceptorFactory {
public: public:
explicit ServerAcceptorFactory( explicit ServerAcceptorFactory(
std::shared_ptr<PipelineFactory<Pipeline>> factory, std::shared_ptr<PipelineFactory<Pipeline>> factory,
std::shared_ptr<PipelineFactory<folly::wangle::Pipeline< std::shared_ptr<PipelineFactory<folly::wangle::Pipeline<void*>>> pipeline)
void*, std::exception>>> pipeline)
: factory_(factory) : factory_(factory)
, pipeline_(pipeline) {} , pipeline_(pipeline) {}
std::shared_ptr<Acceptor> newAcceptor(EventBase* base) { std::shared_ptr<Acceptor> newAcceptor(EventBase* base) {
std::shared_ptr<folly::wangle::Pipeline< std::shared_ptr<folly::wangle::Pipeline<void*>> pipeline(
void*, std::exception>> pipeline( pipeline_->newPipeline(nullptr));
pipeline_->newPipeline(nullptr));
return std::make_shared<ServerAcceptor<Pipeline>>(factory_, pipeline, base); return std::make_shared<ServerAcceptor<Pipeline>>(factory_, pipeline, base);
} }
private: private:
std::shared_ptr<PipelineFactory<Pipeline>> factory_; std::shared_ptr<PipelineFactory<Pipeline>> factory_;
std::shared_ptr<PipelineFactory< std::shared_ptr<PipelineFactory<
folly::wangle::Pipeline< folly::wangle::Pipeline<void*>>> pipeline_;
void*, std::exception>>> pipeline_;
}; };
class ServerWorkerPool : public folly::wangle::ThreadPoolExecutor::Observer { class ServerWorkerPool : public folly::wangle::ThreadPoolExecutor::Observer {
...@@ -179,10 +174,8 @@ void ServerWorkerPool::forEachWorker(F&& f) const { ...@@ -179,10 +174,8 @@ void ServerWorkerPool::forEachWorker(F&& f) const {
} }
class DefaultAcceptPipelineFactory class DefaultAcceptPipelineFactory
: public PipelineFactory<wangle::Pipeline<void*, std::exception>> { : public PipelineFactory<wangle::Pipeline<void*>> {
typedef wangle::Pipeline< typedef wangle::Pipeline<void*> AcceptPipeline;
void*,
std::exception> AcceptPipeline;
public: public:
AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) { AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) {
......
...@@ -52,9 +52,7 @@ class ServerBootstrap { ...@@ -52,9 +52,7 @@ class ServerBootstrap {
join(); join();
} }
typedef wangle::Pipeline< typedef wangle::Pipeline<void*> AcceptPipeline;
void*,
std::exception> AcceptPipeline;
/* /*
* Pipeline used to add connections to event bases. * Pipeline used to add connections to event bases.
* This is used for UDP or for load balancing * This is used for UDP or for load balancing
......
...@@ -101,8 +101,6 @@ class Handler : public HandlerBase<HandlerContext<Rout, Wout>> { ...@@ -101,8 +101,6 @@ class Handler : public HandlerBase<HandlerContext<Rout, Wout>> {
*/ */
}; };
struct Unit{};
template <class Rin, class Rout = Rin> template <class Rin, class Rout = Rin>
class InboundHandler : public HandlerBase<InboundHandlerContext<Rout>> { class InboundHandler : public HandlerBase<InboundHandlerContext<Rout>> {
public: public:
...@@ -110,8 +108,8 @@ class InboundHandler : public HandlerBase<InboundHandlerContext<Rout>> { ...@@ -110,8 +108,8 @@ class InboundHandler : public HandlerBase<InboundHandlerContext<Rout>> {
typedef Rin rin; typedef Rin rin;
typedef Rout rout; typedef Rout rout;
typedef Unit win; typedef Nothing win;
typedef Unit wout; typedef Nothing wout;
typedef InboundHandlerContext<Rout> Context; typedef InboundHandlerContext<Rout> Context;
virtual ~InboundHandler() {} virtual ~InboundHandler() {}
...@@ -129,8 +127,8 @@ class OutboundHandler : public HandlerBase<OutboundHandlerContext<Wout>> { ...@@ -129,8 +127,8 @@ class OutboundHandler : public HandlerBase<OutboundHandlerContext<Wout>> {
public: public:
static const HandlerDir dir = HandlerDir::OUT; static const HandlerDir dir = HandlerDir::OUT;
typedef Unit rin; typedef Nothing rin;
typedef Unit rout; typedef Nothing rout;
typedef Win win; typedef Win win;
typedef Wout wout; typedef Wout wout;
typedef OutboundHandlerContext<Wout> Context; typedef OutboundHandlerContext<Wout> Context;
......
...@@ -26,11 +26,32 @@ ...@@ -26,11 +26,32 @@
namespace folly { namespace wangle { namespace folly { namespace wangle {
// See Pipeline docblock for purpose
struct Nothing{};
namespace detail {
template <class T>
inline void logWarningIfNotNothing(const std::string& warning) {
LOG(WARNING) << warning;
}
template <>
inline void logWarningIfNotNothing<Nothing>(const std::string& warning) {
// do nothing
}
} // detail
/* /*
* R is the inbound type, i.e. inbound calls start with pipeline.read(R) * R is the inbound type, i.e. inbound calls start with pipeline.read(R)
* W is the outbound type, i.e. outbound calls start with pipeline.write(W) * W is the outbound type, i.e. outbound calls start with pipeline.write(W)
*
* Use Nothing for one of the types if your pipeline is unidirectional.
* If R is Nothing, read(), readEOF(), and readException() will be disabled.
* If W is Nothing, write() and close() will be disabled.
*/ */
template <class R, class W> template <class R, class W = Nothing>
class Pipeline : public DelayedDestruction { class Pipeline : public DelayedDestruction {
public: public:
Pipeline() : isStatic_(false) {} Pipeline() : isStatic_(false) {}
...@@ -61,21 +82,27 @@ class Pipeline : public DelayedDestruction { ...@@ -61,21 +82,27 @@ class Pipeline : public DelayedDestruction {
return readBufferSettings_; return readBufferSettings_;
} }
void read(R msg) { template <class T = R>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
read(R msg) {
if (!front_) { if (!front_) {
throw std::invalid_argument("read(): no inbound handler in Pipeline"); throw std::invalid_argument("read(): no inbound handler in Pipeline");
} }
front_->read(std::forward<R>(msg)); front_->read(std::forward<R>(msg));
} }
void readEOF() { template <class T = R>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
readEOF() {
if (!front_) { if (!front_) {
throw std::invalid_argument("readEOF(): no inbound handler in Pipeline"); throw std::invalid_argument("readEOF(): no inbound handler in Pipeline");
} }
front_->readEOF(); front_->readEOF();
} }
void readException(exception_wrapper e) { template <class T = R>
typename std::enable_if<!std::is_same<T, Nothing>::value>::type
readException(exception_wrapper e) {
if (!front_) { if (!front_) {
throw std::invalid_argument( throw std::invalid_argument(
"readException(): no inbound handler in Pipeline"); "readException(): no inbound handler in Pipeline");
...@@ -83,14 +110,18 @@ class Pipeline : public DelayedDestruction { ...@@ -83,14 +110,18 @@ class Pipeline : public DelayedDestruction {
front_->readException(std::move(e)); front_->readException(std::move(e));
} }
Future<void> write(W msg) { template <class T = W>
typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
write(W msg) {
if (!back_) { if (!back_) {
throw std::invalid_argument("write(): no outbound handler in Pipeline"); throw std::invalid_argument("write(): no outbound handler in Pipeline");
} }
return back_->write(std::forward<W>(msg)); return back_->write(std::forward<W>(msg));
} }
Future<void> close() { template <class T = W>
typename std::enable_if<!std::is_same<T, Nothing>::value, Future<void>>::type
close() {
if (!back_) { if (!back_) {
throw std::invalid_argument("close(): no outbound handler in Pipeline"); throw std::invalid_argument("close(): no outbound handler in Pipeline");
} }
...@@ -154,12 +185,14 @@ class Pipeline : public DelayedDestruction { ...@@ -154,12 +185,14 @@ class Pipeline : public DelayedDestruction {
} }
if (!front_) { if (!front_) {
LOG(WARNING) << "No inbound handler in Pipeline, " detail::logWarningIfNotNothing<R>(
"inbound operations will throw std::invalid_argument"; "No inbound handler in Pipeline, inbound operations will throw "
"std::invalid_argument");
} }
if (!back_) { if (!back_) {
LOG(WARNING) << "No outbound handler in Pipeline, " detail::logWarningIfNotNothing<W>(
"outbound operations will throw std::invalid_argument"; "No outbound handler in Pipeline, outbound operations will throw "
"std::invalid_argument");
} }
for (auto it = ctxs_.rbegin(); it != ctxs_.rend(); it++) { for (auto it = ctxs_.rbegin(); it != ctxs_.rend(); it++) {
......
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