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    * The NUMA node, the size of stack on huge pages, and the size of an
 110    * additional thread-local space collocated with the stack can be specified.
 111    */
 112   AsyncFuncImpl(void *obj, PFN_THREAD_FUNC *func,
 113                 int numaNode, unsigned hugeStackKb, unsigned tlExtraKb);
 114   ~AsyncFuncImpl();
 115
 116   /**
 117    * Starts this thread.
 118    */
 119   void start();
 120
 121   /**
 122    * Sends a cancellation request to the thread. NB: Do not use this unless
 123    * the function is known to support cancellation and known to leave shared
 124    * state in a consistent state (alternatively, the caller should proceed to
 125    * shut down the process as well). Also, call waitForEnd following this call
 126    * before proceeding as if the async func has stopped executing.
 127    */
 128   void cancel();
 129
 130   /**
 131    * Waits until this thread finishes running.
 132    *
 133    * If `seconds' is positive, we wait that many seconds.  If `seconds' is
 134    * zero, we wait without a timeout.  If `seconds' is negative, we don't wait
 135    * at all, and return false if we aren't already stopped.
 136    */
 137   bool waitForEnd(int seconds = 0);
 138
 139   /**
 140    * Starts and waits until this thread finishes running.
 141    */
 142   void run() {
 143     start();
 144     waitForEnd();
 145   }
 146
 147   pthread_attr_t *getThreadAttr() {
 148     return &m_attr;
 149   }
 150
 151   static void SetThreadInitFunc(PFN_THREAD_FUNC* func, void *arg) {
 152     s_initFunc = func;
 153     s_initFuncArg = arg;
 154   }
 155
 156   static void SetThreadFiniFunc(PFN_THREAD_FUNC* func, void *arg) {
 157     s_finiFunc = func;
 158     s_finiFuncArg = arg;
 159   }
 160
 161   static PFN_THREAD_FUNC* GetThreadInitFunc() {
 162     return s_initFunc;
 163   }
 164
 165   static PFN_THREAD_FUNC* GetThreadFiniFunc() {
 166     return s_finiFunc;
 167   }
 168
 169   void setNoInitFini() { m_noInitFini = true; }
 170
 171   void setThreadName();
 172
 173   static uint32_t count() { return s_count; }
 174 private:
 175   Synchronizable m_stopMonitor;
 176
 177   void* m_obj{nullptr};
 178   PFN_THREAD_FUNC* m_func{nullptr};
 179   static PFN_THREAD_FUNC* s_initFunc;
 180   static PFN_THREAD_FUNC* s_finiFunc;
 181   static void* s_initFuncArg;
 182   static void* s_finiFuncArg;
 183   static std::atomic<uint32_t> s_count;
 184
 185   char* m_threadStack{nullptr};
 186   size_t m_stackAllocSize{0};
 187   int m_node{0};
 188   unsigned m_hugeStackKb{0};
 189   char* m_tlExtraBase{nullptr};
 190   unsigned m_tlExtraKb{0};
 191   MemBlock m_hugePages{nullptr, 0};
 192   pthread_attr_t m_attr;
 193   pthread_t m_threadId{0};
 194   // exception was thrown and thread was terminated
 195   Exception* m_exception{nullptr};
 196   bool m_stopped{false};
 197   bool m_noInitFini{false};
 198   /**
 199    * Called by ThreadFunc() to delegate the work.
 200    */
 201   void threadFuncImpl();
 202 };
 203
 204 ///////////////////////////////////////////////////////////////////////////////
 205
 206 /**
 207  * We could have written AysncFunc<T> directly with those methods implemented
 208  * inside AsyncFuncImpl class, but this way we reduce sizes of our code by
 209  * only templatizing a very minimal piece of code, sharing everything inside
 210  * AsyncFuncImpl by all AsyncFunc<T> classes.
 211  */
 212 template<class T>
 213 struct AsyncFunc : AsyncFuncImpl {
 214   AsyncFunc(T *obj, void (T::*member_func)(),
 215             int numaNode = -1, unsigned hugeStackKb = 0, unsigned tlExtraKb = 0)
 216     : AsyncFuncImpl((void*)this, run_, numaNode, hugeStackKb, tlExtraKb)
 217     , m_obj(obj)
 218     , m_memberFunc(member_func) {}
 219
 220   static void run_(void *obj) {
 221     AsyncFunc<T> *p = (AsyncFunc<T>*)obj;
 222     (p->m_obj->*(p->m_memberFunc))();
 223   }
 224
 225 private:
 226   T *m_obj;
 227   void (T::*m_memberFunc)();
 228 };
 229
 230 ///////////////////////////////////////////////////////////////////////////////
 231 }
 232
 233 #endif // incl_HPHP_CONCURRENCY_ASYNC_FUNC_H_