Creating a C++ callback function from a class instance method

I have been playing around with the Boost library and, in particular, the boost::thread multithreading library.

There a thread is simply created by constructing an object of type boost::thread, and passing in to the constructor a pointer to the function to be executed in the newly spawned thread.

There is much more to it, and I encourage you to read the documentation, but that gave me an interesting opportunity to ponder the issue of how to create a thread by instead passing in a pointer to a class's method: the need typically arises when one does not want to use global variables (which need to be guarded against race conditions) or where parameters are necessary for the execution of the method and/or return values are expected.

Granted, the boost::thread constructor allows you to pass in up to nine parameters, so that's rarely an issue, but I just wanted to find out a reasonably general way to achieve this; it turns out that this is far from trivial, and I thought I'd share my findings.

For the impatient, this is the solution:

template<typename T, typename V, typename R>
class MakeCallback {

  // Member method in T(ype), takes parameter V(alue),
  // and returns an object of type R(esult)
  // Note the parentheses around (T::*): without them, 
  // the compiler gets confused.
  typedef R(T::*func)(const V&);
  func f_;
  T& t_;
  V value_;
  R* res_;
public:
  MakeCallback(T& type, func f, const V& value, R* res = NULL) :
      f_(f), t_(type), value_(value), res_(res) { }
  virtual ~MakeCallback() { }
  void operator()() {
    if (res_)
      // Note here the parenthesis around t_.*f_ 
      // They are necessary, or a compiler error will be generated
      *res_ = (t_.*f_)(value_);
  }

  // This allows the pointer to the result value to
  // be set after the object has been created
  void set_res(R* res) { res_ = res; }
};

and this is how one uses it:

int main(int argc, char[}* argv) {
  A a(33);
  int res;
  // Generally, you don't need the & operator to take a function's pointer.
  // But in this case it's mandatory, or the compiler will complain
  // about doSomething() not being static.
  // In any event, the use of & makes your intent clearer (you are taking
  // the method's address,
  // not invoking it) and I encourage you to use it consistently, even where
  // this is not stricly necessary.
  MakeCallback<>A, int, int> mc(a, &A::doSomething, 22, &res);
  boost::thread do_it(mc);
  do_it.join();
  std::cout << "And the result is : " << res << std::endl;

  B b;
  boost::thread another(MakeCallback<A, std::string, B>(a, &A::doSomethingElse, "22.13", &b));
  another.join();
  std::cout << "..and B is " << b.get() << std::endl;
}

given the following declarations for A and B:

class B {
  float x;
public:
  B(float s = 0.0) : x(s) {}
  float get() { return x; }
};

class A {
  int num_;
public:
  A(int num) : num_(num) { }

  int doSomething(const int& k) {
    return num_ + k;
  }

  B doSomethingElse(const std::string& s) {
    double b = ::atof(s.c_str()) + num_;
    return B(b);
  }
};

A couple of points that have caused me much head-scratching and I thought it worth passing on:

