thrift/concurrency/Thread.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_CONCURRENCY_THREAD_H_
0021 #define _THRIFT_CONCURRENCY_THREAD_H_ 1
0022
0023 #include <memory>
0024 #include <thread>
0025
0026 #include <thrift/concurrency/Monitor.h>
0027
0028 namespace apache {
0029 namespace thrift {
0030 namespace concurrency {
0031
0032 class Thread;
0033
0034 /**
0035  * Minimal runnable class.  More or less analogous to java.lang.Runnable.
0036  *
0037  * @version $Id:$
0038  */
0039 class Runnable {
0040
0041 public:
0042   virtual ~Runnable() = default;
0043   virtual void run() = 0;
0044
0045   /**
0046    * Gets the thread object that is hosting this runnable object  - can return
0047    * an empty boost::shared pointer if no references remain on that thread object
0048    */
0049   virtual std::shared_ptr<Thread> thread() { return thread_.lock(); }
0050
0051   /**
0052    * Sets the thread that is executing this object.  This is only meant for
0053    * use by concrete implementations of Thread.
0054    */
0055   virtual void thread(std::shared_ptr<Thread> value) { thread_ = value; }
0056
0057 private:
0058   std::weak_ptr<Thread> thread_;
0059 };
0060
0061 /**
0062  * Minimal thread class. Returned by thread factory bound to a Runnable object
0063  * and ready to start execution.  More or less analogous to java.lang.Thread
0064  * (minus all the thread group, priority, mode and other baggage, since that
0065  * is difficult to abstract across platforms and is left for platform-specific
0066  * ThreadFactory implementations to deal with
0067  *
0068  * @see apache::thrift::concurrency::ThreadFactory)
0069  */
0070 class Thread : public std::enable_shared_from_this<Thread> {
0071
0072 public:
0073   typedef std::thread::id id_t;
0074   typedef void (*thread_funct_t)(std::shared_ptr<Thread> );
0075
0076   enum STATE { uninitialized, starting, started, stopping, stopped };
0077
0078   static void threadMain(std::shared_ptr<Thread> thread);
0079
0080   static inline bool is_current(id_t t) { return t == std::this_thread::get_id(); }
0081   static inline id_t get_current() { return std::this_thread::get_id(); }
0082
0083   Thread(bool detached, std::shared_ptr<Runnable> runnable)
0084     : state_(uninitialized), detached_(detached) {
0085     this->_runnable = runnable;
0086   }
0087
0088   virtual ~Thread() {
0089     if (!detached_ && thread_->joinable()) {
0090       try {
0091         join();
0092       } catch (...) {
0093         // We're really hosed.
0094       }
0095     }
0096   }
0097
0098   STATE getState() const
0099   {
0100     Synchronized sync(monitor_);
0101     return state_;
0102   }
0103
0104   void setState(STATE newState)
0105   {
0106     Synchronized sync(monitor_);
0107     state_ = newState;
0108
0109     // unblock start() with the knowledge that the thread has actually
0110     // started running, which avoids a race in detached threads.
0111     if (newState == started) {
0112       monitor_.notify();
0113     }
0114   }
0115
0116   /**
0117    * Starts the thread. Does platform specific thread creation and
0118    * configuration then invokes the run method of the Runnable object bound
0119    * to this thread.
0120    */
0121   virtual void start() {
0122     if (getState() != uninitialized) {
0123       return;
0124     }
0125
0126     std::shared_ptr<Thread> selfRef = shared_from_this();
0127     setState(starting);
0128
0129     Synchronized sync(monitor_);
0130     thread_ = std::unique_ptr<std::thread>(new std::thread(getThreadFunc(), selfRef));
0131
0132     if (detached_)
0133       thread_->detach();
0134
0135     // Wait for the thread to start and get far enough to grab everything
0136     // that it needs from the calling context, thus absolving the caller
0137     // from being required to hold on to runnable indefinitely.
0138     monitor_.wait();
0139   }
0140
0141   /**
0142    * Join this thread. If this thread is joinable, the calling thread blocks
0143    * until this thread completes.  If the target thread is not joinable, then
0144    * nothing happens.
0145    */
0146   virtual void join() {
0147     if (!detached_ && state_ != uninitialized) {
0148       thread_->join();
0149     }
0150   }
0151
0152   /**
0153    * Gets the thread's platform-specific ID
0154    */
0155   Thread::id_t getId() const { return thread_.get() ? thread_->get_id() : std::thread::id(); }
0156
0157   /**
0158    * Gets the runnable object this thread is hosting
0159    */
0160   std::shared_ptr<Runnable> runnable() const { return _runnable; }
0161
0162 protected:
0163
0164   virtual thread_funct_t getThreadFunc() const {
0165       return threadMain;
0166   }
0167
0168 private:
0169   std::shared_ptr<Runnable> _runnable;
0170   std::unique_ptr<std::thread> thread_;
0171   Monitor monitor_;
0172   STATE state_;
0173   bool detached_;
0174 };
0175
0176
0177 }
0178 }
0179 } // apache::thrift::concurrency
0180
0181 #endif // #ifndef _THRIFT_CONCURRENCY_THREAD_H_