/* Keep waking workers until the queue is empty */
while (!_queue.empty ()) {
- _condition.notify_all ();
- _condition.wait (lock);
+ _empty_condition.notify_all ();
+ _full_condition.wait (lock);
}
lock.unlock ();
/* Wait until the queue has gone down a bit */
while (_queue.size() >= _threads.size() * 2 && !_terminate) {
LOG_TIMING ("decoder sleeps with queue of %1", _queue.size());
- _condition.wait (lock);
+ _full_condition.wait (lock);
LOG_TIMING ("decoder wakes with queue of %1", _queue.size());
}
_film->j2k_bandwidth(), _film->resolution(), _film->log()
)
));
-
- _condition.notify_all ();
+
+ /* The queue might not be empty any more, so notify anything which is
+ waiting on that.
+ */
+ _empty_condition.notify_all ();
_have_a_real_frame[pvf->eyes()] = true;
}
{
boost::mutex::scoped_lock lock (_mutex);
_terminate = true;
- _condition.notify_all ();
+ _full_condition.notify_all ();
+ _empty_condition.notify_all ();
}
for (list<boost::thread *>::iterator i = _threads.begin(); i != _threads.end(); ++i) {
*/
int remote_backoff = 0;
- while (1) {
+ while (true) {
LOG_TIMING ("[%1] encoder thread sleeps", boost::this_thread::get_id());
boost::mutex::scoped_lock lock (_mutex);
while (_queue.empty () && !_terminate) {
- _condition.wait (lock);
+ _empty_condition.wait (lock);
}
if (_terminate) {
dcpomatic_sleep (remote_backoff);
}
+ /* The queue might not be full any more, so notify anything that is waiting on that */
lock.lock ();
- _condition.notify_all ();
+ _full_condition.notify_all ();
}
}
catch (...)
std::list<boost::shared_ptr<DCPVideoFrame> > _queue;
std::list<boost::thread *> _threads;
mutable boost::mutex _mutex;
- boost::condition _condition;
+ /** condition to manage thread wakeups when we have nothing to do */
+ boost::condition _empty_condition;
+ /** condition to manage thread wakeups when we have too much to do */
+ boost::condition _full_condition;
boost::shared_ptr<Writer> _writer;
Waker _waker;
while (1) {
boost::mutex::scoped_lock lock (_worker_mutex);
while (_queue.empty ()) {
- _worker_condition.wait (lock);
+ _empty_condition.wait (lock);
}
shared_ptr<Socket> socket = _queue.front ();
LOG_GENERAL_NC (message.str ());
}
- _worker_condition.notify_all ();
+ _full_condition.notify_all ();
}
}
/* Wait until the queue has gone down a bit */
while (int (_queue.size()) >= num_threads * 2) {
- _worker_condition.wait (lock);
+ _full_condition.wait (lock);
}
_queue.push_back (socket);
- _worker_condition.notify_all ();
+ _empty_condition.notify_all ();
}
}
std::vector<boost::thread *> _worker_threads;
std::list<boost::shared_ptr<Socket> > _queue;
boost::mutex _worker_mutex;
- boost::condition _worker_condition;
+ boost::condition _full_condition;
+ boost::condition _empty_condition;
boost::shared_ptr<Log> _log;
bool _verbose;