Блокировка чтения/записи. Часть 2.

Рассмотрим повторное блокирование (reentrant locking) для читающих потоков.

По материалам Read/Write Locks in Java - Read Reentrance.

Рассмотрим ситуацию:

• Поток 1 получил доступ на чтение
• Поток 2 запросил доступ на запись (изменение)
• Поток 1 повторно запросил получил доступ на чтение, но он окажется заблокированным, т.к. уже имеется запрос на изменение данных от Потока 2

В этом случае мы столкнемся с вариантом дедлока - Поток 2 окажется заблокированным, т.к. уже имеется читающий поток, а Поток 1 будет заблокирован, т.к. есть запрос на изменение данных от Потока 2.

Введем правило для повторного блокирования: поток может получить повторную блокировку на чтение, если он может получить доступ на чтение (нет записывающих потоков, и также нет запросов на изменение) или он уже имеет доступ на чтение (вне зависимости от имеющихся запросов на изменение).

Если поток уже имеет доступ на чтение, это означает, что гарантированно нет потоков, записывающих данные. Но могут быть запросы на изменение данных. Мы их не будем учитывать при отдаче повторной блокировки для уже читающего потока. Ибо учет этих запросов будет приводить в мертвой блокировки читающего потока. А это как раз та ситуация, во избежание которой мы и вводим повторную блокировку.

Чтобы определить, имеет ли поток доступ на чтение, придется держать табличку из потоков, читающих данные. Для каждого потока в этой табличке будем держать счетчик блокировок для этого потока. Когда поток получает доступ на чтение, он добавляется в эту табличку, и счетчик для него устанавливается в 1. Когда поток запрашивает повторный доступ на чтение, счетчик увеличивается на 1. Когда поток отпускает блокировку, то счетчик уменьшается на 1. Если при этом он обнуляется, то поток удаляется из таблички.

Код:

public class ReadWriteLock {
  private Map<Thread, Integer> readingThreads = new HashMap<Thread, Integer>();
  private int writers        = 0;
  private int writeRequests  = 0;
 
  public synchronized void lockRead() throws InterruptedException {
    Thread callingThread = Thread.currentThread();
    while(!canGrantReadAccess(callingThread)){
      wait();                                                                   
    }
    int accessCount = getReadAccessCount(callingThread);
    readingThreads.put(callingThread,accessCount+1);
  }
 
  public synchronized void unlockRead() {
    Thread callingThread = Thread.currentThread();
    int accessCount = getReadAccessCount(callingThread);
    if(accessCount == 1) { 
      readingThreads.remove(callingThread); 
    } else { 
      readingThreads.put(callingThread, accessCount-1); 
    }
    notifyAll();
  }
 
  private boolean canGrantReadAccess(Thread callingThread) {
    if(writers > 0)            return false;
    if(isReader(callingThread) return true;
    if(writeRequests > 0)      return false;
    return true;
  }
 
  private int getReadAccessCount(Thread callingThread) {
    Integer accessCount = readingThreads.get(callingThread);
    if(accessCount == null) return 0;
    return accessCount.intValue();
  }
 
  private boolean isReader(Thread callingThread){
    return readingThreads.get(callingThread) != null;
  }
 
  public synchronized void lockWrite() throws InterruptedException{
    writeRequests++;
 
    while(readingThreads.size() > 0 || writers > 0){
      wait();
    }
    writeRequests--;
    writers++;
  }
 
  public synchronized void unlockWrite() {
    writers--;
    notifyAll();
  }
}