commons-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ozeigerm...@apache.org
Subject svn commit: r558810 - in /jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking: LockException.java RWLockManager.java
Date Mon, 23 Jul 2007 17:40:11 GMT
Author: ozeigermann
Date: Mon Jul 23 10:40:10 2007
New Revision: 558810

URL: http://svn.apache.org/viewvc?view=rev&rev=558810
Log:
Added first version of lock manager that supports deadlock detection. 
This has not been properly tested and still misses reasonable thread-safety protection.

Modified:
    jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/LockException.java
    jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/RWLockManager.java

Modified: jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/LockException.java
URL: http://svn.apache.org/viewvc/jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/LockException.java?view=diff&rev=558810&r1=558809&r2=558810
==============================================================================
--- jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/LockException.java
(original)
+++ jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/LockException.java
Mon Jul 23 10:40:10 2007
@@ -43,7 +43,7 @@
         /**
          * Locking request canceled because of deadlock.
          */
-        DEADLOCK_VICTIM,
+        WOULD_DEADLOCK,
         
         /**
          * A conflict between two optimistic transactions occured.

Modified: jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/RWLockManager.java
URL: http://svn.apache.org/viewvc/jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/RWLockManager.java?view=diff&rev=558810&r1=558809&r2=558810
==============================================================================
--- jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/RWLockManager.java
(original)
+++ jakarta/commons/proper/transaction/branches/TRANSACTION_2/src/java/org/apache/commons/transaction/locking/RWLockManager.java
Mon Jul 23 10:40:10 2007
@@ -16,10 +16,13 @@
  */
 package org.apache.commons.transaction.locking;
 
+import java.util.ArrayList;
+import java.util.Collection;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReadWriteLock;
@@ -31,45 +34,15 @@
 
     protected ConcurrentHashMap<KeyEntry<K, M>, ReadWriteLock> locks = new ConcurrentHashMap<KeyEntry<K,
M>, ReadWriteLock>();
 
-    protected Map<Thread, Set<Lock>> locksForThreads = new ConcurrentHashMap<Thread,
Set<Lock>>();
+    protected Map<Thread, CopyOnWriteArraySet<Lock>> locksForThreads = new ConcurrentHashMap<Thread,
CopyOnWriteArraySet<Lock>>();
 
-    protected Map<ReadWriteLock, Set<Thread>> threadsForLocks = new ConcurrentHashMap<ReadWriteLock,
Set<Thread>>();
+    protected ConcurrentHashMap<Lock, Set<Thread>> threadsForLocks = new ConcurrentHashMap<Lock,
Set<Thread>>();
 
     protected Map<Thread, Long> effectiveGlobalTimeouts = new ConcurrentHashMap<Thread,
Long>();
 
-    // TODO
-    public Iterable<ReadWriteLock> orderLocks() {
-        Set<Lock> locks = locksForThreads.get(Thread.currentThread());
-        if (locks == null) {
-            throw new IllegalStateException("lock() can only be called after startWork()");
-        }
-
-        return null;
-
-    }
-
     @Override
     public void endWork() {
-        Set<Lock> locks = locksForThreads.get(Thread.currentThread());
-        // graceful reaction...
-        if (locks == null) {
-            return;
-        }
-        for (Lock lock : locks) {
-            lock.unlock();
-
-            // FIXME: We need to do this atomically
-            Set<Thread> threadsForThisLock = threadsForLocks.get(lock);
-            if (threadsForThisLock != null) {
-                threadsForThisLock.remove(Thread.currentThread());
-                if (threadsForThisLock.isEmpty()) {
-                    threadsForLocks.remove(lock);
-                    locks.remove(lock);
-                }
-            }
-        }
-
-        locksForThreads.remove(Thread.currentThread());
+        release(Thread.currentThread());
     }
 
     @Override
@@ -77,7 +50,7 @@
         if (isWorking()) {
             throw new IllegalStateException("work has already been started");
         }
