Core Java. Lecture #11

Part 3. Thread-safe data structures

Non-blocking algorithms

Atomics

Atomic operations in atomic variable classes

getAndSet(newValue)    compareAndSet(expect, update)

incrementAndGet()      decrementAndGet()

getAndIncrement()      getAndDecrement()

getAndAdd(delta)       addAndGet(delta)

getAndUpdate(updateFunction)
updateAndGet(updateFunction)

getAndAccumulate(x, accumulatorBiFunction)
accumulateAndGet(x, accumulatorBiFunction)

Thread-safe collections

CopyOnWriteArrayList and CopyOnWriteArraySet

class CopyOnWriteArrayList<E> implements List<E>
class CopyOnWriteArraySet<E> extends AbstractSet<E>

ConcurrentLinkedQueue/Deque

ConcurrentLinkedQueue<E> extends AbstractQueue<E>
        implements Queue<E>
ConcurrentLinkedDeque<E> extends AbstractCollection<E>
        implements Deque<E>

ConcurrentLinkedQueue/Deque

Blocking queues: a means of implementing the producer-consumer pattern

ConcurrentHashMap

class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>

ConcurrentHashMap: Java 7-

ConcurrentHashMap: Java 8+

ConcurrentSkipListMap

class ConcurrentSkipListMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentNavigableMap<K,V>

ConcurrentSkipListMap: the principle

ConcurrentSkipListMap

ConcurrentSkipListMap

ConcurrentSkipListMap

ConcurrentSkipListMap

ConcurrentSkipListMap

ConcurrentSkipListMap

ConcurrentSkipListMap

ConcurrentSkipListMap

Part 4. Executor Framework

Executor Framework

Executor Framework

//Abstraction of a computational task returning a result
public interface Callable<V> {
  V call() throws Exception;
}</V>

//Abstraction of the "manager of threads"
public interface ExecutorService {
  <T> Future<T> submit(Callable<T> task);
  /*... there are many other methods, will discuss later...*/
}

Future

//Abstraction of the "in progress" result that you can wait for
//finalization or just cancel
public interface Future<V> {
  V get() throws InterruptedException, ExecutionException;
  V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException;
  boolean cancel(boolean mayInterruptIfRunning);
  boolean isDone();
  boolean isCancelled();
}

Task, Executor, Future

How do I create an ExecutorService?

public class Executors {
  //fixed pool size
  public static ExecutorService newFixedThreadPool(int nThreads)
  public static ExecutorService newSingleThreadExecutor()
  //pool grows as needed, keeps inactive thread for 60 seconds
  public static ExecutorService newCachedThreadPool()
  //allows you to perform tasks with a delay or periodically
  public static
    ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
  ...
}

Processing tasks in batches

//Run and wait for everything to finish
//List<Future<T>>, not List<T>, because exceptions are possible
<T>List <Future <T>> invokeAll(
    Collection <? extends Callable <T>> tasks)
        throws InterruptedException

//Run, return the first successful result
//cancel the rest
<T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

Cancellation of tasks and "termination of service"

//cancel a task if it has not already started
future.cancel(false)
//request task interruption (details ahead)
future.cancel(true)

//Prevent new tasks from being accepted
executorService.shutdown();
//Wait until the accepted tasks are completed
if (!service.awaitTermination(10, TimeUnit.SECONDS)){
    Abort tasks
    service.shutdownNow();
}

Interruption of tasks

Interrupting threads

Cooperative termination mechanism

What should we do with InterruptedException

} catch (InterruptedException e) {
    //record the fact of interruption to the log...
    ...
    //restore the interrupted status
    Thread.currentThread().interrupt();
    //exit the interrupted method
    return;
}

CompletableFuture

CompletableFuture<Integer> f = new CompletableFuture<>();
executor.execute(() -> {
  int n = workHard(arg);
  f.complete(n); });

executor.execute(() -> {
  int n = workSmart(arg);
  f.complete(n); });

executor.execute(() -> {
  Throwable t = ...;
  f.completeExceptionally(t); });

CompletableFuture composition with action

CompletableFuture composition with action

