Java ReadWriteLock и ReentrantReadWriteLock

ReentrantLock (или блок synchronized) даёт одному потоку эксклюзивный доступ — читатели и писатели делят один слот. Для рабочих нагрузок с преобладанием чтения это расточительно: если сотня потоков хочет читать значение и один хочет его записать, настоящего конфликта между читателями нет — конфликт только между читателями и писателем. Интерфейс ReadWriteLock и его стандартная реализация ReentrantReadWriteLock разделяют блокировку на две части — блокировка чтения, которую несколько потоков могут удерживать одновременно, и блокировка записи, которая является эксклюзивной для всех. При правильном использовании это значительно снижает конкуренцию. При неправильном — медленнее обычной блокировки.

Интерфейс

public interface ReadWriteLock {
  Lock readLock();
  Lock writeLock();
}

Два Lock-а, жёстко связанных своим родителем: блокировка чтения и блокировка записи подчиняются правилу — либо блокировка записи удерживается ровно одним потоком, либо блокировку чтения удерживают ноль или более потоков. Никогда обе одновременно, никогда по одной из каждой.

Стандартная реализация:

ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
Lock r = rw.readLock();
Lock w = rw.writeLock();

Обе блокировки имеют стандартный Lock API — lock, tryLock, lockInterruptibly, unlock. Контракт для пары rw:

Множество потоков могут удерживать r одновременно. Они не блокируют друг друга.
Только один поток может удерживать w в один момент времени.
Поток, пытающийся получить w, ждёт, пока все текущие читатели не освободят блокировку.
Потоки, пытающиеся получить r, ждут, если w удерживается или (в зависимости от политики) если писатель уже стоит в очереди.

Последний пункт — это политика предотвращения голодания писателей, рассмотренная ниже.

Пример использования: кэш

Классический пример. Кэш, который постоянно читается и изредка обновляется:

class ConfigCache {
  private final ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
  private final Lock r = rw.readLock();
  private final Lock w = rw.writeLock();
  private Map<String, String> data = new HashMap<>();

  public String get(String k) {
    r.lock();
    try {
      return data.get(k);                               // many readers, no contention
    } finally { r.unlock(); }
  }

  public void reload(Map<String, String> fresh) {
    w.lock();
    try {
      data = new HashMap<>(fresh);                       // exclusive: blocks readers and other writers
    } finally { w.unlock(); }
  }
}

При типичной нагрузке с 99% чтений это масштабируется намного лучше, чем единственный ReentrantLock. Читатели не блокируют друг друга; редкий писатель ненадолго останавливает всё, а затем все продолжают работу.

Та же дисциплина try/finally, что и для Lock, применяется здесь — каждый lock() должен быть сопряжён с unlock() в блоке finally. Блокировка чтения столь же нетерпима к утечкам, как и блокировка записи.

Справедливость и голодание писателей

ReentrantReadWriteLock поддерживает две политики:

new ReentrantReadWriteLock();          // non-fair (default)
new ReentrantReadWriteLock(true);      // fair (FIFO)

Политика по умолчанию (несправедливая) позволяет новым входящим читателям получить блокировку, даже если писатель уже ждёт — высокая пропускная способность, но писатели могут испытывать голодание при непрерывной нагрузке чтения. Справедливая политика ставит каждого запрашивающего в очередь FIFO: ожидающий писатель блокирует последующих читателей, и читатели ждут своей очереди.

По умолчанию всё же используется несправедливая политика. Если в продакшене вы наблюдаете писателей, вечно стоящих в очереди (одна из вещей, которую getQueueLength помогает обнаружить), переключитесь на справедливую.

Есть также более тонкая защита. Даже в несправедливом режиме, если писатель «следующий в очереди» (возглавляет очередь), входящие читатели блокируются. Это предотвращает наихудшую форму голодания; новые читатели всё равно могут пробиться вперёд, если писателей в очереди нет.

Понижение блокировки: запись → чтение

Полезный приём: можно удерживать блокировку записи, захватить также блокировку чтения, а затем освободить блокировку записи — не допуская при этом других писателей. Это называется понижением:

w.lock();
try {
  data = recompute();                                   // exclusive write
  r.lock();                                              // before releasing w
} finally { w.unlock(); }
// now holding only r — readers can join, but no writer can sneak in until I release r
try {
  process(data);                                         // read-only work, multiple threads can do it
} finally { r.unlock(); }

Смысл понижения: произвести фактическую мутацию под блокировкой записи, а затем продолжить чтение результата, не отстраняя других от данных. Промежуточное «захватить чтение, удерживая запись» работает, потому что блокировка это допускает — вы меняете резервирование блокировки записи с «эксклюзивного» на «разделяемое, с вашим персональным разрешением».

