Миграция java-приложения на Fork/Join или о чём нужно помнить

С выходом седьмой версии JDK нам, счастливым разработчикам на Java, стал доступен из коробки фреймворк Fork/Join, о котором уже писали на хабре тут. Фреймворк в плане API очень похож на уже привычный ExecutorServices, но даёт весьма ощутимый прирост производительности и действительную «легковесность» потоков.

Здесь, я бы хотел рассмотреть на что стоит обратить внимание при переходе на Fork/Join.

ThreadLocal переменные

C ExecutorService у нас была гарантия, что одна задача от начала и до конца выполняется одним потоком.
В Fork/Join работа с потоками претерпела сильные изменения. Задачи (ForkJoinTask’s) имеют уже другую семантику нежели потоки: один поток может выполнять несколько задач пересекающихся по времени.

Например, при вызове task.invoke(), вполне возможен сценарий, когда исходная задача выполнялась одним потоком, затем тот же поток начал выполнять новую задачу task. Это быстрее: не нужно стартовать ещё один поток и мы избегаем переключения контекста. После окончания task исходная задача продолжила своё выполнение.

Таким образом следует пересмотреть подход к использованию локальных переменных потока.
ThreadLocal могут использоваться в нескольких случаях:

1. Для хранение каких-либо непотокобезопасных утилитных классов. Например, SimpleDateFormat. Создание которого весьма дорого, а использование несколькими потоками чревато некорректной работой и исключениями.
2. Для хранения какого-либо контекста выполнения потока. Например, текущая транзакция, соединение с базой данных и т. д. Или данных, которые просто решили передавать не через сигнатуру методов или setter’ы, а через локальный для потока контекст. Например, ActionContext в Struts2.

Если в первом случае при переходе на Fork/Join ничего страшного не произойдёт, то во втором, локальные переменные одной задачи станут доступны другой.

Мораль: не использовать ThreadLocal переменные в этом случае, или реализовать свой аналог ThreadLocal, поддерживающий ForkJoinTask.

Блокировки

В целом, фреймворк не накладывает ограничений на использование других средств блокировки и синхронизации. Более того, помогает избежать ситуаций зависания потоков в ожидании других потоков (thread starvation).

Например, у нас есть ThreadPoolExecutor, с ограниченным сверху размером пула. Допустим это один, для простоты. Мы запускаем один поток, который в свою очередь добавляет в очередь ещё один поток и ждёт его завершения. В таком случае, мы никогда не дождёмся обоих потоков. Если пул больше, то можно рассмотреть случай, когда остальные потоки ожидают друг друга и находятся в deadlock’e. Или ещё проще, одна задача породила вторую, та третью и т. д., и все ждут результатов выполнения запущенных задач.

Часть проблемы решается тем, что join(), по сути, возвращает поток в пул.

Для обеспечения необходимого уровня параллелизма в ForkJoinPool’е предусмотрен механизм контроллируемой блокировки. С помощью класса ForkJoinPool.ManagedBlocker мы можем сказать ForkJoinPool’y, что поток может заблокироваться в ожидании лока и ForkJoinPool создаст дополнительный поток для обеспечения заданного уровня параллельности.

Допустим, мы хотим использовать ReentrantLock в своём коде. Нам нужно реализовать интерфейс ForkJoinPool.ManagedBlocker следующим образом (взято из javadoc'ов):

class ManagedLocker implements ManagedBlocker {
   final ReentrantLock lock;
   boolean hasLock = false;
   ManagedLocker(ReentrantLock lock) { 
      this.lock = lock; 
   }
   public boolean block() {
     if (!hasLock)
       lock.lock();
     return true;
   }
   public boolean isReleasable() {
     return hasLock || (hasLock = lock.tryLock());
   }
 }

и использовать лок следующим образом в коде:

ReentrantLock lock = new ReentrantLock();

//Somewhere in thread
try{
   ForkJoinPool.managedBlock(new ManagedLocker(lock));
   //Guarded code goes here
}finally{
   lock.unlock();
}

Всё.

И да прибудет с вами сила!