added a consistent hash implementation. Not ready for live action yet

git-svn-id: https://hudson.dev.java.net/svn/hudson/trunk/hudson/main@17476 71c3de6d-444a-0410-be80-ed276b4c234a

core/src/main/java/hudson/model/Hudson.java

@@ -495,7 +495,7 @@ public final class Hudson extends Node implements ItemGroup<TopLevelItem>, Stapl
             log.load();
 
             Trigger.timer = new Timer("Hudson cron thread");
-            queue = new Queue(LoadBalancer.DEFAULT); // TODO: make this somehow pluggable
+            queue = new Queue(CONSISTENT_HASH?LoadBalancer.CONSISTENT_HASH:LoadBalancer.DEFAULT);
 
             try {
                 dependencyGraph = DependencyGraph.EMPTY;
@@ -3265,6 +3265,7 @@ public final class Hudson extends Node implements ItemGroup<TopLevelItem>, Stapl
     public static boolean PARALLEL_LOAD = !"false".equals(System.getProperty(Hudson.class.getName()+".parallelLoad"));
     public static boolean KILL_AFTER_LOAD = Boolean.getBoolean(Hudson.class.getName()+".killAfterLoad");
     public static boolean LOG_STARTUP_PERFORMANCE = Boolean.getBoolean(Hudson.class.getName()+".logStartupPerformance");
+    private static final boolean CONSISTENT_HASH = Boolean.getBoolean(Hudson.class.getName()+".consistentHash");
 
     private static final Logger LOGGER = Logger.getLogger(Hudson.class.getName());
 

core/src/main/java/hudson/model/LoadBalancer.java

 package hudson.model;
 
-import hudson.model.Queue.Task;
-import hudson.model.Queue.JobOffer;
 import hudson.model.Node.Mode;
+import hudson.model.Queue.AvailableJobOfferList;
+import hudson.model.Queue.JobOffer;
+import hudson.model.Queue.Task;
+import hudson.util.ConsistentHash;
+import hudson.util.ConsistentHash.Hash;
 
-import java.util.Collection;
 import java.util.logging.Logger;
 
 /**
  * Strategy that decides which {@link Task} gets run on which {@link Executor}.
  *
  * @author Kohsuke Kawaguchi
- * @since 1.301 (but this is not yet open for plugins to change)
+ * @since 1.301
  */
 public abstract class LoadBalancer /*implements ExtensionPoint*/ {
     /**
@@ -35,13 +37,13 @@ public abstract class LoadBalancer /*implements ExtensionPoint*/ {
      *      the task to be executed right now, in which case this method will be called
      *      some time later with the same task.
      */
-    protected abstract JobOffer choose(Task task, Collection<JobOffer> available);
+    protected abstract JobOffer choose(Task task, AvailableJobOfferList available);
 
     /**
      * Traditional implementation of this.
      */
     public static final LoadBalancer DEFAULT = new LoadBalancer() {
-        protected JobOffer choose(Task task, Collection<JobOffer> available) {
+        protected JobOffer choose(Task task, AvailableJobOfferList available) {
             Label l = task.getAssignedLabel();
             if (l != null) {
                 // if a project has assigned label, it can be only built on it
@@ -94,6 +96,34 @@ public abstract class LoadBalancer /*implements ExtensionPoint*/ {
         }
     };
 
+    /**
+     * Work in progress implementation that uses a consistent hash for scheduling.
+     */
+    public static final LoadBalancer CONSISTENT_HASH = new LoadBalancer() {
+        protected JobOffer choose(Task task, AvailableJobOfferList available) {
+            // populate a consistent hash linear to the # of executors
+            // TODO: there's a lot of room for innovations here
+            // TODO: do this upfront and reuse
+            ConsistentHash<Node> hash = new ConsistentHash<Node>(new Hash<Node>() {
+                public String hash(Node node) {
+                    return node.getNodeName();
+                }
+            });
+            for (Node n : Hudson.getInstance().getNodes())
+                hash.add(n,n.getNumExecutors()*100);
+
+            // TODO: add some salt as a query point so that the user can tell Hudson to hop the project to a new node
+            for(Node n : hash.list(task.getFullDisplayName())) {
+                JobOffer o = available._for(n);
+                if(o!=null)
+                    return o;
+            }
+
+            // nothing available
+            return null;
+        }
+    };
+
     /**
      * Wraps this {@link LoadBalancer} into a decorator that tests the basic sanity of the implementation.
      * Only override this if you find some of the checks excessive, but beware that it's like driving without a seat belt.
@@ -102,7 +132,7 @@ public abstract class LoadBalancer /*implements ExtensionPoint*/ {
         final LoadBalancer base = this;
         return new LoadBalancer() {
             @Override
-            protected JobOffer choose(Task task, Collection<JobOffer> available) {
+            protected JobOffer choose(Task task, AvailableJobOfferList available) {
                 if (Hudson.getInstance().isQuietingDown()) {
                     // if we are quieting down, don't start anything new so that
                     // all executors will be eventually free.
@@ -111,9 +141,8 @@ public abstract class LoadBalancer /*implements ExtensionPoint*/ {
 
                 JobOffer target = base.choose(task, available);
                 if(target!=null) {
-                    Label l = task.getAssignedLabel();
-                    if(l!=null && !l.contains(target.getNode())) {
-                        LOGGER.severe(task.getFullDisplayName()+" needs to be assigned to a node that has a label "+l);
+                    if(!target.canTake(task)) {
+                        LOGGER.severe(target.getNode().getDisplayName()+" cannot take "+task.getFullDisplayName());
                         return null;
                     }
                     // Queue should guarantee that, but just in case.

core/src/main/java/hudson/model/Queue.java

@@ -150,6 +150,23 @@ public class Queue extends ResourceController implements Saveable {
             event.signal();
         }
 
+        /**
+         * Verifies that the {@link Executor} represented by this object is capable of executing the given task.
+         */
+        public boolean canTake(Task task) {
+            Label l = task.getAssignedLabel();
+            if(l!=null && !l.contains(getNode()))
+                return false;   // the task needs to be executed on label that this node doesn't have.
+
+            if(l==null && getNode().getMode()== Mode.EXCLUSIVE)
+                return false;   // this node is reserved for tasks that are tied to it
+
+            return isAvailable();
+        }
+
+        /**
+         * Is this executor ready to accept some tasks?
+         */
         public boolean isAvailable() {
             return item == null && !executor.getOwner().isOffline() && executor.getOwner().isAcceptingTasks();
         }
@@ -168,7 +185,7 @@ public class Queue extends ResourceController implements Saveable {
      */
     private final Map<Executor,JobOffer> parked = new HashMap<Executor,JobOffer>();
 
-    private final LoadBalancer loadBalancer;
+    private volatile transient LoadBalancer loadBalancer;
 
     public Queue(LoadBalancer loadBalancer) {
         this.loadBalancer =  loadBalancer.sanitize();
@@ -177,6 +194,15 @@ public class Queue extends ResourceController implements Saveable {
         new MaintainTask(this);
     }
 
+    public LoadBalancer getLoadBalancer() {
+        return loadBalancer;
+    }
+
+    public void setLoadBalancer(LoadBalancer loadBalancer) {
+        if(loadBalancer==null)  throw new IllegalArgumentException();
+        this.loadBalancer = loadBalancer;
+    }
+
     /**
      * Loads the queue contents that was {@link #save() saved}.
      */
@@ -572,7 +598,7 @@ public class Queue extends ResourceController implements Saveable {
                             continue;
                         }
 
-                        JobOffer runner = loadBalancer.choose(p.task, getAvailableJobOffers());
+                        JobOffer runner = loadBalancer.choose(p.task, new AvailableJobOfferList());
                         if (runner == null)
                             // if we couldn't find the executor that fits,
                             // just leave it in the buildables list and
@@ -641,14 +667,60 @@ public class Queue extends ResourceController implements Saveable {
     }
 
     /**
-     * Creates a sublist of {@link #parked} that only consists of available offers.
+     * Represents a list of {@linkplain JobOffer#isAvailable() available} {@link JobOffer}s
+     * and provides various typical 
      */
-    private List<JobOffer> getAvailableJobOffers() {
-        List<JobOffer> availables = new ArrayList<JobOffer>(parked.size());
-        for (JobOffer j : parked.values())
-            if(j.isAvailable())
-                availables.add(j);
-        return availables;
+    public final class AvailableJobOfferList implements Iterable<JobOffer> {
+        private final List<JobOffer> list;
+        // laziy filled
+        private Map<Node,List<JobOffer>> nodes;
+
+        private AvailableJobOfferList() {
+            list = new ArrayList<JobOffer>(parked.size());
+            for (JobOffer j : parked.values())
+                if(j.isAvailable())
+                    list.add(j);
+        }
+
+        /**
+         * Returns all the {@linkplain JobOffer#isAvailable() available} {@link JobOffer}s.
+         */
+        public List<JobOffer> all() {
+            return list;
+        }
+
+        public Iterator<JobOffer> iterator() {
+            return list.iterator();
+        }
+
+        /**
+         * List up all the {@link Node}s that have some available offers.
+         */
+        public Set<Node> nodes() {
+            return byNodes().keySet();
+        }
+
+        /**
+         * Gets a {@link JobOffer} for an executor of the given node, if any.
+         * Otherwise null. 
+         */
+        public JobOffer _for(Node n) {
+            List<JobOffer> r = byNodes().get(n);
+            if(r==null) return null;
+            return r.get(0);
+        }
+
+        public Map<Node,List<JobOffer>> byNodes() {
+            if(nodes==null) {
+                nodes = new HashMap<Node,List<JobOffer>>();
+                for (JobOffer o : list) {
+                    List<JobOffer> l = nodes.get(o.getNode());
+                    if(l==null) nodes.put(o.getNode(),l=new ArrayList<JobOffer>());
+                    l.add(o);
+                }
+            }
+            return nodes;
+        }
     }
 
     /**