In my application, I have a message dispatcher. Each message gets relayed to a dispatcher thread.
In some scenarios, I can get two responses from the third party in the same millisecond:
- An "I received your message"
ACKnowledgement - "My response to your message"
RESPONSE
When I get both messages at the same time, this can create a race condition. It is important to have my EventHandler handle the ACK before I get the RESPONSE, but I cannot guarantee ordering from the Asynchronous networking IO library; I must do it myself.
Below is a working MVCE of this scenario. Most of it is just framework, it's not how it works in the real code, but it's enough to get the bit I'm asking about - the ordering logic - to show it's value. The Proposed Solution to this race condition is the ensureAck method, which is called in the handleResponse method. If you comment out that call, and run the application repeatedly, you will see that there is no guarantee of method ordering. Additionally, and this is the difficult bit: I don't want the RESPONSE to be ignored if, for some reason, the ACK never shows up (a dropped network packet or something) - I want the RESPONSE to get handled, reluctantly, eventually.
Aside: the bit about NOTHING messages are just to warm up the ExecutorService so you can actually see the race condition.
How is my solution?
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class RaceSolution {
private static enum State {
NEW, READY;
}
private static enum Type {
ACK, RESPONSE, NOTHING;
}
private static class EventHandler {
private volatile State state = State.NEW;
private final Object lock = new Object();
public void handleEvent(Event event) {
switch(event.t) {
case ACK:
handleAck(event);
break;
case RESPONSE:
handleResponse(event);
break;
default:
}
}
private void handleResponse(Event event) {
// COMMENT OUT THE NEXT LINE TO SEE THE RACE CONDITION
ensureAck();
synchronized(lock) {
System.out.println("Received response event: " + state);
}
}
private void handleAck(Event event) {
synchronized(lock) {
System.out.println("Received ack Event: " + state);
state = State.READY;
}
}
/* THIS IS THE IMPORTANT BIT */
private void ensureAck() {
int tries = 0;
try {
while(state == State.NEW && tries < 10) {
Thread.sleep(1);
tries++;
}
} catch (InterruptedException e) {
}
}
}
private static class Event {
Type t;
public Event(Type t) {
this.t = t;
}
}
public static void main(String... args) throws InterruptedException {
RaceSolution solution = new RaceSolution();
Event nothing = new Event(Type.NOTHING);
Event ack = new Event(Type.ACK);
Event response = new Event(Type.RESPONSE);
for(int i = 0; i < 10; i++) {
solution.submit(nothing);
}
Thread.sleep(100);
solution.submit(ack);
solution.submit(response);
solution.shutdown();
}
private final ExecutorService dispatcher = Executors.newCachedThreadPool();
private final EventHandler handler = new EventHandler();
public void shutdown() {
if (!dispatcher.isShutdown()) {
dispatcher.shutdownNow();
}
}
public void submit(final Event e) {
dispatcher.execute(new Runnable() {
@Override
public void run() {
handler.handleEvent(e);
}
});
}
}
