hphp/util/async-job.h

   1 /*
   2    +----------------------------------------------------------------------+
   3    | HipHop for PHP                                                       |
   4    +----------------------------------------------------------------------+
   5    | Copyright (c) 2010-2013 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_ASYNC_JOB_H_
  18 #define incl_HPHP_ASYNC_JOB_H_
  19
  20 #include "hphp/util/base.h"
  21 #include "hphp/util/async-func.h"
  22 #include "hphp/util/lock.h"
  23 #include <algorithm>
  24 #include <sys/time.h>
  25
  26 namespace HPHP {
  27 ///////////////////////////////////////////////////////////////////////////////
  28 // helper, skip to JobDispatcher at below to read comments and usages...
  29
  30 template<class TJob, class TWorker>
  31 class JobDispatcher;
  32
  33 template<class TJob>
  34 class WorkerInfo {
  35  public:
  36   enum { DoInit = true };
  37 };
  38
  39 template<class TJob, class TWorker>
  40 class WorkerWrapper {
  41 public:
  42   explicit WorkerWrapper(JobDispatcher<TJob, TWorker> &dispatcher)
  43     : m_dispatcher(dispatcher)
  44     , m_func(this, &WorkerWrapper<TJob, TWorker>::doJob)
  45   {
  46     if (!WorkerInfo<TJob>::DoInit) {
  47       m_func.setNoInit();
  48     }
  49   }
  50
  51   TWorker *getWorker() { return &m_worker;}
  52   AsyncFunc<WorkerWrapper<TJob, TWorker> > &getAsyncFunc() {
  53     return m_func;
  54   }
  55
  56   void doJob() {
  57     m_worker.onThreadEnter();
  58     while (true) {
  59       std::shared_ptr<TJob> job = m_dispatcher.getNextJob();
  60       if (!job) break;
  61       m_worker.doJob(job);
  62     }
  63     m_worker.onThreadExit();
  64   }
  65
  66 private:
  67   JobDispatcher<TJob, TWorker> &m_dispatcher;
  68   AsyncFunc<WorkerWrapper<TJob, TWorker> > m_func;
  69   TWorker m_worker;
  70 };
  71
  72 ///////////////////////////////////////////////////////////////////////////////
  73
  74 /**
  75  * The promise is to make job processing cross multi-worker as easy as
  76  * possible. The final syntax is like,
  77  *
  78  * MyJobPtrVec jobs;
  79  * // prepare job list in "jobs"
  80  *
  81  * JobDispatcher<MyJob, MyWorker>(jobs, 10).run();
  82  *
  83  * or
  84  *
  85  * JobDispatcher<MyJob, MyWorker> dispatcher(jobs, 10);
  86  * dispatcher.start();
  87  * // do something else;
  88  * dispatcher.waitForEnd();
  89  *
  90  *
  91  * The only requirement is MyWorker has to be a class that implements this:
  92  *
  93  *  class MyWorker {
  94  *   public:
  95  *     void doJob(MyJobPtr job);
  96  *  };
  97  */
  98 template<class TJob, class TWorker>
  99 class JobDispatcher {
 100  public:
 101   JobDispatcher(std::vector<std::shared_ptr<TJob> > &jobs,
 102                 unsigned int workerCount, bool showStatus = false)
 103     : m_index(0), m_jobs(jobs), m_showStatus(showStatus), m_lastPercent(0) {
 104     std::random_shuffle(m_jobs.begin(), m_jobs.end());
 105     if (workerCount > jobs.size()) {
 106       workerCount = jobs.size();
 107     }
 108     m_workers.resize(workerCount);
 109     for (unsigned int i = 0; i < m_workers.size(); i++) {
 110       m_workers[i] = std::shared_ptr<WorkerWrapper<TJob, TWorker> >
 111         (new WorkerWrapper<TJob, TWorker>(*this));
 112     }
 113   }
 114
 115   /**
 116    * Start job processing asynchronously.
 117    */
 118   void start() {
 119     gettimeofday(&m_start, 0);
 120     for (unsigned int i = 0; i < m_workers.size(); i++) {
 121       m_workers[i]->getAsyncFunc().start();
 122     }
 123   }
 124
 125   /**
 126    * Wait for all jobs finish running.
 127    */
 128   void waitForEnd() {
 129     for (unsigned int i = 0; i < m_workers.size(); i++) {
 130       m_workers[i]->getAsyncFunc().waitForEnd();
 131     }
 132   }
 133
 134   /**
 135    * Start job processing and wait for all jobs finish running.
 136    */
 137   void run() {
 138     if (m_workers.size() == 1) {
 139       m_workers[0]->doJob();
 140     } else {
 141       start();
 142       waitForEnd();
 143     }
 144   }
 145
 146   unsigned int getWorkerCount() {
 147     return m_workers.size();
 148   }
 149
 150   TWorker *getWorker(unsigned int i) {
 151     assert(i < m_workers.size());
 152     return m_workers[i]->getWorker();
 153   }
 154
 155   void setStatus(int percent) {
 156     if (percent > m_lastPercent) {
 157       struct timeval tv;
 158       gettimeofday(&tv, 0);
 159       long diff = (tv.tv_sec  - m_start.tv_sec ) * 1000000 +
 160         (tv.tv_usec - m_start.tv_usec);
 161       m_lastPercent = percent;
 162       int seconds = diff / (percent * 10000);
 163       printf("%d%% (ETA: %d sec or %d min or %d hours)\n", percent,
 164              seconds, seconds/60, seconds/3600);
 165     }
 166   }
 167
 168   std::shared_ptr<TJob> getNextJob() {
 169     Lock lock(m_mutex);
 170     if (m_index >= m_jobs.size()) {
 171       return std::shared_ptr<TJob>();
 172     }
 173     if (m_showStatus && m_index > m_workers.size()) {
 174       setStatus((m_index - m_workers.size()) * 100 / m_jobs.size());
 175     }
 176     return m_jobs[m_index++];
 177   }
 178
 179  private:
 180   Mutex m_mutex;
 181   unsigned int m_index;
 182   std::vector<std::shared_ptr<TJob> > &m_jobs;
 183   std::vector<std::shared_ptr<WorkerWrapper<TJob, TWorker> > > m_workers;
 184   bool m_showStatus;
 185   int m_lastPercent;
 186   struct timeval m_start;
 187 };
 188
 189 ///////////////////////////////////////////////////////////////////////////////
 190 }
 191
 192 #endif // incl_HPHP_ASYNC_JOB_H_