svn commit: r950254 - in /ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent: DependencyPool.java test/DependencyPoolTests.java

[doogie-3]

Author: doogie
Date: Tue Jun  1 21:46:19 2010
New Revision: 950254

URL: http://svn.apache.org/viewvc?rev=950254&view=rev
Log:
Start work in provides in inflight features.

Modified:
    ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/DependencyPool.java
    ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/test/DependencyPoolTests.java

Modified: ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/DependencyPool.java
URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/DependencyPool.java?rev=950254&r1=950253&r2=950254&view=diff
==============================================================================
--- ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/DependencyPool.java (original)
+++ ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/DependencyPool.java Tue Jun  1 21:46:19 2010
@@ -10,6 +10,7 @@ import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
@@ -26,15 +27,18 @@ public class DependencyPool<K, I extends
     private final Executor executor;
     private final ConcurrentMap<K, I> allItems = new ConcurrentHashMap<K, I>();
     private final ConcurrentMap<K, Future<V>> results = new ConcurrentHashMap<K, Future<V>>();
+    private final Set<K> provides = new ConcurrentSkipListSet<K>();
     private final ReentrantLock submitLock = new ReentrantLock();
     private final Condition submitCondition = submitLock.newCondition();
+    private final int inflight;
     @LockedBy("submitLock")
     private final Set<I> outstanding = new HashSet<I>();
     @LockedBy("submitLock")
     private final List<I> pending = new LinkedList<I>();
 
-    public DependencyPool(Executor executor) {
+    public DependencyPool(Executor executor, int inflight) {
         this.executor = executor;
+        this.inflight = inflight;
     }
 
     public I add(I item) {
@@ -98,7 +102,7 @@ OUTER:
         while (pendingIt.hasNext()) {
             I item = pendingIt.next();
             for (K dep: item.getDependencies()) {
-                if (!results.containsKey(dep)) {
+                if (!results.containsKey(dep) && !provides.contains(dep)) {
                     continue OUTER;
                 }
             }
@@ -121,10 +125,11 @@ OUTER:
         protected void done() {
             super.done();
             results.put(item.getKey(), this);
+            provides.addAll(item.getProvides());
             submitLock.lock();
             try {
                 outstanding.remove(item);
-                if (submitWork() == 0 && outstanding.isEmpty()) {
+                if (outstanding.size() < inflight && submitWork() == 0 && outstanding.isEmpty()) {
                     submitCondition.signal();
                 }
             } finally {
@@ -136,5 +141,6 @@ OUTER:
     public interface Item<I extends Item<I, K, V>, K, V> extends Callable<V> {
         K getKey();
         Collection<K> getDependencies();
+        Collection<K> getProvides();
     }
 }

Modified: ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/test/DependencyPoolTests.java
URL: http://svn.apache.org/viewvc/ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/test/DependencyPoolTests.java?rev=950254&r1=950253&r2=950254&view=diff
==============================================================================
--- ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/test/DependencyPoolTests.java (original)
+++ ofbiz/trunk/framework/base/src/org/ofbiz/base/concurrent/test/DependencyPoolTests.java Tue Jun  1 21:46:19 2010
@@ -23,7 +23,7 @@ public class DependencyPoolTests extends
     public void testDependencyPool() throws Exception {
         // always use more threads than cpus, so that the single-cpu case can be tested
         ScheduledExecutorService executor = ExecutionPool.getNewOptimalExecutor(getName());
-        DependencyPool pool = new DependencyPool(executor);
+        DependencyPool pool = new DependencyPool(executor, 5);
         int itemSize = 100, depMax = 5, subMax = 3;
         List<TestItem> items = new ArrayList<TestItem>(itemSize);
         List<TestItem> previousItems = new ArrayList<TestItem>(itemSize);
@@ -74,6 +74,7 @@ OUTER:
         private final Integer key;
         private final String result;
         private final Collection<Integer> dependencies;
+        private final Collection<Integer> provides;
         private final Collection<TestItem> subItems;
 
         protected TestItem(DependencyPool pool, Integer key, String result, Collection<Integer> dependencies, Collection<TestItem> subItems) {
@@ -81,6 +82,7 @@ OUTER:
             this.key = key;
             this.result = result;
             this.dependencies = dependencies;
+            this.provides = java.util.Collections.emptyList();
             this.subItems = subItems;
         }
 
@@ -92,6 +94,10 @@ OUTER:
             return dependencies;
         }
 
+        public Collection<Integer> getProvides() {
+            return provides;
+        }
+
         public Collection<TestItem> getSubItems() {
             return subItems;
         }