Сбой, связанный с использованием boost::function в пуле потоков

Я пытаюсь реализовать пул потоков на С++, используя pthread. Я хочу инкапсулировать логику, связанную с управлением потоками, в одном объекте, который становится владельцем этих потоков. Это означает, что всякий раз, когда этот объект уничтожается, потоки должны быть остановлены и очищены.

Я тестировал свой код, и оказалось, что я получаю ошибку сегментации, когда уничтожаю объект WorkerThreadManager, когда вызывается функция boost::. См. код и обратную трассировку из GDB. Я не очень понимаю, почему это происходит, насколько я знаю, boost::function является копируемым, поэтому, как только я получаю его копию из очереди, я могу ее вытолкнуть () и даже уничтожить всю очередь (я доказал это в небольшой тест), а затем вызвать копию функции.

WorkerThreadManager.h:

#include "WorkerThreadManagerInterface.h"
#include "utils/mutex.h"
#include <queue>
#include <semaphore.h>

#include <iostream>

class WorkerThreadManager : public WorkerThreadManagerInterface
{
    public:
        WorkerThreadManager(unsigned threadsNumber = 5);
        virtual ~WorkerThreadManager();

        virtual void    PushTask(thread_function_t A_threadFun, result_function_t A_resultFun);
        void    SignalResults();

    private:
        static void*    WorkerThread(void* A_data);

        void    PushResult(int A_result, result_function_t A_resultFun);

        typedef boost::function<void ()> signal_function_t;

        struct worker_thread_data_t
        {
            worker_thread_data_t(thread_function_t A_threadFun, result_function_t A_resultFun) :
                threadFun(A_threadFun), resultFun(A_resultFun) {}
            worker_thread_data_t() {}

            thread_function_t       threadFun;
            result_function_t       resultFun;
        };


        const unsigned                      m_threadsNumber;
        pthread_t*                          m_pthreads;

        utils::Mutex                        m_tasksMutex;
        sem_t                               m_tasksSem;
        std::queue<worker_thread_data_t>    m_tasks;

        utils::Mutex                        m_resultsMutex;
        std::queue<signal_function_t>       m_results;
};

WorkerThreadManager.cpp:

#include "WorkerThreadManager.h"
#include "gateway_log.h"
#include <pthread.h>

/**
 * @brief Creates semaphore and starts threads.
 */
WorkerThreadManager::WorkerThreadManager(unsigned threadsNumber) : m_threadsNumber(threadsNumber)
{
    if ( sem_init(&m_tasksSem, 0, 0) )
    {
        std::stringstream ss;
        ss << "Semaphore could not be initialized: " << errno << " - " << strerror(errno);
        LOG_FATAL(ss);
        throw std::runtime_error(ss.str());
    }

    m_pthreads = new pthread_t[m_threadsNumber];
    for (unsigned i = 0; i < m_threadsNumber; ++i)
    {
        int rc = pthread_create(&m_pthreads[i], NULL, WorkerThreadManager::WorkerThread, (void*) this );
        if(rc)
        {
            std::stringstream ss;
            ss << "Pthread could not be started: " << errno << " - " << strerror(errno);
            LOG_FATAL(ss.str());

            if ( sem_destroy(&m_tasksSem) )
                LOG_ERROR("Semaphore could not be destroyed: " << errno << " - " << strerror(errno));

            delete [] m_pthreads;

            throw std::runtime_error(ss.str());
        }
        else
        {
            LOG_DEBUG("Worker thread started " << m_pthreads[i]);

            if(pthread_detach(m_pthreads[i]))
                LOG_WARN("Failed to detach worker thread");
        }
    }
}

/**
 * @brief Cancels all threads, destroys semaphore
 */
WorkerThreadManager::~WorkerThreadManager()
{
    LOG_DEBUG("~WorkerThreadManager()");

    for(unsigned i = 0; i < m_threadsNumber; ++i)
    {
        if ( pthread_cancel(m_pthreads[i]) )
            LOG_ERROR("Worker thread cancellation failed");
    }

    if ( sem_destroy(&m_tasksSem) )
        LOG_ERROR("Semaphore could not be destroyed: " << errno << " - " << strerror(errno));

    delete [] m_pthreads;
}

/**
 * @brief Adds new task to queue, so worker threads can
 * @param A_threadFun function which will be executed by thread
 * @param A_resultFun function which will be enqueued for calling with return value of A_threadFun as parameter
 *          after worker thread executes A_threadFun.
 */
