How to make a thread wait until a variable reaches a specific value (Multi-threaded Java)

Question

I have a server program which accepts client connections. These client connections can belong to many streams. For example two or more clients can belong to the same stream. Out of these streams one message I have to pass but I have to wait until all the streams are established. For this I maintain the following data structure.

ConcurrentHashMap<Integer, AtomicLong> conhasmap = new ConcurrentHashMap<Integer, AtomicLong>();

Integer is the stream ID and Long is the client number. To make one thread for a given stream to wait till AtomicLong reach a specific value I used the following loop. Actually the first packet of the stream puts it stream ID and the number of connections to wait. With each connection I decrease the connections to wait.

while(conhasmap.get(conectionID) != new AtomicLong(0)){
       // Do nothing
}

However this loop blocks the other threads. According to this answer it does a volatile read. How can I modify the code to wait the correct thread for a given stream until it reaches a specific value?

Answer 1

If you're using Java 8, CompletableFuture could be a good fit. Here's a complete, contrived example which is waiting for 5 clients to connect and send a message to a server (simulated using a BlockingQueue with offer/poll).

In this example, when the expected client connected message count is reached, a CompletableFuture hook is completed, which then runs arbitrary code on any thread of your choice.

In this example, you don't have any complex thread wait/notify setups or busy wait loops.

package so.thread.state;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

public class Main {

  public static String CONNETED_MSG = "CONNETED";
  public static Long EXPECTED_CONN_COUNT = 5L;

  public static ExecutorService executor = Executors.newCachedThreadPool();
  public static BlockingQueue<String> queue = new LinkedBlockingQueue<>();

  public static AtomicBoolean done = new AtomicBoolean(false);

  public static void main(String[] args) throws Exception {

    // add a "server" thread
    executor.submit(() -> server());

    // add 5 "client" threads
    for (int i = 0; i < EXPECTED_CONN_COUNT; i++) {
      executor.submit(() -> client());
    }

    // clean shut down
    Thread.sleep(TimeUnit.SECONDS.toMillis(1));
    done.set(true);
    Thread.sleep(TimeUnit.SECONDS.toMillis(1));
    executor.shutdown();
    executor.awaitTermination(1, TimeUnit.SECONDS);

  }

  public static void server() {

    print("Server started up");
    // track # of client connections established
    AtomicLong connectionCount = new AtomicLong(0L);

    // at startup, create my "hook"
    CompletableFuture<Long> hook = new CompletableFuture<>();
    hook.thenAcceptAsync(Main::allClientsConnected, executor);

    // consume messages
    while (!done.get()) {
      try {
        String msg = queue.poll(5, TimeUnit.MILLISECONDS);
        if (null != msg) {
          print("Server received client message");
          if (CONNETED_MSG.equals(msg)) {
            long count = connectionCount.incrementAndGet();

            if (count >= EXPECTED_CONN_COUNT) {
              hook.complete(count);
            }
          }
        }

      } catch (Exception e) {
        e.printStackTrace();
      }
    }

    print("Server shut down");

  }

  public static void client() {
    queue.offer(CONNETED_MSG);
    print("Client sent message");
  }

  public static void allClientsConnected(Long count) {
    print("All clients connected, count: " + count);
  }


  public static void print(String msg) {
    System.out.println(String.format("[%s] %s", Thread.currentThread().getName(), msg));
  }
}

You get output like this

[pool-1-thread-1] Server started up
[pool-1-thread-5] Client sent message
[pool-1-thread-3] Client sent message
[pool-1-thread-2] Client sent message
[pool-1-thread-6] Client sent message
[pool-1-thread-4] Client sent message
[pool-1-thread-1] Server received client message
[pool-1-thread-1] Server received client message
[pool-1-thread-1] Server received client message
[pool-1-thread-1] Server received client message
[pool-1-thread-1] Server received client message
[pool-1-thread-4] All clients connected, count: 5
[pool-1-thread-1] Server shut down

Answer 2

Your expression:

conhasmap.get(conectionID) != new AtomicLong(0)

will always be true because you are comparing the object references, which will never be equal, instead of the values. The better expression would be:

conhasmap.get(conectionID).longValue() != 0L)

, but looping like this without wait/notify logic within the loop is not a good practice because it uses CPU time constantly. Instead, each thread should call .wait() on the AtomicLong instance, and when it is decremented or incremented, you should call .notifyAll() on the AtomicLong instance to wake up each waiting thread to check the expression. The AtomicLong class may already be calling the notifyAll() method whenever it is modified, but I don't know for sure.

AtomicLong al = conhasmap.get(conectionID);
synchronized(al) {
    while(al.longValue() != 0L) {
        al.wait(100); //wait up to 100 millis to be notified
    }
}

In the code that increments/decrements, it will look like:

AtomicLong al = conhasmap.get(conectionID);
synchronized(al) {
    if(al.decrementAndGet() == 0L) {
        al.notifyAll();
    }
}

I personally would not use an AtomicLong for this counter because you are not benefiting from the lock-less behavior of the AtomicLong. Just use a java.lang.Long instead because you will need to synchronize on the counter object for the wait()/notify() logic anyway.