src/library/workthread.cpp

   1 #include "workthread.hpp"
   2 #include <stdexcept>
   3 #include <sys/time.h>
   4
   5 namespace
   6 {
   7         uint64_t ticks()
   8         {
   9                 struct timeval tv;
  10                 gettimeofday(&tv, NULL);
  11                 return static_cast<uint64_t>(tv.tv_sec) * 1000000 + tv.tv_usec;
  12         }
  13 }
  14
  15 struct worker_thread_reflector
  16 {
  17         int operator()(worker_thread* x)
  18         {
  19                 (*x)(42);
  20         }
  21 };
  22
  23 worker_thread::worker_thread()
  24 {
  25         thread = NULL;
  26         reflector = NULL;
  27         workflag = 0;
  28         busy = false;
  29         waitamt_busy = 0;
  30         waitamt_work = 0;
  31         exception_caught = false;
  32         exception_oom = false;
  33         joined = false;
  34 }
  35
  36 worker_thread::~worker_thread()
  37 {
  38         set_workflag(WORKFLAG_QUIT_REQUEST);
  39         if(!joined && thread)
  40                 thread->join();
  41         delete thread;
  42         delete reflector;
  43 }
  44
  45 void worker_thread::request_quit()
  46 {
  47         {
  48                 //If the thread isn't there yet, wait for it.
  49                 umutex_class h(mutex);
  50                 if(!thread)
  51                         condition.wait(h);
  52         }
  53         set_workflag(WORKFLAG_QUIT_REQUEST);
  54         if(!joined)
  55                 thread->join();
  56         joined = true;
  57 }
  58
  59 void worker_thread::set_busy()
  60 {
  61         busy = true;
  62 }
  63
  64 void worker_thread::clear_busy()
  65 {
  66         umutex_class h(mutex);
  67         busy = false;
  68         condition.notify_all();
  69 }
  70
  71 void worker_thread::wait_busy()
  72 {
  73         umutex_class h(mutex);
  74         if(busy) {
  75                 uint64_t tmp = ticks();
  76                 while(busy)
  77                         condition.wait(h);
  78                 waitamt_busy += (ticks() - tmp);
  79         }
  80 }
  81
  82 void worker_thread::rethrow()
  83 {
  84         if(exception_caught) {
  85                 if(exception_oom)
  86                         throw std::bad_alloc();
  87                 else
  88                         throw std::runtime_error(exception_text);
  89         }
  90 }
  91
  92 void worker_thread::set_workflag(uint32_t flag)
  93 {
  94         umutex_class h(mutex);
  95         workflag |= flag;
  96         condition.notify_all();
  97 }
  98
  99 uint32_t worker_thread::clear_workflag(uint32_t flag)
 100 {
 101         umutex_class h(mutex);
 102         uint32_t tmp = workflag;
 103         workflag &= ~flag;
 104         return tmp;
 105 }
 106
 107 uint32_t worker_thread::wait_workflag()
 108 {
 109         umutex_class h(mutex);
 110         if(!workflag) {
 111                 uint64_t tmp = ticks();
 112                 while(!workflag)
 113                         condition.wait(h);
 114                 waitamt_work += (ticks() - tmp);
 115         }
 116         return workflag;
 117 }
 118
 119 std::pair<uint64_t, uint64_t> worker_thread::get_wait_count()
 120 {
 121         umutex_class h(mutex);
 122         return std::make_pair(waitamt_busy, waitamt_work);
 123 }
 124
 125 int worker_thread::operator()(int dummy)
 126 {
 127         try {
 128                 entry();
 129         } catch(std::bad_alloc& e) {
 130                 exception_oom = true;
 131                 exception_caught = true;
 132                 return 1;
 133         } catch(std::exception& e) {
 134                 exception_text = e.what();
 135                 exception_caught = true;
 136                 return 1;
 137         }
 138         return 0;
 139 }
 140
 141 void worker_thread::fire()
 142 {
 143         reflector = new worker_thread_reflector;
 144         thread = new thread_class(*reflector, this);
 145 }