Hi Gottlob,
 
   Thanks for your detailed reply.
   I will go through it and see if I can implement what you have suggested.
   In the mean time, please find below a complete program.
   I did not send it in the previous mail just to save on the space.
   Let me know if you have any more comments.
 
-----------------------------------------------------------------------
# include <iostream>
# include <vector>
# include <string>
# include <boost/thread/thread.hpp>
# include <boost/thread/pthread/condition_variable.hpp>
 
using namespace std;
using namespace boost;
 
boost::condition_variable new_job_condition;
boost::mutex new_job_mutex;
boost::mutex new_job_notification_reached_flag_mutex;
bool new_job_notification_reached_flag;
 
boost::condition_variable job_read_condition;
boost::mutex job_read_mutex;
boost::mutex job_read_notification_reached_flag_mutex;
bool job_read_notification_reached_flag;
 
boost::condition_variable job_done_condition;
boost::mutex job_done_mutex;
boost::mutex job_done_notification_reached_flag_mutex;
bool job_done_notification_reached_flag;
 
boost::condition_variable worker_exit_notification_condition;
boost::mutex worker_exit_mutex;
boost::mutex worker_exit_notification_flag_mutex;
bool worker_exit_notification_reached_flag;
 
boost::mutex job_string_mutex;
string job_string;
 
boost::mutex whole_job_completed_mutex;
bool whole_job_completed_flag;
 
void enable_flag (boost::mutex &mut, bool &flag)
{
mut.lock ();
flag = 1;
mut.unlock ();
}
 
void disable_flag (boost::mutex &mut, bool &flag)
{
mut.lock ();
flag = 0;
mut.unlock ();
}
 
void mywait (boost::condition_variable &cond, boost::mutex &mut, boost::mutex &flag_mut, bool &flag)
{
boost::unique_lock<boost::mutex> lock(mut);
 
enable_flag (flag_mut, flag);
cond.wait (lock);
}
 
void notify_until_success (boost::condition_variable &cond, boost::mutex &mut, boost::mutex &flag_mut, bool &flag)
{
for (;;)
{
mut.lock ();
flag_mut.lock ();
if (flag == 1)
{
cond.notify_one();
 
flag = 0;
flag_mut.unlock ();
mut.unlock ();
break;
}
flag_mut.unlock ();
mut.unlock ();
}
}
 
void worker_function (void)
{
for (;;)
{
/* wait for the new job */
mywait (new_job_condition, new_job_mutex, new_job_notification_reached_flag_mutex, new_job_notification_reached_flag);
 
whole_job_completed_mutex.lock();
if (whole_job_completed_flag == 1)
{
whole_job_completed_mutex.unlock();
notify_until_success (worker_exit_notification_condition, worker_exit_mutex, worker_exit_notification_flag_mutex, worker_exit_notification_reached_flag);
return;
}
whole_job_completed_mutex.unlock();
 
/* read the new job */
job_string_mutex.lock();
string local_job_string = job_string;
job_string = "";
job_string_mutex.unlock();
 
/* notify job read successfully */
notify_until_success (job_read_condition, job_read_mutex, job_read_notification_reached_flag_mutex, job_read_notification_reached_flag);
 
/* execute job */
system (local_job_string.c_str ());
 
/* notify job done successfully */
notify_until_success (job_done_condition, job_done_mutex, job_done_notification_reached_flag_mutex, job_done_notification_reached_flag);
}
}
 
class MANAGER
{
private :
thread_group G;
int nWorkers;
 
public :
MANAGER (int num) : nWorkers (num)
{
for (int i=0; i<nWorkers; i++)
{
G.create_thread (worker_function);
}
}
 
int run_job (vector <string> &joblist);
void join_all (void)
{
G.join_all ();
}
};
 
