@Test public void testOnDemandTasksPersist() { SingularityRequestBuilder bldr = new SingularityRequestBuilder(requestId, RequestType.ON_DEMAND); requestResource.postRequest(bldr.build()); deploy("d2"); deployChecker.checkDeploys(); requestResource.scheduleImmediately(requestId); resourceOffers(); requestResource.scheduleImmediately(requestId); resourceOffers(); Assert.assertEquals(2, taskManager.getActiveTaskIds().size()); requestResource.scheduleImmediately(requestId); scheduler.drainPendingQueue(stateCacheProvider.get()); requestResource.scheduleImmediately(requestId); scheduler.drainPendingQueue(stateCacheProvider.get()); Assert.assertEquals(2, taskManager.getPendingTaskIds().size()); resourceOffers(); Assert.assertEquals(4, taskManager.getActiveTaskIds().size()); }
@Timed public void drainPendingQueue(final SingularitySchedulerStateCache stateCache) { final long start = System.currentTimeMillis(); final ImmutableList<SingularityPendingRequest> pendingRequests = ImmutableList.copyOf(requestManager.getPendingRequests()); if (pendingRequests.isEmpty()) { LOG.trace("Pending queue was empty"); return; } LOG.info("Pending queue had {} requests", pendingRequests.size()); int totalNewScheduledTasks = 0; int heldForScheduledActiveTask = 0; int obsoleteRequests = 0; for (SingularityPendingRequest pendingRequest : pendingRequests) { Optional<SingularityRequestWithState> maybeRequest = requestManager.getRequest(pendingRequest.getRequestId()); if (!isRequestActive(maybeRequest)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } Optional<SingularityRequestDeployState> maybeRequestDeployState = deployManager.getRequestDeployState(pendingRequest.getRequestId()); Optional<SingularityPendingDeploy> maybePendingDeploy = deployManager.getPendingDeploy(maybeRequest.get().getRequest().getId()); if (!shouldScheduleTasks(maybeRequest.get().getRequest(), pendingRequest, maybePendingDeploy, maybeRequestDeployState)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } final List<SingularityTaskId> matchingTaskIds = getMatchingTaskIds(stateCache, maybeRequest.get().getRequest(), pendingRequest); final SingularityDeployStatistics deployStatistics = getDeployStatistics(pendingRequest.getRequestId(), pendingRequest.getDeployId()); final RequestState requestState = checkCooldown(maybeRequest.get(), deployStatistics); int numScheduledTasks = scheduleTasks(stateCache, maybeRequest.get().getRequest(), requestState, deployStatistics, pendingRequest, matchingTaskIds, maybePendingDeploy); if (numScheduledTasks == 0 && !matchingTaskIds.isEmpty() && maybeRequest.get().getRequest().isScheduled() && pendingRequest.getPendingType() == PendingType.NEW_DEPLOY) { LOG.trace("Holding pending request {} because it is scheduled and has an active task", pendingRequest); heldForScheduledActiveTask++; continue; } LOG.debug("Pending request {} resulted in {} new scheduled tasks", pendingRequest, numScheduledTasks); totalNewScheduledTasks += numScheduledTasks; requestManager.deletePendingRequest(pendingRequest); } LOG.info("Scheduled {} new tasks ({} obsolete requests, {} held) in {}", totalNewScheduledTasks, obsoleteRequests, heldForScheduledActiveTask, JavaUtils.duration(start)); }
SingularityScheduler { @Timed public void drainPendingQueue(final SingularitySchedulerStateCache stateCache) { final long start = System.currentTimeMillis(); final ImmutableList<SingularityPendingRequest> pendingRequests = ImmutableList.copyOf(requestManager.getPendingRequests()); if (pendingRequests.isEmpty()) { LOG.trace("Pending queue was empty"); return; } LOG.info("Pending queue had {} requests", pendingRequests.size()); int totalNewScheduledTasks = 0; int heldForScheduledActiveTask = 0; int obsoleteRequests = 0; for (SingularityPendingRequest pendingRequest : pendingRequests) { Optional<SingularityRequestWithState> maybeRequest = requestManager.getRequest(pendingRequest.getRequestId()); if (!isRequestActive(maybeRequest)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } Optional<SingularityRequestDeployState> maybeRequestDeployState = deployManager.getRequestDeployState(pendingRequest.getRequestId()); Optional<SingularityPendingDeploy> maybePendingDeploy = deployManager.getPendingDeploy(maybeRequest.get().getRequest().getId()); if (!shouldScheduleTasks(maybeRequest.get().getRequest(), pendingRequest, maybePendingDeploy, maybeRequestDeployState)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } final List<SingularityTaskId> matchingTaskIds = getMatchingTaskIds(stateCache, maybeRequest.get().getRequest(), pendingRequest); final SingularityDeployStatistics deployStatistics = getDeployStatistics(pendingRequest.getRequestId(), pendingRequest.getDeployId()); final RequestState requestState = checkCooldown(maybeRequest.get(), deployStatistics); int numScheduledTasks = scheduleTasks(stateCache, maybeRequest.get().getRequest(), requestState, deployStatistics, pendingRequest, matchingTaskIds, maybePendingDeploy); if (numScheduledTasks == 0 && !matchingTaskIds.isEmpty() && maybeRequest.get().getRequest().isScheduled() && pendingRequest.getPendingType() == PendingType.NEW_DEPLOY) { LOG.trace("Holding pending request {} because it is scheduled and has an active task", pendingRequest); heldForScheduledActiveTask++; continue; } LOG.debug("Pending request {} resulted in {} new scheduled tasks", pendingRequest, numScheduledTasks); totalNewScheduledTasks += numScheduledTasks; requestManager.deletePendingRequest(pendingRequest); } LOG.info("Scheduled {} new tasks ({} obsolete requests, {} held) in {}", totalNewScheduledTasks, obsoleteRequests, heldForScheduledActiveTask, JavaUtils.duration(start)); } }
SingularityScheduler { @Timed public void drainPendingQueue(final SingularitySchedulerStateCache stateCache) { final long start = System.currentTimeMillis(); final ImmutableList<SingularityPendingRequest> pendingRequests = ImmutableList.copyOf(requestManager.getPendingRequests()); if (pendingRequests.isEmpty()) { LOG.trace("Pending queue was empty"); return; } LOG.info("Pending queue had {} requests", pendingRequests.size()); int totalNewScheduledTasks = 0; int heldForScheduledActiveTask = 0; int obsoleteRequests = 0; for (SingularityPendingRequest pendingRequest : pendingRequests) { Optional<SingularityRequestWithState> maybeRequest = requestManager.getRequest(pendingRequest.getRequestId()); if (!isRequestActive(maybeRequest)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } Optional<SingularityRequestDeployState> maybeRequestDeployState = deployManager.getRequestDeployState(pendingRequest.getRequestId()); Optional<SingularityPendingDeploy> maybePendingDeploy = deployManager.getPendingDeploy(maybeRequest.get().getRequest().getId()); if (!shouldScheduleTasks(maybeRequest.get().getRequest(), pendingRequest, maybePendingDeploy, maybeRequestDeployState)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } final List<SingularityTaskId> matchingTaskIds = getMatchingTaskIds(stateCache, maybeRequest.get().getRequest(), pendingRequest); final SingularityDeployStatistics deployStatistics = getDeployStatistics(pendingRequest.getRequestId(), pendingRequest.getDeployId()); final RequestState requestState = checkCooldown(maybeRequest.get(), deployStatistics); int numScheduledTasks = scheduleTasks(stateCache, maybeRequest.get().getRequest(), requestState, deployStatistics, pendingRequest, matchingTaskIds, maybePendingDeploy); if (numScheduledTasks == 0 && !matchingTaskIds.isEmpty() && maybeRequest.get().getRequest().isScheduled() && pendingRequest.getPendingType() == PendingType.NEW_DEPLOY) { LOG.trace("Holding pending request {} because it is scheduled and has an active task", pendingRequest); heldForScheduledActiveTask++; continue; } LOG.debug("Pending request {} resulted in {} new scheduled tasks", pendingRequest, numScheduledTasks); totalNewScheduledTasks += numScheduledTasks; requestManager.deletePendingRequest(pendingRequest); } LOG.info("Scheduled {} new tasks ({} obsolete requests, {} held) in {}", totalNewScheduledTasks, obsoleteRequests, heldForScheduledActiveTask, JavaUtils.duration(start)); } @Inject  SingularityScheduler(TaskRequestManager taskRequestManager, SingularityConfiguration configuration, SingularityCooldown cooldown, DeployManager deployManager,
    TaskManager taskManager, RequestManager requestManager, SlaveManager slaveManager, RackManager rackManager, SingularityMailer mailer); }
