Заметки о синхронизации. Deadlock

В наше суровое время, когда мощности процессоров перестали расти ввысь( частота ) и начали расти вширь( число потоков ), проблема синхронизации стоит как нельзя остро. Столкнувшись с этой проблемой на практике, я на себе ощутил, что задача эта намного сложнее, чем кажется на первый взгляд, и какое число подводных граблей она скрывает. В процессе работы над этой проблемой, у меня родилось несколько интересных паттернов, с которым я хочу познакомить хабрасообщество.


Одна из самых неприятных проблем, с которыми приходится столкнуться при программировании многопоточности, это Deadlock. Чаще всего он случается, когда поток уже заблокировал ресурс А, после чего пытается провести блокировку Б, другой же поток заблокировал ресурс Б, после чего пытается заблокировать ресурс А. Самое печальное в этом типе багов то, что, как правило, ситуация происходит очень редко, и поймать его при тестировании практически невозможно. Значит, нужно не допустить даже возможности появления этого бага! Как? Очень просто. Тут нам поможет паттерн, который я назвал SynchronizationManager, т.е. Диспетчер Синхронизации. Паттерн предназначен для случаев, когда нужно обеспечить синхронизацию достаточно большого количества разнообразных ресурсов.

Начнём с того, что все ресурсы, к которым необходим общий доступ, должны наследоваться от специального базового класса.

class SynchPrim
{
    enum{ NOT_LOCKED=0 };
public:
    SynchPrim(): lockId_( NOT_LOCKED ){}
    SynchPrim( const SynchPrim& ): lockId_( NOT_LOCKED ){}
    SynchPrim operator =( const SynchPrim& ){}
private:
    template< typename T >friend class SynchronizationManager;
    int getLockId()const{ return lockId_; }
    int lockId_;
};

Ну пока всё просто. А что делать в случаи, если мы не можем или не хотим добавлять лишнего предка в класс? Ничего страшного, нужно просто написать для него обёртку:

template< typename T >
class SynchPrimWrapper : public T, public SynchPrim{};

И использовать её при объявлении объекта:

SynchPrimWrapper< Resource1 > resource1;

Основная идея паттерна заключается в том, что все разделяемые ресурсы, которые могут понадобиться в текущий момент, необходим блокировать одновременно, одновременно и освобождать, тогда deadlock не может возникнуть. В случаи, если мы попытаемся заблокировать новые ресурсы, не освободив перед этим предыдущие — сработает макрос assert.

Для начала мы сформируем список ресурсов, с которыми мы будем работать. Это может быть как С-array, так и std::vector.

SynchPrim* lock_list[] = { &resource1, &resource2, &resource3 };

После чего одним вызовом блокируем их все:

SynchronizationManager::getInstance().lock( lock_list+0, lock_list+3 );

Кстати говоря, функция lock поддерживает асинхронную блокировку. Если вы не хотите, чтобы поток ждал, пока занятые ресурсы освободятся — просто передайте последним параметром false.
После того, как ресурсы больше не нужны — мы освобождаем их вызовом:

SynchronizationManager::getInstance().unlock( lock_list+0, lock_list+3 );

Ну и собственно класс:

template< class TSynchronizator >// Стратегия синхронизации должна реализовывать методы lock, unlock, wait, signal и тип ThreadType
class SynchronizationManager : public Singelton< SynchronizationManager< TSynchronizator > >
{
public:
    /*
       Блокировка
       Принимает итераторы начала и конца списка блокировки
       needWait - если установлен в true - функция будет ждать, пока необходимые ресурсы не освободятся
       возвращает false, если ресурсы заняты
    */
    template< typename Iterator >
    bool lock( const Iterator& begin, const Iterator& end, bool needWait= true );
    /*
       Снятие блокировка
       Принимает итераторы начала и конца списка блокировки
    */
    template< typename Iterator >
    void unlock( const Iterator& begin, const Iterator& end );

private:
    TSynchronizator synh_;
#ifdef _DEBUG
    typedef typename TSynchronizator::ThreadType ThreadId;
    typedef std::set< ThreadId > LockedThreads;
    LockedThreads lthread_;
#endif
};

Стратегию синхронизации можно написать, используя любые средства, и практически на любой платформе. Как правило, это mutex и condition.

Реализация:

template< class TSynchronizator >
template< typename Iterator >
bool SynchronizationManager< TSynchronizator  >::lock( const Iterator& begin, const Iterator& end, bool needWait )
{
    synh_.lock();
    ThreadId threadId = synh_.getThreadId();
    bool isFree = false;

#ifdef _DEBUG
    assert( lthread_.find( threadId ) == lthread_.end() );
#endif

    while( !( isFree = std::find_if( begin, end, std::mem_fun( &SynchPrim::getLockId ) ) == end ) && needWait )
       synh_.wait();

    if( isFree )
       for( Iterator it = begin; it != end; it++ )
           (*it)->lockId_ = threadId;

#ifdef _DEBUG
    if( isFree )
       lthread_.insert( threadId );
#endif
    synh_.unlock();
    return isFree;
}

template< class TSynchronizator >
template< typename Iterator >
void SynchronizationManager< TSynchronizator  >::unlock( const Iterator& begin, const Iterator& end )
{
    synh_.lock();

#ifdef _DEBUG
    ThreadId threadId = synh_.getThreadId();
    assert( lthread_.find( threadId ) != lthread_.end() );
    lthread_.erase( threadId );
#endif

    for( Iterator it = begin; it != end; it++ )
       (*it)->lockId_ = SynchPrim::NOT_LOCKED;

    synh_.signal();
    synh_.unlock();
}

Естественно, такой подход не лишён недостатков. Не всегда возможно знать заранее, какие ресурсы нам понадобятся. В данной ситуации придётся разблокировать уже выделенные ресурсы и заблокировать их по новой, но это есть накладные издержки, стоят они того, чтобы избавиться от потенциального деадлока — решать вам. Кроме того, в текущем виде паттерн не позволяет контролировать обращение к незаблокированным ресурсам и не гарантирует разблокировку после использования ресурсов. О том, как это можно сделать, я напишу в следующей статье.