Composition of several CompletableFutures

Composition of several CompletableFutures

If that didn’t seem enough…​

We just scratched the surface

Brian Goetz et al., Java Concurrency in Practice. Released in 2006, in the era of Java 5.0. No new editions were published. It was released in Russian language in 2020! Still the most comprehensive guide to different aspects of multithreaded Java programming.

Code Review Checklist

Acknowledgements

Sometimes you feel the smell while reading the code…​

import org.json.JSONObject;
class NaivePersonSerializer {
  String toJSON(Person person) {
    JSONObject result = new JSONObject();
      result.put("firstName", person.getFirstName());
      result.put("lastName", person.getLastName());
      result.put("birthDate",
        person.getBirthDate()
        .format(DateTimeFormatter.ISO_LOCAL_DATE));
      result.put(...)
      result.put(...)
      //20 more fields
      result.put(...)
      return result.toString();
  }
}

There is a feeling that something is wrong…​

public class NaiveController {
  private final Service service;
  public void executeCommand(String command) {
    switch (command) {
      case "foo":
        service.foo(); break;
      case "bar":
        service.bar(); break;
      case ...
      //15 more branches...
      case "help":
      default:
        service.help(); break;
    }
  }
}

Coupling & Cohesion

Coupling vs. Cohesion

Questions

How would we like to solve this problem?

//Mark what we want to serialize, do not serialize by default.
//(Option: mark what we do not want to serialize, serialize by default)
public class Person {
    @Published
    private final String firstName;
    @Published
    private final String lastName;
    @Published
    private final LocalDate birthDate;
    //... as many other fields as you like
}

//As an option...
JsonSerializer<Person> ser = new JsonSerializer<>(Person.class);
JSONObject json = ser.serialize(p);

Controller

public class Controller {
  private final Service service;

  @Command("foo")
  void doSomething() { service.foo(); }

  @Command("bar")
  void bar() { service.bar(); }

  //15 more commands

  @Command() //default one
  void help() { service.help(); }
}

new CommandDispatcher(new Controller(srv)).executeCommand("foo");

Annotations

Annotation definition syntax

modifiers @interface AnnotationName {
  type elementName();
  ...
  type elementName default value;
}

Example

public @interface BugReport {
  String assignedTo() default "";
  int severity();
}

Annotation Interfaces

Valid return types of annotation methods (JLS 9.6.1)

Annotation Interface Definition Example

public @interface BugReport {
  //nested types are valid!
  enum Status {UNCONFIRMED, CONFIRMED, FIXED, NOTABUG}
  boolean showStopper() default false;
  String assignedTo() default "";

  //metaclass, to be explained later
  Class<?> testCase() default Void.class;
  Status status() default Status.UNCONFIRMED;

  //Annotation type. We use a special syntax to construct an instance ,
  //to be explained later
  Reference ref() default @Reference(id = "");
  String[] reportedBy();
}

Use of annotations

@AnnotationName(elementName1=value1, elementName2=value2,...)

@AnnotationName(singleValue)

@AnnotationName //no values, no parens

Use of annotations

@BugReport(assignedTo="Harry", severity=10)

//order doesn't matter
@BugReport(severity=10, assignedTo="Harry")

//if default values are set, you can skip the parameters
@BugReport(severity=10)

//if there are no parameters or all of them have default,
//then brackets can be omitted
@BugReport

//if there is no default value for the parameter, it must be defined
//(otherwise a compilation error).

Single-value annotations

public @interface ActionListenerFor {
  String value();
  int priority() default 0;
}

//use
@ActionListenerFor("yellowButton")

//which is equivalent to
@ActionListenerFor(value = "yellowButton")

//at the same time:
@ActionListenerFor(value = "yellowButton", priority = 1)

Arrays and other annotations

//Arrays are specified in curly braces
@BugReport(..., reportedBy = {"Harry", "Carl"})

//Same as {"Harry"}
@BugReport(..., reportedBy = "Harry")

//Constructing nested annotations
@BugReport(..., ref = @Reference(id = "12345"))

//... As you can see, using the allowed types,
//you can specify an arbitrarily complex data structure