void WorkerThreadManager::PushTask(thread_function_t A_threadFun, result_function_t A_resultFun)
{
    utils::ScopedLock mutex(m_tasksMutex);

    worker_thread_data_t    data(A_threadFun, A_resultFun);
    m_tasks.push( data );
    sem_post(&m_tasksSem);
    LOG_DEBUG("Task for worker threads has been added to queue");
}

/**
 * @brief   Executes result functions (if there are any) to give feedback 
 *  to classes which requested task execution in worker thread.
 */
void WorkerThreadManager::SignalResults()
{
    while(true)
    {
        signal_function_t signal;
        {
            utils::ScopedLock mutex(m_resultsMutex);
            if(m_results.size())
            {
                signal = m_results.front();
                m_results.pop();
            }
            else
                return;
        }

        signal();
    }
}

/**
 * @brief Enqueues result of function executed in worker thread.
 * @param A_result return value of function executed in worker thread
 * @param A_resultFun function which will be enqueued for calling with A_result as a parameter.
 */
void WorkerThreadManager::PushResult(int A_result, result_function_t A_resultFun)
{
    utils::ScopedLock mutex(m_resultsMutex);

    signal_function_t signal = boost::bind(A_resultFun, A_result);
    m_results.push( signal );
}


/**
 * @brief   worker thread body
 * @param A_data pointer to WorkerThreadManager instance
 */
void* WorkerThreadManager::WorkerThread(void* A_data)
{
    WorkerThreadManager* manager = reinterpret_cast<WorkerThreadManager*>(A_data);
    LOG_DEBUG("Starting worker thread loop");
    while (1)
    {
        if ( -1 == sem_wait(&manager->m_tasksSem) && errno == EINTR )
        {
            LOG_DEBUG("sem_wait interrupted with signal");
            continue;
        }
        LOG_DEBUG("WorkerThread:::::: about to call lock mutex");

        worker_thread_data_t data;
        {
            utils::ScopedLock mutex(manager->m_tasksMutex);
            data = manager->m_tasks.front();
            manager->m_results.pop();
        }

        LOG_DEBUG("WorkerThread:::::: about to call resultFun");
        int result  = data.threadFun();
        LOG_DEBUG("WorkerThread:::::: after call resultFun");
        pthread_testcancel();

        manager->PushResult(result, data.resultFun);
    }

    return NULL;
}

основной.cpp:

#include "gateway_log.h"
#include "WorkerThreadManager.h"
#include <memory>

class A {
public:
    int Fun() { LOG_DEBUG("Fun before sleep"); sleep(8); LOG_DEBUG("Fun after sleep");return 0; }
    void Result(int a) { LOG_DEBUG("Result: " << a); }
};


int main()
{
    sd::auto_ptr<WorkerThreadManager> workerThreadManager = new WorkerThreadManager;
    A a;
    workerThreadManager->PushTask(boost::bind(&A::Fun, &a), boost::bind(&A::Result, &a, _1));
    sleep(3);
    LOG_DEBUG("deleting workerThreadManager");
    workerThreadManager.reset();                    // <<<--- CRASH
    LOG_DEBUG("deleted workerThreadManager");
    sleep(10);
    LOG_DEBUG("after sleep");    

    return 0;
}

ГБД:

(gdb) bt
#0  0xb7ad33a0 in ?? () from /lib/i386-linux-gnu/libc.so.6
#1  0x0807d3a7 in boost::function0<void>::clear (this=0x858db48) at /home/marcin/intel_build/boost_1_42_0/boost/function/function_template.hpp:856
#2  0x0807d17b in boost::function0<void>::~function0 (this=0x858db48, __in_chrg=<optimized out>) at /home/marcin/intel_build/boost_1_42_0/boost/function/function_template.hpp:752
#3  0x0807cec5 in boost::function<void()>::~function(void) (this=0x858db48, __in_chrg=<optimized out>) at /home/marcin/intel_build/boost_1_42_0/boost/function/function_template.hpp:1043
#4  0x0807ced8 in std::_Destroy<boost::function<void ()> >(boost::function<void ()>*) (__pointer=0x858db48) at /usr/include/c++/4.6/bits/stl_construct.h:94
#5  0x0807c868 in std::_Destroy_aux<false>::__destroy<boost::function<void ()>*>(boost::function<void ()>*, boost::function<void ()>*) (__first=0x858db48, __last=0x858d928) at /usr/include/c++/4.6/bits/stl_construct.h:104
#6  0x0807bd05 in std::_Destroy<boost::function<void ()>*>(boost::function<void ()>*, boost::function<void ()>*) (__first=0x858d938, __last=0x858d928) at /usr/include/c++/4.6/bits/stl_construct.h:127
#7  0x0807af23 in std::_Destroy<boost::function<void ()>*, boost::function<void ()> >(boost::function<void ()>*, boost::function<void ()>*, std::allocator<boost::function<void ()> >&) (__first=0x858d938, __last=0x858d928)
    at /usr/include/c++/4.6/bits/stl_construct.h:153
