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 065 final String id = resourceStatusEvent.getIdentifier(); 066 final Optional<EvaluatorManager> evaluatorManager = this.evaluators.get(id); 067 068 LOG.log(Level.FINEST, "Evaluator {0} status: {1}", 069 new Object[] {evaluatorManager, resourceStatusEvent.getState()}); 070 071 if (evaluatorManager.isPresent()) { 072 final EvaluatorManager evaluatorManagerImpl = evaluatorManager.get(); 073 evaluatorManagerImpl.onResourceStatusMessage(resourceStatusEvent); 074 if (evaluatorManagerImpl.isClosed()) { 075 this.evaluators.removeClosedEvaluator(evaluatorManagerImpl); 076 } 077 078 } else { 079 080 if (this.evaluators.wasClosed(id)) { 081 LOG.log(Level.WARNING, 082 "Unexpected resource status from closed evaluator {0} with state {1}", 083 new Object[] {id, resourceStatusEvent.getState()}); 084 } 085 086 if (driverRestartManager.get().getEvaluatorRestartState(id).isFailedOrExpired()) { 087 088 final EvaluatorManager previousEvaluatorManager = this.evaluatorManagerFactory 089 .getNewEvaluatorManagerForEvaluatorFailedDuringDriverRestart(resourceStatusEvent); 090 091 previousEvaluatorManager.onResourceStatusMessage(resourceStatusEvent); 092 093 } else { 094 throw new RuntimeException( 095 "Unknown resource status from evaluator " + id + " with state " + resourceStatusEvent.getState()); 096 } 097 } 098 } 099}