hphp/util/async-func.h

   1 /*
   2    +----------------------------------------------------------------------+
   3    | HipHop for PHP                                                       |
   4    +----------------------------------------------------------------------+
   5    | Copyright (c) 2010-present Facebook, Inc. (http://www.facebook.com)  |
   6    +----------------------------------------------------------------------+
   7    | This source file is subject to version 3.01 of the PHP license,      |
   8    | that is bundled with this package in the file LICENSE, and is        |
   9    | available through the world-wide-web at the following url:           |
  10    | http://www.php.net/license/3_01.txt                                  |
  11    | If you did not receive a copy of the PHP license and are unable to   |
  12    | obtain it through the world-wide-web, please send a note to          |
  13    | license@php.net so we can mail you a copy immediately.               |
  14    +----------------------------------------------------------------------+
  15 */
  16
  17 #ifndef incl_HPHP_CONCURRENCY_ASYNC_FUNC_H_
  18 #define incl_HPHP_CONCURRENCY_ASYNC_FUNC_H_
  19
  20 #include <pthread.h>
  21
  22 #include <atomic>
  23
  24 #include <folly/Portability.h>
  25
  26 #include "hphp/util/alloc.h"
  27 #include "hphp/util/exception.h"
  28 #include "hphp/util/lock.h"
  29 #include "hphp/util/synchronizable.h"
  30
  31 namespace HPHP {
  32 ///////////////////////////////////////////////////////////////////////////////
  33
  34 /**
  35  * Invokes a function asynchrously. For example,
  36  *
  37  *   struct MyClass {
  38  *     void doJob();
  39  *   };
  40  *
  41  *   MyClass obj;
  42  *   AsyncFunc<MyClass> func(&obj, &MyClasss::doJob);
  43  *   func.start(); // this will call obj.doJob() in a separate thread
  44  *   // do something else
  45  *   func.waitForEnd();
  46  *
  47  * Asynchronous function is a slightly different way of thinking about threads.
  48  * Maybe this can help people understand asynchronous function is actually a
  49  * broader/identical view of running threads,
  50  *
  51  *   struct MyRunnable {
  52  *     void run();
  53  *   };
  54  *
  55  *   MyRunnable thread;
  56  *   AsyncFunc<Runnable> func(&thread, &MyRunnable::run);
  57  *   thread.run();
  58  *
  59  * Well, asynchronous function is sometimes more flexible in writing a server,
  60  * because it can bind different threads to methods on the same object:
  61  *
  62  *   struct MyServer {
  63  *     void thread1();
  64  *     void thread2();
  65  *   };
  66  *
  67  *   MyServer server;
  68  *   AsyncFunc<MyServer> func1(&server, &MyServer::thread1);
  69  *   AsyncFunc<MyServer> func2(&server, &MyServer::thread2);
  70  *   func1.start();
  71  *   func2.start();
  72  *   ...now both threads are running, accessing the same server object.
  73  *
  74  * There is nothing wrong embedding the async function object itself in the
  75  * class like this,
  76  *
  77  *   struct MyServer {
  78  *     MyServer()
  79  *       : m_thread1(this, &MyServer::thread1)
  80  *       , m_thread2(this, &MyServer::thread2)
  81  *     {}
  82  *
  83  *     void thread1();
  84  *     void thread2();
  85  *
  86  *     void start() {
  87  *       m_thread1.start();
  88  *       m_thread2.start();
  89  *     }
  90  *
  91  *   private:
  92  *     AsyncFunc<MyServer> m_thread1;
  93  *     AsyncFunc<MyServer> m_thread2;
  94  *   };
  95  *
  96  */
  97 struct AsyncFuncImpl {
  98   typedef void PFN_THREAD_FUNC(void *);
  99
 100   /**
 101    * The global static to feed into pthread_create(), and this will delegate
 102    * the work to AsyncFuncImpl::threadFuncImpl().
 103    */
 104   static void *ThreadFunc(void *obj);
 105
 106   /**
 107    * Called by AsyncFunc<T> so we can call func(obj) back on thread running.
 108    */
 109   AsyncFuncImpl(void *obj, PFN_THREAD_FUNC *func, bool hugify);
 110   ~AsyncFuncImpl();
 111
 112   /**
 113    * Starts this thread.
 114    */
 115   void start();
 116
 117   /**
 118    * Sends a cancellation request to the thread. NB: Do not use this unless
 119    * the function is known to support cancellation and known to leave shared
 120    * state in a consistent state (alternatively, the caller should proceed to
 121    * shut down the process as well). Also, call waitForEnd following this call
 122    * before proceeding as if the async func has stopped executing.
 123    */
 124   void cancel();
 125
 126   /**
 127    * Waits until this thread finishes running.
 128    */
 129   bool waitForEnd(int seconds = 0);
 130
 131   /**
 132    * Starts and waits until this thread finishes running.
 133    */
 134   void run() {
 135     start();
 136     waitForEnd();
 137   }
 138
 139   pthread_attr_t *getThreadAttr() {
 140     return &m_attr;
 141   }
 142
 143   static void SetThreadInitFunc(PFN_THREAD_FUNC* func, void *arg) {
 144     s_initFunc = func;
 145     s_initFuncArg = arg;
 146   }
 147
 148   static void SetThreadFiniFunc(PFN_THREAD_FUNC* func, void *arg) {
 149     s_finiFunc = func;
 150     s_finiFuncArg = arg;
 151   }
 152
 153   static PFN_THREAD_FUNC* GetThreadInitFunc() {
 154     return s_initFunc;
 155   }
 156
 157   static PFN_THREAD_FUNC* GetThreadFiniFunc() {
 158     return s_finiFunc;
 159   }
 160
 161   void setNoInitFini() { m_noInitFini = true; }
 162
 163   static uint32_t count() { return s_count; }
 164 private:
 165   Synchronizable m_stopMonitor;
 166
 167   void* m_obj{nullptr};
 168   PFN_THREAD_FUNC* m_func{nullptr};
 169   static PFN_THREAD_FUNC* s_initFunc;
 170   static PFN_THREAD_FUNC* s_finiFunc;
 171   static void* s_initFuncArg;
 172   static void* s_finiFuncArg;
 173   static std::atomic<uint32_t> s_count;
 174   static std::atomic_int s_curr_numa_node; // for round robin NUMA binding
 175
 176   char* m_threadStack{nullptr};
 177   size_t m_stackAllocSize{0};
 178   MemBlock m_firstSlab{nullptr, 0};
 179   pthread_attr_t m_attr;
 180   pthread_t m_threadId{0};
 181   // exception was thrown and thread was terminated
 182   Exception* m_exception{nullptr};
 183   int m_node{0};
 184   bool m_stopped{false};
 185   bool m_noInitFini{false};
 186   bool m_hugeStack{false};
 187   /**
 188    * Called by ThreadFunc() to delegate the work.
 189    */
 190   void threadFuncImpl();
 191 };
 192
 193 ///////////////////////////////////////////////////////////////////////////////
 194
 195 /**
 196  * We could have written AysncFunc<T> directly with those methods implemented
 197  * inside AsyncFuncImpl class, but this way we reduce sizes of our code by
 198  * only templatizing a very minimal piece of code, sharing everything inside
 199  * AsyncFuncImpl by all AsyncFunc<T> classes.
 200  */
 201 template<class T>
 202 struct AsyncFunc : AsyncFuncImpl {
 203   AsyncFunc(T *obj, void (T::*member_func)(), bool hugify = false)
 204     : AsyncFuncImpl((void*)this, run_, hugify),
 205       m_obj(obj), m_memberFunc(member_func) {
 206   }
 207
 208   static void run_(void *obj) {
 209     AsyncFunc<T> *p = (AsyncFunc<T>*)obj;
 210     (p->m_obj->*(p->m_memberFunc))();
 211   }
 212
 213 private:
 214   T *m_obj;
 215   void (T::*m_memberFunc)();
 216 };
 217
 218 ///////////////////////////////////////////////////////////////////////////////
 219 }
 220
 221 #endif // incl_HPHP_CONCURRENCY_ASYNC_FUNC_H_