• you must use the '&' operator in front of A::doSomething, or the compiler will complain about it not being a static function;
• note we are, in fact, calling an instance method: in other words, we expect to have instance-specific values stored in the class (A in this case) that our method (doSomething()) will use
• note also the use of a pointer (R* res_) to store the return value: this is important; without it, the return value of the method call ((t_.*f_)(value_)) will be lost!
  
  
  This is because, in the call to boost::thread(mc) the compiler automatically created a copy of our object (using the compiler-generated copy constructor; see Scott Meyer's "Effective C++", Item 5)
• Note also the calls to thread::join() - without them, you'd have a race on the returned values: possibly reading them, before the actual method (doSomething()) had any chance of updating it;
  

Just to be clear, this rather trivial implementation is not meant to be used in real code (if you need something along these lines, boost::bind is what you want) but it was an interesting exercise in how to deal with generics, functors and callbacks.

J2ME Multi-Threading and HTTP (Part 2)

In part 1 of this series, I have explained how to implement a LongRunningTask.
All the magic, however is done in the ControllerThread class, which takes care of executing the task, keeping tabs on it and, eventually, terminating it if it overstays its welcome.

Before getting on to ControllerThread, it is worth, however, to briefly review the Observer interface that will have to be implemented by the calling class.

We need an Observer here mostly for three reasons:

• because of the asynchronous nature of the call, we need a way to communicate back when we are done with our task (if you look back at the last snippet of code in Part 1, you'll notice that startRecording() returns immediately: however the recording itself will last a few seconds, until either the user stops it, or the ControllerThread "expires" - in this case, after 10 seconds at most);
• we may also want to be notified of progress (eg, by showing a "progress bar" to the user);
• most importantly, we need to be notified when the task failed (that is, a TaskException was thrown).

Bear in mind that we cannot "wrap" the call to Thread.start() with a try/catch block, as no exception will ever be thrown by this code - catching the TaskException is a job for the ControllerThread (and the reason why LongRunningTask does not implement Runnable).

The Observer interface is defined as follows:

public interface Observer {
  /**
   * A simple logging ability, this has to be implemented in a
   * thread-safe manner, eg wrapping the logging with synchronized blocks
   *
   * @param msg a simple logging message
   */
  public void addMsg(String msg);

  /**
   * callback for signaling progress of time-consuming process, provides
   * an opportunity to either allow stopping the background operation or
   * update an UI element providing visual indication that the application
   * is not 'stuck'
   */
  public void idle();

  /** time-consuming operation complete, results, if any, are available */
  public void complete();

  /** acknowledgement of interruption, optional */
  public void interrupted();

  /**
   * This is called by the LongRunningTask to signal progress, could
   * be used to update some UI element (eg a progress bar)
   *
   * @param progress a value (meaningful to the application's semantics)
   *      that indicates progress.  Most typically, a value between 0
   *      and 100, to indicate percentage progress towards completion.
   */
   public void announceProgress(int progress);

  /** Called when the task fails */
  public void fail(TaskException tex);
}

Once a LongRunningTask is started, it will either complete() or fail(): these methods will be called by ControllerThread (not LongRunningTask); the latter will either complete the execute() method normally, or throw a TaskException to indicate failure.

ControllerThread is a pretty long class: the full source code can be found here, only snippets of code will be shown in the following.

public ControllerThread(LongRunningTask task, long waitInterval,
long timeout, Observer observer)

The constructor takes a reference to the task to execute, an (optional) reference to an Observer and two numeric values that indicate, respectively, how often it should check on progress and how long the task is given to complete its work.

ControllerThread does implement Runnable - it is executed in its very own thread and creates (and executes) in turn another thread: as I mentioned in the first part of this article, to execute a background thread, you need two threads, just one won't do!

public void run() {
  long start = System.currentTimeMillis();
  long elapsed;
  Thread t = new Thread(new ControllerRunnable());
  t.start();
  while (!task.isDone() && !task.isStopped() && !failed) {
    wait_state = THREAD_STATE_WAIT;
    elapsed = System.currentTimeMillis()-start;
    if (elapsed > totalTimeout){
      task.cancel();
      wait_state = THREAD_STATE_EXPIRED;
      getObserver().interrupted();
      return;
    }
    synchronized (lock) {
      try {
        // double-check to avoid race conditions
        if (!task.isDone() && !task.isStopped())
          lock.wait(waitTimeout);
      } catch (InterruptedException ex) {
        failed = true;
        failureMessage = "ControllerThread interrupted";
        t.interrupt();
        getObserver().interrupted();
        return;
      }
    }
  } // while
}

All the magic is done in the internal class ControllerRunnable that implements Runnable itself and inside its run() method actually executes the LongRunningTask, wrapped in a try/catch block to catch failures.

private class ControllerRunnable implements Runnable {

  public void run() {
    if(task == null) {
      failureMessage = "No task to execute";
      failed = true;
      getObserver().fail(new TaskException("No task to execute"));
      return;
    }
    try {
      task.execute(getObserver());
      synchronized (lock) {
        lock.notify();
      }
      getObserver().complete();
    } catch(TaskException ex) {
      failureMessage = new String(ex.getMessage());
      failed = true;
      getObserver().fail(ex);
    }
  }
}

As you can see, it is ControllerRunnable.run() that calls the Observer's complete() or fail() methods, and ensures that:

• one and only one of those methods will be called;
• that they will be called at most once.

This is critically important in the very frequent case of real-life applications that have multiple tasks that may be executing at any one time (and, possibly, even concurrently) and hence are likely to have state variables to control what's going on: in this case, calling twice complete() is almost as bad as not calling it at all.

Note also the wait/notify mechanism (lock is simply an Object, private to ControllerThread, used to synchronize the two threads) to signal the controlling thread that we are done and that is no longer necessary to wait. If the notification fails to arrive within the timeout value set when creating the controller thread, the task will be terminated (see ControllerThread.run()).

From an application developer's point of view, however, all of the above can be treated as "magic."

All one has to do to use this framework is:

• write a time-consuming task as a single execute() method;
• wrap the time-consuming task in a class that implements LongRunningTask (bearing in mind that most of its methods are optional - only execute() is, rather obviously, mandatory);
• implement Observer in the calling class (most likely, a Controller class in an MVC pattern) - this will also include, probably, adding a "Stop" command to the UI as well as some sort of progress indicator (eg, a Gauge);
• wrap the call to the time-consuming task in the following code:

public class Multimedia implements Observer {

  // Observer interface implementation omitted here
  // ref. source code & Part II of this article

  public void startRecording() {
    task = new RecordTask();
    thread = new ControllerThread(task, 1000L, 10000L, this);
    new Thread(thread).start();
  }

  // Other stuff goes here...
}

Update: I have uploaded the source code to BitBucket, you can easily download it using Mercurial (see here for an intro) by simply using:

https://marco@bitbucket.org/marco/j2me_threads

J2ME Multi-Threading and HTTP

Update: This is Part I of a two-part series about multi-threading in the J2ME environment, Part II can be found here.
I have uploaded the source code to BitBucket, you can easily download it using Mercurial (see here for an intro) by simply using:

https://marco@bitbucket.org/marco/j2me_threads

...

In my experience, one of the most difficult things to achieve in software (and even
more so in "limited" environments, such as J2ME) is to harness certain "long-running tasks" so that:

• they have enough time to complete their task, yet they do not
  hog the system so as to make it unresponsive to users (who would
  then have the impression the application is 'hung');
• it is easy to both check on whether they are still running or are, indeed, 'hung';
• it is relatively straightforward to set a timeout, expired which, they will be, rather uncerimoniously aborted;
• it is relatively straightforward to enable a 'Cancel' option for the user to press if they get bored before the timeout expires.

The hortodox answer to this has been (and quite rightly so) to have the long running task run in a separate thread from the main (GUI) thread, with a boolean variable to control execution.

Something like this:

public class MyLongTask implements Runnable {

  private volatile boolean stopped = false;
    public void stop() {
    stopped = true;
  }

   public void run() {
     boolean finished = false;
     // do some setup here
     // ...

     while (!stopped && !finished) {
       doAChunkOfWork();
       finished = amIDone();
     }
   }
}

or something similar to that.
The controlling thread (typically the one that runs the GUI too, which has probably a "Cancel" button hooked to a cancelTask() method) looks something like this:

public class Controller {
  private MyFrameView theView;
  private MyLongTask task = null;

  public Controller(MyFrameView theView) {
    this.theView = theView;
  }

  // various other stuff...

  public void doStartTimeConsumingTask() {
    task = new MyLongTask();
    new Thread(task).start();
  }

  public void doCancel() {
    if(task != null)
      if (! task.isFinished())
        task.stop();
  }

  // other Controller "stuff"

}

This is, in fact, what the good folks at Sun in the NetBeans team have essentially done, by defining the CancellableTask interface and a generic SimpleCancellableTask class (although, the latter, quite sadly has not implemented... the cancel() method!).

This was mostly done to supporting NetBeans' WaitScreen component and, even then, I found it not terribly compelling. (see Lukas Hasik's blog here).

Initially, when confronted with the problem, I simply extended SimpleCancellableTask, and used the stopped/stop() mechanism to enable the cancel() mechanism - that enabled my application to stop a long running task that had overstayed its welcome; however, at the price of having to throw a RuntimeException to communicate back to the WaitScreen that it was done, but had not finished its task.

So far, so good.

However, there is still no easy way to set a timeout task, nor we have implemented any mechanism to let MyLongTask tell Controller they're done (let alone let the latter communicate back to the the View, which, most likely, will display something to the user to tell her we're done).