int MANAGER :: run_job (vector <string> &joblist)
{
int free_workers = nWorkers;
vector<string> :: iterator it;
 
for (it = joblist.begin (); it != joblist.end (); ++it)
{
/* read the job */
job_string_mutex.lock();
job_string = *it;
job_string_mutex.unlock();
 
/* notify any one of the free threads about new_job */
notify_until_success (new_job_condition, new_job_mutex, new_job_notification_reached_flag_mutex, new_job_notification_reached_flag);
 
/* wait for job_read notification from that thread */
mywait (job_read_condition, job_read_mutex, job_read_notification_reached_flag_mutex, job_read_notification_reached_flag);
 
/* worker got allocated to the job so decrement number of free_workers */
free_workers--;
 
if (free_workers > 0)
{
/* enable flag new_job_notification_reached_flag here; as it got disabled in notify_until_success() */
enable_flag (new_job_notification_reached_flag_mutex, new_job_notification_reached_flag);
}
 
/* if there are free workers and there are jobs available */
if (free_workers > 0 && ((it + 1) != joblist.end ()))
continue;
 
/* wait for job_done notification from any of the busy thread */
mywait (job_done_condition, job_done_mutex, job_done_notification_reached_flag_mutex, job_done_notification_reached_flag);
 
/* worker is idle now so increment number of free_workers */
free_workers++;
}
 
/* at this points all the jobs have exhausted and some workers may be idle (listening for new job) and some may be busy (still doing their job)
* hence wait() for all of them to report job_done
*/
 
while (free_workers != nWorkers)
{
mywait (job_done_condition, job_done_mutex, job_done_notification_reached_flag_mutex, job_done_notification_reached_flag);
free_workers++;
}
 
/* at this point all worker threads have done their job */
 
/* now ask all of them about new_job so that they will unblock and will check for whole_job_completed_flag for exit */
enable_flag (whole_job_completed_mutex, whole_job_completed_flag);
 
int i=0;
while (i<nWorkers)
{
notify_until_success (new_job_condition, new_job_mutex, new_job_notification_reached_flag_mutex, new_job_notification_reached_flag);
mywait (worker_exit_notification_condition, worker_exit_mutex, worker_exit_notification_flag_mutex, worker_exit_notification_reached_flag);
i++;
 
/* enable flag new_job_notification_reached_flag here; as it got disabled in notify_until_success() */
enable_flag (new_job_notification_reached_flag_mutex, new_job_notification_reached_flag);
}
 
join_all ();
return 0;
}
 
int main (void)
{
int nWorkers = 3;
MANAGER M(nWorkers);
vector<string> tclist;
 
tclist.push_back ("sleep 1");
tclist.push_back ("sleep 2");
tclist.push_back ("sleep 3");
tclist.push_back ("sleep 4");
tclist.push_back ("sleep 5");
tclist.push_back ("sleep 6");
tclist.push_back ("sleep 7");
tclist.push_back ("sleep 8");
tclist.push_back ("sleep 9");
tclist.push_back ("sleep 10");
tclist.push_back ("sleep 1");
tclist.push_back ("sleep 2");
tclist.push_back ("sleep 3");
tclist.push_back ("sleep 4");
tclist.push_back ("sleep 5");
tclist.push_back ("sleep 6");
tclist.push_back ("sleep 7");
tclist.push_back ("sleep 8");
tclist.push_back ("sleep 9");
tclist.push_back ("sleep 10");
 
M.run_job (tclist);
}
-----------------------------------------------------------------------
Regards,
Girish

--- On Tue, 1/19/10, Gottlob Frege <gottlobfrege@gmail.com> wrote:

From: Gottlob Frege <gottlobfrege@gmail.com>
Subject: Re: [Boost-users] wait/notify problem
To: boost-users@lists.boost.org
Date: Tuesday, January 19, 2010, 5:09 PM



On Mon, Jan 18, 2010 at 8:00 AM, girish hilage <girish_hilage@yahoo.com> wrote:
   The code I have now written is as follows (which seems to work fine) :

This code is broken.  I find it hard to read and understand, and it contains 4 different mutexes, which is asking for trouble.  But besides those 'aesthetic' complaints, it really is broken.

(IIUC) It appears that you are trying very hard to make sure someone is waiting on the condition before notifying.  I was tempted to say this can't be done, but you have done it nonetheless.  Although I am not sure it helps.

