Ответ 1
Это хорошая идея для выделения ресурсов, создаваемых классом при уничтожении класса, даже если один из ресурсов является потоком. Если ресурс создается явно через пользовательский вызов, например Worker::Start(), тогда также должен быть явный способ его выпуска, например Worker::Stop(). Также было бы неплохо либо выполнить очистку в деструкторе в случае, если пользователь не вызывает Worker::Stop(), и/или предоставит пользователю скопированный вспомогательный класс, который реализует RAII -idiom, вызывая Worker::Start() в своем конструкторе и Worker::Stop() в своем деструкторе. Однако, если выделение ресурсов выполняется неявно, например, в конструкторе Worker, то выпуск ресурса также должен быть неявным, оставив деструктор в качестве основного кандидата для этой ответственности.
Разрушение
Рассмотрим Worker::~Worker(). Общее правило заключается в не генерировать исключения в деструкторах. Если объект Worker находится в стеке, который отключается от другого исключения, а Worker::~Worker() генерирует исключение, то std::terminate() будет вызываться, убивая приложение. В то время как Worker::~Worker() явно не бросает исключение, важно учитывать, что некоторые из функций, которые он вызывает, могут бросать:
-
m_Condition.notify_one()не бросает. -
m_pThread->join()может выбрасыватьboost::thread_interrupted.
Если std::terminate() - желаемое поведение, то никаких изменений не требуется. Однако, если std::terminate() не желательно, то поймайте boost::thread_interrupted и подавите его.
Worker::~Worker()
{
m_Running = false;
m_Condition.notify_one();
try
{
m_pThread->join();
}
catch ( const boost::thread_interrupted& )
{
/* suppressed */
}
}
Concurrency
Управление потоками может быть сложным. Важно определить точное требуемое поведение функций, таких как Worker::Wake(), а также понять поведение типов, которые облегчают потоки и синхронизацию. Например, boost::condition_variable::notify_one() не действует, если в boost::condition_variable::wait() не заблокированы нити. Давайте рассмотрим возможные параллельные пути для Worker::Wake().
Ниже приведена грубая попытка диаграммы concurrency для двух сценариев:
- Порядок работы происходит сверху вниз. (то есть операции сверху отображаются перед операциями внизу.
- Параллельные операции записываются в одной строке.
-
<и>используются, чтобы выделить, когда один поток просыпается или разблокирует другой поток. Например,A > Bуказывает, что потокAявляется разблокирующим потокомB.
Сценарий: Worker::Wake(), вызываемый при Worker::ThreadProc() заблокирован на m_Condition.
Other Thread | Worker::ThreadProc
-----------------------------------+------------------------------------------
| lock( m_Mutex )
| `-- m_Mutex.lock()
| m_Condition::wait( lock )
| |-- m_Mutex.unlock()
| |-- waits on notification
Worker::Wake() | |
|-- lock( m_Mutex ) | |
| `-- m_Mutex.lock() | |
|-- m_Condition::notify_one() > |-- wakes up from notification
`-- ~lock() | `-- m_Mutex.lock() // blocks
`-- m_Mutex.unlock() > `-- // acquires lock
| // do some work here
| ~lock() // end of for loop scope
| `-- m_Mutex.unlock()
Результат: Worker::Wake() возвращается довольно быстро, а Worker::ThreadProc выполняется.
Сценарий: Worker::Wake() вызывается, а Worker::ThreadProc() не заблокирован на m_Condition.
Other Thread | Worker::ThreadProc
-----------------------------------+------------------------------------------
| lock( m_Mutex )
| `-- m_Mutex.lock()
| m_Condition::wait( lock )
| |-- m_Mutex.unlock()
Worker::Wake() > |-- wakes up
| `-- m_Mutex.lock()
Worker::Wake() | // do some work here
|-- lock( m_Mutex ) | // still doing work...
| |-- m_Mutex.lock() // block | // hope we do not block on a system call
| | | // and more work...
| | | ~lock() // end of for loop scope
| |-- // still blocked < `-- m_Mutex.unlock()
| `-- // acquires lock | lock( m_Mutex ) // next 'for' iteration.
|-- m_Condition::notify_one() | `-- m_Mutex.lock() // blocked
`-- ~lock() | |-- // still blocked
`-- m_Mutex.unlock() > `-- // acquires lock
| m_Condition::wait( lock )
| |-- m_Mutex.unlock()
| `-- waits on notification
| `-- still waiting...
Результат: Worker::Wake() заблокирован, поскольку Worker::ThreadProc работал, но был не-op, поскольку он отправил уведомление m_Condition, когда его никто не ожидал.
Это не особенно опасно для Worker::Wake(), но это может вызвать проблемы в Worker::~Worker(). Если Worker::~Worker() работает, а Worker::ThreadProc выполняет работу, то Worker::~Worker() может блокироваться бесконечно при присоединении к потоку, поскольку поток может не ждать на m_Condition в том месте, где он уведомлен, и Worker::ThreadProc только проверяет m_Running после завершения ожидания m_Condition.
Работа над решением
В этом примере можно определить следующие требования:
-
Worker::~Worker()не приведет к вызовуstd::terminate(). -
Worker::Wake()не будет блокироваться, покаWorker::ThreadProcвыполняет работу. - Если
Worker::Wake()вызывается, аWorker::ThreadProcне выполняет работу, он будет уведомлятьWorker::ThreadProcо выполнении работы. - Если
Worker::Wake()вызывается, аWorker::ThreadProcвыполняет работу, он уведомляетWorker::ThreadProcо выполнении другой итерации работы. - Несколько вызовов
Worker::Wake()в то время какWorker::ThreadProcвыполняет работу, это приведет к тому, чтоWorker::ThreadProcвыполнит одну дополнительную итерацию работы.
код:
#include <boost/thread.hpp>
class Worker
{
public:
Worker();
~Worker();
void Wake();
private:
Worker(Worker const& rhs); // prevent copying
Worker& operator=(Worker const& rhs); // prevent assignment
void ThreadProc();
enum state { HAS_WORK, NO_WORK, SHUTDOWN };
state m_State;
boost::mutex m_Mutex;
boost::condition_variable m_Condition;
boost::thread m_Thread;
};
Worker::Worker()
: m_State(NO_WORK)
, m_Mutex()
, m_Condition()
, m_Thread()
{
m_Thread = boost::thread(&Worker::ThreadProc, this);
}
Worker::~Worker()
{
// Create scope so that the mutex is only locked when changing state and
// notifying the condition. It would result in a deadlock if the lock was
// still held by this function when trying to join the thread.
{
boost::lock_guard<boost::mutex> lock(m_Mutex);
m_State = SHUTDOWN;
m_Condition.notify_one();
}
try { m_Thread.join(); }
catch ( const boost::thread_interrupted& ) { /* suppress */ };
}
void Worker::Wake()
{
boost::lock_guard<boost::mutex> lock(m_Mutex);
m_State = HAS_WORK;
m_Condition.notify_one();
}
void Worker::ThreadProc()
{
for (;;)
{
// Create scope to only lock the mutex when checking for the state. Do
// not continue to hold the mutex wile doing busy work.
{
boost::unique_lock<boost::mutex> lock(m_Mutex);
// While there is no work (implies not shutting down), then wait on
// the condition.
while (NO_WORK == m_State)
{
m_Condition.wait(lock);
// Will wake up from either Wake() or ~Worker() signaling the condition
// variable. At that point, m_State will either be HAS_WORK or
// SHUTDOWN.
}
// On shutdown, break out of the for loop.
if (SHUTDOWN == m_State) break;
// Set state to indicate no work is queued.
m_State = NO_WORK;
}
// do some work here
}
}
Примечание. В качестве личного предпочтения я решил не выделять boost::thread в кучу, и в результате мне не нужно управлять им через boost::scoped_ptr. boost::thread имеет конструктор по умолчанию, который будет ссылаться на Not-a-Thread и ход переуступке.