The latter problem is easily solved with the Observer pattern (see my other post - to be published later): this can be implemented by either the view or the controller - my preference has always been for the Controller to act as the observer (again, I seem to be in broad agreement with most practitioners - those times when, for expediency, I have implemented that in the View, I have always come to regret it... Thank God for Refactoring!).

Setting a timeout and generally keeping tabs on the long-running task, in fact, requires that one uses an additional, controlling thread.
In other words, to properly manage a time consuming task, you need two additional threads - just one won't do.

I decided then that a more general solution was needed, one that would allow an easy, programmatic way of controlling a long running task and was, ideally, also "pre-wired" with an observer.

My approach uses a ControllerThread, a LongRunningTask interface and an Observer interface.

The code can be downloaded here, please note that this is written specifically for 'microedition' environments(J2ME), hence no generics or any other Java 5 niceties are used.

The LongRunningTask interface is reviewed below, the ControllerThread will be analysed in Part 2 of this article.

LongRunningTask is a relatively straightforward interface, that defines a couple of 'check-me' methods (isDone() and isStopped()) the obvious cancel() one and an optional progress() method, that returns a percentage value to completion (the actual semantic of the value returned by progress(), is worth noting, is entirely application-dependent, although I like to use it as a % value, to update, for example, a Gauge UI indicator).

