C ++ pthread блокировка очереди тупик (я думаю) - PullRequest
Новая
       43

C ++ pthread блокировка очереди тупик (я думаю)

6 голосов
/ 16 февраля 2011

У меня проблема с pthreads, где я думаю, что захожу в тупик. Я создал очередь блокировки, которая, как мне показалось, работала, но после проведения еще одного тестирования я обнаружил, что если я пытаюсь отменить несколько потоков, которые блокируют блокировку_ очереди, я, похоже, захожу в тупик.

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

template <class T> class Blocking_Queue
{
public:
    Blocking_Queue()
    {
        pthread_mutex_init(&_lock, NULL);
        pthread_cond_init(&_cond, NULL);
    }

    ~Blocking_Queue()
    {
        pthread_mutex_destroy(&_lock);
        pthread_cond_destroy(&_cond);
    }

    void put(T t)
    {
        pthread_mutex_lock(&_lock);
        _queue.push(t);
        pthread_cond_signal(&_cond);
        pthread_mutex_unlock(&_lock);
    }

     T pull()
     {
        pthread_mutex_lock(&_lock);
        while(_queue.empty())
        {
            pthread_cond_wait(&_cond, &_lock);
        }

        T t = _queue.front();
        _queue.pop();

        pthread_mutex_unlock(&_lock);

        return t;
     }

priavte:
    std::queue<T> _queue;
    pthread_cond_t _cond;
    pthread_mutex_t _lock;
}

Для тестирования я создал 4 потока, которые тянут эту очередь блокировки. Я добавил несколько операторов печати в очередь блокировки, и каждый поток получает метод pthread_cond_wait (). Однако, когда я пытаюсь вызвать pthread_cancel () и pthread_join () в каждом потоке, программа просто зависает.

Я также проверил это только с одной нитью, и она отлично работает.

Согласно документации, pthread_cond_wait () является точкой отмены, поэтому вызов метода cancel для этих потоков должен привести к остановке их выполнения (и это работает только с 1 потоком). Однако pthread_mutex_lock не является точкой отмены. Может ли что-то происходить по аналогии с вызовом pthread_cancel (), отмененный поток извлекает мьютекс перед завершением и не разблокирует его, а затем, когда следующий поток отменяется, он не может получить мьютекс и взаимоблокировки? Или есть что-то еще, что я делаю не так.

Любой совет был бы прекрасен. Спасибо:)

Ответы [ 3 ]

5 голосов
/ 16 февраля 2011

pthread_cancel() лучше избегать.

Вы можете разблокировать все свои потоки, заблокированные в Blocking_Queue :: pull (), сгенерировав исключение оттуда.

Одно слабое место в очередиT t = _queue.front(); вызывает конструктор копирования T, который может вызвать исключение, что делает мьютекс очереди заблокированным навсегда.Лучше использовать блокировку в C ++.

Вот пример изящного завершения потока: 

$ cat test.cc
#include <boost/thread/mutex.hpp>
#include <boost/thread/thread.hpp>
#include <boost/thread/condition_variable.hpp>
#include <exception>
#include <list>
#include <stdio.h>

struct BlockingQueueTerminate
    : std::exception
{};

template<class T>
class BlockingQueue
{
private:
    boost::mutex mtx_;
    boost::condition_variable cnd_;
    std::list<T> q_;
    unsigned blocked_;
    bool stop_;

public:
    BlockingQueue()
        : blocked_()
        , stop_()
    {}

    ~BlockingQueue()
    {
        this->stop(true);
    }

    void stop(bool wait)
    {
        // tell threads blocked on BlockingQueue::pull() to leave
        boost::mutex::scoped_lock lock(mtx_);
        stop_ = true;
        cnd_.notify_all();

        if(wait) // wait till all threads blocked on the queue leave BlockingQueue::pull()
            while(blocked_)
                cnd_.wait(lock);
    }

    void put(T t)
    {
        boost::mutex::scoped_lock lock(mtx_);
        q_.push_back(t);
        cnd_.notify_one();
    }

    T pull()
    {
        boost::mutex::scoped_lock lock(mtx_);

        ++blocked_;
        while(!stop_ && q_.empty())
            cnd_.wait(lock);
        --blocked_;

        if(stop_) {
            cnd_.notify_all(); // tell stop() this thread has left
            throw BlockingQueueTerminate();
        }

        T front = q_.front();
        q_.pop_front();
        return front;
    }
};

void sleep_ms(unsigned ms)
{
    // i am using old boost
    boost::thread::sleep(boost::get_system_time() + boost::posix_time::milliseconds(ms));
    // with latest one you can do this
    //boost::thread::sleep(boost::posix_time::milliseconds(10));
}

void thread(int n, BlockingQueue<int>* q)
try
{
    for(;;) {
        int m = q->pull();
        printf("thread %u: pulled %d\n", n, m);
        sleep_ms(10);
    }
}
catch(BlockingQueueTerminate&)
{
    printf("thread %u: finished\n", n);
}

int main()
{
    BlockingQueue<int> q;

    // create two threads
    boost::thread_group tg;
    tg.create_thread(boost::bind(thread, 1, &q));
    tg.create_thread(boost::bind(thread, 2, &q));
    for(int i = 1; i < 10; ++i)
        q.put(i);
    sleep_ms(100); // let the threads do something
    q.stop(false); // tell the threads to stop
    tg.join_all(); // wait till they stop
}

$ g++ -pthread -Wall -Wextra -o test -lboost_thread-mt test.cc

$ ./test
thread 2: pulled 1
thread 1: pulled 2
thread 1: pulled 3
thread 2: pulled 4
thread 1: pulled 5
thread 2: pulled 6
thread 1: pulled 7
thread 2: pulled 8
thread 1: pulled 9
thread 2: finished
thread 1: finished
1 голос
/ 16 февраля 2011

У меня был похожий опыт работы с pthread_cond_wait () / pthread_cancel (). У меня были проблемы с блокировкой, все еще удерживаемой после того, как поток вернулся по какой-то причине, и было невозможно разблокировать ее, так как вы должны разблокировать в том же потоке, что и заблокированный. Я заметил эти ошибки при выполнении pthread_mutex_destroy (), так как у меня был один производитель, ситуация с одним потребителем, поэтому тупика не возникало.

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

1 голос
/ 16 февраля 2011

Я не совсем знаком с pthread_cancel () - я предпочитаю совместное завершение.

Не оставит ли pthread_cancel () ваш мьютекс заблокированным?Я полагаю, вам нужно выполнить очистку с помощью обработчика отмены.

...