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; 020 021import org.apache.reef.driver.catalog.NodeDescriptor; 022import org.apache.reef.driver.catalog.RackDescriptor; 023import org.apache.reef.driver.catalog.ResourceCatalog; 024import org.apache.reef.driver.evaluator.EvaluatorRequest; 025import org.apache.reef.driver.evaluator.EvaluatorRequestor; 026import org.apache.reef.proto.DriverRuntimeProtocol; 027import org.apache.reef.runtime.common.driver.api.ResourceRequestHandler; 028import org.apache.reef.util.logging.LoggingScope; 029import org.apache.reef.util.logging.LoggingScopeFactory; 030 031import javax.inject.Inject; 032import java.util.logging.Level; 033import java.util.logging.Logger; 034 035 036/** 037 * Implementation of the EvaluatorRequestor that translates the request and hands it down to the underlying RM. 038 */ 039public final class EvaluatorRequestorImpl implements EvaluatorRequestor { 040 041 private static final Logger LOG = Logger.getLogger(EvaluatorRequestorImpl.class.getName()); 042 043 private final ResourceCatalog resourceCatalog; 044 private final ResourceRequestHandler resourceRequestHandler; 045 private final LoggingScopeFactory loggingScopeFactory; 046 047 /** 048 * @param resourceCatalog 049 * @param resourceRequestHandler 050 */ 051 @Inject 052 public EvaluatorRequestorImpl(final ResourceCatalog resourceCatalog, 053 final ResourceRequestHandler resourceRequestHandler, 054 final LoggingScopeFactory loggingScopeFactory) { 055 this.resourceCatalog = resourceCatalog; 056 this.resourceRequestHandler = resourceRequestHandler; 057 this.loggingScopeFactory = loggingScopeFactory; 058 } 059 060 @Override 061 public synchronized void submit(final EvaluatorRequest req) { 062 LOG.log(Level.FINEST, "Got an EvaluatorRequest: number: {0}, memory = {1}, cores = {2}.", new Object[]{req.getNumber(), req.getMegaBytes(), req.getNumberOfCores()}); 063 064 if (req.getMegaBytes() <= 0) { 065 throw new IllegalArgumentException("Given an unsupported memory size: " + req.getMegaBytes()); 066 } 067 if (req.getNumberOfCores() <= 0) { 068 throw new IllegalArgumentException("Given an unsupported core number: " + req.getNumberOfCores()); 069 } 070 if (req.getNumber() <= 0) { 071 throw new IllegalArgumentException("Given an unsupported number of evaluators: " + req.getNumber()); 072 } 073 074 try (LoggingScope ls = loggingScopeFactory.evaluatorSubmit(req.getNumber())) { 075 final DriverRuntimeProtocol.ResourceRequestProto.Builder request = DriverRuntimeProtocol.ResourceRequestProto 076 .newBuilder() 077 .setResourceCount(req.getNumber()) 078 .setVirtualCores(req.getNumberOfCores()) 079 .setMemorySize(req.getMegaBytes()); 080 081 final ResourceCatalog.Descriptor descriptor = req.getDescriptor(); 082 if (descriptor != null) { 083 if (descriptor instanceof RackDescriptor) { 084 request.addRackName(descriptor.getName()); 085 } else if (descriptor instanceof NodeDescriptor) { 086 request.addNodeName(descriptor.getName()); 087 } else { 088 throw new IllegalArgumentException("Unable to operate on descriptors of type " + descriptor.getClass().getName()); 089 } 090 } 091 092 this.resourceRequestHandler.onNext(request.build()); 093 } 094 } 095}