Как я могу добавить элемент в две очереди и гарантировать, что он существует в обеих или нет (многопоточный)

Question

У меня следующая проблема.

У меня есть два класса, в данном случае A и B, которые оба имеют concurrent_queue. Здесь предполагается, что concurrent_queue - это потокобезопасная очередь с блокирующей функцией push (). Когда объект помещен в очередь в B, он получает доступ к одиночке A, и он также ставится в очередь в A. Это приводит к тому, что целая куча B имеет небольшие очереди со своими собственными объектами и одну большую очередь в A, которая содержит их все. Каждый экземпляр B может жить в отдельном потоке.

С чем я сталкиваюсь, так это то, что часто поток пропускается между двумя строками кода в B :: foo (), то есть A :: mQueue содержит объект, но B :: mQueue еще не содержит объект.

Что мне интересно, так это то, как я могу гарантировать, что при вызове B :: foo () объект либо помещается в обе очереди, либо ни в одну из очередей. Мне кажется, что я должен был бы иметь мьютекс в A, который B мог бы захватить, и заблокировать мьютекс A в B :: foo ().

Есть ли у кого-нибудь предложения, как мне это сделать или как я могу реструктурировать свой код для этого? Я использую библиотеку boost :: threading.

Class A
{    
public:
    A& instance(){/* return singleton */}        
    void addToQueue(SomeObject const& obj)
    {
        mQueue.push(obj);
    }        
private:
    concurrent_queue<SomeObject> mQueue;
};

Class B
{
public:
    void foo()
    {
        SomeObject obj;
        //I would like to guarantee that obj is either present in both queues or neither queue
        A::instance().addToQueue(obj);
        mQueue.push(obj);
    }        
private:
    concurrent_queue<SomeObject> mQueue;
};

В моей реальной реализации это не тот же объект, который ставится в очередь в A и B, а структуры очередей A, которые содержат указатели на B, что позволяет мне удалять все из A и удалять из всех B в в том же порядке, в котором они были поставлены в очередь, но это не должно относиться к вопросу.

Answer 1

Вам необходимо атомизировать операцию «добавления объектов в обе очереди». Вам понадобится блокировка или какой-либо другой вид примитива синхронизации вокруг ваших двух вызовов функций. То же самое для удаления элементов из очередей.

boost :: mutex выглядит подходящим для работы. Вам понадобится один экземпляр, и он должен быть доступен везде, где изменяются очереди. Поскольку он также будет иметь то же время жизни, что и очередь A, я предлагаю вам поместить его в A. Затем измените доступ к очереди так, чтобы он выглядел так:

A::instance().lockQueue(); //calls A.mQueueAccessMutex.lock(), probably
    A::instance().addToQueue(obj);
    mQueue.push(obj);
A::instance().unlockQueue();

Или в стиле RAII:

{
    LockHolder lh(A::instance().getLock()); //lock called in lh's constructor

    A::instance().addToQueue(obj);
    mQueue.push(obj);

    //unlock called in lh's destructor
}

Обратите внимание, что тогда concurrent_queue будет избыточным, поскольку никакие два потока не будут одновременно обращаться к очереди.

-

И, конечно, всегда есть очень маленький шанс, что простое изменение порядка, в котором вы помещаете элементы в очереди, решит ваши проблемы. :)

Answer 2

Вам, вероятно, нужна некоторая форма мьютекса, чтобы гарантировать атомарность (по сравнению с остальной частью вашего приложения). Boost :: threading предоставляет объекты мьютекса iirc, так что вы можете посмотреть на это.

Answer 3

Насколько я понял, B: foo должен гарантировать, что объект добавляется в обе очереди, но впоследствии доступ к этим очередям должен быть независимым.

В этом случае вы должны усовершенствовать A некоторым методом, чтобы напрямую заблокировать его очередь или который возвращает мьютекс, используемый в этой очереди (я предполагаю, что ваш concurent_queue основан на мьютексе). После этого b :: foo должен сначала заблокировать оба мьютекса, сделать push, освободить оба мьютекса.

И не забывайте обрабатывать исключения, например если не удалось добавить во вторую очередь, следует удалить из первой.