thread-safe timer for C++ callback

[arjunaw]

In a unix pthreads based app I'm working on, I have objects of a particular class (call it class foo) being created in multiple threads. I need a specific public method of class foo invoked at or after 60 seconds of the object coming into existence (it is not imperative that it happens at precisely 60s, just that it happens at either 60s or very shortly thereafter).

Any ideas on what thread-safe timers are available that could I use to achieve this? Looking for either something that I could just drop right in to my class foo or which I could derive from.

[template<>]

You could create one thread as a service thread that calls the public methods of your objects every minute...

[mike_2000_17]

I think that just an ordinary sleep function would do, no?

For example, using boost::thread (I'm not so familiar with native pthread stuff):

#include <boost/thread/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/bind/bind.hpp>
#include <iostream>

class foo {
  public:
    void my_delayed_method() {
      std::cout << "Hello World!" << std::endl;
    };
  private:
    boost::thread delay_thread;
    void call_method_with_delay() {
      boost::this_thread::sleep(boost::posix_time::seconds(60));
      my_delayed_method();
    };
  public:
    foo() : delay_thread(boost::bind(&foo::call_method_with_delay,this)) { };
    ~foo() { delay_thread.join(); };
};

If you need the public method to be called from the same thread as the foo object got created on, then you need to use some sort of signal instead of a direct call:

#include <boost/thread/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/bind/bind.hpp>
#include <iostream>

class foo {
  public:
    void my_delayed_method() {
      std::cout << "Hello World!" << std::endl;
    };
  private:
    boost::thread delay_thread;
    bool should_call;
    void call_method_with_delay() {
      boost::this_thread::sleep(boost::posix_time::seconds(60));
      should_call = true;
    };
  public:
    foo() : delay_thread(boost::bind(&foo::call_method_with_delay,this)), should_call(false) { };
    ~foo() { delay_thread.join(); };
    
    void call_if_delay_passed() {
      if(should_call) {
        my_delayed_method();
        should_call = false;
      };
    };
};

//for sake of example, here is a creator thread:
void create_and_loop() {
  foo f;
  while(true) {
    f.call_if_delay_passed();
    boost::this_thread::sleep(boost::posix_time::seconds(1));
  };
};

int main() {
  std::cout << "About to create several threads... press ctrl-C to stop!" << std::endl;
  
  boost::thread* t[10];
  for(int i=0;i<10;++i)
    t[i] = new boost::thread(&create_and_loop);

  std::cin.get();
  return 0;
};

You will observe that there is nothing thread-unsafe in the above.

[arjunaw]

Thanks very much for your reply. I haven't used boost::thread before but I think I understand the gist of it from your two examples.

If my understanding is correct, will a new boost thread be created per foo object created? That is my only concern because in this program's intended usage there will be times when many foo objects are being created (in the region of about 1000 or more per minute). I'm not sure I would want a 1000 new threads being created where each one is essentially a timer for its corresponding foo object. Do you think what you've proposed could be modified so that there is just a single new thread created that will act as the timer for all foo objects created?

[Agni]

It seems like you want a subscriber-publisher model. There will be one publisher thread which calls a particular function on all subscribers every 60 seconds. The publisher can be a Singleton and subscribers could just subscribe with a callback function to it. However if there is only 1 publisher then it could be that a new foo object subscribes just a fraction of second before the publisher is going to wake up so the first call can be at any time after you subscribe. All other calls will happen at 60 seconds periods though.

[mike_2000_17]

As Agni suggested, you can group all those delays/timers into one singleton object. The same mechanism applies when it comes to calling from the same thread as foo was created or calling from any thread. I will just make an example for calling from the timer thread (you would just have to replace the callback by setting of a flag to signal the call on the original thread).

#include <boost/thread/thread.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/bind/bind.hpp>
#include <iostream>
#include <map>

class publisher {
  private:
    typedef std::multimap< boost::posix_time::ptime, boost::function<void()> > CallBackMap;
    typedef CallBackMap::iterator CallBackIter;
    CallBackMap callbacks;
    boost::mutex callbacks_mutex;
    boost::thread publishing_thread;
    void execute() {
      while(true) {
        {
          boost::unique_lock<boost::mutex> lock(callbacks_mutex);
          CallBackIter it_end = callbacks.upper_bound(boost::posix_time::millisec_clock::local_time() + boost::posix_time::milliseconds(20));
          for(CallBackIter it = callbacks.begin(); it != it_end; ++it)
            it->second(); //call the callbacks that are scheduled.
          callbacks.erase(callbacks.begin(), it_end); //erase callbacks that were called.
        }; //release lock.
        boost::this_thread::sleep(boost::posix_time::milliseconds(20)); //sleep 20ms 
      };
    };

    //private constructor. Will start the execution thread.
    publisher() : publishing_thread(boost::bind(&publisher::execute,this)) { }; 
    publisher(const publisher&); 
    publisher& operator=(const publisher&); //non-copyable.
  public:
    void subscribe(boost::function<void()> f, boost::posix_time::time_duration dt) {
      boost::unique_lock<boost::mutex> lock(callbacks_mutex);
      callbacks.insert(std::make_pair(boost::posix_time::microsec_clock::local_time() + dt, f));
    };

    static publisher& getPublisher(); //singleton.
};

publisher& publisher::getPublisher() {
  static publisher p; //will be created on first use.
  return p;
};

class foo {
  public:
    void my_delayed_method() {
      std::cout << "Hello World!" << std::endl;
    };

    foo() {
      publisher::getPublisher().subscribe(boost::bind(&foo::call_method_with_delay,this), boost::posix_time::seconds(60)); //subscribe for a callback in 60 seconds.
    };
};

[arjunaw]

Many thanks Mike (and Agni). I appreciate your taking the time to write sample code showing what you mean. Cheers!