Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

boost::thread_group doesn't automatically clean up contained threads when they end, so I needed something like that described in this boost-users list post. However, in my opinion that code is not safe, because there's no guarantee that tp will be .reset before the thread reaches its th_ptr.get() call.

After an initial attempt at re-inventing boost::thread_group entirely (which failed), I came up with this wrapper and, after much sweat, I believe it's complete and correct. It seems to fix every deadlock that was generally reproducible during its development, so I think I've got them all.

Of course, I'm not sure.

What do you think of my wrapper?

(It also performs some signalling sanity — I only want signals to be handled in the "root"/parent thread, otherwise all hell breaks loose.)

has-threads.h

#ifndef HASTHREADS_H
#define HASTHREADS_H

#include <set>
#include <iostream>
#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>

/**
 * Concept to be derived by any class that should be capable of launching and tracking
 * worker threads, and waiting for workers to complete at object destruction.
 * 
 * Non-copyable.
 * 
 * 
 * ** NOTE REGARDING DESTRUCTION **
 * 
 * Threads are automatically waited for on destruction, but they are not interrupted. So:
 *  - Deriving classes should call interruptThreads() if any are persistent and would not terminate
 *    soon on their own.
 *  - They should also call waitForThreads() if any callback uses member data, to ensure that
 *    the thread finished execution before the member data is destroyed.
 * 
 * 
 * ** ANOTHER NOTE **
 * 
 * Be careful not to expose boost::thread_group::size(); if a thread uses it, this could cause
 * a deadlock with boost::thread_group::join_all(). Instead, we can track it ourselves, safely.
 */
struct has_threads
{
    public:

        /**
         * Asks all pending threads to terminate.
         * 
         * See http://www.boost.org/doc/libs/1_40_0/doc/html/thread/thread_management.html
         * for what this actually means.
         */
        void interruptThreads();

        /**
         * Wait for all pending threads to terminate.
         */
        void waitForThreads();

    protected:
        has_threads();
        ~has_threads();

        /**
         * Create and run a tracked thread.
         */
        template <typename Callable>
        void createThread(Callable f, bool allowSignals = false);

        /**
         * Blocks SIGINT, SIGTERM and SIGQUIT signals in the current thread.
         * Should always be used in a new thread, since we have a dedicated
         * signal handling thread that we want to always receive any of these
         * signals.
         */
        static void blockSignalsInThisThread();

        /**
         * Returns the number of threads currently managed by this object.
         */
        size_t threadCount() const;

    private:

        has_threads(has_threads const&);

        /**
         * Entrypoint function for a thread.
         * Wraps the user-provided worker function in signal and exception handling.
         */
        template <typename Callable>
        void runThread(Callable f, bool allowSignals, boost::thread* tp);


        boost::thread_group* grp;
        size_t count;

        mutable boost::mutex thread_create_lock;
        mutable boost::mutex thread_start_lock;
        mutable boost::mutex thread_removal_lock;
        mutable boost::mutex count_lock;
};


template <typename Callable>
void has_threads::createThread(Callable f, bool allowSignals)
{
    {
        // Although boost::thread_group does a good job of looking after threads for us,
        // it doesn't auto-remove objects wrapping terminated threads. So we must do that... carefully.

        thread_start_lock.lock(); // thread creation starts
        boost::mutex::scoped_lock l(thread_create_lock);

        // We need `tp` to be valid *before* the thread starts, so that its third argument is
        // always guaranteed to be valid. But we also need `tp` to be the pointer we put into the
        // thread_group. So we swap a running thread object into the original non-running one.
        boost::thread* tp = new boost::thread;

        boost::thread tmp(&has_threads::runThread<Callable>, this, f, allowSignals, tp);
        tp->swap(tmp);
        grp->add_thread(tp);

        {
            boost::mutex::scoped_lock l(count_lock);
            count++;
        }
    }
}


