1.1 --- a/core/logic/thread/ThreadWorker.cpp Mon Aug 19 12:59:10 2013 -0700
1.2 +++ b/core/logic/thread/ThreadWorker.cpp Mon Aug 19 20:58:02 2013 -0700
1.3 @@ -33,10 +33,6 @@
1.4
1.5 ThreadWorker::ThreadWorker(IThreadWorkerCallbacks *hooks) : BaseWorker(hooks),
1.6 m_Threader(NULL),
1.7 - m_QueueLock(NULL),
1.8 - m_StateLock(NULL),
1.9 - m_PauseSignal(NULL),
1.10 - m_AddSignal(NULL),
1.11 me(NULL),
1.12 m_think_time(DEFAULT_THINK_TIME_MS)
1.13 {
1.14 @@ -46,32 +42,19 @@
1.15 ThreadWorker::ThreadWorker(IThreadWorkerCallbacks *hooks, IThreader *pThreader, unsigned int thinktime) :
1.16 BaseWorker(hooks),
1.17 m_Threader(pThreader),
1.18 - m_QueueLock(NULL),
1.19 - m_StateLock(NULL),
1.20 - m_PauseSignal(NULL),
1.21 - m_AddSignal(NULL),
1.22 me(NULL),
1.23 m_think_time(thinktime)
1.24 {
1.25 - if (m_Threader)
1.26 - {
1.27 - m_state = Worker_Stopped;
1.28 - } else {
1.29 - m_state = Worker_Invalid;
1.30 - }
1.31 + m_state = m_Threader ? Worker_Stopped : Worker_Invalid;
1.32 }
1.33
1.34 ThreadWorker::~ThreadWorker()
1.35 {
1.36 if (m_state != Worker_Stopped || m_state != Worker_Invalid)
1.37 - {
1.38 Stop(true);
1.39 - }
1.40 -
1.41 +
1.42 if (m_ThreadQueue.size())
1.43 - {
1.44 Flush(true);
1.45 - }
1.46 }
1.47
1.48 void ThreadWorker::OnTerminate(IThreadHandle *pHandle, bool cancel)
1.49 @@ -82,131 +65,88 @@
1.50
1.51 void ThreadWorker::RunThread(IThreadHandle *pHandle)
1.52 {
1.53 - WorkerState this_state = Worker_Running;
1.54 - size_t num;
1.55 + if (m_pHooks)
1.56 + m_pHooks->OnWorkerStart(this);
1.57
1.58 - if (m_pHooks)
1.59 - {
1.60 - m_pHooks->OnWorkerStart(this);
1.61 - }
1.62 -
1.63 + ke::AutoLock lock(&monitor_);
1.64 +
1.65 while (true)
1.66 {
1.67 - /**
1.68 - * Check number of items in the queue
1.69 - */
1.70 - m_StateLock->Lock();
1.71 - this_state = m_state;
1.72 - m_StateLock->Unlock();
1.73 - if (this_state != Worker_Stopped)
1.74 + if (m_state == Worker_Paused)
1.75 {
1.76 - m_QueueLock->Lock();
1.77 - num = m_ThreadQueue.size();
1.78 - if (!num)
1.79 - {
1.80 - /**
1.81 - * if none, wait for an item
1.82 - */
1.83 - m_Waiting = true;
1.84 - m_QueueLock->Unlock();
1.85 - /* first check if we should end again */
1.86 - if (this_state == Worker_Stopped)
1.87 - {
1.88 - break;
1.89 - }
1.90 - m_AddSignal->Wait();
1.91 - m_Waiting = false;
1.92 - } else {
1.93 - m_QueueLock->Unlock();
1.94 - }
1.95 + // Wait until we're told to wake up.
1.96 + monitor_.Wait();
1.97 + continue;
1.98 }
1.99 - m_StateLock->Lock();
1.100 - this_state = m_state;
1.101 - m_StateLock->Unlock();
1.102 - if (this_state != Worker_Running)
1.103 +
1.104 + if (m_state == Worker_Stopped)
1.105 {
1.106 - if (this_state == Worker_Paused || this_state == Worker_Stopped)
1.107 + // We've been told to stop entirely. If we've also been told to
1.108 + // flush the queue, do that now.
1.109 + while (!m_ThreadQueue.empty())
1.110 {
1.111 - //wait until the lock is cleared.
1.112 - if (this_state == Worker_Paused)
1.113 - {
1.114 - m_PauseSignal->Wait();
1.115 - }
1.116 - if (this_state == Worker_Stopped)
1.117 - {
1.118 - //if we're supposed to flush cleanrly,
1.119 - // run all of the remaining frames first.
1.120 - if (!m_FlushType)
1.121 - {
1.122 - while (m_ThreadQueue.size())
1.123 - {
1.124 - RunFrame();
1.125 - }
1.126 - }
1.127 - break;
1.128 - }
1.129 + // Release the lock since PopThreadFromQueue() will re-acquire it. The
1.130 + // main thread is blocking anyway.
1.131 + ke::AutoUnlock unlock(&monitor_);
1.132 + RunFrame();
1.133 }
1.134 + assert(m_state == Worker_Stopped);
1.135 + return;
1.136 }
1.137 - /**
1.138 - * Run the frame.
1.139 - */
1.140 - RunFrame();
1.141 +
1.142 + assert(m_state == Worker_Running);
1.143 +
1.144 + // Process one frame.
1.145 + WorkerState oldstate = m_state;
1.146 + {
1.147 + ke::AutoUnlock unlock(&monitor_);
1.148 + RunFrame();
1.149 + }
1.150
1.151 - /**
1.152 - * wait in between threads if specified
1.153 - */
1.154 - if (m_think_time)
1.155 - {
1.156 - m_Threader->ThreadSleep(m_think_time);
1.157 - }
1.158 + // If the state changed, loop back and process the new state.
1.159 + if (m_state != oldstate)
1.160 + continue;
1.161 +
1.162 + // If the thread queue is now empty, wait for a signal. Otherwise, if
1.163 + // we're on a delay, wait for either a notification or a timeout to
1.164 + // process the next item. If the queue has items and we don't have a
1.165 + // delay, then we just loop around and keep processing.
1.166 + if (m_ThreadQueue.empty())
1.167 + monitor_.Wait();
1.168 + else if (m_think_time)
1.169 + monitor_.Wait(m_think_time);
1.170 }
1.171
1.172 - if (m_pHooks)
1.173 {
1.174 - m_pHooks->OnWorkerStop(this);
1.175 + ke::AutoUnlock unlock(&monitor_);
1.176 + if (m_pHooks)
1.177 + m_pHooks->OnWorkerStop(this);
1.178 }
1.179 }
1.180
1.181 SWThreadHandle *ThreadWorker::PopThreadFromQueue()
1.182 {
1.183 - if (m_state <= Worker_Stopped && !m_QueueLock)
1.184 - {
1.185 + ke::AutoLock lock(&monitor_);
1.186 + if (m_state <= Worker_Stopped)
1.187 return NULL;
1.188 - }
1.189
1.190 - SWThreadHandle *swt;
1.191 - m_QueueLock->Lock();
1.192 - swt = BaseWorker::PopThreadFromQueue();
1.193 - m_QueueLock->Unlock();
1.194 -
1.195 - return swt;
1.196 + return BaseWorker::PopThreadFromQueue();
1.197 }
1.198
1.199 void ThreadWorker::AddThreadToQueue(SWThreadHandle *pHandle)
1.200 {
1.201 + ke::AutoLock lock(&monitor_);
1.202 if (m_state <= Worker_Stopped)
1.203 - {
1.204 return;
1.205 - }
1.206
1.207 - m_QueueLock->Lock();
1.208 BaseWorker::AddThreadToQueue(pHandle);
1.209 - if (m_Waiting)
1.210 - {
1.211 - m_AddSignal->Signal();
1.212 - }
1.213 - m_QueueLock->Unlock();
1.214 + monitor_.Notify();
1.215 }
1.216
1.217 WorkerState ThreadWorker::GetStatus(unsigned int *threads)
1.218 {
1.219 - WorkerState state;
1.220 -
1.221 - m_StateLock->Lock();
1.222 - state = BaseWorker::GetStatus(threads);
1.223 - m_StateLock->Unlock();
1.224 -
1.225 - return state;
1.226 + ke::AutoLock lock(&monitor_);
1.227 + return BaseWorker::GetStatus(threads);
1.228 }
1.229
1.230 void ThreadWorker::SetThinkTimePerFrame(unsigned int thinktime)
1.231 @@ -216,56 +156,32 @@
1.232
1.233 bool ThreadWorker::Start()
1.234 {
1.235 - if (m_state == Worker_Invalid)
1.236 - {
1.237 - if (m_Threader == NULL)
1.238 - {
1.239 - return false;
1.240 - }
1.241 - } else if (m_state != Worker_Stopped) {
1.242 + if (m_state == Worker_Invalid && m_Threader == NULL)
1.243 return false;
1.244 - }
1.245
1.246 - m_Waiting = false;
1.247 - m_QueueLock = m_Threader->MakeMutex();
1.248 - m_StateLock = m_Threader->MakeMutex();
1.249 - m_PauseSignal = m_Threader->MakeEventSignal();
1.250 - m_AddSignal = m_Threader->MakeEventSignal();
1.251 + if (m_state != Worker_Stopped)
1.252 + return false;
1.253 +
1.254 m_state = Worker_Running;
1.255 +
1.256 ThreadParams pt;
1.257 pt.flags = Thread_Default;
1.258 pt.prio = ThreadPrio_Normal;
1.259 me = m_Threader->MakeThread(this, &pt);
1.260 -
1.261 return true;
1.262 }
1.263
1.264 bool ThreadWorker::Stop(bool flush_cancel)
1.265 {
1.266 - if (m_state == Worker_Invalid || m_state == Worker_Stopped)
1.267 + // Change the state to signal a stop, and then trigger a notify.
1.268 {
1.269 - return false;
1.270 - }
1.271 -
1.272 - WorkerState oldstate;
1.273 + ke::AutoLock lock(&monitor_);
1.274 + if (m_state == Worker_Invalid || m_state == Worker_Stopped)
1.275 + return false;
1.276
1.277 - //set new state
1.278 - m_StateLock->Lock();
1.279 - oldstate = m_state;
1.280 - m_state = Worker_Stopped;
1.281 - m_FlushType = flush_cancel;
1.282 - m_StateLock->Unlock();
1.283 -
1.284 - if (oldstate == Worker_Paused)
1.285 - {
1.286 - Unpause();
1.287 - } else {
1.288 - m_QueueLock->Lock();
1.289 - if (m_Waiting)
1.290 - {
1.291 - m_AddSignal->Signal();
1.292 - }
1.293 - m_QueueLock->Unlock();
1.294 + m_state = Worker_Stopped;
1.295 + m_FlushType = flush_cancel;
1.296 + monitor_.Notify();
1.297 }
1.298
1.299 me->WaitForThread();
1.300 @@ -274,33 +190,18 @@
1.301 //flush all remaining events
1.302 Flush(true);
1.303
1.304 - //free mutex locks
1.305 - m_QueueLock->DestroyThis();
1.306 - m_StateLock->DestroyThis();
1.307 - m_PauseSignal->DestroyThis();
1.308 - m_AddSignal->DestroyThis();
1.309 -
1.310 - //invalidizzle
1.311 - m_QueueLock = NULL;
1.312 - m_StateLock = NULL;
1.313 - m_PauseSignal = NULL;
1.314 - m_AddSignal = NULL;
1.315 me = NULL;
1.316 -
1.317 return true;
1.318 }
1.319
1.320 bool ThreadWorker::Pause()
1.321 {
1.322 if (m_state != Worker_Running)
1.323 - {
1.324 return false;
1.325 - }
1.326
1.327 - m_StateLock->Lock();
1.328 + ke::AutoLock lock(&monitor_);
1.329 m_state = Worker_Paused;
1.330 - m_StateLock->Unlock();
1.331 -
1.332 + monitor_.Notify();
1.333 return true;
1.334 }
1.335
1.336 @@ -308,18 +209,10 @@
1.337 bool ThreadWorker::Unpause()
1.338 {
1.339 if (m_state != Worker_Paused)
1.340 - {
1.341 return false;
1.342 - }
1.343
1.344 - m_StateLock->Lock();
1.345 + ke::AutoLock lock(&monitor_);
1.346 m_state = Worker_Running;
1.347 - m_StateLock->Unlock();
1.348 - m_PauseSignal->Signal();
1.349 - if (m_Waiting)
1.350 - {
1.351 - m_AddSignal->Signal();
1.352 - }
1.353 -
1.354 + monitor_.Notify();
1.355 return true;
1.356 }
