boost asio semaphore-like solution

Question

This question is follow up of this question. Anyhow, this is the motivation - a bit of c/p from original question's comment:

I'd like to be able to post one group of jobs on multiple threads (CalcFib functions), and then when the jobs are over, another group of jobs (CalcFib2 functions), also on multiple threads. This cycle loops many times(here is two), so I thought the best thing to do is to create boost::asio::io_service and create threads at the beginning of the loop, so I don't have to create & destroy threads every time the loop starts/ends.

I had created two int vars wrongly named semaphore_**, and decrement them in above mentioned functions. The code that waits for the job group to finish is plain while as seen below. The solution works, at least as I can see.

Is the wait using while really a way to go? What am I missing? Is there a better way to do this?

My code looks like this:

#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/bind.hpp>
#include <iostream>

boost::mutex global_stream_lock;
int semaphore_fib = 0;
int semaphore_fib2 = 0;

void WorkerThread( boost::shared_ptr< boost::asio::io_service > io_service)
{
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Thread Start" << std::endl;
    global_stream_lock.unlock();

    io_service->run();

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Thread Finish" << std::endl;
    global_stream_lock.unlock();
}

size_t fib( size_t n )
{
    if ( n <= 1 )
    {
        return n;
    }
    boost::this_thread::sleep( boost::posix_time::milliseconds( 1000 ) );
    return fib( n - 1 ) + fib( n - 2);
}

void CalcFib( size_t n )
{
    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] Now calculating fib( " << n << " ) " << std::endl;
    global_stream_lock.unlock();

    size_t f = fib( n );

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] fib( " << n << " ) = " << f << std::endl;
    semaphore_fib = semaphore_fib-1;
    global_stream_lock.unlock();
}

void CalcFib2( size_t n )
{
    global_stream_lock.lock();
    std::cout << "\t\t[" << boost::this_thread::get_id()
        << "] Now calculating fib2( " << n << " ) " << std::endl;
    global_stream_lock.unlock();

    size_t f = fib( n );

    global_stream_lock.lock();
    std::cout << "\t\t[" << boost::this_thread::get_id()
        << "] fib2( " << n << " ) = " << f << std::endl;
    semaphore_fib2=semaphore_fib2-1;
    global_stream_lock.unlock();
}

int main( int argc, char * argv[] )
{
    boost::shared_ptr< boost::asio::io_service > io_service(
        new boost::asio::io_service
        );
    boost::shared_ptr< boost::asio::io_service::work > work(
        new boost::asio::io_service::work( *io_service )
        );
    boost::asio::io_service::strand strand( *io_service );

    global_stream_lock.lock();
    std::cout << "[" << boost::this_thread::get_id()
        << "] The program will exit when all work has finished."
        << std::endl;
    global_stream_lock.unlock();

    boost::thread_group worker_threads;
    for( int x = 0; x < 2; ++x )
    {
        worker_threads.create_thread( 
            boost::bind( &WorkerThread, io_service)
            );
    }
    for(int loop_no=0; loop_no<2; ++loop_no)
    {
        semaphore_fib=3;
        io_service->post( boost::bind( CalcFib, 5 ) );
        io_service->post( boost::bind( CalcFib, 4 ) );
        io_service->post( boost::bind( CalcFib, 3 ) );
        while(semaphore_fib>0)
        {
            // waiting
        }

        global_stream_lock.lock();
        std::cout << "[" << boost::this_thread::get_id()
            << "] ******* CalcFib group finished ********" << std::endl;
        global_stream_lock.unlock();

        semaphore_fib2=3;
        io_service->post( boost::bind( CalcFib2, 2 ) );
        io_service->post( boost::bind( CalcFib2, 1 ) );
        io_service->post( boost::bind( CalcFib2, 1 ) );
        while(semaphore_fib2>0)
        {
            // waiting
        }
        global_stream_lock.lock();
        std::cout << "[" << boost::this_thread::get_id()
            << "] ******* CalcFib2 group finished  ********" << std::endl;
        global_stream_lock.unlock();
    }
    work.reset();
    worker_threads.join_all();

    return 0;
}

score 0 · Answer 1 · answered Oct 18 '11 at 12:32

This is how I solved too many if statements in while loop in main().

Added the boost::condition_variable cv; and boost::mutex mx; as global variables
Added boost::lock_guard<boost::mutex> lk(mx);cv.notify_all(); at the end of CalcFib and CalcFib2 functions
Added boost::unique_lock<boost::mutex> lk(mx); cv.wait(lk); in every while loop

That way I had accomplished that the if statement in while loop gets executed only when cv.notify_all() executes.

boost asio semaphore-like solution

1 Answers1