-        locksForThreads.put(Thread.currentThread(), new HashSet<Lock>());
+        locksForThreads.put(Thread.currentThread(), new CopyOnWriteArraySet<Lock>());
 
         long timeoutMSecs = unit.toMillis(timeout);
         long now = System.currentTimeMillis();
@@ -91,8 +64,8 @@
 
     }
 
-    protected long computeRemainingTime() {
-        long timeout = effectiveGlobalTimeouts.get(Thread.currentThread());
+    protected long computeRemainingTime(Thread thread) {
+        long timeout = effectiveGlobalTimeouts.get(thread);
         long now = System.currentTimeMillis();
         long remaining = timeout - now;
         return remaining;
@@ -147,10 +120,8 @@
 
     @Override
     public void lock(M managedResource, K key, boolean exclusive) throws LockException {
-        long remainingTime = computeRemainingTime();
-        if (remainingTime < 0) {
-            throw new LockException(LockException.Code.TIMED_OUT);
-        }
+        long remainingTime = computeRemainingTime(Thread.currentThread());
+
         boolean locked = tryLockInternal(managedResource, key, exclusive, remainingTime,
                 TimeUnit.MILLISECONDS);
         if (!locked) {
@@ -165,6 +136,8 @@
 
     protected boolean tryLockInternal(M managedResource, K key, boolean exclusive, long time,
             TimeUnit unit) throws LockException {
+        reportTimeout(Thread.currentThread());
+
         KeyEntry<K, M> entry = new KeyEntry<K, M>(key, managedResource);
         ReadWriteLock rwlock = putIfAbsent(entry, create());
         Set<Lock> locks = locksForThreads.get(Thread.currentThread());
@@ -174,20 +147,171 @@
 
         Lock lock = exclusive ? rwlock.writeLock() : rwlock.readLock();
 
+        boolean locked;
+        if (time == 0) {
+            locked = lock.tryLock();
+        } else {
+            locked = doTrickyYetEfficientLockOnlyIfThisCanNotCauseADeadlock(lock, unit.toMillis(time));
+        }
+        if (locked) {
+            locks.add(lock);
+            Set<Thread> threads = threadsForLocks.get(lock);
+            if (threads == null) {
+                threads = new HashSet<Thread>();
+                Set<Thread> concurrentlyInsertedThreads = threadsForLocks
+                        .putIfAbsent(lock, threads);
+                if (concurrentlyInsertedThreads != null)
+                    threads = concurrentlyInsertedThreads;
+            }
+            threads.add(Thread.currentThread());
+        }
+        return locked;
+    }
+
+    protected boolean doTrickyYetEfficientLockOnlyIfThisCanNotCauseADeadlock(Lock lock,
+            long timeMsecs) throws LockException {
+
+        // This algorithm is devided into three parts:
+        // Note: We can be interrupted most of the time
+        //
+        // I prewait:
+        // Wait a fraktion of the time to see if we can acquire
+        // the lock in short time. If we can all is good and we exit
+        // signalling success. If not we need to get into a more resource
+        // consuming phase.
+        //
+        // II clearing of timed out thtreads / deadlock detection:
+        // As we have not been able to acquire the lock, yet, maybe there is
+        // deadlock. Clear all threads already timed out and afterwards
+        // check for a deadlock state. If there is one report it with an
+        // exception. If not we enter the final phase.
+        // 
+        // III real wait:
+        // Everything is under control, we were just a little bit too
+        // impatient. So wait for the remaining time and see if the can get
+        // the lock
+        // 
+
         try {
             boolean locked;
-            if (time == 0) {
-                locked = lock.tryLock();
-            } else {
-                locked = lock.tryLock(time, unit);
-            }
-            if (locked) {
-                locks.add(lock);
+
+            // I prewait
+
+            long startTime = System.currentTimeMillis();
+
+            // TODO this heuristic devisor really should be configurable
+            long preWaitTime = timeMsecs / 5;
+            locked = lock.tryLock(preWaitTime, TimeUnit.MILLISECONDS);
+            if (locked)
+                return true;
+
+            // II deadlock detect
+            cancelAllTimedOut();
+            if (wouldDeadlock(Thread.currentThread(), new HashSet<Thread>())) {
+                throw new LockException(LockException.Code.WOULD_DEADLOCK);
             }
+
+            // III real wait
+            long now = System.currentTimeMillis();
+            long remainingWaitTime = timeMsecs - (now - startTime);
+            if (remainingWaitTime < 0)
+                return false;
+
+            locked = lock.tryLock(remainingWaitTime, TimeUnit.MILLISECONDS);
             return locked;
         } catch (InterruptedException e) {
-            throw new LockException(Code.INTERRUPTED, key);
+            throw new LockException(Code.INTERRUPTED);
         }
+
+    }
+
+    protected boolean wouldDeadlock(Thread thread, Set<Thread> path) {
+        path.add(thread);
+        // these are our locks
+        // Note: No need to make a copy as we can be sure to iterate on our
+        // private
+        // version, as this is a CopyOnWriteArraySet!
+        CopyOnWriteArraySet<Lock> locks = locksForThreads.get(thread);
+        for (Lock lock : locks) {
+            // these are the ones waiting for one of our locks
+            // and if they wait, they wait because of me!
+            Collection<Thread> conflicts = getConflictingWaiters((ReentrantReadWriteLock)
lock);
+            for (Thread conflictThread : conflicts) {
+                // this means, we have found a cycle in the wait graph
+                if (path.contains(conflictThread)) {
+                    return true;
+                } else if (wouldDeadlock(conflictThread, path)) {
+                    return true;
+                }
+            }
+        }
+
+        path.remove(thread);
+        return false;
+    }
+
+    protected Collection<Thread> getConflictingWaiters(ReentrantReadWriteLock lock)
{
+        Collection<Thread> result = new ArrayList<Thread>();
+        // Consider every thread that holds at least one lock!
+        // Caution: We can not use "threadsForLocks" as the waiting threads
+        // have not yet acquired the lock and thus are not part of the map.
+        // An alternative algorithm could also remember the threads waiting for
+        // a lock
+        Collection<Thread> threadsWithLocks = locksForThreads.keySet();
+        for (Thread thread : threadsWithLocks) {
+            if (lock.hasQueuedThread(thread)) {
+                result.add(thread);
+            }
+        }
+        return result;
+    }
+
+    protected void reportTimeout(Thread thread) throws LockException {
+        if (hasTimedOut(thread)) {
+            throw new LockException(LockException.Code.TIMED_OUT);
+        }
+    }
+
+    protected void cancelAllTimedOut() {
+        Set<Thread> threads = effectiveGlobalTimeouts.keySet();
+        for (Thread thread : threads) {
+            if (hasTimedOut(thread)) {
+                // TODO: We need to record this thread has timed out to produce
+                // a meaningful exception when it tries to continue its work
+                release(thread);
+                thread.interrupt();
+            }
+
+        }
+    }
+
+    protected boolean hasTimedOut(Thread thread) {
+        long remainingTime = computeRemainingTime(thread);
+        return (remainingTime < 0);
+
+    }
+
+    protected void release(Thread thread) {
+        Set<Lock> locks = locksForThreads.get(thread);
+        // graceful reaction...
+        if (locks == null) {
+            return;
+        }
+        for (Lock lock : locks) {
+            lock.unlock();
+
+            // FIXME: We need to do this atomically
+            Set<Thread> threadsForThisLock = threadsForLocks.get(lock);
+            if (threadsForThisLock != null) {
+                threadsForThisLock.remove(Thread.currentThread());
+                if (threadsForThisLock.isEmpty()) {
+                    threadsForLocks.remove(lock);
+                    locks.remove(lock);
+                }
+            }
+        }
+
+        locksForThreads.remove(thread);
     }
 
 }



---------------------------------------------------------------------
To unsubscribe, e-mail: commons-dev-unsubscribe@jakarta.apache.org
For additional commands, e-mail: commons-dev-help@jakarta.apache.org


Mime
View raw message