Method Worker

core/logic/ThreadSupport.cpp:162–223

}

void CompatWorker::Worker()
{
	// Note: this must be first to ensure an ordering between Worker() and
	// Start(). It must also be outside of the loop to ensure the lock is
	// held across wakeup and retesting the predicates.
	std::unique_lock<std::mutex> lock(mutex_);

	if (callbacks_) {
		lock.unlock();
		callbacks_->OnWorkerStart(this);
		lock.lock();
	}

	typedef std::chrono::system_clock Clock;
	typedef std::chrono::time_point<Clock> TimePoint;

	auto can_work = [this]() -> bool {
		return state_ == Worker_Running && !work_.empty();
	};

	ke::Maybe<TimePoint> wait;
	unsigned int work_in_frame = 0;
	for (;;) {
		if (state_ == Worker_Stopped)
			break;

		if (!can_work()) {
			// Wait for work or a Stop.
			work_cv_.wait(lock);
			continue;
		}

		if (wait.isValid()) {
			// Wait until the specified time has passed. If we wake up with a
			// timeout, then the wait has elapsed, so reset the holder.
			if (work_cv_.wait_until(lock, wait.get()) == std::cv_status::timeout)
				wait = ke::Nothing();
			continue;
		}

		assert(state_ == Worker_Running);
		assert(!work_.empty());

		SWThreadHandle* handle = ke::PopFront(&work_);
		RunWorkLocked(&lock, handle);
		work_in_frame++;

		// If we've reached our max jobs per "frame", signal that the next
		// immediate job must be delayed. We retain the old ThreadWorker
		// behavior by checking if the queue has more work. Thus, a delay
		// only occurs if two jobs would be processed in the same wakeup.
		if (work_in_frame >= jobs_per_wakeup_ && wait_between_jobs_ && can_work())
			wait = ke::Some(Clock::now() + std::chrono::milliseconds(wait_between_jobs_));
	}

	assert(lock.owns_lock());

	while (!work_.empty()) {