SingularityScheduler { @Timed public void drainPendingQueue(final SingularitySchedulerStateCache stateCache) { final long start = System.currentTimeMillis(); final ImmutableList<SingularityPendingRequest> pendingRequests = ImmutableList.copyOf(requestManager.getPendingRequests()); if (pendingRequests.isEmpty()) { LOG.trace("Pending queue was empty"); return; } LOG.info("Pending queue had {} requests", pendingRequests.size()); int totalNewScheduledTasks = 0; int heldForScheduledActiveTask = 0; int obsoleteRequests = 0; for (SingularityPendingRequest pendingRequest : pendingRequests) { Optional<SingularityRequestWithState> maybeRequest = requestManager.getRequest(pendingRequest.getRequestId()); if (!isRequestActive(maybeRequest)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } Optional<SingularityRequestDeployState> maybeRequestDeployState = deployManager.getRequestDeployState(pendingRequest.getRequestId()); Optional<SingularityPendingDeploy> maybePendingDeploy = deployManager.getPendingDeploy(maybeRequest.get().getRequest().getId()); if (!shouldScheduleTasks(maybeRequest.get().getRequest(), pendingRequest, maybePendingDeploy, maybeRequestDeployState)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } final List<SingularityTaskId> matchingTaskIds = getMatchingTaskIds(stateCache, maybeRequest.get().getRequest(), pendingRequest); final SingularityDeployStatistics deployStatistics = getDeployStatistics(pendingRequest.getRequestId(), pendingRequest.getDeployId()); final RequestState requestState = checkCooldown(maybeRequest.get(), deployStatistics); int numScheduledTasks = scheduleTasks(stateCache, maybeRequest.get().getRequest(), requestState, deployStatistics, pendingRequest, matchingTaskIds, maybePendingDeploy); if (numScheduledTasks == 0 && !matchingTaskIds.isEmpty() && maybeRequest.get().getRequest().isScheduled() && pendingRequest.getPendingType() == PendingType.NEW_DEPLOY) { LOG.trace("Holding pending request {} because it is scheduled and has an active task", pendingRequest); heldForScheduledActiveTask++; continue; } LOG.debug("Pending request {} resulted in {} new scheduled tasks", pendingRequest, numScheduledTasks); totalNewScheduledTasks += numScheduledTasks; requestManager.deletePendingRequest(pendingRequest); } LOG.info("Scheduled {} new tasks ({} obsolete requests, {} held) in {}", totalNewScheduledTasks, obsoleteRequests, heldForScheduledActiveTask, JavaUtils.duration(start)); } @Inject  SingularityScheduler(TaskRequestManager taskRequestManager, SingularityConfiguration configuration, SingularityCooldown cooldown, DeployManager deployManager,
    TaskManager taskManager, RequestManager requestManager, SlaveManager slaveManager, RackManager rackManager, SingularityMailer mailer); @Timed void checkForDecomissions(SingularitySchedulerStateCache stateCache); @Timed void drainPendingQueue(final SingularitySchedulerStateCache stateCache); @Timed List<SingularityTaskRequest> getDueTasks(); @Timed void handleCompletedTask(Optional<SingularityTask> task, SingularityTaskId taskId, boolean wasActive, long timestamp, ExtendedTaskState state,
    SingularityCreateResult taskHistoryUpdateCreateResult, SingularitySchedulerStateCache stateCache); }
