2 * Copyright 2015 Facebook, Inc.
4 * Licensed under the Apache License, Version 2.0 (the "License");
5 * you may not use this file except in compliance with the License.
6 * You may obtain a copy of the License at
8 * http://www.apache.org/licenses/LICENSE-2.0
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS,
12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 * See the License for the specific language governing permissions and
14 * limitations under the License.
18 #include <folly/wangle/acceptor/Acceptor.h>
19 #include <folly/wangle/bootstrap/ServerSocketFactory.h>
20 #include <folly/io/async/EventBaseManager.h>
21 #include <folly/wangle/concurrent/IOThreadPoolExecutor.h>
22 #include <folly/wangle/acceptor/ManagedConnection.h>
23 #include <folly/wangle/channel/Pipeline.h>
24 #include <folly/wangle/channel/Handler.h>
28 template <typename Pipeline>
31 , public folly::wangle::HandlerAdapter<void*, std::exception> {
32 typedef std::unique_ptr<Pipeline,
33 folly::DelayedDestruction::Destructor> PipelinePtr;
35 class ServerConnection : public wangle::ManagedConnection {
37 explicit ServerConnection(PipelinePtr pipeline)
38 : pipeline_(std::move(pipeline)) {}
43 void timeoutExpired() noexcept {
46 void describe(std::ostream& os) const {}
50 void notifyPendingShutdown() {}
51 void closeWhenIdle() {}
52 void dropConnection() {
55 void dumpConnectionState(uint8_t loglevel) {}
57 PipelinePtr pipeline_;
61 explicit ServerAcceptor(
62 std::shared_ptr<PipelineFactory<Pipeline>> pipelineFactory,
63 std::shared_ptr<folly::wangle::Pipeline<
64 void*, std::exception>> acceptorPipeline,
66 : Acceptor(ServerSocketConfig())
68 , childPipelineFactory_(pipelineFactory)
69 , acceptorPipeline_(acceptorPipeline) {
70 Acceptor::init(nullptr, base_);
71 CHECK(acceptorPipeline_);
73 acceptorPipeline_->addBack(folly::wangle::HandlerPtr<ServerAcceptor, false>(this));
74 acceptorPipeline_->finalize();
77 void read(Context* ctx, void* conn) {
78 AsyncSocket::UniquePtr transport((AsyncSocket*)conn);
79 std::unique_ptr<Pipeline,
80 folly::DelayedDestruction::Destructor>
81 pipeline(childPipelineFactory_->newPipeline(
82 std::shared_ptr<AsyncSocket>(
84 folly::DelayedDestruction::Destructor())));
85 auto connection = new ServerConnection(std::move(pipeline));
86 Acceptor::addConnection(connection);
89 folly::Future<void> write(Context* ctx, std::exception e) {
90 return ctx->fireWrite(e);
93 /* See Acceptor::onNewConnection for details */
95 AsyncSocket::UniquePtr transport, const SocketAddress* address,
96 const std::string& nextProtocolName, const TransportInfo& tinfo) {
97 acceptorPipeline_->read(transport.release());
101 void onDataAvailable(std::shared_ptr<AsyncUDPSocket> socket,
102 const folly::SocketAddress& addr,
103 std::unique_ptr<folly::IOBuf> buf,
104 bool truncated) noexcept {
105 acceptorPipeline_->read(buf.release());
111 std::shared_ptr<PipelineFactory<Pipeline>> childPipelineFactory_;
112 std::shared_ptr<folly::wangle::Pipeline<
113 void*, std::exception>> acceptorPipeline_;
116 template <typename Pipeline>
117 class ServerAcceptorFactory : public AcceptorFactory {
119 explicit ServerAcceptorFactory(
120 std::shared_ptr<PipelineFactory<Pipeline>> factory,
121 std::shared_ptr<PipelineFactory<folly::wangle::Pipeline<
122 void*, std::exception>>> pipeline)
124 , pipeline_(pipeline) {}
126 std::shared_ptr<Acceptor> newAcceptor(EventBase* base) {
127 std::shared_ptr<folly::wangle::Pipeline<
128 void*, std::exception>> pipeline(
129 pipeline_->newPipeline(nullptr));
130 return std::make_shared<ServerAcceptor<Pipeline>>(factory_, pipeline, base);
133 std::shared_ptr<PipelineFactory<Pipeline>> factory_;
134 std::shared_ptr<PipelineFactory<
135 folly::wangle::Pipeline<
136 void*, std::exception>>> pipeline_;
139 class ServerWorkerPool : public folly::wangle::ThreadPoolExecutor::Observer {
141 explicit ServerWorkerPool(
142 std::shared_ptr<AcceptorFactory> acceptorFactory,
143 folly::wangle::IOThreadPoolExecutor* exec,
144 std::shared_ptr<std::vector<std::shared_ptr<folly::AsyncSocketBase>>> sockets,
145 std::shared_ptr<ServerSocketFactory> socketFactory)
146 : acceptorFactory_(acceptorFactory)
149 , socketFactory_(socketFactory) {
153 template <typename F>
154 void forEachWorker(F&& f) const;
157 folly::wangle::ThreadPoolExecutor::ThreadHandle*);
159 folly::wangle::ThreadPoolExecutor::ThreadHandle*);
160 void threadPreviouslyStarted(
161 folly::wangle::ThreadPoolExecutor::ThreadHandle* thread) {
162 threadStarted(thread);
164 void threadNotYetStopped(
165 folly::wangle::ThreadPoolExecutor::ThreadHandle* thread) {
166 threadStopped(thread);
170 std::map<folly::wangle::ThreadPoolExecutor::ThreadHandle*,
171 std::shared_ptr<Acceptor>> workers_;
172 std::shared_ptr<AcceptorFactory> acceptorFactory_;
173 folly::wangle::IOThreadPoolExecutor* exec_{nullptr};
174 std::shared_ptr<std::vector<std::shared_ptr<folly::AsyncSocketBase>>> sockets_;
175 std::shared_ptr<ServerSocketFactory> socketFactory_;
178 template <typename F>
179 void ServerWorkerPool::forEachWorker(F&& f) const {
180 for (const auto& kv : workers_) {
185 class DefaultAcceptPipelineFactory
186 : public PipelineFactory<wangle::Pipeline<void*, std::exception>> {
187 typedef wangle::Pipeline<
189 std::exception> AcceptPipeline;
192 AcceptPipeline* newPipeline(std::shared_ptr<AsyncSocket>) {
193 return new AcceptPipeline;