myriad-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mo...@apache.org
Subject [15/50] [abbrv] incubator-myriad git commit: Rebased and added changes suggested by @pdread100.
Date Fri, 25 Sep 2015 22:45:55 GMT
Rebased and added changes suggested by @pdread100.

Tested again on 4 node cluster all appears well


Project: http://git-wip-us.apache.org/repos/asf/incubator-myriad/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-myriad/commit/e60bbaa7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-myriad/tree/e60bbaa7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-myriad/diff/e60bbaa7

Branch: refs/heads/master
Commit: e60bbaa7af4dc619e7b9bb8b6d99d95a677ba357
Parents: bfbefc3
Author: DarinJ <no@thanks.com>
Authored: Fri Jul 10 01:00:12 2015 -0400
Committer: DarinJ <no@thanks.com>
Committed: Fri Jul 10 01:19:05 2015 -0400

----------------------------------------------------------------------
 .../handlers/ResourceOffersEventHandler.java    | 113 ++++++++++++-------
 1 file changed, 74 insertions(+), 39 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-myriad/blob/e60bbaa7/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
----------------------------------------------------------------------
diff --git a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
index dae3a03..c009ce8 100644
--- a/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
+++ b/myriad-scheduler/src/main/java/com/ebay/myriad/scheduler/event/handlers/ResourceOffersEventHandler.java
@@ -80,6 +80,7 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
                         NodeTask taskToLaunch = schedulerState
                                 .getTask(pendingTaskId);
                         NMProfile profile = taskToLaunch.getProfile();
+
                         if (matches(offer, profile)
                                 && SchedulerUtils.isUniqueHostname(offer,
                                 schedulerState.getActiveTasks())) {
@@ -124,54 +125,34 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
     }
 
     private boolean matches(Offer offer, NMProfile profile) {
-        double cpus = -1;
-        double mem = -1;
-        int ports = 0;
+        Map<String, Object> results = new HashMap<String, Object>(5);
+
         for (Resource resource : offer.getResourcesList()) {
-            if (resource.getName().equals("cpus")) {
-                if (resource.getType().equals(Value.Type.SCALAR)) {
-                    cpus = resource.getScalar().getValue();
-                } else {
-                    LOGGER.error("Cpus resource was not a scalar: {}", resource
-                            .getType().toString());
-                }
-            } else if (resource.getName().equals("mem")) {
-                if (resource.getType().equals(Value.Type.SCALAR)) {
-                    mem = resource.getScalar().getValue();
-                } else {
-                    LOGGER.error("Mem resource was not a scalar: {}", resource
-                            .getType().toString());
-                }
-            } else if (resource.getName().equals("disk")) {
-                LOGGER.warn("Ignoring disk resources from offer");
-            } else if (resource.getName().equals("ports")) {
-                if (resource.getType().equals(Value.Type.RANGES)) {
-                    Value.Ranges ranges =  resource.getRanges();
-                    for (Value.Range range : ranges.getRangeList()) {
-                        if (range.getBegin() < range.getEnd()) {
-                            ports += range.getEnd() - range.getBegin() + 1;
-                        }
-                    }
-                } else {
-                    LOGGER.error("ports resource was not Ranges: {}", resource
-                            .getType().toString());
-                }
+            if (resourceEvaluators.containsKey(resource.getName())) {
+                resourceEvaluators.get(resource.getName()).eval(resource, results);
             } else {
                 LOGGER.warn("Ignoring unknown resource type: {}",
                         resource.getName());
             }
         }
+        double cpus = (Double) results.get("cpus");
+        double mem = (Double) results.get("mem");
+        int ports = (Integer) results.get("ports");
 
-        if (cpus < 0) {
-            LOGGER.error("No cpus resource present");
-        }
-        if (mem < 0) {
-            LOGGER.error("No mem resource present");
-        }
-        if (ports < 0) {
-            LOGGER.error("No port resources present");
+        checkResource(cpus < 0, "cpus");
+        checkResource(mem < 0, "mem");
+        checkResource(ports < 0, "port");
+
+        return checkAggregates(offer, profile, ports, cpus, mem);
+    }
+
+    private void checkResource(boolean fail, String resource) {
+        if (fail) {
+            LOGGER.info("No " + resource + " resources present");
         }
+    }
 
+    private boolean checkAggregates(Offer offer, NMProfile profile, int ports, double cpus,
double mem) {
         Map<String, String> requestAttributes = new HashMap<>();
 
         if (taskUtils.getAggregateCpus(profile) <= cpus
@@ -185,4 +166,58 @@ public class ResourceOffersEventHandler implements EventHandler<ResourceOffersEv
             return false;
         }
     }
+
+    private static Double scalarToDouble(Resource resource, String id) {
+        Double value = new Double(0.0);
+        if (resource.getType().equals(Value.Type.SCALAR)) {
+            value = new Double(resource.getScalar().getValue());
+        } else {
+            LOGGER.error(id + " resource was not a scalar: {}", resource
+                    .getType().toString());
+        }
+        return value;
+    }
+
+    private interface EvalResources {
+        public void eval(Resource resource, Map<String, Object>results);
+    }
+
+    private static Map<String, EvalResources> resourceEvaluators;
+
+    static {
+        resourceEvaluators = new HashMap<String, EvalResources>(4);
+        resourceEvaluators.put("cpus", new EvalResources() {
+            public void eval(Resource resource, Map<String, Object> results) {
+                results.put("cpus", scalarToDouble(resource, "cpus"));
+            }
+        });
+        resourceEvaluators.put("mem", new EvalResources() {
+            public void eval(Resource resource, Map<String, Object> results) {
+                results.put("mem", scalarToDouble(resource, "mem"));
+            }
+        });
+        resourceEvaluators.put("disk", new EvalResources() {
+            public void eval(Resource resource, Map<String, Object> results) {
+            }
+        });
+        resourceEvaluators.put("ports", new EvalResources() {
+            public void eval(Resource resource, Map<String, Object> results) {
+                int ports = 0;
+                if (resource.getType().equals(Value.Type.RANGES)) {
+                    Value.Ranges ranges = resource.getRanges();
+                    for (Value.Range range : ranges.getRangeList()) {
+                        if (range.getBegin() < range.getEnd()) {
+                            ports += range.getEnd() - range.getBegin() + 1;
+                        }
+                    }
+
+                } else {
+                    LOGGER.error("ports resource was not Ranges: {}", resource
+                            .getType().toString());
+
+                }
+                results.put("ports", Integer.valueOf(ports));
+            }
+        });
+    }
 }


Mime
View raw message