#8  0x0807a037 in std::deque<boost::function<void ()>, std::allocator<boost::function<void ()> > >::_M_destroy_data_aux(std::_Deque_iterator<boost::function<void ()>, boost::function<void ()>&, boost::function<void ()>*>, std::_Deque_iterator<boost::function<void ()>, boost::function<void ()>&, boost::function<void ()>*>) (this=0x858beec, __first=..., __last=...) at /usr/include/c++/4.6/bits/deque.tcc:795
#9  0x08076153 in std::deque<boost::function<void ()>, std::allocator<boost::function<void ()> > >::_M_destroy_data(std::_Deque_iterator<boost::function<void ()>, boost::function<void ()>&, boost::function<void ()>*>, std::_Deque_iterator<boost::function<void ()>, boost::function<void ()>&, boost::function<void ()>*>, std::allocator<boost::function<void ()> > const&) (this=0x858beec, __first=..., __last=...) at /usr/include/c++/4.6/bits/stl_deque.h:1816
#10 0x08073411 in std::deque<boost::function<void()>, std::allocator<boost::function<void()> > >::~deque(void) (this=0x858beec, __in_chrg=<optimized out>) at /usr/include/c++/4.6/bits/stl_deque.h:898
#11 0x0806a355 in std::queue<boost::function<void()>, std::deque<boost::function<void()>, std::allocator<boost::function<void()> > > >::~queue(void) (this=0x858beec, __in_chrg=<optimized out>)
    at /usr/include/c++/4.6/bits/stl_queue.h:92
#12 0x0815a054 in WorkerThreadManager::~WorkerThreadManager (this=0x858be98, __in_chrg=<optimized out>) at WorkerThreadManager.cpp:42
#13 0x0815a1e3 in WorkerThreadManager::~WorkerThreadManager (this=0x858be98, __in_chrg=<optimized out>) at WorkerThreadManager.cpp:56
#14 0x080c6c51 in std::auto_ptr<WorkerThreadManager>::reset (this=0x85463e4, __p=0x0) at /usr/include/c++/4.6/backward/auto_ptr.h:244
#15 0x080604a9 in main ()

Буду очень признателен за любую помощь.


person Marcin    schedule 17.09.2012    source источник


Ответы (2)


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

Я подозреваю, что поскольку деструктор работает в одном потоке, один из потоков просыпается (из-за sem_destroy) и ошибочно пытается прочитать/вытолкнуть очередь. Я не уверен, почему это вызывает сбой в основном потоке, но я бы сначала устранил эту потенциальную проблему.

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

person Dave S    schedule 17.09.2012
comment
В этом примере при вызове pthread_cancel рабочий поток бездействовал в функции A::Fun. sleep() — это точка отмены, поэтому все рабочие потоки, включая занятый, завершаются немедленно. Вот почему sem_destroy не мог вызвать сбой. Во всяком случае, я исправил код, чтобы внедрить пустую функцию boost:: во все рабочие потоки, чтобы запросить выход, а затем присоединиться к этим потокам. - person Marcin; 19.09.2012
comment
Я не понимаю, почему некоторые из этих механизмов семафоров и потоков для своих собственных классов делают код более безопасным для исключений? Функции POSIX не вызывают генерацию ошибок, поскольку они реализованы в C. Мой код вызывает генерацию ошибок в критических ситуациях, которые нельзя обработать, например создание потока, — это нормально. - person Marcin; 19.09.2012
comment
@Marcin: Ну, среди прочего: если какой-либо из pthread_creates выйдет из строя, то вы не сможете уничтожить оставшиеся. Аналогичная ситуация с уничтожением семафора, если в вашем конструкторе возникает исключение - person Dave S; 19.09.2012

Нашел баг, банальный - позор мне :(

В функции void* WorkerThreadManager::WorkerThread(void* A_data) я вытащил очередь m_results вместо m_tasks, как я и предполагал:

worker_thread_data_t data;
{
    utils::ScopedLock mutex(manager->m_tasksMutex);
    data = manager->m_tasks.front();
    manager->m_results.pop();
}

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

person Marcin    schedule 18.09.2012
comment
Звонок pop() из пустой очереди является UB. В этом случае я подозреваю, что это портит внутреннюю структуру, что становится проблемой, когда он пытается очиститься во время разрушения. - person Dave S; 19.09.2012