Java Fundamentals Tutorial: Threading

public class ThreadDemo {
  public static void main(String[] args) throws InterruptedException {
    int numThreads = Integer.parseInt(args[0]);
    int count = Integer.parseInt(args[1]);
    Thread[] threads = new Thread[numThreads];
    System.out.println("Creating threads");
    for (int i = 0; i < threads.length; i++) {
      threads[i] = new Thread(new Runner("Runner " + i, count));
    }
    System.out.println("Starting threads");
    for (int i = 0; i < threads.length; i++) { threads[i].start(); }
    System.out.println("Waiting for threads");
    for (int i = 0; i < threads.length; i++) { threads[i].join(); }
    System.out.println("Done");
  }
}
public class Runner implements Runnable {
  private final String name;
  private final int count;
  public Runner(String name, int count) {
    this.name = name; this.count = count;
  }
  public void run() {
    for (int i = 0; i < count; i++) {
      System.out.println(name + "=" + i);
      Thread.yield();
    }
  }
}

public class Counter {
  private int count = 0;
  public void setCount(int count) { this.count = count; }
  public int getCount() { return this.count; }
}
public class Runner implements Runnable {
  private final int maxCount;
  private final Counter counter;
  public Runner(int maxCount, Counter counter) {
    this.maxCount = maxCount; this.counter = counter;
  }
  public void run() {
    for (int i = 0; i < this.maxCount; i++) {
      int currentCount;
      synchronized(this.counter) {
        currentCount = this.counter.getCount();
        currentCount++;
        this.counter.setCount(currentCount);
      }
      System.out.println(Thread.currentThread().getName()
        + " at count " + currentCount);
    }
  }
}

A better approach would be to add a synchronized count method on the Counter class so that Runners would not have to worry about synchronization.

public synchronized int count() {
  return ++this.count;
}

public class Queue {
  private static class Element {
    final Object value;
    Element next;
    Element(Object value) {this.value = value;}
  }
  private Element first, last;
  private int curSize, maxSize;

  public Queue(int maxSize) { this.maxSize = maxSize; }

  public synchronized void put(Object o) throws InterruptedException {
    while (this.curSize == this.maxSize) { this.wait(); }
    if (this.first == null) {
      this.first = (this.last = new Element(o));
    } else {
      this.last = (this.last.next = new Element(o));
    }
    this.curSize++;
    this.notifyAll();
  }
  public synchronized Object get() throws InterruptedException {
    while (this.curSize == 0) { this.wait(); }
    Object o = this.first.value;
    this.first = this.first.next;
    this.curSize--;
    this.notifyAll();
    return o;
  }
}

public class NetworkedService {
  public void service(int port) throws Exception {
    ServerSocket ss = new ServerSocket(port);
    while(true) {
      new Thread(new Handler(ss.accept())).start(); //
    }
  }
  private static class Handler implements Runnable {
    private final Socket s;
    public Handler(Socket s) { this.s = s; }
    public void run() {
      //
    }
  }
}

	Note
	With no access to a standard thread-pool library, we end up creating Threads on demand. Works well for very simple cases, but does not scale and it is not resource-efficient.

public class NetworkedService {
  public void service(int port) throws Exception {
    ServerSocket ss = new ServerSocket(port);
    ExecutorService pool = Executors.newCachedThreadPool(); //

    while(true) {
      pool.execute(new Handler(ss.accept())); //
    }
  }
  private static class Handler implements Runnable {
    //
  }
}

public interface ExecutorService extends Executor {
  /** Stop accepting new tasks */
  public void shutdown();
  /** Stop accepting new tasks, cancel tasks
      currently waiting, and interrupt active tasks */
  public List<Runnable> shutdownNow();
  public boolean isShutdown();
  public boolean isTerminated();
  public boolean awaitTermination(long timeout, TimeUnit unit);
  public void execute(Runnable c);
  public <T> Future<T> submit(Callable<T> task);
  ...
}

