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 #pragma once
  18
  19 #include <pthread.h>
  20
  21 #include <atomic>
  22
  23 #include <folly/Portability.h>
  24
  25 #include "hphp/util/alloc.h"
  26 #include "hphp/util/exception.h"
  27 #include "hphp/util/lock.h"
  28 #include "hphp/util/synchronizable.h"
  29
  30 namespace HPHP {
  31 ///////////////////////////////////////////////////////////////////////////////
  32
  33 /**
  34  * Invokes a function asynchrously. For example,
  35  *
  36  *   struct MyClass {
  37  *     void doJob();
  38  *   };
  39  *
  40  *   MyClass obj;
  41  *   AsyncFunc<MyClass> func(&obj, &MyClasss::doJob);
  42  *   func.start(); // this will call obj.doJob() in a separate thread
  43  *   // do something else
  44  *   func.waitForEnd();
  45  *
  46  * Asynchronous function is a slightly different way of thinking about threads.
  47  * Maybe this can help people understand asynchronous function is actually a
  48  * broader/identical view of running threads,
  49  *
  50  *   struct MyRunnable {
  51  *     void run();
  52  *   };
  53  *
  54  *   MyRunnable thread;
  55  *   AsyncFunc<Runnable> func(&thread, &MyRunnable::run);
  56  *   thread.run();
  57  *
  58  * Well, asynchronous function is sometimes more flexible in writing a server,
  59  * because it can bind different threads to methods on the same object:
  60  *
  61  *   struct MyServer {
  62  *     void thread1();
  63  *     void thread2();
  64  *   };
  65  *
  66  *   MyServer server;
  67  *   AsyncFunc<MyServer> func1(&server, &MyServer::thread1);
  68  *   AsyncFunc<MyServer> func2(&server, &MyServer::thread2);
  69  *   func1.start();
  70  *   func2.start();
  71  *   ...now both threads are running, accessing the same server object.
  72  *
  73  * There is nothing wrong embedding the async function object itself in the
  74  * class like this,
  75  *
  76  *   struct MyServer {
  77  *     MyServer()
  78  *       : m_thread1(this, &MyServer::thread1)
  79  *       , m_thread2(this, &MyServer::thread2)
  80  *     {}
  81  *
  82  *     void thread1();
  83  *     void thread2();
  84  *
  85  *     void start() {
  86  *       m_thread1.start();
  87  *       m_thread2.start();
  88  *     }
  89  *
  90  *   private:
  91  *     AsyncFunc<MyServer> m_thread1;
  92  *     AsyncFunc<MyServer> m_thread2;
  93  *   };
  94  *
  95  */
  96 struct AsyncFuncImpl {
  97   typedef void PFN_THREAD_FUNC(void *);
  98
  99   /**
 100    * The global static to feed into pthread_create(), and this will delegate
 101    * the work to AsyncFuncImpl::threadFuncImpl().
 102    */
 103   static void *ThreadFunc(void *obj);
 104
 105   /**
 106    * Called by AsyncFunc<T> so we can call func(obj) back on thread running.
 107    *
 108    * The NUMA node, the size of stack on huge pages, and the size of an
 109    * additional thread-local space collocated with the stack can be specified.
 110    */
 111   AsyncFuncImpl(void *obj, PFN_THREAD_FUNC *func,
 112                 int numaNode, unsigned hugeStackKb, unsigned tlExtraKb);
 113   ~AsyncFuncImpl();
 114
 115   /**
 116    * Starts this thread.
 117    */
 118   void start();
 119
 120   /**
 121    * Sends a cancellation request to the thread. NB: Do not use this unless
 122    * the function is known to support cancellation and known to leave shared
 123    * state in a consistent state (alternatively, the caller should proceed to
 124    * shut down the process as well). Also, call waitForEnd following this call
 125    * before proceeding as if the async func has stopped executing.
 126    */
 127   void cancel();
 128
 129   /**
 130    * Waits until this thread finishes running.
 131    *
 132    * If `seconds' is positive, we wait that many seconds.  If `seconds' is
 133    * zero, we wait without a timeout.  If `seconds' is negative, we don't wait
 134    * at all, and return false if we aren't already stopped.
 135    */
 136   bool waitForEnd(int seconds = 0);
 137
 138   /**
 139    * Starts and waits until this thread finishes running.
 140    */
 141   void run() {
 142     start();
 143     waitForEnd();
 144   }
 145
 146   pthread_attr_t *getThreadAttr() {
 147     return &m_attr;
 148   }
 149
 150   static void SetThreadInitFunc(PFN_THREAD_FUNC* func, void *arg) {
 151     s_initFunc = func;
 152     s_initFuncArg = arg;
 153   }
 154
 155   static void SetThreadFiniFunc(PFN_THREAD_FUNC* func, void *arg) {
 156     s_finiFunc = func;
 157     s_finiFuncArg = arg;
 158   }
 159
 160   static PFN_THREAD_FUNC* GetThreadInitFunc() {
 161     return s_initFunc;
 162   }
 163
 164   static PFN_THREAD_FUNC* GetThreadFiniFunc() {
 165     return s_finiFunc;
 166   }
 167
 168   void setNoInitFini() { m_noInitFini = true; }
 169
 170   void setThreadName();
 171
 172   static uint32_t count() { return s_count; }
 173 private:
 174   Synchronizable m_stopMonitor;
 175
 176   void* m_obj{nullptr};
 177   PFN_THREAD_FUNC* m_func{nullptr};
 178   static PFN_THREAD_FUNC* s_initFunc;
 179   static PFN_THREAD_FUNC* s_finiFunc;
 180   static void* s_initFuncArg;
 181   static void* s_finiFuncArg;
 182   static std::atomic<uint32_t> s_count;
 183
 184   char* m_threadStack{nullptr};
 185   size_t m_stackAllocSize{0};
 186   int m_node{0};
 187   unsigned m_hugeStackKb{0};
 188   char* m_tlExtraBase{nullptr};
 189   unsigned m_tlExtraKb{0};
 190   MemBlock m_hugePages{nullptr, 0};
 191   pthread_attr_t m_attr;
 192   pthread_t m_threadId{0};
 193   // exception was thrown and thread was terminated
 194   Exception* m_exception{nullptr};
 195   bool m_stopped{false};
 196   bool m_noInitFini{false};
 197   /**
 198    * Called by ThreadFunc() to delegate the work.
 199    */
 200   void threadFuncImpl();
 201 };
 202
 203 ///////////////////////////////////////////////////////////////////////////////
 204
 205 /**
 206  * We could have written AysncFunc<T> directly with those methods implemented
 207  * inside AsyncFuncImpl class, but this way we reduce sizes of our code by
 208  * only templatizing a very minimal piece of code, sharing everything inside
 209  * AsyncFuncImpl by all AsyncFunc<T> classes.
 210  */
 211 template<class T>
 212 struct AsyncFunc : AsyncFuncImpl {
 213   AsyncFunc(T *obj, void (T::*member_func)(),
 214             int numaNode = -1, unsigned hugeStackKb = 0, unsigned tlExtraKb = 0)
 215     : AsyncFuncImpl((void*)this, run_, numaNode, hugeStackKb, tlExtraKb)
 216     , m_obj(obj)
 217     , m_memberFunc(member_func) {}
 218
 219   static void run_(void *obj) {
 220     AsyncFunc<T> *p = (AsyncFunc<T>*)obj;
 221     (p->m_obj->*(p->m_memberFunc))();
 222   }
 223
 224 private:
 225   T *m_obj;
 226   void (T::*m_memberFunc)();
 227 };
 228
 229 ///////////////////////////////////////////////////////////////////////////////
 230 }
 231