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

Итак, приведем полностью код класса ReadWriteLock, который поддерживает все варианты повторных блокировок:

• для читающих потоков
• для записывающих потоков
• для читающих потоков, получающих также доступ на запись
• для записывающих потоков, получающих также доступ на чтение

Код класса:

public class ReadWriteLock {
  private Map<Thread, Integer> readingThreads = new HashMap<Thread, Integer>();
  private int writers        = 0;
  private int writeRequests  = 0;
  private Thread writingThread;
 
  //
  // read locking
  //
 
  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 (isWriter(callingThread)) return true;
    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;
  }
 
  private boolean hasReaders() {
    return readingThreads.size() > 0;
  }
 
  private boolean isOnlyReader(Thread thread) {
    return readingThreads.size() == 1 && readingThreads.get(thread) != null;
  }
 
  //
  // write locking
  //
 
  public synchronized void lockWrite() throws InterruptedException {
    writeRequests++;
    Thread callingThread = Thread.currentThread();
 
    while(!canGrantWriteAccess(callingThread)){
      wait();
    }
    writeRequests--;
    writers++;
    writingThread = callingThread;
  }
 
  public synchronized void unlockWrite() throws InterruptedException {
    writers--;
    if(writers == 0){
      writingThread = null;
    }
    notifyAll();
  }
 
  private boolean canGrantWriteAccess(Thread callingThread) {
    if(isOnlyReader(callingThread))    return true;
    if(hasReaders())                   return false;
    if(writingThread == null)          return true;
    if(!isWriter(callingThread))       return false;
    return true;
  }
 
  private boolean isWriter(Thread callingThread) {
    return writingThread == callingThread;
  }
}

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

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

Наконец, последний случай, который мы рассмотрим - если потоку, который имеет доступ на запись, понадобилось вдруг получить также доступ на чтение.

Поток, изменяющий данные, должен всегда получать доступ и на чтение, если это понадобилось. Код из предыдущего поста останется без изменений, кроме метода canGrantReadAccess - туда мы добавим проверку того, что текущий поток является записывающим потоком:

private boolean canGrantReadAccess(Thread callingThread) {
  if (isWriter(callingThread)) return true;
  if(writers > 0)              return false;
  if(isReader(callingThread)   return true;
  if(writeRequests > 0)        return false;
  return true;
}

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

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

В предыдущем посте был рассмотрен случай блокировки, который обрабатывает варианты повторной блокировки для чтения и записи. Возьмем код оттуда и немножко модифицируем.

Введем новый метод - isOnlyReader, который проверяет, является ли текущий поток единственным читающим потоком:

private boolean isOnlyReader(Thread thread){
  return readingThreads.size() == 1 && readingThreads.get(thread) != null;
}

Также немножко модифицируем метод canGrantWriteAccess:

private boolean canGrantWriteAccess(Thread callingThread){
  if(isOnlyReader(callingThread))    return true;
  if(hasReaders())                   return false;
  if(writingThread == null)          return true;
  if(!isWriter(callingThread))       return false;
  return true;
}

В остальном, код класса ReadWriteLock такой же, как и в
предыдущем посте. Полный код:

public class ReadWriteLock {
  private Map<Thread, Integer> readingThreads = new HashMap<Thread, Integer>();
  private int writers        = 0;
  private int writeRequests  = 0;
  private Thread writingThread;
 
  //
  // read locking
  //
 
  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;
  }
 
  private boolean hasReaders() {
    return readingThreads.size() > 0;
  }
 
  private boolean isOnlyReader(Thread thread) {
    return readingThreads.size() == 1 && readingThreads.get(thread) != null;
  }
 
  //
  // write locking
  //
 
  public synchronized void lockWrite() throws InterruptedException {
    writeRequests++;
    Thread callingThread = Thread.currentThread();
 
    while(!canGrantWriteAccess(callingThread)){
      wait();
    }
    writeRequests--;
    writers++;
    writingThread = callingThread;
  }
 
  public synchronized void unlockWrite() throws InterruptedException {
    writers--;
    if(writers == 0){
      writingThread = null;
    }
    notifyAll();
  }
 
  private boolean canGrantWriteAccess(Thread callingThread) {
    if(isOnlyReader(callingThread))    return true;
    if(hasReaders())                   return false;
    if(writingThread == null)          return true;
    if(!isWriter(callingThread))       return false;
    return true;
  }
 
  private boolean isWriter(Thread callingThread) {
    return writingThread == callingThread;
  }
}

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

Здесь мы рассмотрим повторное блокирование (reentrant locking) для записывающего потока.

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

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

• Поток получает доступ на запись
• Поток повторно запрашивает доступ на запись
• Из-за того, что счетчик writers отличен от нуля, поток переводится в состояние ожидания и зависает насовсем

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