Обратное — повышение чтение → запись — не работает. Попытка захватить w, удерживая r, вызывает взаимоблокировку: блокировка записи ждёт, пока все читатели освободятся, а вы — один из них. Блокировка будет бесконечно ждать от вас самих.

r.lock();
try {
  if (needsRefresh()) {
    w.lock();                                            // DEADLOCK on the same thread
    ...
  }
} finally { r.unlock(); }

Для перехода чтение → запись нужно сначала освободить блокировку чтения, затем захватить блокировку записи и повторно проверить условие (кто-то другой мог обновить данные, пока вы были без блокировки).

Когда `ReadWriteLock` выигрывает у `ReentrantLock`

Грубое правило. ReentrantReadWriteLock выигрывает, когда:

Чтений значительно больше, чем записей (скажем, 100:1 или более).
Защищённая чтением секция нетривиальна — достаточно длинная, чтобы параллельное выполнение несколькими потоками давало ощутимый выигрыш.
Запись также достаточно длинная, чтобы кратковременная блокировка читателей была приемлема.

Он проигрывает (или выходит вничью), когда:

Чтения чрезвычайно короткие (один поиск в map). Накладные расходы на получение блокировки сопоставимы с самой работой; лучше использовать простой ReentrantLock или иммутабельный снимок через AtomicReference.
Соотношение читателей и писателей не является экстремальным.
Потоков много. Внутренний учёт, который ведёт блокировка чтения/записи для подсчёта читателей, становится всё дороже при масштабировании. Для очень read-heavy структур данных на многих ядрах StampedLock или copy-on-write обычно лучший выбор.

`StampedLock` — современная альтернатива

В Java 8 добавлен java.util.concurrent.locks.StampedLock с тремя режимами — запись, чтение и оптимистичное чтение. Оптимистичный режим позволяет читателю продолжать работу без захвата какой-либо блокировки; после чтения он проверяет, не изменилось ли значение, через штамп (stamp). Если изменилось, читатель откатывается к полноценному захвату блокировки чтения.

StampedLock sl = new StampedLock();
long stamp = sl.tryOptimisticRead();
String val = data.get(k);                                // read without locking
if (!sl.validate(stamp)) {                                // somebody wrote during our read
  stamp = sl.readLock();
  try {
    val = data.get(k);                                    // re-read under proper lock
  } finally { sl.unlockRead(stamp); }
}

Для рабочих нагрузок с преобладанием чтения StampedLock обычно быстрее ReentrantReadWriteLock. Цена: он не является реентерабельным, не поддерживает Condition, и API значительно проще использовать неправильно. Обращайтесь к нему, когда профайлер указывает на ReadWriteLock; по умолчанию используйте ReentrantReadWriteLock для удобства работы.

Рабочий пример: read-heavy кэш, три претендента

Программа ниже сравнивает три реализации одного и того же read-heavy кэша при 16 читателях и 2 писателях: synchronized map, map с защитой ReentrantLock и map с защитой ReentrantReadWriteLock.

java— editable, runs on the server

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

public class ReadWriteLockDemo {
  interface Cache {
    String get(String k);
    void put(String k, String v);
  }

static class SyncCache implements Cache {
    final Map<String, String> m = new HashMap<>();
    public synchronized String get(String k) { return m.get(k); }
    public synchronized void put(String k, String v) { m.put(k, v); }
  }

static class RLCache implements Cache {
    final Lock lock = new ReentrantLock();
    final Map<String, String> m = new HashMap<>();
    public String get(String k) { lock.lock(); try { return m.get(k); } finally { lock.unlock(); } }
    public void put(String k, String v) { lock.lock(); try { m.put(k, v); } finally { lock.unlock(); } }
  }

static class RWCache implements Cache {
    final ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    final Lock r = rw.readLock(), w = rw.writeLock();
    final Map<String, String> m = new HashMap<>();
    public String get(String k) { r.lock(); try { return m.get(k); } finally { r.unlock(); } }
    public void put(String k, String v) { w.lock(); try { m.put(k, v); } finally { w.unlock(); } }
  }

static long bench(Cache c, int readers, int writers, long durationMs) throws InterruptedException {
    AtomicBoolean stop = new AtomicBoolean(false);
    AtomicLong reads = new AtomicLong(), writes = new AtomicLong();

Thread[] rs = new Thread[readers], ws = new Thread[writers];
    for (int i = 0; i < readers; i++) {
      rs[i] = new Thread(() -> {
        long n = 0;
        while (!stop.get()) { c.get("k" + (n & 7)); n++; }
        reads.addAndGet(n);
      });
      rs[i].start();
    }
    for (int i = 0; i < writers; i++) {
      final int idx = i;
      ws[i] = new Thread(() -> {
        long n = 0;
        while (!stop.get()) { c.put("k" + (n & 7), "v" + idx + "-" + n); n++; }
        writes.addAndGet(n);
      });
      ws[i].start();
    }

Thread.sleep(durationMs);
    stop.set(true);
    for (Thread t : rs) t.join();
    for (Thread t : ws) t.join();

System.out.printf("  reads=%,d  writes=%,d%n", reads.get(), writes.get());
    return reads.get();
  }

public static void main(String[] args) throws InterruptedException {
    int readers = 16, writers = 2;
    long dur = 500;
    System.out.println("synchronized:");        bench(new SyncCache(), readers, writers, dur);
    System.out.println("\nReentrantLock:");       bench(new RLCache(),   readers, writers, dur);
    System.out.println("\nReentrantReadWriteLock:"); bench(new RWCache(),   readers, writers, dur);
  }
}