The 'quirk' here is that LongRunningTask does not extend Runnable: the reason for this will be clearer when we'll come to look at ControllerThread (that does implement Runnable). All the work is done in execute() that throws a TaskException (unlike run() in Runnable) and can thus indicate abnormal termination to a 'wrapper' class.

To use the framework, one simply implements LongRunningTask for a class with a time-consuming method, and checks regularly on a boolean variable that may be set by cancel() (or any other sensible way to communicate to a running thread that it's time to quit it, enough is enough).

public class RecordTask implements LongRunningTask {

  private boolean stopped;
  private boolean done;
  private int progressIndicator;

  private byte[] audioData;

  /** Creates a new instance of RecordTask */
  public RecordTask() {
  }

  public boolean cancel() {
    stopped = true;
    return true;
  }

  public boolean isStopped() {
    return stopped;
  }

  public boolean isDone() {
    return done;
  }

  public int progress() {
    return progressIndicator;
  }

  /**
   * This is where all the time-consuming work is done:
   * here we record a speech segment, long at most maxDuration
   * or until the user presses a "Stop" command (that in turn will cause
   * the controller to call cancel()
   *
   * @param observer the observing class
   */
  public void execute(Observer observer) throws TaskException {
    observer.addMsg("Record started");
    try {
      done = false;
      stopped = false;
      Player player = Manager.createPlayer(getAudioMimetype());
      if (player != null) {
        player.realize();
        RecordControl rc = (RecordControl) player.getControl("RecordControl");
        if (rc != null) {
          observer.addMsg("RecordControl: OK");
        } else {
          observer.addMsg("No RecordControl, exiting");
          return;
        }

        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        rc.setRecordStream(bos);
        rc.startRecord();
        player.prefetch();
        player.start();

        long start = System.currentTimeMillis();
        while (!isStopped() && ((System.currentTimeMillis()-start) < getMaxDuration())) { 
          calcProgress(start);
          observer.announceProgress(progress());
        }
        rc.commit();
        player.close();
        audioData = bos.toByteArray();
        observer.announceProgress(100);
      }
      done = true;
    } catch (Exception ex) {
      observer.addMsg("Recorder exception: "+ex.getMessage());
      // The following is necessary to communicate back to ControllerThread
      // that an error condition was encountered
      throw new TaskException("Recorder exception: "+ex.getMessage());
    }
  }
  // Other utility methods follow....
}

Then, an instance of this class is passed, at construction, to a newly created instance of ControllerThread:

public class Multimedia implements Observer {

// Observer interface implementation omitted here
// ref. source code & Part II of this article

public void startRecording() {
  task = new RecordTask();
  thread = new ControllerThread(task, 1000L, 10000L, this);
  new Thread(thread).start();
}

// Other stuff goes here...
}

And that's pretty much about it: with the code above, we have created and started a new task which we'll check every second (1000L msec) and will allow at most 10 sec (10000L msec) to complete.
All the magic is done in ControllerThread - explained in part 2 of this series.