src/library/workthread.cpp

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