diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/SimpleCapacityReplanner.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/SimpleCapacityReplanner.java new file mode 100644 index 0000000..67fb712 --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/SimpleCapacityReplanner.java @@ -0,0 +1,95 @@ +package org.apache.hadoop.yarn.server.resourcemanager.reservation; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.hadoop.yarn.api.records.ReservationDefinition; +import org.apache.hadoop.yarn.api.records.Resource; +import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.PlanQueue; +import org.apache.hadoop.yarn.util.Clock; +import org.apache.hadoop.yarn.util.UTCClock; +import org.apache.hadoop.yarn.util.resource.Resources; + +/** + * This (re)planner scan a period of time from now to a maximum time window (or + * the end of the last session, whichever comes first) checking the overall + * capacity is not violated. + * + * It greedily removes sessions in reversed order of acceptance (latest accepted + * is the first removed). + */ +public class SimpleCapacityReplanner implements Planner { + + private static final Log LOG = LogFactory + .getLog(SimpleCapacityReplanner.class); + + Clock clock; + + // this allows to control to time-span of this replanning + // far into the future time instants might be worth replanning for + // later on + private long lengthOfCheckZone; + + @Override + public void init(String planQueueName, CapacitySchedulerConfiguration conf) { + this.clock = new UTCClock(); + this.lengthOfCheckZone = + ((CapacitySchedulerConfiguration) conf) + .getEnforcementWindow(planQueueName + + PlanQueue.DEFAULT_QUEUE_SUFFIX); + + } + + @Override + public void plan(Plan plan, List contracts) + throws PlanningException { + + if (contracts != null) { + throw new RuntimeException("InventorySizeAdjusmentPolicy " + + "cannot handle new contracts"); + } + + Resource totCap = plan.getTotalCapacity(); + long now = clock.getTime(); + + // loop on all moment in time from now to the end of the check Zone + // or the end of the planned sessions whichever comes first + for (long t = now; (t < plan.getLastEndTime() && t < (now + lengthOfCheckZone)); t += + plan.getStep()) { + Resource excessCap = + Resources.subtract(plan.getTotalCommittedResources(t), totCap); + // if we are violating + if (Resources.greaterThan(plan.getResourceCalculator(), + plan.getTotalCapacity(), excessCap, Resource.newInstance(0, 0))) { + + ArrayList curSessions = + new ArrayList(plan.getReservationsAtTime(t)); + if (curSessions != null && curSessions.size() > 0) { + Collections.sort(curSessions); + + for (ReservationAllocation s : curSessions) { + // remove sessions (in whichever order they are listed) + plan.deleteReservation(s.getReservationId()); + excessCap = + Resources.subtract(plan.getTotalCommittedResources(t), totCap); + LOG.info("Removing reservation " + s.getReservationId() + + " to repair physical-resource constraints in the plan: " + + plan.getQueueName()); + // till we don't violate anymore + if (Resources.lessThan(plan.getResourceCalculator(), plan + .getTotalCapacity(), Resources.subtract(excessCap, + ((InMemoryReservationAllocation) s).getResourcesAtTime(t)), + Resource.newInstance(0, 0))) { + break; + } + } + } + } + } + } +} diff --git hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSimpleCapacityReplanner.java hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSimpleCapacityReplanner.java new file mode 100644 index 0000000..82551df --- /dev/null +++ hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/reservation/TestSimpleCapacityReplanner.java @@ -0,0 +1,153 @@ +package org.apache.hadoop.yarn.server.resourcemanager.reservation; + +import static org.junit.Assert.assertTrue; +import static org.mockito.Matchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.HashSet; +import java.util.Map; +import java.util.TreeMap; + +import org.apache.hadoop.yarn.api.records.ReservationId; +import org.apache.hadoop.yarn.api.records.ReservationRequest; +import org.apache.hadoop.yarn.api.records.Resource; +import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics; +import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration; +import org.apache.hadoop.yarn.util.Clock; +import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator; +import org.apache.hadoop.yarn.util.resource.ResourceCalculator; +import org.junit.Test; + +public class TestSimpleCapacityReplanner { + + @Test + public void test() throws InterruptedException { + + Resource clusterCapacity = Resource.newInstance(100 * 1024, 10); + Resource minAlloc = Resource.newInstance(1024, 1); + Resource maxAlloc = Resource.newInstance(1024 * 8, 8); + + ResourceCalculator res = new DefaultResourceCalculator(); + long step = 1L; + Clock clock = mock(Clock.class); + ReservationAgent agent = mock(ReservationAgent.class); + + SharingPolicy policy = new NoOverCommitPolicy(); + policy.init("root.dedicated", null, new HashSet()); + + QueueMetrics queueMetrics = mock(QueueMetrics.class); + + when(clock.getTime()).thenReturn(0L); + SimpleCapacityReplanner enf = new SimpleCapacityReplanner(); + + CapacitySchedulerConfiguration conf = + mock(CapacitySchedulerConfiguration.class); + when(conf.getEnforcementWindow(any(String.class))).thenReturn(6L); + + conf.setLong(CapacitySchedulerConfiguration.PREFIX + "blah" + + CapacitySchedulerConfiguration.DOT + + CapacitySchedulerConfiguration.RESERVATION_ENFORCEMENT_WINDOW, 6); + enf.init("blah", conf); + + // Initialize the plan with more resources + InMemoryPlan plan = + new InMemoryPlan(queueMetrics, policy, agent, clusterCapacity, step, + res, minAlloc, maxAlloc, "dedicated", enf, true); + try { + + // add reservation filling the plan (separating them 1ms, so we are sure + // s2 follows s1 on acceptance + long ts = System.currentTimeMillis(); + ReservationId r1 = ReservationId.newInstance(ts, 1); + int[] f5 = { 20, 20, 20, 20, 20 }; + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r1, null, "u3", + "dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res, + minAlloc))); + Thread.sleep(1); + ReservationId r2 = ReservationId.newInstance(ts, 2); + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r2, null, "u4", + "dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res, + minAlloc))); + Thread.sleep(1); + ReservationId r3 = ReservationId.newInstance(ts, 3); + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r3, null, "u5", + "dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res, + minAlloc))); + Thread.sleep(1); + ReservationId r4 = ReservationId.newInstance(ts, 4); + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r4, null, "u6", + "dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res, + minAlloc))); + Thread.sleep(1); + ReservationId r5 = ReservationId.newInstance(ts, 5); + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r5, null, "u7", + "dedicated", 0, 0 + f5.length, generateAllocation(0, f5), res, + minAlloc))); + + int[] f6 = { 50, 50, 50, 50, 50 }; + ReservationId r6 = ReservationId.newInstance(ts, 6); + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r6, null, "u3", + "dedicated", 10, 10 + f6.length, generateAllocation(10, f6), res, + minAlloc))); + Thread.sleep(1); + ReservationId r7 = ReservationId.newInstance(ts, 7); + assertTrue(plan.toString(), + plan.addReservation(new InMemoryReservationAllocation(r7, null, "u4", + "dedicated", 10, 10 + f6.length, generateAllocation(10, f6), res, + minAlloc))); + + // remove some of the resources (requires replanning) + plan.setTotalCapacity(Resource.newInstance(70 * 1024, 70)); + + enf.clock = clock; + + // run the replanner + enf.plan(plan, null); + + // check which reservation are still present + assertTrue(plan.getReservationById(r1) != null); + assertTrue(plan.getReservationById(r2) != null); + assertTrue(plan.getReservationById(r3) != null); + assertTrue(plan.getReservationById(r6) != null); + assertTrue(plan.getReservationById(r7) != null); + + // and which ones are removed + assertTrue(plan.getReservationById(r4) == null); + assertTrue(plan.getReservationById(r5) == null); + + // check resources at each moment in time no more exceed capacity + for (int i = 0; i < 20; i++) { + int tot = 0; + for (ReservationAllocation r : plan.getReservationsAtTime(i)) { + tot = r.getResourcesAtTime(i).getMemory(); + } + assertTrue(tot <= 70 * 1024); + } + + } catch (PlanningException e) { + e.printStackTrace(); + } + + } + + Map generateAllocation( + int startTime, int[] alloc) { + Map req = + new TreeMap(); + for (int i = 0; i < alloc.length; i++) { + req.put(new ReservationInterval(startTime + i, startTime + i + 1), + ReservationRequest.newInstance(Resource.newInstance(1024, 1), + alloc[i])); + } + return req; + } + +}