Менеджер ресурсов с ReentrantLocks

Question

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

Я пытаюсь использовать справедливые ReentrantReadWriteLock-s, по одному для каждого ресурса.

Я вижу только журнал клиента.

Существует несколько проблем, иногда поток не прекращает запрашивать и приобретать ресурсы, иногда происходит взаимоблокировка, а иногда и сбой releaseLock. Любые советы оценены.

public class ResHandler { 

//ID-s of the granted resource lists 
private static long lockNum = 0; 

//Resources are identified by strings, each client has a list of demanded resources 
//we store these when granted, along with an ID 
private static ConcurrentHashMap<Long, Set<String>> usedResources 
    = new ConcurrentHashMap<Long, Set<String>>(); 

//We store a lock for each resource 
private static ConcurrentHashMap<String, ReentrantReadWriteLock> resources 
    = new ConcurrentHashMap<String, ReentrantReadWriteLock>(); 

//Filling our resources map with the resources and their locks 
static { 
    for (int i = 0; i < SharedValues.RESOURCE_LIST.length; ++i) { 
     String res = SharedValues.RESOURCE_LIST[i]; 
     //Fair reentrant lock 
     ReentrantReadWriteLock lc = new ReentrantReadWriteLock(true); 
     resources.put(res, lc); 
    } 
} 

//We get a set of the required resources and the type of lock we have to use 
public static long getLock(Set<String> mNeededRes, boolean mMethod) { 
    //!!! 
    if (mMethod == SharedValues.READ_METHOD) { 

     //We try to get the required resources 
     for (String mn : mNeededRes) 
      resources.get(mn).readLock().lock(); 

     //After grandted, we put them in the usedResources map 
     ++lockNum; 
     usedResources.put(lockNum, mNeededRes); 
     return lockNum;   
    } 

    //Same thing, but with write locks 
    else { 

     for (String mn : mNeededRes) 
      resources.get(mn).writeLock().lock(); 

     ++lockNum; 
     usedResources.put(lockNum, mNeededRes); 
     return lockNum;   
    } 
} 

//Releasing a set of locks by the set's ID 
public static void releaseLock(long mLockID) { 
    if (!usedResources.containsKey(mLockID)) { 
     System.out.println("returned, no such key as: " + mLockID); 
     return; 
    } 

    Set<String> toBeReleased = usedResources.get(mLockID); 

    //Unlocking every lock from this set 
    for (String s : toBeReleased) { 
     if (resources.get(s).isWriteLockedByCurrentThread()) 
      resources.get(s).writeLock().unlock(); 
     else 
      resources.get(s).readLock().unlock(); 
    } 

    //Deleting from the map 
    usedResources.remove(mLockID); 
} 
} 

Answer 1

Обновлено решение, дайте ему попробовать:

public class ResHandler { 

private static AtomicLong lockNum = new AtomicLong(0); 
private static Map<Long, Set<String>> usedResources = new ConcurrentHashMap<Long, Set<String>>(); 
private static final Map<String, ReentrantReadWriteLock> resources = new ConcurrentHashMap<String, ReentrantReadWriteLock>(); 
// "priorityResources" to avoid deadlocks and starvation 
private static final Map<String, PriorityBlockingQueue<Long>> priorityResources = new ConcurrentHashMap<String, PriorityBlockingQueue<Long>>(); 

static { 
    for (int i = 0; i < SharedValues.RESOURCE_LIST.length; ++i) { 
     String res = SharedValues.RESOURCE_LIST[i]; 
     ReentrantReadWriteLock lc = new ReentrantReadWriteLock(true); 
     resources.put(res, lc); 
     priorityResources.put(res, new PriorityBlockingQueue<Long>()); 
    } 
} 

public static long getLock(Set<String> mNeededRes, boolean mMethod) { 
    long lockNumLocal = lockNum.addAndGet(1); 
    for (String mn : mNeededRes) { 
     priorityResources.get(mn).offer(lockNumLocal); 
    } 
    boolean tryLockResult; 
    List<String> lockedList = new ArrayList<String>(); 
    boolean allLocked = false; 
    while (!allLocked) { 
     allLocked = true; 
     for (String mn : mNeededRes) { 
      if (lockedList.contains(mn) == true) { 
       continue;//because we already have the lock 
      } 
      try { 
       if (mMethod == SharedValues.READ_METHOD) { 
        tryLockResult = resources.get(mn).readLock().tryLock(1, TimeUnit.MILLISECONDS); 
       } else { 
        tryLockResult = resources.get(mn).writeLock().tryLock(1, TimeUnit.MILLISECONDS); 
       } 
      } catch (InterruptedException ex) { 
       Logger.getLogger(ResHandler.class.getName()).log(Level.SEVERE, null, ex); 
       tryLockResult = false; 
      } 

      if (tryLockResult) { 
       lockedList.add(mn); 
      } else { 
       allLocked = false; 
       for (int i = lockedList.size() - 1; i >= 0; i--) { 
        //if the lock failed, all previous locked resources need to be released, but only if they will be used by higher priority lock operations 
        if (priorityResources.get(lockedList.get(i)).peek() != lockNumLocal) { 
         if (mMethod == SharedValues.READ_METHOD) { 
          resources.get(lockedList.get(i)).readLock().unlock(); 
         } else { 
          resources.get(lockedList.get(i)).writeLock().unlock(); 
         } 
         lockedList.remove(i); 
        } 
       } 
       break; 
      } 
     } 
    } 
    usedResources.put(lockNumLocal, mNeededRes); 
    for (String mn : mNeededRes) { 
     priorityResources.get(mn).remove(lockNumLocal); 
    } 
    return lockNumLocal; 
} 

public static void releaseLock(long mLockID) { 
    if (!usedResources.containsKey(mLockID)) { 
     System.out.println("returned, no such key as: " + mLockID); 
     return; 
    } 

    Set<String> toBeReleased = usedResources.get(mLockID); 

    //Unlocking every lock from this set 
    for (String s : toBeReleased) { 
     if (resources.get(s).isWriteLockedByCurrentThread()) { 
      resources.get(s).writeLock().unlock(); 
     } else { 
      resources.get(s).readLock().unlock(); 
     } 
    } 

    //Deleting from the map 
    usedResources.remove(mLockID); 
} 

}