В коде ниже будем учитывать также повторную блокировку для читающих потоков. Уберем проверку условия на то, можем ли мы дать доступ на запись в отдельный метод - canGrantWriteAccess. В счетчик writers будем записывать число повторных блокировок. Теперь, если writers >0 мы будем проверять, имеет ли этот поток доступ на запись. Если поток имеет доступ на запись, то увеличиваем счетчик writers. Если поток не имеет доступа на запись, то по прежнему переводим его в режим ожидания.

Метод canGrantWriteAccess выглядит так:

private boolean canGrantWriteAccess(Thread callingThread){
  if(hasReaders())             return false;
  if(writingThread == null)    return true;
  if(!isWriter(callingThread)) return false;
  return true;
}

Полный код класса таков:

public class ReadWriteLock {
  private Map<Thread, Integer> readingThreads = new HashMap<Thread, Integer>();
  private int writers        = 0;
  private int writeRequests  = 0;
  private Thread writingThread;
 
  //
  // read locking
  //
 
  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;
  }
 
  private boolean hasReaders(){
    return readingThreads.size() > 0;
  }
 
  //
  // write locking
  //
 
  public synchronized void lockWrite() throws InterruptedException {
    writeRequests++;
    Thread callingThread = Thread.currentThread();
 
    while(!canGrantWriteAccess(callingThread)){
      wait();
    }
    writeRequests--;
    writers++;
    writingThread = callingThread;
  }
 
  public synchronized void unlockWrite() throws InterruptedException{
    writers--;
    if(writers == 0){
      writingThread = null;
    }
    notifyAll();
  }
 
  private boolean canGrantWriteAccess(Thread callingThread){
    if(hasReaders())             return false;
    if(writingThread == null)    return true;
    if(!isWriter(callingThread)) return false;
    return true;
  }
 
  private boolean isWriter(Thread callingThread){
    return writingThread == callingThread;
  }
}

Здесь есть учет повторной блокировки для читающих потоков и учет повторной блокировки для записывающего потока. Мы еще не рассмотрели два случая: когда читающий поток хочет получить повторную блокировку на запись, и когда записывающий поток хочет получить повторную блокировку на чтение данных. Эти случаи будут рассмотрены в следующих постах.

Блокировка чтения/записи. Часть 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();
  }
}

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

Read/write Locks

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

Условия:

Имеются общие данные, к которым потоки обращаются на чтение и изменение (запись). При этом изменение данных происходит относительно редко. Гораздо чаще происходит их чтение.

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

Задача:

Требуется эффективно синхронизировать читающие и изменяющие потоки между собой.

Решение:

Для начала, еще раз подытожим условия блокировки:

Доступ для чтения: если нет потоков, которые изменяют данные, а также нет потоков, которые запросили доступ для изменения данных.

Доступ для записи: если нет потоков, которые читают или изменяют данные.

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

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

Поэтому мы можем написать следующие код:

public class ReadWriteLock{
 
  private int readers       = 0;
  private int writers       = 0;
  private int writeRequests = 0;
 
  public synchronized void lockRead() throws InterruptedException{
    while(writers > 0 || writeRequests > 0){
      wait();
    }
    readers++;
  }
 
  public synchronized void unlockRead(){
    readers--;
    notifyAll();
  }
 
  public synchronized void lockWrite() throws InterruptedException{
    writeRequests++;
 
    while(readers > 0 || writers > 0){
      wait();
    }
    writeRequests--;
    writers++;
  }
 
  public synchronized void unlockWrite() {
    writers--;
    notifyAll();
  }
}

Здесь мы имеем два метода для начала блокировки, и два метода для окончания блокировки. Код, который мы реализовали, не подразумевает возвратного блокирования (reentrant locking).

Ниже приведен пример использования блокировки чтения/записи для потоко-безопасного кеша:

public class Cache {
  HashMap<Integer,String> map = new HashMap<Integer,String>();
  ReadWriteLock lock = new ReadWriteLock();
 
  public String get(Integer key) throws CacheException {
    try {
      lock.lockRead();
      return map.get(key);
    } catch (InterruptedException ie) {
      throw new CacheException(ie);
    } finally {
      lock.unlockRead();
    }
  }
 
  public void put(Integer key, String value) throws CacheException {
    try {
      lock.lockWrite();
      map.put(key,value);
    } catch (InterruptedException ie) {
      throw new CacheException(ie);
    } finally {
      lock.unlockWrite();
    }
  }
 
  public void remove(Integer key) throws CacheException {
    try {
      lock.lockWrite();
      map.remove(key);
    } catch (InterruptedException ie) {
      throw new CacheException(ie);
    } finally {
      lock.unlockWrite();
    }
  }  
}