SingularityScheduler { @Timed public void drainPendingQueue(final SingularitySchedulerStateCache stateCache) { final long start = System.currentTimeMillis(); final ImmutableList<SingularityPendingRequest> pendingRequests = ImmutableList.copyOf(requestManager.getPendingRequests()); if (pendingRequests.isEmpty()) { LOG.trace("Pending queue was empty"); return; } LOG.info("Pending queue had {} requests", pendingRequests.size()); int totalNewScheduledTasks = 0; int heldForScheduledActiveTask = 0; int obsoleteRequests = 0; for (SingularityPendingRequest pendingRequest : pendingRequests) { Optional<SingularityRequestWithState> maybeRequest = requestManager.getRequest(pendingRequest.getRequestId()); if (!isRequestActive(maybeRequest)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } Optional<SingularityRequestDeployState> maybeRequestDeployState = deployManager.getRequestDeployState(pendingRequest.getRequestId()); Optional<SingularityPendingDeploy> maybePendingDeploy = deployManager.getPendingDeploy(maybeRequest.get().getRequest().getId()); if (!shouldScheduleTasks(maybeRequest.get().getRequest(), pendingRequest, maybePendingDeploy, maybeRequestDeployState)) { LOG.debug("Pending request {} was obsolete (request {})", pendingRequest, SingularityRequestWithState.getRequestState(maybeRequest)); obsoleteRequests++; continue; } final List<SingularityTaskId> matchingTaskIds = getMatchingTaskIds(stateCache, maybeRequest.get().getRequest(), pendingRequest); final SingularityDeployStatistics deployStatistics = getDeployStatistics(pendingRequest.getRequestId(), pendingRequest.getDeployId()); final RequestState requestState = checkCooldown(maybeRequest.get(), deployStatistics); int numScheduledTasks = scheduleTasks(stateCache, maybeRequest.get().getRequest(), requestState, deployStatistics, pendingRequest, matchingTaskIds, maybePendingDeploy); if (numScheduledTasks == 0 && !matchingTaskIds.isEmpty() && maybeRequest.get().getRequest().isScheduled() && pendingRequest.getPendingType() == PendingType.NEW_DEPLOY) { LOG.trace("Holding pending request {} because it is scheduled and has an active task", pendingRequest); heldForScheduledActiveTask++; continue; } LOG.debug("Pending request {} resulted in {} new scheduled tasks", pendingRequest, numScheduledTasks); totalNewScheduledTasks += numScheduledTasks; requestManager.deletePendingRequest(pendingRequest); } LOG.info("Scheduled {} new tasks ({} obsolete requests, {} held) in {}", totalNewScheduledTasks, obsoleteRequests, heldForScheduledActiveTask, JavaUtils.duration(start)); } @Inject  SingularityScheduler(TaskRequestManager taskRequestManager, SingularityConfiguration configuration, SingularityCooldown cooldown, DeployManager deployManager,
    TaskManager taskManager, RequestManager requestManager, SlaveManager slaveManager, RackManager rackManager, SingularityMailer mailer); @Timed void checkForDecomissions(SingularitySchedulerStateCache stateCache); @Timed void drainPendingQueue(final SingularitySchedulerStateCache stateCache); @Timed List<SingularityTaskRequest> getDueTasks(); @Timed void handleCompletedTask(Optional<SingularityTask> task, SingularityTaskId taskId, boolean wasActive, long timestamp, ExtendedTaskState state,
    SingularityCreateResult taskHistoryUpdateCreateResult, SingularitySchedulerStateCache stateCache);  }