template <typename Callable>
void has_threads::runThread(Callable f, bool allowSignals, boost::thread* tp)
{
    if (!allowSignals)
        blockSignalsInThisThread();

    {
        // Wait until createThread's work with this thread object is definitely done
        boost::mutex::scoped_lock l(thread_create_lock);
    }

    thread_start_lock.unlock(); // thread creation is now deemed to be utterly complete

    try {
        f();
    }
    catch (boost::thread_interrupted& e) {

        // Yes, we should catch this exception! Letting it bubble over is _potentially_ dangerous:
        // http://stackoverflow.com/questions/6375121

        std::cout << "Thread " << boost::this_thread::get_id() << " interrupted (and ended)." << std::endl;
    }
    catch (std::exception& e) {
        std::cout << "Exception caught from thread " << boost::this_thread::get_id() << ": " << e.what() << std::endl;
    }
    catch (...) {
        std::cout << "Unknown exception caught from thread " << boost::this_thread::get_id() << std::endl;
    }

    // If the group is in the middle of a boost::thread_group::join_all, then boost::thread_group::remove_thread
    // would block as they sit on the same mutex. Silly >.< So we check a "removal" flag first.
    {
        boost::mutex::scoped_try_lock l(thread_removal_lock);
        if (!l) // can only occur when has_threads::waitForThreads() is in progress and we shouldn't be self-deleting
            return;

        grp->remove_thread(tp);
        {
            boost::mutex::scoped_lock l(count_lock);
            count--;
        }

        delete tp;
    }
}


#endif

has-threads.cpp

#include "has-threads.h"
#include <boost/foreach.hpp>
#include <csignal>

#include <iostream>

has_threads::has_threads()
    : grp(NULL), count(0)
{
    grp = new boost::thread_group();
}

has_threads::~has_threads()
{
    waitForThreads();
    delete grp;
}

void has_threads::interruptThreads()
{
    grp->interrupt_all();
}

void has_threads::waitForThreads()
{
    // Let any thread initialisation finish, otherwise
    // we might see deadlock on `thread_create_lock`.
    thread_start_lock.lock();
    thread_start_lock.unlock();

    {
        // Protects from threads being created whilst we join everything we already have.
        boost::mutex::scoped_lock l1(thread_create_lock);

        // Protects from threads in the process of destroying themselves
        // from conflicting with the following logic.
        boost::mutex::scoped_lock l2(thread_removal_lock);

        grp->join_all();

        // Now reset the group, so that any thread objects whose threads ended
        // during that `join_all` are destroyed properly.
        delete grp;
        grp = new boost::thread_group();
    }
}

size_t has_threads::threadCount() const
{
    boost::mutex::scoped_lock l(count_lock);
    return count;
}

void has_threads::blockSignalsInThisThread()
{
    sigset_t signal_set;
    sigemptyset(&signal_set);
    sigaddset(&signal_set, SIGINT);
    sigaddset(&signal_set, SIGTERM);
    sigaddset(&signal_set, SIGHUP);
    sigaddset(&signal_set, SIGPIPE); // http://www.unixguide.net/network/socketfaq/2.19.shtml
    pthread_sigmask(SIG_BLOCK, &signal_set, NULL);
}
share|improve this question
    
Note: Yes, I'm aware that thread pools are "better" in a lot of cases. But I do want to be able to just create and destroy threads at will. –  Lightness Races in Orbit Feb 26 '12 at 0:21
    
And, yes, all this locking makes me think "gosh, there must be a simpler way", but I don't think that there is. Perhaps you can point out something I've missed. –  Lightness Races in Orbit Mar 6 '12 at 14:46
add comment

1 Answer 1

I think you have UB here: mutex::unlock() must only be called (1) if the mutex is presently locked and (2) by the same thread that lock()ed it in the first place (i.e. if the calling thread owns the mutex). Anything else is UB (at least for std::mutex, but I suspect also for boost::mutex).

Your code is sufficiently complex for me not to be sure whether you have this type of UB.

However, if you avoid explicit calls to mutex::lock() and mutex::unlock() but instead use RAII via lock_guard<>, unique_lock<>, or shared_lock<>, you should be on the safe side.

share|improve this answer
    
You're right - I deliberately do the locking in one thread and the locking in another. I had no idea it was undefined but on checking now you're quite right. –  Lightness Races in Orbit Apr 14 at 14:00
    
(Thanks for posting this) –  Lightness Races in Orbit Apr 14 at 14:00
add comment

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

Not the answer you're looking for? Browse other questions tagged or ask your own question.