thrift/server/TThreadedServer.h

0001 /*
0002  * Licensed to the Apache Software Foundation (ASF) under one
0003  * or more contributor license agreements. See the NOTICE file
0004  * distributed with this work for additional information
0005  * regarding copyright ownership. The ASF licenses this file
0006  * to you under the Apache License, Version 2.0 (the
0007  * "License"); you may not use this file except in compliance
0008  * with the License. You may obtain a copy of the License at
0009  *
0010  *   http://www.apache.org/licenses/LICENSE-2.0
0011  *
0012  * Unless required by applicable law or agreed to in writing,
0013  * software distributed under the License is distributed on an
0014  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
0015  * KIND, either express or implied. See the License for the
0016  * specific language governing permissions and limitations
0017  * under the License.
0018  */
0019
0020 #ifndef _THRIFT_SERVER_TTHREADEDSERVER_H_
0021 #define _THRIFT_SERVER_TTHREADEDSERVER_H_ 1
0022
0023 #include <map>
0024 #include <thrift/concurrency/Monitor.h>
0025 #include <thrift/concurrency/ThreadFactory.h>
0026 #include <thrift/concurrency/Thread.h>
0027 #include <thrift/server/TServerFramework.h>
0028
0029 namespace apache {
0030 namespace thrift {
0031 namespace server {
0032
0033 /**
0034  * Manage clients using threads - threads are created one for each client and are
0035  * released when the client disconnects.  This server is used to make a dynamically
0036  * scalable server up to the concurrent connection limit.
0037  */
0038 class TThreadedServer : public TServerFramework {
0039 public:
0040   TThreadedServer(
0041       const std::shared_ptr<apache::thrift::TProcessorFactory>& processorFactory,
0042       const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
0043       const std::shared_ptr<apache::thrift::transport::TTransportFactory>& transportFactory,
0044       const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& protocolFactory,
0045       const std::shared_ptr<apache::thrift::concurrency::ThreadFactory>& threadFactory
0046       = std::shared_ptr<apache::thrift::concurrency::ThreadFactory>(
0047           new apache::thrift::concurrency::ThreadFactory(false)));
0048
0049   TThreadedServer(
0050       const std::shared_ptr<apache::thrift::TProcessor>& processor,
0051       const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
0052       const std::shared_ptr<apache::thrift::transport::TTransportFactory>& transportFactory,
0053       const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& protocolFactory,
0054       const std::shared_ptr<apache::thrift::concurrency::ThreadFactory>& threadFactory
0055       = std::shared_ptr<apache::thrift::concurrency::ThreadFactory>(
0056           new apache::thrift::concurrency::ThreadFactory(false)));
0057
0058   TThreadedServer(
0059       const std::shared_ptr<apache::thrift::TProcessorFactory>& processorFactory,
0060       const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
0061       const std::shared_ptr<apache::thrift::transport::TTransportFactory>& inputTransportFactory,
0062       const std::shared_ptr<apache::thrift::transport::TTransportFactory>& outputTransportFactory,
0063       const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& inputProtocolFactory,
0064       const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& outputProtocolFactory,
0065       const std::shared_ptr<apache::thrift::concurrency::ThreadFactory>& threadFactory
0066       = std::shared_ptr<apache::thrift::concurrency::ThreadFactory>(
0067           new apache::thrift::concurrency::ThreadFactory(false)));
0068
0069   TThreadedServer(
0070       const std::shared_ptr<apache::thrift::TProcessor>& processor,
0071       const std::shared_ptr<apache::thrift::transport::TServerTransport>& serverTransport,
0072       const std::shared_ptr<apache::thrift::transport::TTransportFactory>& inputTransportFactory,
0073       const std::shared_ptr<apache::thrift::transport::TTransportFactory>& outputTransportFactory,
0074       const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& inputProtocolFactory,
0075       const std::shared_ptr<apache::thrift::protocol::TProtocolFactory>& outputProtocolFactory,
0076       const std::shared_ptr<apache::thrift::concurrency::ThreadFactory>& threadFactory
0077       = std::shared_ptr<apache::thrift::concurrency::ThreadFactory>(
0078           new apache::thrift::concurrency::ThreadFactory(false)));
0079
0080   ~TThreadedServer() override;
0081
0082   /**
0083    * Post-conditions (return guarantees):
0084    *   There will be no clients connected.
0085    */
0086   void serve() override;
0087
0088 protected:
0089   /**
0090    * Drain recently connected clients by joining their threads - this is done lazily because
0091    * we cannot do it inside the thread context that is disconnecting.
0092    */
0093   virtual void drainDeadClients();
0094
0095   /**
0096    * Implementation of TServerFramework::onClientConnected
0097    */
0098   void onClientConnected(const std::shared_ptr<TConnectedClient>& pClient) override /* override */;
0099
0100   /**
0101    * Implementation of TServerFramework::onClientDisconnected
0102    */
0103   void onClientDisconnected(TConnectedClient *pClient) override /* override */;
0104
0105   std::shared_ptr<apache::thrift::concurrency::ThreadFactory> threadFactory_;
0106
0107   /**
0108    * A helper wrapper used to wrap the client in something we can use to maintain
0109    * the lifetime of the connected client within a detached thread.  We cannot simply
0110    * track the threads because a shared_ptr<Thread> hangs on to the Runnable it is
0111    * passed, and TServerFramework requires the runnable (TConnectedClient) to be
0112    * destroyed in order to work properly.
0113    */
0114   class TConnectedClientRunner : public apache::thrift::concurrency::Runnable
0115   {
0116   public:
0117     TConnectedClientRunner(const std::shared_ptr<TConnectedClient>& pClient);
0118     ~TConnectedClientRunner() override;
0119     void run() override /* override */;
0120   private:
0121     std::shared_ptr<TConnectedClient> pClient_;
0122   };
0123
0124   apache::thrift::concurrency::Monitor clientMonitor_;
0125
0126   typedef std::map<TConnectedClient *, std::shared_ptr<apache::thrift::concurrency::Thread> > ClientMap;
0127
0128   /**
0129    * A map of active clients
0130    */
0131   ClientMap activeClientMap_;
0132
0133   /**
0134    * A map of clients that have disconnected but their threads have not been joined
0135    */
0136   ClientMap deadClientMap_;
0137 };
0138
0139 }
0140 }
0141 } // apache::thrift::server
0142
0143 #endif // #ifndef _THRIFT_SERVER_TTHREADEDSERVER_H_