Что можно вынести из запуска — и результат, вероятно, окажется неожиданным:

Критическая секция здесь — одиночный HashMap.get — несколько наносекунд. При таком размере ReentrantReadWriteLock фактически проигрывает обычному ReentrantLock (и synchronized). При типичном запуске rwlock выполняет меньше чтений, а не больше, потому что работа внутри блокировки ничтожна по сравнению со стоимостью её получения. Это самое важное, что нужно усвоить: блокировка чтения-записи — не бесплатное улучшение.
Почему она проигрывает здесь: r.lock() должен атомарно увеличить общий счётчик читателей, и этот конкурирующий счётчик становится новым узким местом. Шестнадцать ядер, одновременно молотящих одно поле в стиле AtomicInteger, порождают cache-miss не хуже, чем на одном мьютексе — иногда хуже, потому что учёт rwlock тяжелее, чем у простой блокировки. Теоретический выигрыш «читатели не блокируют друг друга» никогда не реализуется, когда чтение слишком короткое, чтобы реально пересекаться.
rwlock опережает конкурентов только тогда, когда каждое чтение удерживает блокировку достаточно долго для реального перекрытия — думайте о поиске, десериализации или чём-либо в диапазоне микросекунд и более, а не об одном поиске в map. Замените тело get() на по-настоящему дорогостоящую read-only работу и запустите снова: теперь параллельные читатели уверенно выигрывают. Профилируйте реальную нагрузку, прежде чем прибегать к ReadWriteLock.
Стоимость ReadWriteLock — больше состояния для поддержания (счётчик читателей, флаг ожидания писателя, политика справедливости) — каждый r.lock() обходится дороже, чем ReentrantLock.lock(). При низкой конкуренции или очень коротких критических секциях более простая блокировка быстрее. При экстремально read-heavy нагрузке на многих ядрах StampedLock (оптимистичные чтения ничего не захватывают) или иммутабельный снимок за AtomicReference обычно выигрывает у обоих.
Какую бы блокировку вы ни выбрали, писатели всё равно получают эксклюзивный доступ через путь записи и никогда не повреждают map — корректность одинакова для всех трёх вариантов. Разница — исключительно в пропускной способности, которая целиком зависит от того, что находится внутри блокировки.
Понижение (w → r → освобождение w) — это то, что продакшн-код использует, когда перестройка должна происходить под блокировкой записи, а остаток запроса может оставаться под блокировкой чтения. Повышение в обратную сторону вызывает взаимоблокировку; сначала освободите r, возьмите w, проверьте состояние заново, затем продолжайте.

Что дальше

Следующая глава, Java Thread Pools, начинает историю высокоуровневого executor-фреймворка — идею о том, что вы перестаёте создавать потоки вручную и вместо этого отправляете работу в пул, который управляет потоками за вас.

Практика

Вы удерживаете блокировку чтения `ReentrantReadWriteLock` и решаете перейти к блокировке записи, вызвав `w.lock()`, всё ещё удерживая `r`. Что произойдёт?

Взаимоблокировка — блокировка записи ждёт, пока все читатели (включая вас) освободятся, но вы не можете освободиться, не продолжив выполнениеУспех — блокировка обнаруживает, что вы уже удерживаете блокировку чтения, и атомарно повышает еёВыбрасывается `IllegalMonitorStateException`, поскольку блокировки чтения и записи несовместимыВозвращается false — блокировка удерживается читателем, поэтому получение блокировки записи молча не удаётся