public class Executors {
  public static ExecutorService newSingleThreadedExecutor();
  public static ExecutorService newFixedThreadPool(int n);
  public static ExecutorService newCachedThreadPool();
  ...
}

public class Handler implements Runnable {
  private Exception e = null;
  private Long result = null;
  public synchronized void run() {
    long t = System.currentTimeMillis();
    try {
      // handle request
      this.result = System.currentTimeMillis() - t;
    } catch (Exception e) {
      this.e = e;
    } finally {
      this.notifyAll();
    }
  }
  public synchronized long getResult() throws Exception {
    while (result == null && e == null) {
      this.wait();
    }
    if (e != null) {
      throw e;
    } else {
      return result;
    }
  }
}

Our Handler becomes much simpler with a Callable:

public class Handler implements Callable<Long> {
  public Long call() throws Exception {
    long t = System.currentTimeMillis(); //
    return System.currentTimeMillis() - t;
  }
}

Futures make it easy to define asynchronous operations:

public interface Future<V> {
  public V get() throws InterruptedExeption,ExecutionException;
  public V get(long timeout, TimeUnit unit);
  public boolean cancel(boolean mayInterrupt);
  public boolean isCanceled();
  public boolean isDone();
}

And ExecutorService.submit(Callable<T> task) conveniently returns Future<T> for that task. We just have to call Future.get() to wait on the result.

public class BlockingMap<K,V> {
  private final Map<K,V> map = new HashMap<K,V>();

  // only one get() a time - poor read performance
  public synchronized V get(K key) {
    return map.get(key);
  }
  public synchronized V require(K key) {
    V value = null;
    while((value = map.get(key)) == null) {
      try {
        this.wait();
      }
      catch (InterruptedException e) {
        break;
      }
    }
    return value;
  }
  public synchronized void set(K key, V value) {
    map.put(key, value);
    this.notifyAll();
  }
}

public class BlockingMap<K,V> {
  private Map<K,V> map = ...
  private ReadWriteLock lock = new ReentrantReadWriteLock();
  private Condition vAdded = lock.writeLock().newCondition();
  public V get(K key) {
    lock.readLock().lock();
    try {
      return map.get(key);
    } finally {
      lock.readLock().unlock();
    }
  }
  public V require(K key) {
    V value = get(key); // try cheap first
    if (value == null) {
      lock.writeLock().lock();
      try {
        while((value = map.get(key)) == null) {
          vAdded.await();
        }
      } catch (InterruptedException giveup) {
        ...
      } finally {
        lock.writeLock().unlock();
      }
    }
    return value;
  }
  public void set(String key, String value) {
    lock.writeLock().lock();
    try {
      map.put(key, value);
      vAdded.signalAll();
    } finally {
      lock.writeLock().unlock();
    }
  }
}

if (lock.tryLock()) {
  try {
    // do work that requires a lock
  } finally {
    lock.unlock();
  }
} else {
  // the lock was not available
}

try {
  lock.lockInterruptibly();
  try {
    // do work that requires a lock
  } finally {
    lock.unlock();
  }
} catch (InterruptedException e) {
  // interrupted while waiting
}

try {
  if (lock.tryLock(10, TimeUnit.SECONDS)) {
    try {
      // do work that requires a lock
    } finally {
      lock.unlock();
    }
  }
} catch (InterruptedException e) {
  // interrupted while waiting
}

public class AtomicInteger extends Number {
  public int incrementAndGet() {...}
  public int decrementAndGet() {...}
  public int getAndIncrement() {...}
  public int getAndDecrement() {...}
  ...
}
public class AtomicLong extends Number {
  public boolean compareAndSet(long expected,
    long update) {...}
  public long addAndGet(long delta) {...}
  ...
}
public class AtomicBoolean {
  public boolean getAndSet(boolean newValue) {...}
  ...
}