Concurrency#

Todo:#

[ ] lost update
[ ] happens before

Race Condition#

https://web.mit.edu/6.031/www/sp21/classes/20-concurrency/#race_condition

A race condition means that the correctness of the program depends on the relative timing of events in concurrent computations A and B. When this happens we say “A is in race with B.”

check-then-act: a potentially stale observation is used to decide what to do next.
read-modify-write: new state is derived from previous state (may be stale). e.g., count++.

Synchronizers#

`CountDownLatch`#

A synchronization aid that allows one or more threads (by calling await()) to wait until a set of operations being performed in other threads completes.

Initialized with a given count.
The await method blocks until the current count is reached to zero due to invocations of the countDown().
If count is already zero invocation to await returns immediately.
The count cannot be reset.
Threads calling countDown need not wait for the count.

Use cases:

A CountDownLatch with a count of one serves as a simple on/off latch, or gate: all thread invoking await wait at the gate until it is opened by a thread invoking countDown().
A CountDownLatch initialized to N can be used to make one thread wait until N threads have completed some action, or some action has been completed N times.

class Driver {
    private static final int N = 10;

    public static void main() throws InterruptedException {
        CountDownLatch startSignal = new CountDownLatch(1);
        CountDownLatch doneSignal = new CountDownLatch(N);

        for (int i = 0; i < N; i++>) {
            new Thread(new Workder(startSignal, doneSignal)).start();
        }

        doSomeWorkBeforeWorker();
        startSignal.countDown();
        doSomeWork();
        doneSignal.await();
    }
}

class Worker implements Runnable {
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;

    Worker(CountDownLatch startSignal, CountDownLatch doneSignal) {
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
    }

    public void run() {
        try {
            startSignal.await();
            doWork();
            doneSignal.countDown();
        } catch (InterruptedException e) {
            System.out.println("I got interrupted :-()");
            return;
        }
    }

    void doWork() {
        System.out.println("Doing work...");
    }
}

Semaphore#

A counting semaphore maintains a set of permits that can be acquired and released. If a permit is unavailable, the acquire operation will block the thread.”

Liveness#

Ability to make progress.

Deadlock
Livelock
Starvation

Deadlock#

Deadlock is a situation where two or more threads are blocked forever, waiting for each other.

Livelock#

A thread often acts in response to the action of another thread. If the other thread’s action is also a response to the action of another thread, then livelock may result.
As with deadlock, livelocked threads are unable to make further progress. However, the threads are not blocked they are simply too busy responding to each other to resume work.

https://docs.oracle.com/javase/tutorial/essential/concurrency/newlocks.html

BlockingQueue#

A Queue that additionally supports operations that wait for the queue to become non-empty when retrieving an element, and wait for space to become available in the queue when storing an element.

	Throws exception	Special value	Blocks	Times out
Insert	`add(e)`	`offer(e)`	`put(e)`	`offer(e, time, unit)`
Remove	`remove()`	`poll()`	`take()`	`poll(time, unit)`
Examine	`element()`	`peek()`	not applicable	not applicable

Does not accept null elements

Producer Consumer#

public class BlockingQueueUsage {

  public static void main(String[] args) {
    BlockingQueue<Integer> q = new ArrayBlockingQueue<>(5);

    new Thread(new Producer(q)).start();
    new Thread(new Consumer(q)).start();
    new Thread(new Consumer(q)).start();
  }
}

class Producer implements Runnable {

  private final BlockingQueue<Integer> q;

  Producer(BlockingQueue<Integer> q) {
    this.q = q;
  }

  @Override
  public void run() {
    try {
      while (true) {
        int value = ThreadLocalRandom.current().nextInt();
        System.out.println("Produced value: " + value);
        Thread.sleep(1000);
        q.put(value);
      }
    } catch (InterruptedException ignore) {}
  }
}

class Consumer implements Runnable {

  private final BlockingQueue<Integer> q;

  Consumer(BlockingQueue<Integer> q) {
    this.q = q;
  }

  @Override
  public void run() {
    try {
      while (true) {
        int value = q.take();
        System.out.println(
          Thread.currentThread().getName() + " Consumed value: " + value
        );
      }
    } catch (InterruptedException ignore) {}
  }
}

ThreadLocal#

This class provides thread-local variables, which are distinct from regular variables in that each thread accessing one via its get or set method has a separate, independently initialized copy.
ThreadLocal instances are commonly private static fields in classes that aim to associate state with a thread, such as a user ID or transaction ID.
Each thread maintains an implicit reference to its thread-local variable copy as long as the thread is alive and the ThreadLocal instance is accessible. Once the thread ceases, all its thread-local variable copies become eligible for garbage collection, unless there are other references to these copies.

Use cases:

Thread confinement
Per thread object for performance
Per thread context

public class ThreadLocalUsage {

  public static void main(String[] args) throws InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(4);

    IntStream
      .range(1, 10)
      .forEach(i ->
        executor.execute(() -> {
          System.out.println(
            Thread.currentThread().getName() + " Id: " + ThreadId.get()
          );
        })
      );

    executor.shutdown();
    executor.awaitTermination(5, TimeUnit.SECONDS);
  }
}

class ThreadId {

  private static final AtomicInteger id = new AtomicInteger(0);
  private static final ThreadLocal<Integer> threadId = ThreadLocal.withInitial(() ->
    id.incrementAndGet()
  );

  public static int get() {
    return threadId.get();
  }
}

Tips:

Cleanup once not required anymore
Delegate to frameworks e.g., Spring context holders