commons-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ma...@apache.org
Subject svn commit: r775707 - /commons/proper/pool/trunk/src/java/org/apache/commons/pool/impl/GenericObjectPool.java
Date Sun, 17 May 2009 16:42:34 GMT
Author: markt
Date: Sun May 17 16:42:34 2009
New Revision: 775707

URL: http://svn.apache.org/viewvc?rev=775707&view=rev
Log:
Fix POOL-75 for GOP. Ensure objects are allocated in the order in which threads ask for them.

Modified:
    commons/proper/pool/trunk/src/java/org/apache/commons/pool/impl/GenericObjectPool.java

Modified: commons/proper/pool/trunk/src/java/org/apache/commons/pool/impl/GenericObjectPool.java
URL: http://svn.apache.org/viewvc/commons/proper/pool/trunk/src/java/org/apache/commons/pool/impl/GenericObjectPool.java?rev=775707&r1=775706&r2=775707&view=diff
==============================================================================
--- commons/proper/pool/trunk/src/java/org/apache/commons/pool/impl/GenericObjectPool.java
(original)
+++ commons/proper/pool/trunk/src/java/org/apache/commons/pool/impl/GenericObjectPool.java
Sun May 17 16:42:34 2009
@@ -20,6 +20,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.TimerTask;
@@ -539,7 +540,7 @@
      */
     public synchronized void setMaxActive(int maxActive) {
         _maxActive = maxActive;
-        notifyAll();
+        allocate();
     }
 
     /**
@@ -570,7 +571,7 @@
             case WHEN_EXHAUSTED_FAIL:
             case WHEN_EXHAUSTED_GROW:
                 _whenExhaustedAction = whenExhaustedAction;
-                notifyAll();
+                allocate();
                 break;
             default:
                 throw new IllegalArgumentException("whenExhaustedAction " + whenExhaustedAction
+ " not recognized.");
@@ -614,7 +615,7 @@
      */
     public synchronized void setMaxWait(long maxWait) {
         _maxWait = maxWait;
-        notifyAll();
+        allocate();
     }
 
     /**
@@ -641,7 +642,7 @@
      */
     public synchronized void setMaxIdle(int maxIdle) {
         _maxIdle = maxIdle;
-        notifyAll();
+        allocate();
     }
 
     /**
@@ -658,7 +659,7 @@
      */
     public synchronized void setMinIdle(int minIdle) {
         _minIdle = minIdle;
-        notifyAll();
+        allocate();
     }
 
     /**
@@ -917,43 +918,42 @@
         setTimeBetweenEvictionRunsMillis(conf.timeBetweenEvictionRunsMillis);
         setSoftMinEvictableIdleTimeMillis(conf.softMinEvictableIdleTimeMillis);
         setLifo(conf.lifo);
-        notifyAll();
+        allocate();
     }
 
     //-- ObjectPool methods ------------------------------------------
 
     public Object borrowObject() throws Exception {
         long starttime = System.currentTimeMillis();
+        Latch latch = new Latch();
+        synchronized (this) {
+            _allocationQueue.add(latch);
+            allocate();
+        }
+
         for(;;) {
-            ObjectTimestampPair pair = null;
-            
             synchronized (this) {
                 assertOpen();
-                // if there are any sleeping, just grab one of those
-                try {
-                    pair = (ObjectTimestampPair)(_pool.removeFirst());
-                } catch(NoSuchElementException e) {
-                    /* ignored */
-                }
-    
-                // otherwise
-                if(null == pair) {
-                    // check if we can create one
-                    // (note we know that the num sleeping is 0, else we wouldn't be here)
-                    if(_maxActive < 0 || (_numActive + _numInternalProcessing) < _maxActive)
{
-                        // allow new object to be created
-                    } else {
-                        // the pool is exhausted
-                        switch(_whenExhaustedAction) {
-                            case WHEN_EXHAUSTED_GROW:
-                                // allow new object to be created
-                                break;
-                            case WHEN_EXHAUSTED_FAIL:
-                                throw new NoSuchElementException("Pool exhausted");
-                            case WHEN_EXHAUSTED_BLOCK:
-                                try {
+            }
+                
+            // If no object was allocated from the pool above
+            if(latch._pair == null) {
+                // check if we were allowed to create one
+                if(latch._mayCreate) {
+                    // allow new object to be created
+                } else {
+                    // the pool is exhausted
+                    switch(_whenExhaustedAction) {
+                        case WHEN_EXHAUSTED_GROW:
+                            // allow new object to be created
+                            break;
+                        case WHEN_EXHAUSTED_FAIL:
+                            throw new NoSuchElementException("Pool exhausted");
+                        case WHEN_EXHAUSTED_BLOCK:
+                            try {
+                                synchronized (latch) {
                                     if(_maxWait <= 0) {
-                                        wait();
+                                        latch.wait();
                                     } else {
                                         // this code may be executed again after a notify
then continue cycle
                                         // so, need to calculate the amount of time to wait
@@ -961,62 +961,66 @@
                                         final long waitTime = _maxWait - elapsed;
                                         if (waitTime > 0)
                                         {
-                                            wait(waitTime);
+                                            latch.wait(waitTime);
                                         }
                                     }
-                                } catch(InterruptedException e) {
-                                    Thread.currentThread().interrupt();
-                                    throw e; 
-                                }
-                                if(_maxWait > 0 && ((System.currentTimeMillis()
- starttime) >= _maxWait)) {
-                                    throw new NoSuchElementException("Timeout waiting for
idle object");
-                                } else {
-                                    continue; // keep looping
                                 }
-                            default:
-                                throw new IllegalArgumentException("WhenExhaustedAction property
" + _whenExhaustedAction + " not recognized.");
-                        }
+                            } catch(InterruptedException e) {
+                                Thread.currentThread().interrupt();
+                                throw e; 
+                            }
+                            if(_maxWait > 0 && ((System.currentTimeMillis() -
starttime) >= _maxWait)) {
+                                throw new NoSuchElementException("Timeout waiting for idle
object");
+                            } else {
+                                continue; // keep looping
+                            }
+                        default:
+                            throw new IllegalArgumentException("WhenExhaustedAction property
" + _whenExhaustedAction + " not recognized.");
                     }
                 }
-                _numActive++;
             }
 
-            // create new object when needed
             boolean newlyCreated = false;
-            if(null == pair) {
+            if(null == latch._pair) {
                 try {
                     Object obj = _factory.makeObject();
-                    pair = new ObjectTimestampPair(obj);
+                    latch._pair = new ObjectTimestampPair(obj);
                     newlyCreated = true;
                 } finally {
                     if (!newlyCreated) {
                         // object cannot be created
                         synchronized (this) {
-                            _numActive--;
-                            notifyAll();
+                            _numInternalProcessing--;
+                            // No need to reset latch - about to throw exception
+                            allocate();
                         }
                     }
                 }
             }
-
             // activate & validate the object
             try {
-                _factory.activateObject(pair.value);
-                if(_testOnBorrow && !_factory.validateObject(pair.value)) {
+                _factory.activateObject(latch._pair.value);
+                if(_testOnBorrow && !_factory.validateObject(latch._pair.value))
{
                     throw new Exception("ValidateObject failed");
                 }
-                return pair.value;
+                synchronized(this) {
+                    _numInternalProcessing--;
+                    _numActive++;
+                }
+                return latch._pair.value;
             }
             catch (Throwable e) {
                 // object cannot be activated or is invalid
                 try {
-                    _factory.destroyObject(pair.value);
+                    _factory.destroyObject(latch._pair.value);
                 } catch (Throwable e2) {
                     // cannot destroy broken object
                 }
                 synchronized (this) {
-                    _numActive--;
-                    notifyAll();
+                    _numInternalProcessing--;
+                    latch.reset();
+                    _allocationQueue.add(0, latch);
+                    allocate();
                 }
                 if(newlyCreated) {
                     throw new NoSuchElementException("Could not create a validated object,
cause: " + e.getMessage());
@@ -1028,6 +1032,37 @@
         }
     }
 
+    private synchronized void allocate() {
+        // First use any objects in the pool to clear the queue
+        for (;;) {
+            if (isClosed()) return;
+            if (!_pool.isEmpty() && !_allocationQueue.isEmpty()) {
+                Latch latch = (Latch) _allocationQueue.removeFirst();
+                latch._pair = (ObjectTimestampPair) _pool.removeFirst();
+                _numInternalProcessing++;
+                synchronized (latch) {
+                    latch.notify();
+                }
+            } else {
+                break;
+            }
+        }
+        // Second utilise any spare capacity to create new objects
+        for(;;) {
+            if (isClosed()) return;
+            if((!_allocationQueue.isEmpty()) && (_maxActive < 0 || (_numActive
+ _numInternalProcessing) < _maxActive)) {
+                Latch latch = (Latch) _allocationQueue.removeFirst();
+                latch._mayCreate = true;
+                _numInternalProcessing++;
+                synchronized (latch) {
+                    latch.notify();
+                }
+            } else {
+                break;
+            }
+        }
+    }
+
     public void invalidateObject(Object obj) throws Exception {
         try {
             if (_factory != null) {
@@ -1036,7 +1071,7 @@
         } finally {
             synchronized (this) {
                 _numActive--;
-                notifyAll(); // _numActive has changed
+                allocate();
             }
         }
     }
@@ -1068,7 +1103,7 @@
             } finally {
                 synchronized(this) {
                     _numInternalProcessing--;
-                    notifyAll();
+                    allocate();
                 }
             }
         }
@@ -1117,7 +1152,7 @@
                 // "behavior flag",decrementNumActive, from addObjectToPool.
                 synchronized(this) {
                     _numActive--;
-                    notifyAll();
+                    allocate();
                 }
             }
         }
@@ -1152,7 +1187,7 @@
                     if (decrementNumActive) {
                         _numActive--;
                     }
-                    notifyAll();
+                    allocate();
                 }
             }
         }
@@ -1168,7 +1203,7 @@
             if (decrementNumActive) {
                 synchronized(this) {
                     _numActive--;
-                    notifyAll();
+                    allocate();
                 }
             }
         }
@@ -1319,7 +1354,7 @@
             } finally {
                 synchronized (this) {
                     _numInternalProcessing--;
-                    notifyAll();
+                    allocate();
                 }
             }
         }
@@ -1490,6 +1525,26 @@
     
     }
 
+    /**
+     * Latch used to control allocation order of objects to threads to ensure
+     * fairness. ie objects are allocated to threads in the order that threads
+     * request objects.
+     */
+    private static final class Latch {
+        ObjectTimestampPair _pair;
+        boolean _mayCreate = false;
+        
+        /**
+         * Reset the latch data. Used when an allocation fails and the latch
+         * needs to be re-added to the queue. 
+         */
+        private void reset() {
+            _pair = null;
+            _mayCreate = false;
+        }
+    }
+
+
     //--- private attributes ---------------------------------------
 
     /**
@@ -1664,4 +1719,12 @@
      * number of objects but are neither active nor idle.
      */
     private int _numInternalProcessing = 0;
+
+    /**
+     * Used to track the order in which threads call {@link #borrowObject()} so
+     * that objects can be allocated in the order in which the threads requested
+     * them.
+     */
+    private LinkedList _allocationQueue = new LinkedList();
+
 }



Mime
View raw message