YARN-5329. Placement Agent enhancements required to support recurring reservations in ReservationSystem. (Carlo Curino via Subru).

(cherry picked from commit e6e614e380ed1d746973b50f666a9c40d272073e)
This commit is contained in:
Subru Krishnan 2017-10-04 19:28:27 -07:00
parent 7836a6b59a
commit 7fd4a997f4
15 changed files with 443 additions and 128 deletions

View File

@ -723,6 +723,12 @@ public RLESparseResourceAllocation getAvailableResourceOverTime(String user,
+ periodicRle.getTimePeriod() + ")"); + periodicRle.getTimePeriod() + ")");
} }
if (period < (end - start)) {
throw new PlanningException(
"Invalid input: (end - start) = (" + end + " - " + start + ") = "
+ (end - start) + " > period = " + period);
}
// find the minimum resources available among all the instances that fit // find the minimum resources available among all the instances that fit
// in the LCM // in the LCM
long numInstInLCM = periodicRle.getTimePeriod() / period; long numInstInLCM = periodicRle.getTimePeriod() / period;

View File

@ -221,7 +221,8 @@ public RLESparseResourceAllocation getRangeOverlapping(long start, long end) {
NavigableMap<Long, Resource> cumulativeMap = this.getCumulative(); NavigableMap<Long, Resource> cumulativeMap = this.getCumulative();
Long previous = cumulativeMap.floorKey(relativeStart); Long previous = cumulativeMap.floorKey(relativeStart);
previous = (previous != null) ? previous : 0; previous = (previous != null) ? previous : 0;
for (long i = 0; i <= (end - start) / timePeriod; i++) { //make sure to go one past end, to catch end times extending past period
for (long i = 0; i <= 1 + (end - start) / timePeriod; i++) {
for (Map.Entry<Long, Resource> e : cumulativeMap.entrySet()) { for (Map.Entry<Long, Resource> e : cumulativeMap.entrySet()) {
long curKey = e.getKey() + (i * timePeriod); long curKey = e.getKey() + (i * timePeriod);
if (curKey >= previous && (start + curKey - relativeStart) <= end) { if (curKey >= previous && (start + curKey - relativeStart) <= end) {

View File

@ -423,7 +423,8 @@ private static void addIfNeeded(TreeMap<Long, Resource> out, long time,
Resource outRes) { Resource outRes) {
if (out.isEmpty() || (out.lastEntry() != null && outRes == null) if (out.isEmpty() || (out.lastEntry() != null && outRes == null)
|| !Resources.equals(out.lastEntry().getValue(), outRes)) { || (out.lastEntry().getValue() != null
&& !Resources.equals(out.lastEntry().getValue(), outRes))) {
out.put(time, outRes); out.put(time, outRes);
} }
@ -460,7 +461,8 @@ private static Resource combineValue(RLEOperator op,
if (!Resources.fitsIn(b, a)) { if (!Resources.fitsIn(b, a)) {
throw new PlanningException( throw new PlanningException(
"RLESparseResourceAllocation: merge failed as the " "RLESparseResourceAllocation: merge failed as the "
+ "resulting RLESparseResourceAllocation would be negative"); + "resulting RLESparseResourceAllocation would "
+ "be negative, when testing: (" + eB + ") > (" + eA + ")");
} else { } else {
return Resources.subtract(a, b); return Resources.subtract(a, b);
} }

View File

@ -98,6 +98,12 @@ public RLESparseResourceAllocation computeJobAllocation(Plan plan,
// Current stage // Current stage
ReservationRequest currentReservationStage; ReservationRequest currentReservationStage;
// initialize periodicity
long period = 0;
if(reservation.getRecurrenceExpression() != null){
period = Long.parseLong(reservation.getRecurrenceExpression());
}
// Iterate the stages in reverse order // Iterate the stages in reverse order
while (stageProvider.hasNext()) { while (stageProvider.hasNext()) {
@ -117,7 +123,7 @@ public RLESparseResourceAllocation computeJobAllocation(Plan plan,
// Compute stage allocation // Compute stage allocation
Map<ReservationInterval, Resource> curAlloc = Map<ReservationInterval, Resource> curAlloc =
computeStageAllocation(plan, currentReservationStage, stageArrival, computeStageAllocation(plan, currentReservationStage, stageArrival,
stageDeadline, user, reservationId); stageDeadline, period, user, reservationId);
// If we did not find an allocation, return NULL // If we did not find an allocation, return NULL
// (unless it's an ANY job, then we simply continue). // (unless it's an ANY job, then we simply continue).
@ -216,11 +222,10 @@ protected void initialize(Plan plan, ReservationId reservationId,
planLoads = plan.getCumulativeLoadOverTime(jobArrival, jobDeadline); planLoads = plan.getCumulativeLoadOverTime(jobArrival, jobDeadline);
ReservationAllocation oldRes = plan.getReservationById(reservationId); ReservationAllocation oldRes = plan.getReservationById(reservationId);
if (oldRes != null) { if (oldRes != null) {
planLoads = planLoads = RLESparseResourceAllocation.merge(
RLESparseResourceAllocation.merge(plan.getResourceCalculator(), plan.getResourceCalculator(), plan.getTotalCapacity(), planLoads,
plan.getTotalCapacity(), planLoads, oldRes.getResourcesOverTime(jobArrival, jobDeadline),
oldRes.getResourcesOverTime(), RLEOperator.subtract, jobArrival, RLEOperator.subtract, jobArrival, jobDeadline);
jobDeadline);
} }
} }
@ -309,13 +314,13 @@ protected ReservationInterval setStageExecutionInterval(Plan plan,
} }
// Call algStageAllocator // Call algStageAllocator
protected Map<ReservationInterval, Resource> computeStageAllocation( protected Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
Plan plan, ReservationRequest rr, long stageArrivalTime, ReservationRequest rr, long stageArrivalTime, long stageDeadline,
long stageDeadline, String user, ReservationId oldId) long period, String user, ReservationId oldId) throws PlanningException {
throws PlanningException {
return algStageAllocator.computeStageAllocation(plan, planLoads, return algStageAllocator.computeStageAllocation(plan, planLoads,
planModifications, rr, stageArrivalTime, stageDeadline, user, oldId); planModifications, rr, stageArrivalTime, stageDeadline, period, user,
oldId);
} }

View File

@ -26,6 +26,7 @@
import org.apache.hadoop.yarn.api.records.ReservationId; import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.Resource; import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.InMemoryReservationAllocation; import org.apache.hadoop.yarn.server.resourcemanager.reservation.InMemoryReservationAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.PeriodicRLESparseResourceAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.Plan; import org.apache.hadoop.yarn.server.resourcemanager.reservation.Plan;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.RLESparseResourceAllocation; import org.apache.hadoop.yarn.server.resourcemanager.reservation.RLESparseResourceAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationAllocation; import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationAllocation;
@ -65,6 +66,16 @@ protected boolean allocateUser(ReservationId reservationId, String user,
RLESparseResourceAllocation allocation = RLESparseResourceAllocation allocation =
computeJobAllocation(plan, reservationId, adjustedContract, user); computeJobAllocation(plan, reservationId, adjustedContract, user);
long period = Long.parseLong(contract.getRecurrenceExpression());
// Make allocation periodic if request is periodic
if (contract.getRecurrenceExpression() != null) {
if (period > 0) {
allocation =
new PeriodicRLESparseResourceAllocation(allocation, period);
}
}
// If no job allocation was found, fail // If no job allocation was found, fail
if (allocation == null) { if (allocation == null) {
throw new PlanningException( throw new PlanningException(
@ -74,10 +85,12 @@ protected boolean allocateUser(ReservationId reservationId, String user,
// Translate the allocation to a map (with zero paddings) // Translate the allocation to a map (with zero paddings)
long step = plan.getStep(); long step = plan.getStep();
long jobArrival = stepRoundUp(adjustedContract.getArrival(), step); long jobArrival = stepRoundUp(adjustedContract.getArrival(), step);
long jobDeadline = stepRoundUp(adjustedContract.getDeadline(), step); long jobDeadline = stepRoundUp(adjustedContract.getDeadline(), step);
Map<ReservationInterval, Resource> mapAllocations = Map<ReservationInterval, Resource> mapAllocations =
allocationsToPaddedMap(allocation, jobArrival, jobDeadline); allocationsToPaddedMap(allocation, jobArrival, jobDeadline, period);
// Create the reservation // Create the reservation
ReservationAllocation capReservation = ReservationAllocation capReservation =
@ -85,8 +98,7 @@ protected boolean allocateUser(ReservationId reservationId, String user,
adjustedContract, // Contract adjustedContract, // Contract
user, // User name user, // User name
plan.getQueueName(), // Queue name plan.getQueueName(), // Queue name
findEarliestTime(mapAllocations), // Earliest start time adjustedContract.getArrival(), adjustedContract.getDeadline(),
findLatestTime(mapAllocations), // Latest end time
mapAllocations, // Allocations mapAllocations, // Allocations
plan.getResourceCalculator(), // Resource calculator plan.getResourceCalculator(), // Resource calculator
plan.getMinimumAllocation()); // Minimum allocation plan.getMinimumAllocation()); // Minimum allocation
@ -100,33 +112,46 @@ protected boolean allocateUser(ReservationId reservationId, String user,
} }
private Map<ReservationInterval, Resource> private Map<ReservationInterval, Resource> allocationsToPaddedMap(
allocationsToPaddedMap(RLESparseResourceAllocation allocation, RLESparseResourceAllocation allocation, long jobArrival, long jobDeadline,
long jobArrival, long jobDeadline) { long period) {
// Allocate
Map<ReservationInterval, Resource> mapAllocations =
allocation.toIntervalMap();
// Zero allocation // Zero allocation
Resource zeroResource = Resource.newInstance(0, 0); Resource zeroResource = Resource.newInstance(0, 0);
if (period > 0) {
if ((jobDeadline - jobArrival) >= period) {
allocation.addInterval(new ReservationInterval(0L, period),
zeroResource);
}
jobArrival = jobArrival % period;
jobDeadline = jobDeadline % period;
if (jobArrival <= jobDeadline) {
allocation.addInterval(new ReservationInterval(0, jobArrival),
zeroResource);
allocation.addInterval(new ReservationInterval(jobDeadline, period),
zeroResource);
} else {
allocation.addInterval(new ReservationInterval(jobDeadline, jobArrival),
zeroResource);
}
} else {
// Pad at the beginning // Pad at the beginning
long earliestStart = findEarliestTime(mapAllocations); long earliestStart = findEarliestTime(allocation.toIntervalMap());
if (jobArrival < earliestStart) { if (jobArrival < earliestStart) {
mapAllocations.put(new ReservationInterval(jobArrival, earliestStart), allocation.addInterval(
zeroResource); new ReservationInterval(jobArrival, earliestStart), zeroResource);
} }
// Pad at the beginning // Pad at the beginning
long latestEnd = findLatestTime(mapAllocations); long latestEnd = findLatestTime(allocation.toIntervalMap());
if (latestEnd < jobDeadline) { if (latestEnd < jobDeadline) {
mapAllocations.put(new ReservationInterval(latestEnd, jobDeadline), allocation.addInterval(new ReservationInterval(latestEnd, jobDeadline),
zeroResource); zeroResource);
} }
}
return mapAllocations; return allocation.toIntervalMap();
} }
public abstract RLESparseResourceAllocation computeJobAllocation(Plan plan, public abstract RLESparseResourceAllocation computeJobAllocation(Plan plan,

View File

@ -45,6 +45,7 @@ public interface StageAllocator {
* the stage by the two phase planning algorithm * the stage by the two phase planning algorithm
* @param stageDeadline the deadline of the stage set by the two phase * @param stageDeadline the deadline of the stage set by the two phase
* planning algorithm * planning algorithm
* @param period the periodicity with which this stage appears
* @param user name of the user * @param user name of the user
* @param oldId identifier of the old reservation * @param oldId identifier of the old reservation
* *
@ -55,7 +56,7 @@ public interface StageAllocator {
Map<ReservationInterval, Resource> computeStageAllocation(Plan plan, Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
RLESparseResourceAllocation planLoads, RLESparseResourceAllocation planLoads,
RLESparseResourceAllocation planModifications, ReservationRequest rr, RLESparseResourceAllocation planModifications, ReservationRequest rr,
long stageArrival, long stageDeadline, String user, long stageArrival, long stageDeadline, long period, String user,
ReservationId oldId) throws PlanningException; ReservationId oldId) throws PlanningException;
} }

View File

@ -43,7 +43,7 @@ public class StageAllocatorGreedy implements StageAllocator {
public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan, public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
RLESparseResourceAllocation planLoads, RLESparseResourceAllocation planLoads,
RLESparseResourceAllocation planModifications, ReservationRequest rr, RLESparseResourceAllocation planModifications, ReservationRequest rr,
long stageEarliestStart, long stageDeadline, String user, long stageEarliestStart, long stageDeadline, long period, String user,
ReservationId oldId) throws PlanningException { ReservationId oldId) throws PlanningException {
Resource totalCapacity = plan.getTotalCapacity(); Resource totalCapacity = plan.getTotalCapacity();
@ -69,7 +69,7 @@ public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
RLESparseResourceAllocation netAvailable = RLESparseResourceAllocation netAvailable =
plan.getAvailableResourceOverTime(user, oldId, stageEarliestStart, plan.getAvailableResourceOverTime(user, oldId, stageEarliestStart,
stageDeadline, 0); stageDeadline, period);
netAvailable = netAvailable =
RLESparseResourceAllocation.merge(plan.getResourceCalculator(), RLESparseResourceAllocation.merge(plan.getResourceCalculator(),

View File

@ -54,7 +54,7 @@ public StageAllocatorGreedyRLE(boolean allocateLeft) {
public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan, public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
RLESparseResourceAllocation planLoads, RLESparseResourceAllocation planLoads,
RLESparseResourceAllocation planModifications, ReservationRequest rr, RLESparseResourceAllocation planModifications, ReservationRequest rr,
long stageEarliestStart, long stageDeadline, String user, long stageEarliestStart, long stageDeadline, long period, String user,
ReservationId oldId) throws PlanningException { ReservationId oldId) throws PlanningException {
// abort early if the interval is not satisfiable // abort early if the interval is not satisfiable
@ -83,8 +83,9 @@ public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
int gangsToPlace = rr.getNumContainers() / rr.getConcurrency(); int gangsToPlace = rr.getNumContainers() / rr.getConcurrency();
// get available resources from plan // get available resources from plan
RLESparseResourceAllocation netRLERes = plan.getAvailableResourceOverTime( RLESparseResourceAllocation netRLERes =
user, oldId, stageEarliestStart, stageDeadline, 0); plan.getAvailableResourceOverTime(user, oldId, stageEarliestStart,
stageDeadline, period);
// remove plan modifications // remove plan modifications
netRLERes = netRLERes =

View File

@ -70,15 +70,15 @@ public StageAllocatorLowCostAligned(int smoothnessFactor,
public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan, public Map<ReservationInterval, Resource> computeStageAllocation(Plan plan,
RLESparseResourceAllocation planLoads, RLESparseResourceAllocation planLoads,
RLESparseResourceAllocation planModifications, ReservationRequest rr, RLESparseResourceAllocation planModifications, ReservationRequest rr,
long stageArrival, long stageDeadline, String user, ReservationId oldId) long stageArrival, long stageDeadline, long period, String user,
throws PlanningException { ReservationId oldId) throws PlanningException {
// Initialize // Initialize
ResourceCalculator resCalc = plan.getResourceCalculator(); ResourceCalculator resCalc = plan.getResourceCalculator();
Resource capacity = plan.getTotalCapacity(); Resource capacity = plan.getTotalCapacity();
RLESparseResourceAllocation netRLERes = plan.getAvailableResourceOverTime( RLESparseResourceAllocation netRLERes = plan.getAvailableResourceOverTime(
user, oldId, stageArrival, stageDeadline, 0); user, oldId, stageArrival, stageDeadline, period);
long step = plan.getStep(); long step = plan.getStep();

View File

@ -180,8 +180,12 @@ private RLESparseResourceAllocation generateRLEAlloc(long period) {
} }
} }
if(rStart > rEnd){
rle.addInterval(new ReservationInterval(rStart, period), alloc);
rle.addInterval(new ReservationInterval(0, rEnd), alloc);
} else {
rle.addInterval(new ReservationInterval(rStart, rEnd), alloc); rle.addInterval(new ReservationInterval(rStart, rEnd), alloc);
}
return rle; return rle;
} }

View File

@ -22,6 +22,7 @@
import java.util.Collection; import java.util.Collection;
import net.jcip.annotations.NotThreadSafe; import net.jcip.annotations.NotThreadSafe;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.ContractValidationException;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException; import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningQuotaException; import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningQuotaException;
import org.junit.Test; import org.junit.Test;
@ -39,63 +40,67 @@ public class TestCapacityOverTimePolicy extends BaseSharingPolicyTest {
final static long ONEDAY = 86400 * 1000; final static long ONEDAY = 86400 * 1000;
final static long ONEHOUR = 3600 * 1000; final static long ONEHOUR = 3600 * 1000;
final static long ONEMINUTE = 60 * 1000; final static long ONEMINUTE = 60 * 1000;
final static String TWODAYPERIOD = "7200000"; final static String TWOHOURPERIOD = "7200000";
final static String ONEDAYPERIOD = "86400000"; final static String ONEDAYPERIOD = "86400000";
@Parameterized.Parameters(name = "Duration {0}, height {1}," + @Parameterized.Parameters(name = "Duration {0}, height {1}," +
" submission {2}, periodic {3})") " numSubmission {2}, periodic {3})")
public static Collection<Object[]> data() { public static Collection<Object[]> data() {
return Arrays.asList(new Object[][] { return Arrays.asList(new Object[][] {
// easy fit // easy fit
{ONEHOUR, 0.25, 1, null, null }, {ONEHOUR, 0.25, 1, null, null },
{ONEHOUR, 0.25, 1, TWODAYPERIOD, null }, {ONEHOUR, 0.25, 1, TWOHOURPERIOD, null },
{ONEHOUR, 0.25, 1, ONEDAYPERIOD, null }, {ONEHOUR, 0.25, 1, ONEDAYPERIOD, null },
// instantaneous high, but fit integral and inst limits // instantaneous high, but fit integral and inst limits
{ONEMINUTE, 0.74, 1, null, null }, {ONEMINUTE, 0.74, 1, null, null },
{ONEMINUTE, 0.74, 1, TWODAYPERIOD, null }, {ONEMINUTE, 0.74, 1, TWOHOURPERIOD, null },
{ONEMINUTE, 0.74, 1, ONEDAYPERIOD, null }, {ONEMINUTE, 0.74, 1, ONEDAYPERIOD, null },
// barely fit // barely fit
{ONEHOUR, 0.76, 1, null, PlanningQuotaException.class }, {ONEHOUR, 0.76, 1, null, PlanningQuotaException.class },
{ONEHOUR, 0.76, 1, TWODAYPERIOD, PlanningQuotaException.class }, {ONEHOUR, 0.76, 1, TWOHOURPERIOD, PlanningQuotaException.class },
{ONEHOUR, 0.76, 1, ONEDAYPERIOD, PlanningQuotaException.class }, {ONEHOUR, 0.76, 1, ONEDAYPERIOD, PlanningQuotaException.class },
// overcommit with single reservation // overcommit with single reservation
{ONEHOUR, 1.1, 1, null, PlanningQuotaException.class }, {ONEHOUR, 1.1, 1, null, PlanningQuotaException.class },
{ONEHOUR, 1.1, 1, TWODAYPERIOD, PlanningQuotaException.class }, {ONEHOUR, 1.1, 1, TWOHOURPERIOD, PlanningQuotaException.class },
{ONEHOUR, 1.1, 1, ONEDAYPERIOD, PlanningQuotaException.class }, {ONEHOUR, 1.1, 1, ONEDAYPERIOD, PlanningQuotaException.class },
// barely fit with multiple reservations (instantaneously, lowering to // barely fit with multiple reservations (instantaneously, lowering to
// 1min to fit integral) // 1min to fit integral)
{ONEMINUTE, 0.25, 3, null, null }, {ONEMINUTE, 0.25, 3, null, null },
{ONEMINUTE, 0.25, 3, TWODAYPERIOD, null }, {ONEMINUTE, 0.25, 3, TWOHOURPERIOD, null },
{ONEMINUTE, 0.25, 3, ONEDAYPERIOD, null }, {ONEMINUTE, 0.25, 3, ONEDAYPERIOD, null },
// overcommit with multiple reservations (instantaneously) // overcommit with multiple reservations (instantaneously)
{ONEMINUTE, 0.25, 4, null, PlanningQuotaException.class }, {ONEMINUTE, 0.25, 4, null, PlanningQuotaException.class },
{ONEMINUTE, 0.25, 4, TWODAYPERIOD, PlanningQuotaException.class }, {ONEMINUTE, 0.25, 4, TWOHOURPERIOD, PlanningQuotaException.class },
{ONEMINUTE, 0.25, 4, ONEDAYPERIOD, PlanningQuotaException.class }, {ONEMINUTE, 0.25, 4, ONEDAYPERIOD, PlanningQuotaException.class },
// (non-periodic) reservation longer than window // (non-periodic) reservation longer than window
{25 * ONEHOUR, 0.25, 1, null, PlanningQuotaException.class }, {25 * ONEHOUR, 0.25, 1, null, PlanningQuotaException.class },
{25 * ONEHOUR, 0.25, 1, TWODAYPERIOD, PlanningQuotaException.class }, // NOTE: we generally don't accept periodicity < duration but the test
// generator will "wrap" this correctly
{25 * ONEHOUR, 0.25, 1, TWOHOURPERIOD, PlanningQuotaException.class },
{25 * ONEHOUR, 0.25, 1, ONEDAYPERIOD, PlanningQuotaException.class }, {25 * ONEHOUR, 0.25, 1, ONEDAYPERIOD, PlanningQuotaException.class },
// (non-periodic) reservation longer than window // (non-periodic) reservation longer than window
{25 * ONEHOUR, 0.05, 5, null, PlanningQuotaException.class }, {25 * ONEHOUR, 0.05, 5, null, PlanningQuotaException.class },
{25 * ONEHOUR, 0.05, 5, TWODAYPERIOD, PlanningQuotaException.class }, // NOTE: we generally don't accept periodicity < duration but the test
// generator will "wrap" this correctly
{25 * ONEHOUR, 0.05, 5, TWOHOURPERIOD, PlanningQuotaException.class },
{25 * ONEHOUR, 0.05, 5, ONEDAYPERIOD, PlanningQuotaException.class }, {25 * ONEHOUR, 0.05, 5, ONEDAYPERIOD, PlanningQuotaException.class },
// overcommit integral // overcommit integral
{ONEDAY, 0.26, 1, null, PlanningQuotaException.class }, {ONEDAY, 0.26, 1, null, PlanningQuotaException.class },
{2 * ONEHOUR, 0.26, 1, TWODAYPERIOD, PlanningQuotaException.class }, {2 * ONEHOUR, 0.26, 1, TWOHOURPERIOD, PlanningQuotaException.class },
{2 * ONEDAY, 0.26, 1, ONEDAYPERIOD, PlanningQuotaException.class }, {2 * ONEDAY, 0.26, 1, ONEDAYPERIOD, PlanningQuotaException.class },
// overcommit integral // overcommit integral
{ONEDAY / 2, 0.51, 1, null, PlanningQuotaException.class }, {ONEDAY / 2, 0.51, 1, null, PlanningQuotaException.class },
{2 * ONEHOUR / 2, 0.51, 1, TWODAYPERIOD, {2 * ONEHOUR / 2, 0.51, 1, TWOHOURPERIOD,
PlanningQuotaException.class }, PlanningQuotaException.class },
{2 * ONEDAY / 2, 0.51, 1, ONEDAYPERIOD, PlanningQuotaException.class } {2 * ONEDAY / 2, 0.51, 1, ONEDAYPERIOD, PlanningQuotaException.class }

View File

@ -17,6 +17,7 @@
*******************************************************************************/ *******************************************************************************/
package org.apache.hadoop.yarn.server.resourcemanager.reservation; package org.apache.hadoop.yarn.server.resourcemanager.reservation;
import static org.junit.Assert.assertEquals;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
@ -119,6 +120,18 @@ public void testAddReservation() {
checkAllocation(plan, alloc, start, 0); checkAllocation(plan, alloc, start, 0);
} }
@Test(expected = PlanningException.class)
public void testOutOfRange() throws PlanningException {
maxPeriodicity = 100;
Plan plan = new InMemoryPlan(queueMetrics, policy, agent, totalCapacity, 1L,
resCalc, minAlloc, maxAlloc, planName, replanner, true, maxPeriodicity,
context, new UTCClock());
// we expect the plan to complaint as the range 330-150 > 50
RLESparseResourceAllocation availableBefore =
plan.getAvailableResourceOverTime(user, null, 150, 330, 50);
}
@Test @Test
public void testAddPeriodicReservation() throws PlanningException { public void testAddPeriodicReservation() throws PlanningException {
@ -146,8 +159,14 @@ public void testAddPeriodicReservation() throws PlanningException {
checkAllocation(plan, alloc, start, period); checkAllocation(plan, alloc, start, period);
RLESparseResourceAllocation available = RLESparseResourceAllocation available =
plan.getAvailableResourceOverTime(user, reservationID, 150, 330, 50); plan.getAvailableResourceOverTime(user, null, 130, 170, 50);
System.out.println(available);
// the reservation has period 20 starting at 10, and the interaction with
// the period 50 request means that every 10 we expect a "90GB" point
assertEquals(92160, available.getCapacityAtTime(130).getMemorySize());
assertEquals(92160, available.getCapacityAtTime(140).getMemorySize());
assertEquals(92160, available.getCapacityAtTime(150).getMemorySize());
} }
private void checkAllocation(Plan plan, int[] alloc, int start, private void checkAllocation(Plan plan, int[] alloc, int start,
@ -162,18 +181,18 @@ private void checkAllocation(Plan plan, int[] alloc, int start,
for (int i = 0; i < alloc.length; i++) { for (int i = 0; i < alloc.length; i++) {
// only one instance for non-periodic reservation // only one instance for non-periodic reservation
if (periodicity <= 0) { if (periodicity <= 0) {
Assert.assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])), assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])),
plan.getTotalCommittedResources(start + i)); plan.getTotalCommittedResources(start + i));
Assert.assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])), assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])),
userCons.getCapacityAtTime(start + i)); userCons.getCapacityAtTime(start + i));
} else { } else {
// periodic reservations should repeat // periodic reservations should repeat
long y = 0; long y = 0;
Resource res = Resource.newInstance(1024 * (alloc[i]), (alloc[i])); Resource res = Resource.newInstance(1024 * (alloc[i]), (alloc[i]));
while (y <= end * 2) { while (y <= end * 2) {
Assert.assertEquals("At time: " + start + i + y, res, assertEquals("At time: " + start + i + y, res,
plan.getTotalCommittedResources(start + i + y)); plan.getTotalCommittedResources(start + i + y));
Assert.assertEquals(" At time: " + (start + i + y), res, assertEquals(" At time: " + (start + i + y), res,
userCons.getCapacityAtTime(start + i + y)); userCons.getCapacityAtTime(start + i + y));
y = y + periodicity; y = y + periodicity;
} }
@ -253,9 +272,9 @@ public void testUpdateReservation() {
RLESparseResourceAllocation userCons = RLESparseResourceAllocation userCons =
plan.getConsumptionForUserOverTime(user, start, start + alloc.length); plan.getConsumptionForUserOverTime(user, start, start + alloc.length);
for (int i = 0; i < alloc.length; i++) { for (int i = 0; i < alloc.length; i++) {
Assert.assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])), assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])),
plan.getTotalCommittedResources(start + i)); plan.getTotalCommittedResources(start + i));
Assert.assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])), assertEquals(Resource.newInstance(1024 * (alloc[i]), (alloc[i])),
userCons.getCapacityAtTime(start + i)); userCons.getCapacityAtTime(start + i));
} }
@ -275,9 +294,9 @@ public void testUpdateReservation() {
start + updatedAlloc.length); start + updatedAlloc.length);
for (int i = 0; i < updatedAlloc.length; i++) { for (int i = 0; i < updatedAlloc.length; i++) {
Assert.assertEquals(Resource.newInstance(1024 * (updatedAlloc[i] + i), assertEquals(Resource.newInstance(1024 * (updatedAlloc[i] + i),
updatedAlloc[i] + i), plan.getTotalCommittedResources(start + i)); updatedAlloc[i] + i), plan.getTotalCommittedResources(start + i));
Assert.assertEquals(Resource.newInstance(1024 * (updatedAlloc[i] + i), assertEquals(Resource.newInstance(1024 * (updatedAlloc[i] + i),
updatedAlloc[i] + i), userCons.getCapacityAtTime(start + i)); updatedAlloc[i] + i), userCons.getCapacityAtTime(start + i));
} }
} }
@ -371,10 +390,10 @@ public void testDeleteReservation() {
plan.getConsumptionForUserOverTime(user, start, start + alloc.length); plan.getConsumptionForUserOverTime(user, start, start + alloc.length);
for (int i = 0; i < alloc.length; i++) { for (int i = 0; i < alloc.length; i++) {
Assert.assertEquals( assertEquals(
Resource.newInstance(1024 * (alloc[i] + i), (alloc[i] + i)), Resource.newInstance(1024 * (alloc[i] + i), (alloc[i] + i)),
plan.getTotalCommittedResources(start + i)); plan.getTotalCommittedResources(start + i));
Assert.assertEquals( assertEquals(
Resource.newInstance(1024 * (alloc[i] + i), (alloc[i] + i)), Resource.newInstance(1024 * (alloc[i] + i), (alloc[i] + i)),
userCons.getCapacityAtTime(start + i)); userCons.getCapacityAtTime(start + i));
} }
@ -389,9 +408,9 @@ public void testDeleteReservation() {
userCons = userCons =
plan.getConsumptionForUserOverTime(user, start, start + alloc.length); plan.getConsumptionForUserOverTime(user, start, start + alloc.length);
for (int i = 0; i < alloc.length; i++) { for (int i = 0; i < alloc.length; i++) {
Assert.assertEquals(Resource.newInstance(0, 0), assertEquals(Resource.newInstance(0, 0),
plan.getTotalCommittedResources(start + i)); plan.getTotalCommittedResources(start + i));
Assert.assertEquals(Resource.newInstance(0, 0), assertEquals(Resource.newInstance(0, 0),
userCons.getCapacityAtTime(start + i)); userCons.getCapacityAtTime(start + i));
} }
} }
@ -492,11 +511,11 @@ public void testArchiveCompletedReservations() {
plan.getConsumptionForUserOverTime(user, start, start + alloc2.length); plan.getConsumptionForUserOverTime(user, start, start + alloc2.length);
for (int i = 0; i < alloc2.length; i++) { for (int i = 0; i < alloc2.length; i++) {
Assert.assertEquals( assertEquals(
Resource.newInstance(1024 * (alloc1[i] + alloc2[i] + i), Resource.newInstance(1024 * (alloc1[i] + alloc2[i] + i),
alloc1[i] + alloc2[i] + i), alloc1[i] + alloc2[i] + i),
plan.getTotalCommittedResources(start + i)); plan.getTotalCommittedResources(start + i));
Assert.assertEquals( assertEquals(
Resource.newInstance(1024 * (alloc1[i] + alloc2[i] + i), Resource.newInstance(1024 * (alloc1[i] + alloc2[i] + i),
alloc1[i] + alloc2[i] + i), alloc1[i] + alloc2[i] + i),
userCons.getCapacityAtTime(start + i)); userCons.getCapacityAtTime(start + i));
@ -530,9 +549,9 @@ public void testArchiveCompletedReservations() {
Assert.assertNull(plan.getReservationById(reservationID1)); Assert.assertNull(plan.getReservationById(reservationID1));
for (int i = 0; i < alloc1.length; i++) { for (int i = 0; i < alloc1.length; i++) {
Assert.assertEquals(Resource.newInstance(0, 0), assertEquals(Resource.newInstance(0, 0),
plan.getTotalCommittedResources(start + i)); plan.getTotalCommittedResources(start + i));
Assert.assertEquals(Resource.newInstance(0, 0), assertEquals(Resource.newInstance(0, 0),
userCons.getCapacityAtTime(start + i)); userCons.getCapacityAtTime(start + i));
} }
} }
@ -721,16 +740,16 @@ public void testGetReservationsWithNoReservation() {
private void doAssertions(Plan plan, ReservationAllocation rAllocation) { private void doAssertions(Plan plan, ReservationAllocation rAllocation) {
ReservationId reservationID = rAllocation.getReservationId(); ReservationId reservationID = rAllocation.getReservationId();
Assert.assertNotNull(plan.getReservationById(reservationID)); Assert.assertNotNull(plan.getReservationById(reservationID));
Assert.assertEquals(rAllocation, plan.getReservationById(reservationID)); assertEquals(rAllocation, plan.getReservationById(reservationID));
Assert.assertTrue(((InMemoryPlan) plan).getAllReservations().size() == 1); Assert.assertTrue(((InMemoryPlan) plan).getAllReservations().size() == 1);
if (rAllocation.getPeriodicity() <= 0) { if (rAllocation.getPeriodicity() <= 0) {
Assert.assertEquals(rAllocation.getEndTime(), plan.getLastEndTime()); assertEquals(rAllocation.getEndTime(), plan.getLastEndTime());
} }
Assert.assertEquals(totalCapacity, plan.getTotalCapacity()); assertEquals(totalCapacity, plan.getTotalCapacity());
Assert.assertEquals(minAlloc, plan.getMinimumAllocation()); assertEquals(minAlloc, plan.getMinimumAllocation());
Assert.assertEquals(maxAlloc, plan.getMaximumAllocation()); assertEquals(maxAlloc, plan.getMaximumAllocation());
Assert.assertEquals(resCalc, plan.getResourceCalculator()); assertEquals(resCalc, plan.getResourceCalculator());
Assert.assertEquals(planName, plan.getQueueName()); assertEquals(planName, plan.getQueueName());
Assert.assertTrue(plan.getMoveOnExpiry()); Assert.assertTrue(plan.getMoveOnExpiry());
} }

View File

@ -26,11 +26,14 @@
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Arrays; import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap; import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Random; import java.util.Random;
import net.jcip.annotations.NotThreadSafe;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.ReservationDefinition; import org.apache.hadoop.yarn.api.records.ReservationDefinition;
import org.apache.hadoop.yarn.api.records.ReservationId; import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.ReservationRequest; import org.apache.hadoop.yarn.api.records.ReservationRequest;
@ -54,11 +57,27 @@
import org.apache.hadoop.yarn.util.resource.Resources; import org.apache.hadoop.yarn.util.resource.Resources;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
/**
* This class tests the {@code AlignedPlannerWithGreedy} agent.
*/
@RunWith(value = Parameterized.class)
@NotThreadSafe
@SuppressWarnings("VisibilityModifier")
public class TestAlignedPlanner { public class TestAlignedPlanner {
@Parameterized.Parameter(value = 0)
public String recurrenceExpression;
final static String NONPERIODIC = "0";
final static String THREEHOURPERIOD = "10800000";
final static String ONEDAYPERIOD = "86400000";
private static final Logger LOG = LoggerFactory private static final Logger LOG = LoggerFactory
.getLogger(TestAlignedPlanner.class); .getLogger(TestAlignedPlanner.class);
@ -72,6 +91,16 @@ public class TestAlignedPlanner {
private Resource clusterCapacity; private Resource clusterCapacity;
private long step; private long step;
@Parameterized.Parameters(name = "Testing: periodicity {0})")
public static Collection<Object[]> data() {
return Arrays.asList(new Object[][]{
{NONPERIODIC},
{THREEHOURPERIOD},
{ONEDAYPERIOD}
});
}
@Test @Test
public void testSingleReservationAccept() throws PlanningException { public void testSingleReservationAccept() throws PlanningException {
@ -107,6 +136,9 @@ public void testSingleReservationAccept() throws PlanningException {
assertTrue(alloc1.toString(), assertTrue(alloc1.toString(),
check(alloc1, 10 * step, 20 * step, 10, 2048, 2)); check(alloc1, 10 * step, 20 * step, 10, 2048, 2));
System.out.println("--------AFTER AGENT----------");
System.out.println(plan.toString());
} }
@Test @Test
@ -1139,9 +1171,11 @@ private ReservationDefinition createReservationDefinition(long arrival,
long deadline, ReservationRequest[] reservationRequests, long deadline, ReservationRequest[] reservationRequests,
ReservationRequestInterpreter rType, String username) { ReservationRequestInterpreter rType, String username) {
return ReservationDefinition.newInstance(arrival, deadline,
ReservationRequests.newInstance(Arrays.asList(reservationRequests), return ReservationDefinition.newInstance(arrival,
rType), username); deadline, ReservationRequests
.newInstance(Arrays.asList(reservationRequests), rType),
username, recurrenceExpression, Priority.UNDEFINED);
} }

View File

@ -59,13 +59,20 @@
import org.junit.Test; import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
import org.junit.runners.Parameterized; import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@RunWith(Parameterized.class) @RunWith(Parameterized.class)
@SuppressWarnings("VisibilityModifier")
public class TestGreedyReservationAgent { public class TestGreedyReservationAgent {
@Parameterized.Parameter(value = 0)
public boolean allocateLeft;
@Parameterized.Parameter(value = 1)
public String recurrenceExpression;
private static final Logger LOG = LoggerFactory private static final Logger LOG = LoggerFactory
.getLogger(TestGreedyReservationAgent.class); .getLogger(TestGreedyReservationAgent.class);
@ -76,16 +83,18 @@ public class TestGreedyReservationAgent {
Resource maxAlloc = Resource.newInstance(1024 * 8, 8); Resource maxAlloc = Resource.newInstance(1024 * 8, 8);
Random rand = new Random(); Random rand = new Random();
long step; long step;
boolean allocateLeft;
public TestGreedyReservationAgent(Boolean b){ @Parameterized.Parameters(name = "Testing: allocateLeft {0}," +
this.allocateLeft = b; " recurrenceExpression {1})")
}
@Parameters
public static Collection<Object[]> data() { public static Collection<Object[]> data() {
return Arrays.asList(new Object[][] { return Arrays.asList(new Object[][] {
{true}, {false}}); {true, "0"},
{false, "0"},
{true, "7200000"},
{false, "7200000"},
{true, "86400000"},
{false, "86400000"}
});
} }
@Before @Before
@ -134,6 +143,7 @@ public void testSimple() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(5 * step); rr.setArrival(5 * step);
rr.setDeadline(20 * step); rr.setDeadline(20 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
Resource.newInstance(2048, 2), 10, 5, 10 * step); Resource.newInstance(2048, 2), 10, 5, 10 * step);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
@ -193,6 +203,7 @@ public void testSharingPolicyFeedback() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(5 * step); rr.setArrival(5 * step);
rr.setDeadline(100 * step); rr.setDeadline(100 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequest r = ReservationRequest r =
ReservationRequest.newInstance(Resource.newInstance(2048, 2), 20, 20, ReservationRequest.newInstance(Resource.newInstance(2048, 2), 20, 20,
10 * step); 10 * step);
@ -283,8 +294,9 @@ public void testOrder() throws PlanningException {
int[] f = { 100, 100 }; int[] f = { 100, 100 };
ReservationDefinition rDef = ReservationDefinition rDef =
ReservationSystemTestUtil.createSimpleReservationDefinition( ReservationSystemTestUtil.createSimpleReservationDefinition(30 * step,
30 * step, 30 * step + f.length * step, f.length * step); 30 * step + f.length * step, f.length * step, 1,
recurrenceExpression);
assertTrue(plan.toString(), assertTrue(plan.toString(),
plan.addReservation(new InMemoryReservationAllocation( plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1", ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
@ -296,6 +308,7 @@ public void testOrder() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(0 * step); rr.setArrival(0 * step);
rr.setDeadline(70 * step); rr.setDeadline(70 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ORDER); reqs.setInterpreter(ReservationRequestInterpreter.R_ORDER);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
@ -346,10 +359,11 @@ public void testOrderNoGapImpossible() throws PlanningException {
prepareBasicPlan(); prepareBasicPlan();
// create a completely utilized segment at time 30 // create a completely utilized segment at time 30
int[] f = { 100, 100 }; int[] f = { 100, 100 };
ReservationDefinition rDef = ReservationDefinition rDef = ReservationSystemTestUtil
ReservationSystemTestUtil.createSimpleReservationDefinition( .createSimpleReservationDefinition(30, 30 * step + f.length * step,
30, 30 * step + f.length * step, f.length * step); f.length * step, 1, recurrenceExpression);
assertTrue(plan.toString(), assertTrue(
plan.toString(),
plan.addReservation(new InMemoryReservationAllocation( plan.addReservation(new InMemoryReservationAllocation(
ReservationSystemTestUtil.getNewReservationId(), rDef, "u1", ReservationSystemTestUtil.getNewReservationId(), rDef, "u1",
"dedicated", 30 * step, 30 * step + f.length * step, "dedicated", 30 * step, 30 * step + f.length * step,
@ -406,6 +420,7 @@ public void testOrderNoGap() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(0 * step); rr.setArrival(0 * step);
rr.setDeadline(60 * step); rr.setDeadline(60 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ORDER_NO_GAP); reqs.setInterpreter(ReservationRequestInterpreter.R_ORDER_NO_GAP);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
@ -453,6 +468,7 @@ public void testSingleSliding() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(100 * step); rr.setArrival(100 * step);
rr.setDeadline(120 * step); rr.setDeadline(120 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ALL); reqs.setInterpreter(ReservationRequestInterpreter.R_ALL);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
@ -494,6 +510,7 @@ public void testAny() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(100 * step); rr.setArrival(100 * step);
rr.setDeadline(120 * step); rr.setDeadline(120 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ANY); reqs.setInterpreter(ReservationRequestInterpreter.R_ANY);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
@ -542,6 +559,7 @@ public void testAnyImpossible() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(100L); rr.setArrival(100L);
rr.setDeadline(120L); rr.setDeadline(120L);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ANY); reqs.setInterpreter(ReservationRequestInterpreter.R_ANY);
@ -587,6 +605,7 @@ public void testAll() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(100 * step); rr.setArrival(100 * step);
rr.setDeadline(120 * step); rr.setDeadline(120 * step);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ALL); reqs.setInterpreter(ReservationRequestInterpreter.R_ALL);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
@ -635,6 +654,7 @@ public void testAllImpossible() throws PlanningException {
ReservationDefinition rr = new ReservationDefinitionPBImpl(); ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(100L); rr.setArrival(100L);
rr.setDeadline(120L); rr.setDeadline(120L);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl(); ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ALL); reqs.setInterpreter(ReservationRequestInterpreter.R_ALL);
ReservationRequest r = ReservationRequest.newInstance( ReservationRequest r = ReservationRequest.newInstance(
@ -759,25 +779,4 @@ public void testStress(int numJobs) throws PlanningException, IOException {
+ " in " + (end - start) + "ms"); + " in " + (end - start) + "ms");
} }
public static void main(String[] arg) {
boolean left = false;
// run a stress test with by default 1000 random jobs
int numJobs = 1000;
if (arg.length > 0) {
numJobs = Integer.parseInt(arg[0]);
}
if (arg.length > 1) {
left = Boolean.parseBoolean(arg[1]);
}
try {
TestGreedyReservationAgent test = new TestGreedyReservationAgent(left);
test.setup();
test.testStress(numJobs);
} catch (Exception e) {
e.printStackTrace();
}
}
} }

View File

@ -0,0 +1,213 @@
/*****************************************************************************
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*****************************************************************************/
package org.apache.hadoop.yarn.server.resourcemanager.reservation.planning;
import org.apache.hadoop.yarn.api.records.ReservationDefinition;
import org.apache.hadoop.yarn.api.records.ReservationId;
import org.apache.hadoop.yarn.api.records.ReservationRequest;
import org.apache.hadoop.yarn.api.records.ReservationRequestInterpreter;
import org.apache.hadoop.yarn.api.records.ReservationRequests;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.impl.pb.ReservationDefinitionPBImpl;
import org.apache.hadoop.yarn.api.records.impl.pb.ReservationRequestsPBImpl;
import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.CapacityOverTimePolicy;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.InMemoryPlan;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.Plan;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.RLESparseResourceAllocation;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSchedulerConfiguration;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.ReservationSystemTestUtil;
import org.apache.hadoop.yarn.server.resourcemanager.reservation.exceptions.PlanningException;
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
import org.apache.hadoop.yarn.util.resource.DefaultResourceCalculator;
import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
import org.apache.hadoop.yarn.util.resource.Resources;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Random;
import static org.mockito.Mockito.mock;
/**
* General purpose ReservationAgent tester.
*/
@RunWith(Parameterized.class)
@SuppressWarnings("VisibilityModifier")
public class TestReservationAgents {
@Parameterized.Parameter(value = 0)
public Class agentClass;
@Parameterized.Parameter(value = 1)
public boolean allocateLeft;
@Parameterized.Parameter(value = 2)
public String recurrenceExpression;
@Parameterized.Parameter(value = 3)
public int numOfNodes;
private long step;
private Random rand = new Random(2);
private ReservationAgent agent;
private Plan plan;
private ResourceCalculator resCalc = new DefaultResourceCalculator();
private Resource minAlloc = Resource.newInstance(1024, 1);
private Resource maxAlloc = Resource.newInstance(32 * 1023, 32);
private long timeHorizon = 2 * 24 * 3600 * 1000; // 2 days
private static final Logger LOG =
LoggerFactory.getLogger(TestReservationAgents.class);
@Parameterized.Parameters(name = "Testing: agent {0}, allocateLeft: {1}," +
" recurrenceExpression: {2}, numNodes: {3})")
public static Collection<Object[]> data() {
return Arrays.asList(
new Object[][] {{GreedyReservationAgent.class, true, "0", 100 },
{GreedyReservationAgent.class, false, "0", 100 },
{GreedyReservationAgent.class, true, "7200000", 100 },
{GreedyReservationAgent.class, false, "7200000", 100 },
{GreedyReservationAgent.class, true, "86400000", 100 },
{GreedyReservationAgent.class, false, "86400000", 100 },
{AlignedPlannerWithGreedy.class, true, "0", 100 },
{AlignedPlannerWithGreedy.class, false, "0", 100 },
{AlignedPlannerWithGreedy.class, true, "7200000", 100 },
{AlignedPlannerWithGreedy.class, false, "7200000", 100 },
{AlignedPlannerWithGreedy.class, true, "86400000", 100 },
{AlignedPlannerWithGreedy.class, false, "86400000", 100 } });
}
@Before
public void setup() throws Exception {
long seed = rand.nextLong();
rand.setSeed(seed);
LOG.info("Running with seed: " + seed);
// setting completely loose quotas
long timeWindow = 1000000L;
Resource clusterCapacity =
Resource.newInstance(numOfNodes * 1024, numOfNodes);
step = 1000L;
String reservationQ =
ReservationSystemTestUtil.getFullReservationQueueName();
float instConstraint = 100;
float avgConstraint = 100;
ReservationSchedulerConfiguration conf = ReservationSystemTestUtil
.createConf(reservationQ, timeWindow, instConstraint, avgConstraint);
CapacityOverTimePolicy policy = new CapacityOverTimePolicy();
policy.init(reservationQ, conf);
// setting conf to
conf.setBoolean(GreedyReservationAgent.FAVOR_EARLY_ALLOCATION,
allocateLeft);
agent = (ReservationAgent) agentClass.newInstance();
agent.init(conf);
QueueMetrics queueMetrics = mock(QueueMetrics.class);
RMContext context = ReservationSystemTestUtil.createMockRMContext();
plan = new InMemoryPlan(queueMetrics, policy, agent, clusterCapacity, step,
resCalc, minAlloc, maxAlloc, "dedicated", null, true, context);
}
@Test
public void test() throws Exception {
long period = Long.parseLong(recurrenceExpression);
for (int i = 0; i < 1000; i++) {
ReservationDefinition rr = createRandomRequest(i);
if (rr != null) {
ReservationId reservationID =
ReservationSystemTestUtil.getNewReservationId();
try {
agent.createReservation(reservationID, "u1", plan, rr);
} catch (PlanningException p) {
// happens
}
}
}
}
private ReservationDefinition createRandomRequest(int i)
throws PlanningException {
long arrival = (long) Math.floor(rand.nextDouble() * timeHorizon);
long period = Long.parseLong(recurrenceExpression);
// min between period and rand around 30min
long duration =
(long) Math.round(Math.min(rand.nextDouble() * 3600 * 1000, period));
// min between period and rand around 5x duration
long deadline = (long) Math
.ceil(arrival + Math.min(duration * rand.nextDouble() * 10, period));
assert((deadline - arrival) <= period);
RLESparseResourceAllocation available = plan
.getAvailableResourceOverTime("u1", null, arrival, deadline, period);
NavigableMap<Long, Resource> availableMap = available.getCumulative();
// look at available space, and for each segment, use half of it with 50%
// probability
List<ReservationRequest> reservationRequests = new ArrayList<>();
for (Map.Entry<Long, Resource> e : availableMap.entrySet()) {
if (e.getValue() != null && rand.nextDouble() > 0.001) {
int numContainers = (int) Math.ceil(Resources.divide(resCalc,
plan.getTotalCapacity(), e.getValue(), minAlloc) / 2);
long tempDur =
Math.min(duration, availableMap.higherKey(e.getKey()) - e.getKey());
reservationRequests.add(ReservationRequest.newInstance(minAlloc,
numContainers, 1, tempDur));
}
}
if (reservationRequests.size() < 1) {
return null;
}
ReservationDefinition rr = new ReservationDefinitionPBImpl();
rr.setArrival(arrival);
rr.setDeadline(deadline);
rr.setRecurrenceExpression(recurrenceExpression);
ReservationRequests reqs = new ReservationRequestsPBImpl();
reqs.setInterpreter(ReservationRequestInterpreter.R_ORDER);
reqs.setReservationResources(reservationRequests);
rr.setReservationRequests(reqs);
rr.setReservationName("res_" + i);
return rr;
}
}