001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, 013 * software distributed under the License is distributed on an 014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 015 * KIND, either express or implied. See the License for the 016 * specific language governing permissions and limitations 017 * under the License. 018 */ 019package org.apache.reef.runtime.common.driver.resourcemanager; 020 021import org.apache.reef.annotations.audience.Private; 022import org.apache.reef.driver.restart.DriverRestartManager; 023import org.apache.reef.runtime.common.driver.evaluator.EvaluatorManager; 024import org.apache.reef.runtime.common.driver.evaluator.EvaluatorManagerFactory; 025import org.apache.reef.runtime.common.driver.evaluator.Evaluators; 026import org.apache.reef.tang.InjectionFuture; 027import org.apache.reef.util.Optional; 028import org.apache.reef.wake.EventHandler; 029 030import javax.inject.Inject; 031import java.util.logging.Level; 032import java.util.logging.Logger; 033 034/** 035 * A ResourceStatusProto message comes from the ResourceManager layer to indicate what it thinks 036 * about the current state of a given resource. Ideally, we should think the same thing. 037 */ 038@Private 039public final class ResourceStatusHandler implements EventHandler<ResourceStatusEvent> { 040 private static final Logger LOG = Logger.getLogger(Evaluators.class.getName()); 041 042 private final Evaluators evaluators; 043 private final EvaluatorManagerFactory evaluatorManagerFactory; 044 private final InjectionFuture<DriverRestartManager> driverRestartManager; 045 046 @Inject 047 ResourceStatusHandler(final Evaluators evaluators, 048 final EvaluatorManagerFactory evaluatorManagerFactory, 049 final InjectionFuture<DriverRestartManager> driverRestartManager) { 050 this.evaluators = evaluators; 051 this.evaluatorManagerFactory = evaluatorManagerFactory; 052 this.driverRestartManager = driverRestartManager; 053 } 054 055 /** 056 * This resource status message comes from the ResourceManager layer, telling me what it thinks 057 * about the state of the resource executing an Evaluator. This method simply passes the message 058 * off to the referenced EvaluatorManager 059 * 060 * @param resourceStatusEvent resource status message from the ResourceManager 061 */ 062 @Override 063 public void onNext(final ResourceStatusEvent resourceStatusEvent) { 064 final Optional<EvaluatorManager> evaluatorManager = this.evaluators.get(resourceStatusEvent.getIdentifier()); 065 if (evaluatorManager.isPresent()) { 066 evaluatorManager.get().onResourceStatusMessage(resourceStatusEvent); 067 068 if (evaluatorManager.get().isClosed()) { 069 this.evaluators.removeClosedEvaluator(evaluatorManager.get()); 070 } 071 } else { 072 if (this.evaluators.wasClosed(resourceStatusEvent.getIdentifier())) { 073 LOG.log(Level.WARNING, "Unexpected resource status from closed evaluator " + 074 resourceStatusEvent.getIdentifier() + " with state " + resourceStatusEvent.getState()); 075 } 076 077 if (driverRestartManager.get().getEvaluatorRestartState(resourceStatusEvent.getIdentifier()) 078 .isFailedOrExpired()) { 079 final EvaluatorManager previousEvaluatorManager = this.evaluatorManagerFactory 080 .getNewEvaluatorManagerForEvaluatorFailedDuringDriverRestart(resourceStatusEvent); 081 082 previousEvaluatorManager.onResourceStatusMessage(resourceStatusEvent); 083 } else { 084 throw new RuntimeException( 085 "Unknown resource status from evaluator " + resourceStatusEvent.getIdentifier() + 086 " with state " + resourceStatusEvent.getState() 087 ); 088 } 089 } 090 } 091}