Best way to synchronize asynchronous task

Question

Hi there,

My Problem: I've got 802.15.4 Wireless network connected to a serial port (using a wrapper). I can send packages into the network and listen for incoming packages. As you can imagine this is highly asynchronous.

Here comes the task: I want to send commands to the network and wait for the response in one function call. For Example: I want to get the temperature from the Node with the network ID 1338.

double getTemperature(int id) throws Exception { .... }

Is there a better way of waiting for a response message other than doing all this "synchonized(object) wait(..) notify(..)" stuff?

Best regards, bigbohne

Maybe to add some spice:

This should all end in a webinterface where user can request these commands (either through ajax or directly). I've also thought of caching responce values in a database. But for some commands you MUSS have a direct answer of success/failure

Answer 1

The better way is to wrap it up in some reusable request/response class. This way you can have a standard way to query the serial port and wait for a response. This class, however, will undoubtedly have to use the synchronized keyword and wait and notify commands somewhere in its implementation. This is a key part of writing Java and it's important to understand how they work.

If you were to design a function like you say, you have to make sure that you call it from a thread which can block waiting for the result. The function will have to implement some sort of a wait loop inside of it waiting for the response from the serial port. So it's important that whatever thread it's running on be able to wait as well.

This does bring up one additional point, too. If you have multiple threads doing this, you will need to have some class which handles the serial port communication that can deal with multiple requests coming in from multiple threads and queue them as necessary to handle the asynchronous nature of the serial port and the specific device to which you are connecting. For example, it may not be appropriate to send a second command before the response from the first is read completely.

There are several complicated issues here, and it's important to have a good design in place which addresses them all in a reasonable way.

Answer 2

I'm not sure I understand the question correcly, but I usually use a Future<T> for these sorts of tasks.

Internally the Future<T> implementation uses wait and notify and all of that, but the interface itself becomes rather clean.

Future<Double> getTemperature(int id);

In your communication code you then can map incoming messages towards unfulfilled futures. If ordering is guaranteed you could do something like

class Something {
    Map<Integer, Queue<Object>> requests;

    synchronized Future<?> request(int id, Object data) {
        MyFutureImpl future = new MyFuture();
        requests.get(id).add(future);
        serializeAndSend(id, data);
        return future;
    }

    void serializeAndSend(id, data) {...}

    synchronized void response(int id, Object data) {
        MyFutureImpl future = requests.get(id).remove();
        future.setValue(data); // This would fulfill the future, and any
                               // threads waiting in a get() will get the
                               // value and continue.
    }
}

Where MyFutureImpl is a very basic future-implementation. I'm assuming there's a communications thread that calls response() when a packet is received. I'm also assuming that the serializeAndSend()-function handles the writing to the client or blocks until the write operation can be made or handed off to a communication thread.

The use of concurrency-capable map and queue structures could make some synchronizations unnecessary. If there's only ever one outstanding call per id, the Queue becomes unnecessary of course.

Answer 3

You can do this with a BlockingQueue so you will not have to manually manage any of the synchronization. The request can send the message, then call take() on a BlockingQueue which will wait for an element to be present. The element is the reply which is inserted into the queue by whatever listener you have on the port when the reply is returned.

You just have to coordinate so that the listener and requester share the same queue. You only need a queue of size one for each request to support this scenario.

// Send method
BlockingQueue<Reply> q = new ArrayBlockingQueue<Reply>(1);
serialCom.registerQueue(myRequest.getId());
serialCom.sendRequest(myRequest);
return q.take();

//Listener
BlockingQueue<Reply> q = queueMap.get(incomingMessage.getId());
q.add(incomingMessage.getReply());