Answer 2

Я предполагаю, что различные клиенты могут позвонить getLock из разных потоков. Если это так, то первая проблема заключается в том, что доступ к lockNum не синхронизирован. Два потока могут одновременно вызвать getLock, поэтому, в зависимости от времени, оба они могут быть возвращены одному номеру блокировки. Это объясняет, почему иногда блокируется блокировка релиза.

Если вы можете устранить эту проблему, должно быть проще определить, что еще происходит.

Answer 3

Чтобы избежать тупика, ваши ресурсы должны быть приобретены в том же порядке, поэтому вам необходимо отсортировать Set<String> mNeededRes перед выполнением блокировок в цикле. Метод сортировки не имеет большого значения.

Это описано в Chapter10. Avoiding Liveness Hazards от Java Concurrency In Practice Brian Göetz.

Идентифицируйте вас, чтобы удалить getLock и releaseLock или сделать их частными. И заверните всю свою логику в Runnable. Если вы контролируете все блокировки, невозможно забыть их освободить. Сделать что-то вроде этого:

public void performMethod(List<String> mNeededRes, boolean mMethod, Runnable r){ 
    List sortd = Collections.sort(mNeededRes); 
    try{ 
     getLock(mNeededRes, mMethod); 
     r.run(); 
    }finally { 
     releaseLock(mNeededRes); 
    } 
} 

Answer 4

Есть несколько вопросов происходит в вашей программе, которые являются причиной зависания и ошибки:

• В общем: объявить глобальные переменные окончательными. Вы не хотите случайно с ними связываться. Кроме того, это позволяет использовать их в качестве объектов синхронизации.

• long не гарантированно является атомарным и не является оператором ++. 32-разрядная JVM должна записать это за 2 шага и, следовательно, теоретически может вызвать серьезную зависть в вашей системе. Лучше использовать AtomicLong.

• GetLock не является потокобезопасным.Пример:

Автор А вызывает getLock для ресурса 1, 3, 5
резьбы B вызывает getLock в то же самое время для ресурса 2,5,3
Автор А предоставляется замок на 1, 3, а затем надевается пауза
Thread B предоставляется замок на 2, 5, то ставится на паузу
Поток а теперь ждет на 5 из темы B и B Thread теперь ждет на 3 из темы А.
взаимоблокировки.

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

• ++lockNum приведет к тому, что два потока испортить свое значение блокировки, если вызываются одновременно с глобальной переменной.

Вот код в рабочем состоянии:

private static final AtomicLong lockNum = new AtomicLong(0); 
    private static final ConcurrentHashMap<Long, Set<String>> usedResources = new ConcurrentHashMap<Long, Set<String>>(); 
    private static final ConcurrentHashMap<String, ReentrantReadWriteLock> resources = new ConcurrentHashMap<String, ReentrantReadWriteLock>(); 

    static { 
    for (int i = 0; i < SharedValues.RESOURCE_LIST.length; ++i) { 
     String res = SharedValues.RESOURCE_LIST[i]; 
     ReentrantReadWriteLock lc = new ReentrantReadWriteLock(true); 
     resources.put(res, lc); 
    } 
    } 

    public static long getLock(Set<String> mNeededRes, boolean mMethod) { 
    synchronized (resources) { 
     if (mMethod == SharedValues.READ_METHOD) { 
     for (String mn : mNeededRes) { 
      resources.get(mn).readLock().lock(); 
     } 
     } else { 
     for (String mn : mNeededRes) { 
      resources.get(mn).writeLock().lock(); 
     } 
     } 
    } 
    final long lockNumber = lockNum.getAndIncrement(); 
    usedResources.put(lockNumber, mNeededRes); 
    return lockNumber; 
    } 

    public static void releaseLock(final long mLockID) { 
    if (!usedResources.containsKey(mLockID)) { 
     System.out.println("returned, no such key as: " + mLockID); 
     return; 
    } 

    final Set<String> toBeReleased = usedResources.remove(mLockID); 

    for (String s : toBeReleased) { 
     final ReentrantReadWriteLock lock = resources.get(s); 
     if (lock.isWriteLockedByCurrentThread()) { 
     lock.writeLock().unlock(); 
     } else { 
     lock.readLock().unlock(); 
     } 
    } 
    }