7
\$\begingroup\$

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);
      }
    }); 
  }
}
\$\endgroup\$
7
\$\begingroup\$

Concurrency Strategy

In general, using volatile is complicated, partially because the meaning of volatile changed in Java 1.4, and also because it is hard to spot. I recommend against using it at all. Instead, you should use a more visible concurrency mechanism like synchronization, java.util.concurrent.* classes, java.util.concurrent.locks.*, or java.util.concurrent.atomics.*

So, while I recommend against volatile, what's a real problem is using multiple different locking schemes in the same code, and you use both volatile and synchronized.

The use of synchronization as your concurrency strategy should be fine all on its own in this case.

General

  • The private final Object lock = new Object() is great. Good to be final, and I prefer that to locking on the instance itself (like synchronized(this)). Makes it more clear.

  • Println in a synchronized block is almost never a good idea:

    synchronized(lock) {
      System.out.println("Received ack Event: " + state);
      state = State.READY;
    }
    

    A better solution would be:

    State copy = null;
    synchronized(lock) {
      copy = state;
      state = State.READY;
    }
    System.out.println("Received ack Event: " + copy);
    
  • Timed sleeps while waiting for a lock are a horrible solution:

    Thread.sleep(1);
    

Better algorithm

The right way to solve this problem, though is to used timed-waits on the lock:

private void handleAck(Event event) {
    State copy = null;
    synchronized(lock) {
      copy = state;
      state = State.READY;
      // tell any waiting threads the state has changed.
      lock.notifyAll();
    }
    System.out.println("Received ack Event: " + copy);
}

/* THIS IS THE IMPORTANT BIT */
private void ensureAck() {
  // how long to wait... 10 milliseconds
  try {
      synchronized(lock) {
          if (state == State.NEW) {
              long waited = 0; // milliseconds
              long start = System.currentTimeMillis();
              while (state == State.NEW && waited < 10) {
                  lock.wait(10 - waited);
                  waited = System.currentTimeMillis() - start;
              }
          }
      }
  } catch (InterruptedException e) {
      // do more than nothing.....
  }
}

The above strategy does not need anything to be volatile, and it does not 'poll' the state, it returns immediately when the state changes....

\$\endgroup\$
4
  • \$\begingroup\$ Great answer! One question: why doesn't state still need to be volatile here? \$\endgroup\$
    – durron597
    Oct 16 '14 at 18:16
  • 2
    \$\begingroup\$ The memory model used by Java ensures that all variables used ina a synchronized block are always synchronized with their most recently active state. It is as if all variables in a synchronized block are volatile (as long as all other places those variables are used are also in a synchronized block). \$\endgroup\$
    – rolfl
    Oct 16 '14 at 18:19
  • 2
    \$\begingroup\$ System.currentTimeMillis() should be replaced with System.nanoTime(). currentTimeMillis() can jump forward or backwards if the hardware clock is changed, while nanoTime() cannot. \$\endgroup\$
    – Dan Getz
    Nov 29 '14 at 0:13
  • \$\begingroup\$ Interesting observation, @DanGetz - thanks. You're right, it can, and I had never considered the implications of that. \$\endgroup\$
    – rolfl
    Nov 29 '14 at 0:14
1
\$\begingroup\$

I think you are solving the problem backwards.

Consider the following code, run in a single thread....

Event ack = new Event(Type.ACK);
Event response = new Event(Type.RESPONSE);

EventHandler handler = new EventHandler();

// Wrong Order!
handler.handleEvent(response);
handler.handleEvent(ack);

In your proposed solution, the expected behavior would be to sleep for about 10 seconds, then process the two events in wrong order.

Which means that you are taking broken code, and trying to "fix" it by running it in multiple threads. That's the wrong hammer. Instead, you should implementing your protocol more explicitly.

To wit: you've got three states, not merely two. State.NEW and State.READY you've already recognized, but you've elided State.OUT_OF_ORDER. handleResponse checks to see the current state; if READY, then you can process the response at once. If NEW, then you cache the response and flip the state to OUT_OF_ORDER. Similarly, handleAck checks the state, processing the ACK first, then looking at the state to know if there is a cached response to process as well.

Depending on the protocol, it may be worth while to add State.DONE as well, or to explicitly reset to State.NEW after response processing.

There's a temptation to use the response cache as your means of tracking the out of order state. Resist it - first, because the states and state transitions document the protocol you are implementing; second, because it is useful to be able to query the state of the handler without blocking it. For instance, it's common to implement monitor threads that watch your state to report error conditions, and if you have a single memory location to watch, you don't have to interrupt processing to get an accurate understanding of the current state.

Stateless4j is a pretty good library for documenting (and implementing) your stateful protocol.

Once you've got the protocol implemented, you can write single threaded unit tests to ensure that all of the orderings of events are properly handled. You won't need Thread.sleep() any longer, and the synchronization mechanisms don't interfere with correctness in a single threaded test.

Having done that, you can call the event handler from multiple threads, and everything works. But it's not clear that you should. The experts sharing their experiences from the low latency space teach that the single writer principle makes reasoning about the state of a program a lot easier to reason about.

Using that approach, you would have a single thread reading a queue of events, and passing them in a fixed order to the event handler. The queue would be shared by this thread, and by RaceSolution.submit() which would write new events into the queue rather than dispatching new Runnables for each event.

If you want more fun than a queue, you can go whole hog by using the Disruptor pattern.

\$\endgroup\$
1
  • \$\begingroup\$ Some great insight here. I'll keep it in mind as I continue along / refactor. \$\endgroup\$
    – durron597
    Oct 17 '14 at 13:25

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.