Win32 reset event like synchronization class with boost C++

Question

I need some mechanism reminiscent of Win32 reset events that I can check via functions having the same semantics with WaitForSingleObject() and WaitForMultipleObjects() (Only need the ..SingleObject() version for the moment) . But I am targeting multiple platforms so all I have is boost::threads (AFAIK) . I came up with the following class and wanted to ask about the potential problems and whether it is up to the task or not. Thanks in advance.

class reset_event
{
 bool flag, auto_reset;
 boost::condition_variable cond_var;
 boost::mutex mx_flag;

public:
 reset_event(bool _auto_reset = false) : flag(false), auto_reset(_auto_reset)
 {
 }

 void wait()
 {
  boost::unique_lock<boost::mutex> LOCK(mx_flag);
  if (flag)
   return;

  cond_var.wait(LOCK);
  if (auto_reset)
   flag = false;
 }

 bool wait(const boost::posix_time::time_duration& dur)
 {
  boost::unique_lock<boost::mutex> LOCK(mx_flag);
  bool ret = cond_var.timed_wait(LOCK, dur) || flag;
  if (auto_reset && ret)
   flag = false;

  return ret;
 }

 void set()
 {
  boost::lock_guard<boost::mutex> LOCK(mx_flag);
  flag = true;
  cond_var.notify_all();
 }

 void reset()
 {
  boost::lock_guard<boost::mutex> LOCK(mx_flag);
  flag = false;
 }
};

Example usage;

reset_event terminate_thread;

void fn_thread()
{
 while(!terminate_thread.wait(boost::posix_time::milliseconds(10)))
 {
  std::cout << "working..." << std::endl;
  boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
 }

 std::cout << "thread terminated" << std::endl;
}

int main()
{
 boost::thread worker(fn_thread);

 boost::this_thread::sleep(boost::posix_time::seconds(1));
 terminate_thread.set();

 worker.join();

 return 0;
}

EDIT

I have fixed the code according to Michael Burr's suggestions. My "very simple" tests indicate no problems.

class reset_event
{
    bool flag, auto_reset;
    boost::condition_variable cond_var;
    boost::mutex mx_flag;

public:
    explicit reset_event(bool _auto_reset = false) : flag(false), auto_reset(_auto_reset)
    {
    }

    void wait()
    {
        boost::unique_lock<boost::mutex> LOCK(mx_flag);
        if (flag)
        {
            if (auto_reset)
                flag = false;
            return;
        }

        do
        {
            cond_var.wait(LOCK);
        } while(!flag);

        if (auto_reset)
            flag = false;
    }

    bool wait(const boost::posix_time::time_duration& dur)
    {
        boost::unique_lock<boost::mutex> LOCK(mx_flag);
        if (flag)
        {
            if (auto_reset)
                flag = false;
            return true;
        }

        bool ret = cond_var.timed_wait(LOCK, dur);
        if (ret && flag)
        {
            if (auto_reset)
                flag = false;

            return true;
        }

        return false;
    }

    void set()
    {
        boost::lock_guard<boost::mutex> LOCK(mx_flag);
        flag = true;
        cond_var.notify_all();
    }

    void reset()
    {
        boost::lock_guard<boost::mutex> LOCK(mx_flag);
        flag = false;
    }
};

Standice

Answer 1

A few things that you'll want to check/fix (note - I'm by no means saying that these are the only things - I've only had a quick look):

• in your wait() function, you don't reset on an already signaled event if it's set up for auto_reset:

   void wait()
   {
    boost::unique_lock<boost::mutex> LOCK(mx_flag);
    if (flag) {
     if (auto_reset) flag = false;    // <-- I think you need this
     return;
    }
  
    cond_var.wait(LOCK);
    if (auto_reset)
     flag = false;
   }
  

• in wait(const boost::posix_time::time_duration& dur) you should check flag before waiting on the condition variable.

• in both wait functions, if you wait on the condition variable, you might need to recheck the flag to make sure that some other thread hasn't reset the event in the meantime. This is particularly true for auto_reset events which should only free up a single waiter even when multiple threads are waiting on the event.

Krute

Thank you so much Michael. Can you explain the last item a bit more please? – Standice

Answer 2

Here is my version which was slightly tweaked to achieve the following.

• No blocking the producer using set(),reset(),etc instead it counts the number of "releases" rather than losing the 1:1 mapping after the boolean condition is true.
• Allows outside callers to specify the mutex for wait(), it's often an outside resource anyways in my use scenarios and can be separate from the internal mutex.
• Moved set(),reset_one(),reset_all() calls to an internal mutex, now they don't block when repeatedly called before the consumer calls wait().

Now my loading thread can queue up multiple long-lived requests without dropping any tasks while busy processing.

Progression in my project Using....

Boost Condition Variable: -> send 3 load requests, thread is busy and only sees 1 or 2.
Posted Answer using Bool: -> send 3 load requests, producer blocks on 2nd request due to shared mutex. producer doesn't unblock until 1st load request processed.
My version -> send 3 load requests, producer immediately returns from all 3, consumer sees 3 load request slowly but surely :)

Hopefully it helps someone out there.

    class cNonLossyCondition
    {
        bool flag, auto_reset;
        boost::condition_variable cond_var;
        int lost_signals;
        boost::mutex internal_mutex;

    public:
        cNonLossyCondition(bool _auto_reset)
        {
            this->flag = false;
            this->auto_reset = auto_reset;
            this->lost_signals = 0;
        }

        void wait(boost::mutex* mx_flag)
        {
            boost::unique_lock LOCK(*mx_flag);
            if (flag)
            {
                if (auto_reset)
                    this->reset_one();
                return;
            }

            do
            {
                cond_var.wait(LOCK);
            } while(!flag);

            if (auto_reset)
                this->reset_one();
        }

        bool wait(boost::mutex* mx_flag,const boost::posix_time::time_duration& dur)
        {
            boost::unique_lock LOCK(*mx_flag);
            if (flag)
            {
                if (auto_reset)
                    this->reset_one();
                return true;
            }

            bool ret = cond_var.timed_wait(LOCK, dur);
            if (ret && flag)
            {
                if (auto_reset)
                    this->reset_one();

                return true;
            }

            return false;
        }

        void set()
        {
            boost::lock_guard LOCK(this->internal_mutex);
            flag = true;
            if (this->lost_signals lost_signals = 1; //already incremented
            } else {
                this->lost_signals = this->lost_signals + 1;
            }

            cond_var.notify_all();
        }

        void reset_one()
        {
            boost::lock_guard LOCK(this->internal_mutex);
            this->lost_signals = this->lost_signals - 1;
            if (this->lost_signals lost_signals = 0;
                flag = false;
            }

        }
        void reset_all()
        {
            boost::lock_guard LOCK(this->internal_mutex);
            flag = false;
            this->lost_signals = 0;
        }
    };

Rosenquist