Please research 'spurious wakeups' of condition variables:
 
void mywait (boost::condition_variable &cond, boost::mutex &mut, boost::mutex &flag_mut, bool &flag)
{
   boost::unique_lock<boost::mutex> lock(mut);

   enable_flag (flag_mut, flag);
   cond.wait (lock);

This cond.wait(lock) may wake up even though cond.notify_one() was NOT called.  This may seem strange, but it is allowed because otherwise implementing conditions becomes very hard.  And by allowing it, it encourages developers to code with conditions properly.  To use a condition you MUST do so in a loop which checks the global *state* that the condition is communicating.  ie in your case 'new job available'.  The standard could have stored state inside the condition variable, but it was harder, and it was also assumed that you can determine the state better on your own, so it is outside the condvar. It also makes it easier for multiple threads to agree - they can all look at the external state (available jobs) and decide if something needs to be done, or if another thread has already done it.

Another way to look at things: spurious wakeups mean that, in effect, notify_one might wake up more than one thread (they still battle for the mutex of course), but notify_one is essentially (worse case) the same as notify_all - ie all the other threads could 'spuriously' wake up.  ie notify_one could be implemented by just calling notify_all (it would be a poor implementation, but valid).  So notify_one is just an optimization (*probably* only one thread will awaken), not a guarantee.
 
So what happens in your code if cond.wait() wakens spuriously?  It appears a job is 'done' (or attempted) even though no job exists.

If you can tell (from the worker thread) that there is no job to do, then make THAT your external state, and wrap the cond.wait() inside a loop:


}

void worker_function (void)
{ 
 for (;;)
   { 
         while ( ! /* there is work to be done */)  // note the '!' at the front
            cond.wait(mutex);

      ..... read new job .....
 
         // probably need to hold lock while reading job (from some shared queue or whatever)
         // then unlock so you can do the work *in parallel* with other threads

         mutex.unlock(); // unlock before doing work
 
      ..... do the work .....

         mutex.lock();   // lock again before checking /* there is work to be done */, unlocked inside cond.wait()
   }
}

int main (void)
{
   ..... create worker threads .....

   while (/*there is work to be done */)
   {

                 cond.notify_all();  // there may be more than one job to do, so notify all
 
      /* wait for job_done notification from any of the busy thread */
      mywait (job_done_condition, job_done_mutex, job_done_notification_reached_flag_mutex, job_done_notification_reached_flag);

Why do you wait for the job to be done?
If you send a job to another thread, then do nothing but wait for it to be done, why not just do it yourself?
ie it appears that your code will run like this:

Thread 1          Thread 2
                        wait:
check for job     .....
notify                wake
wait for done:     do job
.....                   keep doing that job
.....                   keep doing that job
.....                   job done!
wake!                wait for new job (ie repeat)

do the 2 threads ever do any real work at the same time, or is one always waiting for the other?

 
   }
}

   I hope this will not create any problems.
   (I have also edited the problem statement in the mail below to make it more clear. Shown by __TEXT_EDITED__.)
Regards,
Girish


I honestly had (still have?) trouble understanding the code, but I don't think it works, and I think it could be much simpler.

Basically, just use the condvars to tell threads 'hey *CHECK* if there is something to do'.  Do NOT try to use condvars to actually communicate state.  Just 'hey check the state'.  Then let the threads decide what to do based on that state.  condvars also nicely handle the lock/unlock/relock that you probably needed to do around the check and wait anyhow.  So the primary purpose of the lock is to protect your state, not for the condvar.  The condvar is just nice enough to know how to work with that.

In your case your state is the state of the job list.  Put a mutex around that, and have the condvar use that mutex as well.  Check the state of the job list to decide what (if anything) to do.

There is only one state (job list) and one mutex and one condvar.

Tony

Tony

-----Inline Attachment Follows-----

_______________________________________________
Boost-users mailing list
Boost-users@lists.boost.org
http://lists.boost.org/mailman/listinfo.cgi/boost-users