You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by ss...@apache.org on 2016/02/01 18:45:58 UTC
[04/10] hive git commit: HIVE-12934. Refactor llap module structure
to allow for a usable client. (Siddharth Seth, reviewed by Sergey Shelukhin)
http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java b/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
deleted file mode 100644
index 6beb4f8..0000000
--- a/llap-server/src/java/org/apache/tez/dag/app/rm/LlapTaskSchedulerService.java
+++ /dev/null
@@ -1,1512 +0,0 @@
-/*
- * Licensed 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.tez.dag.app.rm;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Comparator;
-import java.util.EnumSet;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedHashMap;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-import java.util.NavigableMap;
-import java.util.Random;
-import java.util.Set;
-import java.util.TreeMap;
-import java.util.TreeSet;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.DelayQueue;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import com.google.common.collect.Sets;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
-import org.apache.commons.lang.mutable.MutableInt;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
-import org.apache.hadoop.hive.llap.registry.ServiceInstance;
-import org.apache.hadoop.hive.llap.registry.ServiceInstanceSet;
-import org.apache.hadoop.hive.llap.registry.impl.LlapRegistryService;
-import org.apache.hadoop.yarn.api.records.Container;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.Priority;
-import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.util.Clock;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Preconditions;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.ListeningExecutorService;
-import com.google.common.util.concurrent.MoreExecutors;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.apache.hadoop.yarn.util.SystemClock;
-import org.apache.tez.common.TezUtils;
-import org.apache.tez.dag.api.TezUncheckedException;
-import org.apache.tez.serviceplugins.api.TaskAttemptEndReason;
-import org.apache.tez.serviceplugins.api.TaskScheduler;
-import org.apache.tez.serviceplugins.api.TaskSchedulerContext;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class LlapTaskSchedulerService extends TaskScheduler {
-
- private static final Logger LOG = LoggerFactory.getLogger(LlapTaskSchedulerService.class);
-
- private final Configuration conf;
-
- // interface into the registry service
- private ServiceInstanceSet activeInstances;
-
- // Tracks all instances, including ones which have been disabled in the past.
- // LinkedHashMap to provide the same iteration order when selecting a random host.
- @VisibleForTesting
- final Map<ServiceInstance, NodeInfo> instanceToNodeMap = new LinkedHashMap<>();
- // TODO Ideally, remove elements from this once it's known that no tasks are linked to the instance (all deallocated)
-
- // Tracks tasks which could not be allocated immediately.
- @VisibleForTesting
- final TreeMap<Priority, List<TaskInfo>> pendingTasks = new TreeMap<>(new Comparator<Priority>() {
- @Override
- public int compare(Priority o1, Priority o2) {
- return o1.getPriority() - o2.getPriority();
- }
- });
-
- // Tracks running and queued tasks. Cleared after a task completes.
- private final ConcurrentMap<Object, TaskInfo> knownTasks = new ConcurrentHashMap<>();
- private final TreeMap<Integer, TreeSet<TaskInfo>> runningTasks = new TreeMap<>();
- private static final TaskStartComparator TASK_INFO_COMPARATOR = new TaskStartComparator();
-
- // Queue for disabled nodes. Nodes make it out of this queue when their expiration timeout is hit.
- @VisibleForTesting
- final DelayQueue<NodeInfo> disabledNodesQueue = new DelayQueue<>();
-
- private final boolean forceLocation;
-
- private final ContainerFactory containerFactory;
- private final Random random = new Random();
- private final Clock clock;
-
- private final ListeningExecutorService nodeEnabledExecutor;
- private final NodeEnablerCallable nodeEnablerCallable =
- new NodeEnablerCallable();
-
- private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
- private final ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
- private final ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
-
- private final Lock scheduleLock = new ReentrantLock();
- private final Condition scheduleCondition = scheduleLock.newCondition();
- private final AtomicBoolean pendingScheduleInvodations = new AtomicBoolean(false);
- private final ListeningExecutorService schedulerExecutor;
- private final SchedulerCallable schedulerCallable = new SchedulerCallable();
-
- private final AtomicBoolean isStopped = new AtomicBoolean(false);
- // Tracks total pending preemptions.
- private final AtomicInteger pendingPreemptions = new AtomicInteger(0);
- // Tracks pending preemptions per host, using the hostname || Always to be accessed inside a lock
- private final Map<String, MutableInt> pendingPreemptionsPerHost = new HashMap<>();
-
- private final NodeBlacklistConf nodeBlacklistConf;
-
- // Per daemon
- private final int memoryPerInstance;
- private final int coresPerInstance;
- private final int executorsPerInstance;
-
- private final int numSchedulableTasksPerNode;
-
- // Per Executor Thread
- private final Resource resourcePerExecutor;
-
- private final LlapRegistryService registry = new LlapRegistryService(false);
-
- private volatile ListenableFuture<Void> nodeEnablerFuture;
- private volatile ListenableFuture<Void> schedulerFuture;
-
- @VisibleForTesting
- private final AtomicInteger dagCounter = new AtomicInteger(1);
- // Statistics to track allocations
- // All of stats variables are visible for testing.
- @VisibleForTesting
- StatsPerDag dagStats = new StatsPerDag();
-
- public LlapTaskSchedulerService(TaskSchedulerContext taskSchedulerContext) {
- this(taskSchedulerContext, new SystemClock());
- }
-
- @VisibleForTesting
- public LlapTaskSchedulerService(TaskSchedulerContext taskSchedulerContext, Clock clock) {
- super(taskSchedulerContext);
- this.clock = clock;
- try {
- this.conf = TezUtils.createConfFromUserPayload(taskSchedulerContext.getInitialUserPayload());
- } catch (IOException e) {
- throw new TezUncheckedException(
- "Failed to parse user payload for " + LlapTaskSchedulerService.class.getSimpleName(), e);
- }
- this.containerFactory = new ContainerFactory(taskSchedulerContext.getApplicationAttemptId(),
- taskSchedulerContext.getCustomClusterIdentifier());
- this.memoryPerInstance = HiveConf.getIntVar(conf, ConfVars.LLAP_DAEMON_MEMORY_PER_INSTANCE_MB);
- this.coresPerInstance = HiveConf.getIntVar(conf, ConfVars.LLAP_DAEMON_VCPUS_PER_INSTANCE);
- this.executorsPerInstance = HiveConf.getIntVar(conf, ConfVars.LLAP_DAEMON_NUM_EXECUTORS);
- this.nodeBlacklistConf = new NodeBlacklistConf(
- HiveConf.getTimeVar(conf, ConfVars.LLAP_TASK_SCHEDULER_NODE_REENABLE_MIN_TIMEOUT_MS,
- TimeUnit.MILLISECONDS),
- HiveConf.getTimeVar(conf, ConfVars.LLAP_TASK_SCHEDULER_NODE_REENABLE_MAX_TIMEOUT_MS,
- TimeUnit.MILLISECONDS),
- HiveConf.getFloatVar(conf, ConfVars.LLAP_TASK_SCHEDULER_NODE_DISABLE_BACK_OFF_FACTOR));
-
- this.numSchedulableTasksPerNode =
- HiveConf.getIntVar(conf, ConfVars.LLAP_TASK_SCHEDULER_NUM_SCHEDULABLE_TASKS_PER_NODE);
-
- long localityDelayMs = HiveConf
- .getTimeVar(conf, ConfVars.LLAP_TASK_SCHEDULER_LOCALITY_DELAY, TimeUnit.MILLISECONDS);
- if (localityDelayMs == -1) {
- this.forceLocation = true;
- } else {
- this.forceLocation = false;
- }
-
- int memoryPerExecutor = (int) (memoryPerInstance / (float) executorsPerInstance);
- int coresPerExecutor = (int) (coresPerInstance / (float) executorsPerInstance);
- this.resourcePerExecutor = Resource.newInstance(memoryPerExecutor, coresPerExecutor);
-
- String instanceId = HiveConf.getTrimmedVar(conf, ConfVars.LLAP_DAEMON_SERVICE_HOSTS);
-
- Preconditions.checkNotNull(instanceId, ConfVars.LLAP_DAEMON_SERVICE_HOSTS.varname
- + " must be defined");
-
- ExecutorService executorServiceRaw =
- Executors.newFixedThreadPool(1,
- new ThreadFactoryBuilder().setDaemon(true).setNameFormat("LlapSchedulerNodeEnabler").build());
- nodeEnabledExecutor = MoreExecutors.listeningDecorator(executorServiceRaw);
-
- ExecutorService schedulerExecutorServiceRaw = Executors.newFixedThreadPool(1,
- new ThreadFactoryBuilder().setDaemon(true).setNameFormat("LlapScheduler").build());
- schedulerExecutor = MoreExecutors.listeningDecorator(schedulerExecutorServiceRaw);
-
- LOG.info("Running with configuration: " + "memoryPerInstance=" + memoryPerInstance
- + ", vCoresPerInstance=" + coresPerInstance + ", executorsPerInstance="
- + executorsPerInstance + ", resourcePerInstanceInferred=" + resourcePerExecutor
- + ", nodeBlacklistConf=" + nodeBlacklistConf
- + ", forceLocation=" + forceLocation);
- }
-
- @Override
- public void initialize() {
- registry.init(conf);
- }
-
- @Override
- public void start() throws IOException {
- writeLock.lock();
- try {
- nodeEnablerFuture = nodeEnabledExecutor.submit(nodeEnablerCallable);
- Futures.addCallback(nodeEnablerFuture, new FutureCallback<Void>() {
- @Override
- public void onSuccess(Void result) {
- LOG.info("NodeEnabledThread exited");
- }
-
- @Override
- public void onFailure(Throwable t) {
- LOG.warn("NodeEnabledThread exited with error", t);
- }
- });
- schedulerFuture = schedulerExecutor.submit(schedulerCallable);
- Futures.addCallback(schedulerFuture, new FutureCallback<Void>() {
- @Override
- public void onSuccess(Void result) {
- LOG.info("SchedulerThread exited");
- }
-
- @Override
- public void onFailure(Throwable t) {
- LOG.warn("SchedulerThread exited with error", t);
- }
- });
- registry.start();
- activeInstances = registry.getInstances();
- for (ServiceInstance inst : activeInstances.getAll().values()) {
- addNode(inst, new NodeInfo(inst, nodeBlacklistConf, clock, numSchedulableTasksPerNode));
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- @Override
- public void shutdown() {
- writeLock.lock();
- try {
- if (!this.isStopped.getAndSet(true)) {
- nodeEnablerCallable.shutdown();
- if (nodeEnablerFuture != null) {
- nodeEnablerFuture.cancel(true);
- }
- nodeEnabledExecutor.shutdownNow();
-
- schedulerCallable.shutdown();
- if (schedulerFuture != null) {
- schedulerFuture.cancel(true);
- }
- schedulerExecutor.shutdownNow();
-
- if (registry != null) {
- registry.stop();
- }
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- @Override
- public Resource getTotalResources() {
- int memory = 0;
- int vcores = 0;
- readLock.lock();
- try {
- for (ServiceInstance inst : activeInstances.getAll().values()) {
- if (inst.isAlive()) {
- Resource r = inst.getResource();
- LOG.info("Found instance " + inst);
- memory += r.getMemory();
- vcores += r.getVirtualCores();
- } else {
- LOG.info("Ignoring dead instance " + inst);
- }
- }
- } finally {
- readLock.unlock();
- }
-
- return Resource.newInstance(memory, vcores);
- }
-
- /**
- * The difference between this and getTotalResources() is that this only gives currently free
- * resource instances, while the other lists all the instances that may become available in a
- * while.
- */
- @Override
- public Resource getAvailableResources() {
- // need a state store eventually for current state & measure backoffs
- int memory = 0;
- int vcores = 0;
- readLock.lock();
- try {
- for (Entry<ServiceInstance, NodeInfo> entry : instanceToNodeMap.entrySet()) {
- if (entry.getKey().isAlive() && !entry.getValue().isDisabled()) {
- Resource r = entry.getKey().getResource();
- memory += r.getMemory();
- vcores += r.getVirtualCores();
- }
- }
- } finally {
- readLock.unlock();
- }
-
- return Resource.newInstance(memory, vcores);
- }
-
- @Override
- public int getClusterNodeCount() {
- readLock.lock();
- try {
- int n = 0;
- for (ServiceInstance inst : activeInstances.getAll().values()) {
- if (inst.isAlive()) {
- n++;
- }
- }
- return n;
- } finally {
- readLock.unlock();
- }
- }
-
- @Override
- public void dagComplete() {
- // This is effectively DAG completed, and can be used to reset statistics being tracked.
- LOG.info("DAG: " + dagCounter.get() + " completed. Scheduling stats: " + dagStats);
- dagCounter.incrementAndGet();
- dagStats = new StatsPerDag();
- }
-
- @Override
- public void blacklistNode(NodeId nodeId) {
- LOG.info("BlacklistNode not supported");
- }
-
- @Override
- public void unblacklistNode(NodeId nodeId) {
- LOG.info("unBlacklistNode not supported");
- }
-
- @Override
- public void allocateTask(Object task, Resource capability, String[] hosts, String[] racks,
- Priority priority, Object containerSignature, Object clientCookie) {
- TaskInfo taskInfo =
- new TaskInfo(task, clientCookie, priority, capability, hosts, racks, clock.getTime());
- writeLock.lock();
- try {
- dagStats.registerTaskRequest(hosts, racks);
- } finally {
- writeLock.unlock();
- }
- addPendingTask(taskInfo);
- trySchedulingPendingTasks();
- }
-
- @Override
- public void allocateTask(Object task, Resource capability, ContainerId containerId,
- Priority priority, Object containerSignature, Object clientCookie) {
- // Container affinity can be implemented as Host affinity for LLAP. Not required until
- // 1:1 edges are used in Hive.
- TaskInfo taskInfo =
- new TaskInfo(task, clientCookie, priority, capability, null, null, clock.getTime());
- writeLock.lock();
- try {
- dagStats.registerTaskRequest(null, null);
- } finally {
- writeLock.unlock();
- }
- addPendingTask(taskInfo);
- trySchedulingPendingTasks();
- }
-
-
- // This may be invoked before a container is ever assigned to a task. allocateTask... app decides
- // the task is no longer required, and asks for a de-allocation.
- @Override
- public boolean deallocateTask(Object task, boolean taskSucceeded, TaskAttemptEndReason endReason, String diagnostics) {
- writeLock.lock(); // Updating several local structures
- TaskInfo taskInfo;
- try {
- taskInfo = unregisterTask(task);
- if (taskInfo == null) {
- LOG.error("Could not determine ContainerId for task: "
- + task
- + " . Could have hit a race condition. Ignoring."
- + " The query may hang since this \"unknown\" container is now taking up a slot permanently");
- return false;
- }
- if (taskInfo.containerId == null) {
- if (taskInfo.assigned) {
- LOG.error("Task: "
- + task
- + " assigned, but could not find the corresponding containerId."
- + " The query may hang since this \"unknown\" container is now taking up a slot permanently");
- } else {
- LOG.info("Ignoring deallocate request for task " + task
- + " which hasn't been assigned to a container");
- removePendingTask(taskInfo);
- }
- return false;
- }
- ServiceInstance assignedInstance = taskInfo.assignedInstance;
- assert assignedInstance != null;
-
- NodeInfo nodeInfo = instanceToNodeMap.get(assignedInstance);
- assert nodeInfo != null;
-
- // Re-enable the node if preempted
- if (taskInfo.preempted) {
- LOG.info("Processing deallocateTask for {} which was preempted, EndReason={}", task, endReason);
- unregisterPendingPreemption(taskInfo.assignedInstance.getHost());
- nodeInfo.registerUnsuccessfulTaskEnd(true);
- if (nodeInfo.isDisabled()) {
- // Re-enable the node. If a task succeeded, a slot may have become available.
- // Also reset commFailures since a task was able to communicate back and indicate success.
- nodeInfo.enableNode();
- // Re-insert into the queue to force the poll thread to remove the element.
- if ( disabledNodesQueue.remove(nodeInfo)) {
- disabledNodesQueue.add(nodeInfo);
- }
- }
- // In case of success, trigger a scheduling run for pending tasks.
- trySchedulingPendingTasks();
- } else {
- if (taskSucceeded) {
- // The node may have been blacklisted at this point - which means it may not be in the
- // activeNodeList.
-
- nodeInfo.registerTaskSuccess();
-
- if (nodeInfo.isDisabled()) {
- // Re-enable the node. If a task succeeded, a slot may have become available.
- // Also reset commFailures since a task was able to communicate back and indicate success.
- nodeInfo.enableNode();
- // Re-insert into the queue to force the poll thread to remove the element.
- if (disabledNodesQueue.remove(nodeInfo)) {
- disabledNodesQueue.add(nodeInfo);
- }
- }
- // In case of success, trigger a scheduling run for pending tasks.
- trySchedulingPendingTasks();
-
- } else if (!taskSucceeded) {
- nodeInfo.registerUnsuccessfulTaskEnd(false);
- if (endReason != null && EnumSet
- .of(TaskAttemptEndReason.EXECUTOR_BUSY, TaskAttemptEndReason.COMMUNICATION_ERROR)
- .contains(endReason)) {
- if (endReason == TaskAttemptEndReason.COMMUNICATION_ERROR) {
- dagStats.registerCommFailure(taskInfo.assignedInstance.getHost());
- } else if (endReason == TaskAttemptEndReason.EXECUTOR_BUSY) {
- dagStats.registerTaskRejected(taskInfo.assignedInstance.getHost());
- }
- }
- boolean commFailure =
- endReason != null && endReason == TaskAttemptEndReason.COMMUNICATION_ERROR;
- disableInstance(assignedInstance, commFailure);
- }
- }
- } finally {
- writeLock.unlock();
- }
- getContext().containerBeingReleased(taskInfo.containerId);
- return true;
- }
-
- @Override
- public Object deallocateContainer(ContainerId containerId) {
- LOG.debug("Ignoring deallocateContainer for containerId: " + containerId);
- // Containers are not being tracked for re-use.
- // This is safe to ignore since a deallocate task will come in.
- return null;
- }
-
- @Override
- public void setShouldUnregister() {
-
- }
-
- @Override
- public boolean hasUnregistered() {
- // Nothing to do. No registration involved.
- return true;
- }
-
- private ExecutorService createAppCallbackExecutorService() {
- return Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
- .setNameFormat("TaskSchedulerAppCaller #%d").setDaemon(true).build());
- }
-
- /**
- * @param request the list of preferred hosts. null implies any host
- * @return
- */
- private SelectHostResult selectHost(TaskInfo request) {
- String[] requestedHosts = request.requestedHosts;
- readLock.lock(); // Read-lock. Not updating any stats at the moment.
- try {
- // Check if any hosts are active.
- if (getAvailableResources().getMemory() <= 0) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Refreshing instances since total memory is 0");
- }
- refreshInstances();
- }
-
- // If there's no memory available, fail
- if (getTotalResources().getMemory() <= 0) {
- return SELECT_HOST_RESULT_INADEQUATE_TOTAL_CAPACITY;
- }
-
- if (requestedHosts != null && requestedHosts.length > 0) {
- int prefHostCount = -1;
- boolean requestedHostExists = false;
- for (String host : requestedHosts) {
- prefHostCount++;
- // Pick the first host always. Weak attempt at cache affinity.
- Set<ServiceInstance> instances = activeInstances.getByHost(host);
- if (!instances.isEmpty()) {
- requestedHostExists = true;
- for (ServiceInstance inst : instances) {
- NodeInfo nodeInfo = instanceToNodeMap.get(inst);
- if (nodeInfo != null && nodeInfo.canAcceptTask()) {
- LOG.info("Assigning " + inst + " when looking for " + host + "." +
- " FirstRequestedHost=" + (prefHostCount == 0) +
- (requestedHosts.length > 1 ? "#prefLocations=" + requestedHosts.length : ""));
- return new SelectHostResult(inst, nodeInfo);
- }
- }
- }
- }
- // Check if forcing the location is required.
- if (forceLocation) {
- if (requestedHostExists) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Skipping non-local location allocation for [" + request.task +
- "] when trying to allocate on [" + Arrays.toString(requestedHosts) + "]");
- }
- return SELECT_HOST_RESULT_DELAYED_LOCALITY;
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Not skipping non-local location allocation for [" + request.task +
- "] when trying to allocate on [" + Arrays.toString(requestedHosts) +
- "] since none of these hosts are part of the known list");
- }
- }
- }
- }
- /* fall through - miss in locality (random scheduling) */
- Entry<ServiceInstance, NodeInfo>[] all =
- instanceToNodeMap.entrySet().toArray(new Entry[instanceToNodeMap.size()]);
- // Check again
- if (all.length > 0) {
- int n = random.nextInt(all.length);
- // start at random offset and iterate whole list
- for (int i = 0; i < all.length; i++) {
- Entry<ServiceInstance, NodeInfo> inst = all[(i + n) % all.length];
- if (inst.getValue().canAcceptTask()) {
- LOG.info("Assigning " + inst + " when looking for any host, from #hosts=" + all.length +
- ", requestedHosts=" +
- ((requestedHosts == null || requestedHosts.length == 0) ? "null" :
- Arrays.toString(requestedHosts)));
- return new SelectHostResult(inst.getKey(), inst.getValue());
- }
- }
- }
- return SELECT_HOST_RESULT_DELAYED_RESOURCES;
- } finally {
- readLock.unlock();
- }
- }
-
- // TODO Each refresh operation should addNodes if they don't already exist.
- // Even better would be to get notifications from the service impl when a node gets added or removed.
- // Instead of having to walk through the entire list. The computation of a node getting added or
- // removed already exists in the DynamicRegistry implementation.
- private void refreshInstances() {
- try {
- activeInstances.refresh(); // handles its own sync
- } catch (IOException ioe) {
- LOG.warn("Could not refresh list of active instances", ioe);
- }
- }
-
- private void scanForNodeChanges() {
- /* check again whether nodes are disabled or just missing */
- writeLock.lock();
- try {
- for (ServiceInstance inst : activeInstances.getAll().values()) {
- if (inst.isAlive() && instanceToNodeMap.containsKey(inst) == false) {
- /* that's a good node, not added to the allocations yet */
- LOG.info("Found a new node: " + inst + ".");
- addNode(inst, new NodeInfo(inst, nodeBlacklistConf, clock, numSchedulableTasksPerNode));
- }
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- private void addNode(ServiceInstance inst, NodeInfo node) {
- LOG.info("Adding node: " + inst);
- instanceToNodeMap.put(inst, node);
- // Trigger scheduling since a new node became available.
- trySchedulingPendingTasks();
- }
-
- private void reenableDisabledNode(NodeInfo nodeInfo) {
- writeLock.lock();
- try {
- if (nodeInfo.hadCommFailure()) {
- // If the node being re-enabled was not marked busy previously, then it was disabled due to
- // some other failure. Refresh the service list to see if it's been removed permanently.
- refreshInstances();
- }
- LOG.info("Attempting to re-enable node: " + nodeInfo.getServiceInstance());
- if (nodeInfo.getServiceInstance().isAlive()) {
- nodeInfo.enableNode();
- } else {
- if (LOG.isInfoEnabled()) {
- LOG.info("Removing dead node " + nodeInfo);
- }
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- private void disableInstance(ServiceInstance instance, boolean isCommFailure) {
- writeLock.lock();
- try {
- NodeInfo nodeInfo = instanceToNodeMap.get(instance);
- if (nodeInfo == null || nodeInfo.isDisabled()) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Node: " + instance + " already disabled, or invalid. Not doing anything.");
- }
- } else {
- nodeInfo.disableNode(isCommFailure);
- // TODO: handle task to container map events in case of hard failures
- disabledNodesQueue.add(nodeInfo);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- private void addPendingTask(TaskInfo taskInfo) {
- writeLock.lock();
- try {
- List<TaskInfo> tasksAtPriority = pendingTasks.get(taskInfo.priority);
- if (tasksAtPriority == null) {
- tasksAtPriority = new LinkedList<>();
- pendingTasks.put(taskInfo.priority, tasksAtPriority);
- }
- tasksAtPriority.add(taskInfo);
- knownTasks.putIfAbsent(taskInfo.task, taskInfo);
- } finally {
- writeLock.unlock();
- }
- }
-
- /* Remove a task from the pending list */
- private void removePendingTask(TaskInfo taskInfo) {
- writeLock.lock();
- try {
- Priority priority = taskInfo.priority;
- List<TaskInfo> taskInfoList = pendingTasks.get(priority);
- if (taskInfoList == null || taskInfoList.isEmpty() || !taskInfoList.remove(taskInfo)) {
- LOG.warn("Could not find task: " + taskInfo.task + " in pending list, at priority: "
- + priority);
- }
- } finally {
- writeLock.unlock();
- }
- }
-
- /* Register a running task into the runningTasks structure */
- private void registerRunningTask(TaskInfo taskInfo) {
- writeLock.lock();
- try {
- int priority = taskInfo.priority.getPriority();
- TreeSet<TaskInfo> tasksAtpriority = runningTasks.get(priority);
- if (tasksAtpriority == null) {
- tasksAtpriority = new TreeSet<>(TASK_INFO_COMPARATOR);
- runningTasks.put(priority, tasksAtpriority);
- }
- tasksAtpriority.add(taskInfo);
- } finally {
- writeLock.unlock();
- }
- }
-
- /* Unregister a task from the known and running structures */
- private TaskInfo unregisterTask(Object task) {
- writeLock.lock();
- try {
- TaskInfo taskInfo = knownTasks.remove(task);
- if (taskInfo != null) {
- if (taskInfo.assigned) {
- // Remove from the running list.
- int priority = taskInfo.priority.getPriority();
- Set<TaskInfo> tasksAtPriority = runningTasks.get(priority);
- Preconditions.checkState(tasksAtPriority != null,
- "runningTasks should contain an entry if the task was in running state. Caused by task: {}", task);
- tasksAtPriority.remove(taskInfo);
- if (tasksAtPriority.isEmpty()) {
- runningTasks.remove(priority);
- }
- }
- } else {
- LOG.warn("Could not find TaskInfo for task: {}. Not removing it from the running set", task);
- }
- return taskInfo;
- } finally {
- writeLock.unlock();
- }
- }
-
- private enum ScheduleResult {
- // Successfully scheduled
- SCHEDULED,
-
- // Delayed to find a local match
- DELAYED_LOCALITY,
-
- // Delayed due to temporary resource availability
- DELAYED_RESOURCES,
-
- // Inadequate total resources - will never succeed / wait for new executors to become available
- INADEQUATE_TOTAL_RESOURCES,
- }
-
- @VisibleForTesting
- protected void schedulePendingTasks() {
- writeLock.lock();
- try {
- Iterator<Entry<Priority, List<TaskInfo>>> pendingIterator =
- pendingTasks.entrySet().iterator();
- while (pendingIterator.hasNext()) {
- Entry<Priority, List<TaskInfo>> entry = pendingIterator.next();
- List<TaskInfo> taskListAtPriority = entry.getValue();
- Iterator<TaskInfo> taskIter = taskListAtPriority.iterator();
- boolean scheduledAllAtPriority = true;
- while (taskIter.hasNext()) {
-
- // TODO Optimization: Add a check to see if there's any capacity available. No point in
- // walking through all active nodes, if they don't have potential capacity.
-
- TaskInfo taskInfo = taskIter.next();
- if (taskInfo.getNumPreviousAssignAttempts() == 1) {
- dagStats.registerDelayedAllocation();
- }
- taskInfo.triedAssigningTask();
- ScheduleResult scheduleResult = scheduleTask(taskInfo);
- if (scheduleResult == ScheduleResult.SCHEDULED) {
- taskIter.remove();
- } else {
- // TODO Handle INADEQUATE_TOTAL_RESOURCES eventually - either by throwin an error immediately,
- // or waiting for some timeout for new executors and then throwing an error
-
- // Try pre-empting a task so that a higher priority task can take it's place.
- // Preempt only if there's no pending preemptions to avoid preempting twice for a task.
- String[] potentialHosts;
- if (scheduleResult == ScheduleResult.DELAYED_LOCALITY) {
- // preempt only on specific hosts, if no preemptions already exist on those.
- potentialHosts = taskInfo.requestedHosts;
- //Protect against a bad location being requested.
- if (potentialHosts == null || potentialHosts.length == 0) {
- potentialHosts = null;
- }
- } else {
- // preempt on any host.
- potentialHosts = null;
- }
-
- if (potentialHosts != null) {
- // Preempt on specific host
- boolean shouldPreempt = true;
- for (String host : potentialHosts) {
- // Preempt only if there are not pending preemptions on the same host
- // When the premption registers, the request at the highest priority will be given the slot,
- // even if the initial request was for some other task.
- // TODO Maybe register which task the preemption was for, to avoid a bad non-local allocation.
- MutableInt pendingHostPreemptions = pendingPreemptionsPerHost.get(host);
- if (pendingHostPreemptions != null && pendingHostPreemptions.intValue() > 0) {
- shouldPreempt = false;
- break;
- }
- }
- if (shouldPreempt) {
- LOG.info("Attempting to preempt for {}, pendingPreemptions={} on hosts={}",
- taskInfo.task, pendingPreemptions.get(), Arrays.toString(potentialHosts));
- preemptTasks(entry.getKey().getPriority(), 1, potentialHosts);
- }
- } else {
- // Request for a preemption if there's none pending. If a single preemption is pending,
- // and this is the next task to be assigned, it will be assigned once that slot becomes available.
- if (pendingPreemptions.get() == 0) {
- LOG.info("Attempting to preempt for {}, pendingPreemptions={} on any host",
- taskInfo.task, pendingPreemptions.get());
- preemptTasks(entry.getKey().getPriority(), 1, null);
- }
- }
- // Since there was an allocation failure - don't try assigning tasks at the next priority.
-
- scheduledAllAtPriority = false;
- // Don't break if this allocation failure was a result of a LOCALITY_DELAY. Others could still be allocated.
- if (scheduleResult != ScheduleResult.DELAYED_LOCALITY) {
- break;
- }
- } // end of else - i.e. could not allocate
- } // end of loop over pending tasks
- if (taskListAtPriority.isEmpty()) {
- // Remove the entry, if there's nothing left at the specific priority level
- pendingIterator.remove();
- }
- if (!scheduledAllAtPriority) {
- // Don't attempt scheduling for additional priorities
- break;
- }
- }
- } finally {
- writeLock.unlock();
- }
- }
-
-
- private ScheduleResult scheduleTask(TaskInfo taskInfo) {
- SelectHostResult selectHostResult = selectHost(taskInfo);
- if (selectHostResult.scheduleResult == ScheduleResult.SCHEDULED) {
- NodeServiceInstancePair nsPair = selectHostResult.nodeServiceInstancePair;
- Container container =
- containerFactory.createContainer(resourcePerExecutor, taskInfo.priority,
- nsPair.getServiceInstance().getHost(),
- nsPair.getServiceInstance().getRpcPort());
- writeLock.lock(); // While updating local structures
- try {
- LOG.info("Assigned task {} to container {}", taskInfo, container.getId());
- dagStats.registerTaskAllocated(taskInfo.requestedHosts, taskInfo.requestedRacks,
- nsPair.getServiceInstance().getHost());
- taskInfo.setAssignmentInfo(nsPair.getServiceInstance(), container.getId(), clock.getTime());
- registerRunningTask(taskInfo);
- nsPair.getNodeInfo().registerTaskScheduled();
- } finally {
- writeLock.unlock();
- }
- getContext().taskAllocated(taskInfo.task, taskInfo.clientCookie, container);
- }
- return selectHostResult.scheduleResult;
- }
-
- // Removes tasks from the runningList and sends out a preempt request to the system.
- // Subsequent tasks will be scheduled again once the de-allocate request for the preempted
- // task is processed.
- private void preemptTasks(int forPriority, int numTasksToPreempt, String []potentialHosts) {
- Set<String> preemptHosts;
- if (potentialHosts == null) {
- preemptHosts = null;
- } else {
- preemptHosts = Sets.newHashSet(potentialHosts);
- }
- writeLock.lock();
- List<TaskInfo> preemptedTaskList = null;
- try {
- NavigableMap<Integer, TreeSet<TaskInfo>> orderedMap = runningTasks.descendingMap();
- Iterator<Entry<Integer, TreeSet<TaskInfo>>> iterator = orderedMap.entrySet().iterator();
- int preemptedCount = 0;
- while (iterator.hasNext() && preemptedCount < numTasksToPreempt) {
- Entry<Integer, TreeSet<TaskInfo>> entryAtPriority = iterator.next();
- if (entryAtPriority.getKey() > forPriority) {
- Iterator<TaskInfo> taskInfoIterator = entryAtPriority.getValue().iterator();
- while (taskInfoIterator.hasNext() && preemptedCount < numTasksToPreempt) {
- TaskInfo taskInfo = taskInfoIterator.next();
- if (preemptHosts == null || preemptHosts.contains(taskInfo.assignedInstance.getHost())) {
- // Candidate for preemption.
- preemptedCount++;
- LOG.info("preempting {} for task at priority {} with potentialHosts={}", taskInfo,
- forPriority, potentialHosts == null ? "" : Arrays.toString(potentialHosts));
- taskInfo.setPreemptedInfo(clock.getTime());
- if (preemptedTaskList == null) {
- preemptedTaskList = new LinkedList<>();
- }
- dagStats.registerTaskPreempted(taskInfo.assignedInstance.getHost());
- preemptedTaskList.add(taskInfo);
- registerPendingPreemption(taskInfo.assignedInstance.getHost());
- // Remove from the runningTaskList
- taskInfoIterator.remove();
- }
- }
-
- // Remove entire priority level if it's been emptied.
- if (entryAtPriority.getValue().isEmpty()) {
- iterator.remove();
- }
- } else {
- // No tasks qualify as preemptable
- LOG.info("DBG: No tasks qualify as killable to schedule tasks at priority {}", forPriority);
- break;
- }
- }
- } finally {
- writeLock.unlock();
- }
- // Send out the preempted request outside of the lock.
- if (preemptedTaskList != null) {
- for (TaskInfo taskInfo : preemptedTaskList) {
- LOG.info("DBG: Preempting task {}", taskInfo);
- getContext().preemptContainer(taskInfo.containerId);
- // Preemption will finally be registered as a deallocateTask as a result of preemptContainer
- // That resets preemption info and allows additional tasks to be pre-empted if required.
- }
- }
- // The schedule loop will be triggered again when the deallocateTask request comes in for the
- // preempted task.
- }
-
- private void registerPendingPreemption(String host) {
- writeLock.lock();
- try {
- pendingPreemptions.incrementAndGet();
- MutableInt val = pendingPreemptionsPerHost.get(host);
- if (val == null) {
- val = new MutableInt(1);
- pendingPreemptionsPerHost.put(host, val);
- }
- val.increment();
- } finally {
- writeLock.unlock();
- }
- }
-
- private void unregisterPendingPreemption(String host) {
- writeLock.lock();
- try {
- pendingPreemptions.decrementAndGet();
- MutableInt val = pendingPreemptionsPerHost.get(host);
- Preconditions.checkNotNull(val);
- val.decrement();
- // Not bothering with removing the entry. There's a limited number of hosts, and a good
- // chance that the entry will make it back in when the AM is used for a long duration.
- } finally {
- writeLock.unlock();
- }
- }
-
- private class NodeEnablerCallable implements Callable<Void> {
-
- private AtomicBoolean isShutdown = new AtomicBoolean(false);
- private static final long REFRESH_INTERVAL = 10000l;
- long nextPollInterval = REFRESH_INTERVAL;
- long lastRefreshTime;
-
- @Override
- public Void call() {
-
- lastRefreshTime = System.currentTimeMillis();
- while (!isShutdown.get() && !Thread.currentThread().isInterrupted()) {
- try {
- while (true) {
- NodeInfo nodeInfo = disabledNodesQueue.poll(nextPollInterval, TimeUnit.MILLISECONDS);
- if (nodeInfo != null) {
- long currentTime = System.currentTimeMillis();
- // A node became available. Enable the node and try scheduling.
- reenableDisabledNode(nodeInfo);
- trySchedulingPendingTasks();
-
- nextPollInterval -= (currentTime - lastRefreshTime);
- }
-
- if (nextPollInterval < 0 || nodeInfo == null) {
- // timeout expired. Reset the poll interval and refresh nodes.
- nextPollInterval = REFRESH_INTERVAL;
- lastRefreshTime = System.currentTimeMillis();
- // TODO Get rid of this polling once we have notificaitons from the registry sub-system
- if (LOG.isDebugEnabled()) {
- LOG.debug("Refreshing instances based on poll interval");
- }
- refreshInstances();
- scanForNodeChanges();
- }
- }
- } catch (InterruptedException e) {
- if (isShutdown.get()) {
- LOG.info("NodeEnabler thread interrupted after shutdown");
- break;
- } else {
- LOG.warn("NodeEnabler thread interrupted without being shutdown");
- throw new RuntimeException("NodeEnabler thread interrupted without being shutdown", e);
- }
- }
- }
- return null;
- }
-
- // Call this first, then send in an interrupt to the thread.
- public void shutdown() {
- isShutdown.set(true);
- }
- }
-
- private void trySchedulingPendingTasks() {
- scheduleLock.lock();
- try {
- pendingScheduleInvodations.set(true);
- scheduleCondition.signal();
- } finally {
- scheduleLock.unlock();
- }
- }
-
- private class SchedulerCallable implements Callable<Void> {
- private AtomicBoolean isShutdown = new AtomicBoolean(false);
-
- @Override
- public Void call() {
- while (!isShutdown.get() && !Thread.currentThread().isInterrupted()) {
- scheduleLock.lock();
- try {
- while (!pendingScheduleInvodations.get()) {
- scheduleCondition.await();
- }
- } catch (InterruptedException e) {
- if (isShutdown.get()) {
- LOG.info("Scheduler thread interrupted after shutdown");
- break;
- } else {
- LOG.warn("Scheduler thread interrupted without being shutdown");
- throw new RuntimeException("Scheduler thread interrupted without being shutdown", e);
- }
- } finally {
- scheduleLock.unlock();
- }
-
- // Set pending to false since scheduling is about to run. Any triggers up to this point
- // will be handled in the next run.
- // A new request may come in right after this is set to false, but before the actual scheduling.
- // This will be handled in this run, but will cause an immediate run after, which is harmless.
- pendingScheduleInvodations.set(false);
- // Schedule outside of the scheduleLock - which should only be used to wait on the condition.
- schedulePendingTasks();
- }
- return null;
- }
-
- // Call this first, then send in an interrupt to the thread.
- public void shutdown() {
- isShutdown.set(true);
- }
- }
-
- @VisibleForTesting
- static class NodeInfo implements Delayed {
- private final NodeBlacklistConf blacklistConf;
- final ServiceInstance serviceInstance;
- private final Clock clock;
-
- long expireTimeMillis = -1;
- private long numSuccessfulTasks = 0;
- private long numSuccessfulTasksAtLastBlacklist = -1;
- float cumulativeBackoffFactor = 1.0f;
-
- // Indicates whether a node had a recent communication failure.
- private boolean hadCommFailure = false;
-
- // Indicates whether a node is disabled - for whatever reason - commFailure, busy, etc.
- private boolean disabled = false;
-
- private int numPreemptedTasks = 0;
- private int numScheduledTasks = 0;
- private final int numSchedulableTasks;
-
-
- /**
- * Create a NodeInfo bound to a service instance
- *
- * @param serviceInstance the associated serviceInstance
- * @param blacklistConf blacklist configuration
- * @param clock clock to use to obtain timing information
- * @param numSchedulableTasksConf number of schedulable tasks on the node. 0 represents auto
- * detect based on the serviceInstance, -1 indicates indicates
- * unlimited capacity
- */
- NodeInfo(ServiceInstance serviceInstance, NodeBlacklistConf blacklistConf, Clock clock, int numSchedulableTasksConf) {
- Preconditions.checkArgument(numSchedulableTasksConf >= -1, "NumSchedulableTasks must be >=-1");
- this.serviceInstance = serviceInstance;
- this.blacklistConf = blacklistConf;
- this.clock = clock;
-
- if (numSchedulableTasksConf == 0) {
- int pendingQueueuCapacity = 0;
- String pendingQueueCapacityString = serviceInstance.getProperties()
- .get(ConfVars.LLAP_DAEMON_TASK_SCHEDULER_WAIT_QUEUE_SIZE.varname);
- if (LOG.isDebugEnabled()) {
- LOG.debug("Setting up node: " + serviceInstance + ", with available capacity=" +
- serviceInstance.getResource().getVirtualCores() + ", pendingQueueCapacity=" +
- pendingQueueCapacityString);
- }
-
- if (pendingQueueCapacityString != null) {
- pendingQueueuCapacity = Integer.parseInt(pendingQueueCapacityString);
- }
- this.numSchedulableTasks = serviceInstance.getResource().getVirtualCores() + pendingQueueuCapacity;
- } else {
- this.numSchedulableTasks = numSchedulableTasksConf;
- }
- LOG.info("Setting up node: " + serviceInstance + " with schedulableCapacity=" + this.numSchedulableTasks);
- }
-
- ServiceInstance getServiceInstance() {
- return serviceInstance;
- }
-
- void enableNode() {
- expireTimeMillis = -1;
- disabled = false;
- hadCommFailure = false;
- }
-
- void disableNode(boolean commFailure) {
- long duration = blacklistConf.minDelay;
- long currentTime = clock.getTime();
- this.hadCommFailure = commFailure;
- disabled = true;
- if (numSuccessfulTasksAtLastBlacklist == numSuccessfulTasks) {
- // Relying on a task succeeding to reset the exponent.
- // There's no notifications on whether a task gets accepted or not. That would be ideal to
- // reset this.
- cumulativeBackoffFactor = cumulativeBackoffFactor * blacklistConf.backoffFactor;
- } else {
- // Was able to execute something before the last blacklist. Reset the exponent.
- cumulativeBackoffFactor = 1.0f;
- }
-
- long delayTime = (long) (duration * cumulativeBackoffFactor);
- if (delayTime > blacklistConf.maxDelay) {
- delayTime = blacklistConf.maxDelay;
- }
- if (LOG.isInfoEnabled()) {
- LOG.info("Disabling instance " + serviceInstance + " for " + delayTime + " milli-seconds");
- }
- expireTimeMillis = currentTime + delayTime;
- numSuccessfulTasksAtLastBlacklist = numSuccessfulTasks;
- }
-
- void registerTaskScheduled() {
- numScheduledTasks++;
- }
-
- void registerTaskSuccess() {
- numSuccessfulTasks++;
- numScheduledTasks--;
- }
-
- void registerUnsuccessfulTaskEnd(boolean wasPreempted) {
- numScheduledTasks--;
- if (wasPreempted) {
- numPreemptedTasks++;
- }
- }
-
- public boolean isDisabled() {
- return disabled;
- }
-
- public boolean hadCommFailure() {
- return hadCommFailure;
- }
- /* Returning true does not guarantee that the task will run, considering other queries
- may be running in the system. Also depends upon the capacity usage configuration
- */
- public boolean canAcceptTask() {
- boolean result = !hadCommFailure && !disabled && serviceInstance.isAlive()
- &&(numSchedulableTasks == -1 || ((numSchedulableTasks - numScheduledTasks) > 0));
- LOG.info("canAcceptTask={}, numScheduledTasks={}, numSchedulableTasks={}, hadCommFailure={}, disabled={}, serviceInstance.isAlive={}", result, numScheduledTasks, numSchedulableTasks, hadCommFailure, disabled, serviceInstance.isAlive());
- return result;
- }
-
- @Override
- public long getDelay(TimeUnit unit) {
- return unit.convert(expireTimeMillis - clock.getTime(), TimeUnit.MILLISECONDS);
- }
-
- @Override
- public int compareTo(Delayed o) {
- NodeInfo other = (NodeInfo) o;
- if (other.expireTimeMillis > this.expireTimeMillis) {
- return -1;
- } else if (other.expireTimeMillis < this.expireTimeMillis) {
- return 1;
- } else {
- return 0;
- }
- }
-
- @Override
- public String toString() {
- return "NodeInfo{" + "instance=" + serviceInstance
- + ", expireTimeMillis=" + expireTimeMillis + ", numSuccessfulTasks=" + numSuccessfulTasks
- + ", numSuccessfulTasksAtLastBlacklist=" + numSuccessfulTasksAtLastBlacklist
- + ", cumulativeBackoffFactor=" + cumulativeBackoffFactor
- + ", numSchedulableTasks=" + numSchedulableTasks
- + ", numScheduledTasks=" + numScheduledTasks
- + ", disabled=" + disabled
- + ", commFailures=" + hadCommFailure
- +'}';
- }
- }
-
- @VisibleForTesting
- static class StatsPerDag {
- int numRequestedAllocations = 0;
- int numRequestsWithLocation = 0;
- int numRequestsWithoutLocation = 0;
- int numTotalAllocations = 0;
- int numLocalAllocations = 0;
- int numNonLocalAllocations = 0;
- int numAllocationsNoLocalityRequest = 0;
- int numRejectedTasks = 0;
- int numCommFailures = 0;
- int numDelayedAllocations = 0;
- int numPreemptedTasks = 0;
- Map<String, AtomicInteger> localityBasedNumAllocationsPerHost = new HashMap<>();
- Map<String, AtomicInteger> numAllocationsPerHost = new HashMap<>();
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder();
- sb.append("NumPreemptedTasks=").append(numPreemptedTasks).append(", ");
- sb.append("NumRequestedAllocations=").append(numRequestedAllocations).append(", ");
- sb.append("NumRequestsWithlocation=").append(numRequestsWithLocation).append(", ");
- sb.append("NumLocalAllocations=").append(numLocalAllocations).append(",");
- sb.append("NumNonLocalAllocations=").append(numNonLocalAllocations).append(",");
- sb.append("NumTotalAllocations=").append(numTotalAllocations).append(",");
- sb.append("NumRequestsWithoutLocation=").append(numRequestsWithoutLocation).append(", ");
- sb.append("NumRejectedTasks=").append(numRejectedTasks).append(", ");
- sb.append("NumCommFailures=").append(numCommFailures).append(", ");
- sb.append("NumDelayedAllocations=").append(numDelayedAllocations).append(", ");
- sb.append("LocalityBasedAllocationsPerHost=").append(localityBasedNumAllocationsPerHost)
- .append(", ");
- sb.append("NumAllocationsPerHost=").append(numAllocationsPerHost);
- return sb.toString();
- }
-
- void registerTaskRequest(String[] requestedHosts, String[] requestedRacks) {
- numRequestedAllocations++;
- // TODO Change after HIVE-9987. For now, there's no rack matching.
- if (requestedHosts != null && requestedHosts.length != 0) {
- numRequestsWithLocation++;
- } else {
- numRequestsWithoutLocation++;
- }
- }
-
- void registerTaskAllocated(String[] requestedHosts, String[] requestedRacks,
- String allocatedHost) {
- // TODO Change after HIVE-9987. For now, there's no rack matching.
- if (requestedHosts != null && requestedHosts.length != 0) {
- Set<String> requestedHostSet = new HashSet<>(Arrays.asList(requestedHosts));
- if (requestedHostSet.contains(allocatedHost)) {
- numLocalAllocations++;
- _registerAllocationInHostMap(allocatedHost, localityBasedNumAllocationsPerHost);
- } else {
- numNonLocalAllocations++;
- }
- } else {
- numAllocationsNoLocalityRequest++;
- }
- numTotalAllocations++;
- _registerAllocationInHostMap(allocatedHost, numAllocationsPerHost);
- }
-
- void registerTaskPreempted(String host) {
- numPreemptedTasks++;
- }
-
- void registerCommFailure(String host) {
- numCommFailures++;
- }
-
- void registerTaskRejected(String host) {
- numRejectedTasks++;
- }
-
- void registerDelayedAllocation() {
- numDelayedAllocations++;
- }
-
- private void _registerAllocationInHostMap(String host, Map<String, AtomicInteger> hostMap) {
- AtomicInteger val = hostMap.get(host);
- if (val == null) {
- val = new AtomicInteger(0);
- hostMap.put(host, val);
- }
- val.incrementAndGet();
- }
- }
-
- private static class TaskInfo {
- // IDs used to ensure two TaskInfos are different without using the underlying task instance.
- // Required for insertion into a TreeMap
- static final AtomicLong ID_GEN = new AtomicLong(0);
- final long uniqueId;
- final Object task;
- final Object clientCookie;
- final Priority priority;
- final Resource capability;
- final String[] requestedHosts;
- final String[] requestedRacks;
- final long requestTime;
- long startTime;
- long preemptTime;
- ContainerId containerId;
- ServiceInstance assignedInstance;
- private boolean assigned = false;
- private boolean preempted = false;
-
- private int numAssignAttempts = 0;
-
- // TaskInfo instances for two different tasks will not be the same. Only a single instance should
- // ever be created for a taskAttempt
- public TaskInfo(Object task, Object clientCookie, Priority priority, Resource capability,
- String[] hosts, String[] racks, long requestTime) {
- this.task = task;
- this.clientCookie = clientCookie;
- this.priority = priority;
- this.capability = capability;
- this.requestedHosts = hosts;
- this.requestedRacks = racks;
- this.requestTime = requestTime;
- this.uniqueId = ID_GEN.getAndIncrement();
- }
-
- void setAssignmentInfo(ServiceInstance instance, ContainerId containerId, long startTime) {
- this.assignedInstance = instance;
- this.containerId = containerId;
- this.startTime = startTime;
- assigned = true;
- }
-
- void setPreemptedInfo(long preemptTime) {
- this.preempted = true;
- this.assigned = false;
- this.preemptTime = preemptTime;
- }
-
- void triedAssigningTask() {
- numAssignAttempts++;
- }
-
- int getNumPreviousAssignAttempts() {
- return numAssignAttempts;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- TaskInfo taskInfo = (TaskInfo) o;
-
- if (uniqueId != taskInfo.uniqueId) {
- return false;
- }
- return task.equals(taskInfo.task);
-
- }
-
- @Override
- public int hashCode() {
- int result = (int) (uniqueId ^ (uniqueId >>> 32));
- result = 31 * result + task.hashCode();
- return result;
- }
-
- @Override
- public String toString() {
- return "TaskInfo{" +
- "task=" + task +
- ", priority=" + priority +
- ", startTime=" + startTime +
- ", containerId=" + containerId +
- ", assignedInstance=" + assignedInstance +
- ", uniqueId=" + uniqueId +
- '}';
- }
- }
-
- // Newer tasks first.
- private static class TaskStartComparator implements Comparator<TaskInfo> {
-
- @Override
- public int compare(TaskInfo o1, TaskInfo o2) {
- if (o1.startTime > o2.startTime) {
- return -1;
- } else if (o1.startTime < o2.startTime) {
- return 1;
- } else {
- // Comparing on time is not sufficient since two may be created at the same time,
- // in which case inserting into a TreeSet/Map would break
- if (o1.uniqueId > o2.uniqueId) {
- return -1;
- } else if (o1.uniqueId < o2.uniqueId) {
- return 1;
- } else {
- return 0;
- }
- }
- }
- }
-
- private static class SelectHostResult {
- final NodeServiceInstancePair nodeServiceInstancePair;
- final ScheduleResult scheduleResult;
-
- SelectHostResult(ServiceInstance serviceInstance, NodeInfo nodeInfo) {
- this.nodeServiceInstancePair = new NodeServiceInstancePair(serviceInstance, nodeInfo);
- this.scheduleResult = ScheduleResult.SCHEDULED;
- }
-
- SelectHostResult(ScheduleResult scheduleResult) {
- this.nodeServiceInstancePair = null;
- this.scheduleResult = scheduleResult;
- }
- }
-
- private static final SelectHostResult SELECT_HOST_RESULT_INADEQUATE_TOTAL_CAPACITY =
- new SelectHostResult(ScheduleResult.INADEQUATE_TOTAL_RESOURCES);
- private static final SelectHostResult SELECT_HOST_RESULT_DELAYED_LOCALITY =
- new SelectHostResult(ScheduleResult.DELAYED_LOCALITY);
- private static final SelectHostResult SELECT_HOST_RESULT_DELAYED_RESOURCES =
- new SelectHostResult(ScheduleResult.DELAYED_RESOURCES);
-
- private static class NodeServiceInstancePair {
- final NodeInfo nodeInfo;
- final ServiceInstance serviceInstance;
-
- private NodeServiceInstancePair(ServiceInstance serviceInstance, NodeInfo nodeInfo) {
- this.serviceInstance = serviceInstance;
- this.nodeInfo = nodeInfo;
- }
-
- public ServiceInstance getServiceInstance() {
- return serviceInstance;
- }
-
- public NodeInfo getNodeInfo() {
- return nodeInfo;
- }
- }
-
- private static final class NodeBlacklistConf {
- private final long minDelay;
- private final long maxDelay;
- private final float backoffFactor;
-
- public NodeBlacklistConf(long minDelay, long maxDelay, float backoffFactor) {
- this.minDelay = minDelay;
- this.maxDelay = maxDelay;
- this.backoffFactor = backoffFactor;
- }
-
- @Override
- public String toString() {
- return "NodeBlacklistConf{" +
- "minDelay=" + minDelay +
- ", maxDelay=" + maxDelay +
- ", backoffFactor=" + backoffFactor +
- '}';
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-server/src/protobuf/LlapDaemonProtocol.proto
----------------------------------------------------------------------
diff --git a/llap-server/src/protobuf/LlapDaemonProtocol.proto b/llap-server/src/protobuf/LlapDaemonProtocol.proto
deleted file mode 100644
index 944c96c..0000000
--- a/llap-server/src/protobuf/LlapDaemonProtocol.proto
+++ /dev/null
@@ -1,148 +0,0 @@
-/**
- * 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.
- */
-
-option java_package = "org.apache.hadoop.hive.llap.daemon.rpc";
-option java_outer_classname = "LlapDaemonProtocolProtos";
-option java_generic_services = true;
-option java_generate_equals_and_hash = true;
-
-// TODO Change this as the interface evolves. Currently using Tez constructs.
-
-message UserPayloadProto {
- optional bytes user_payload = 1;
- optional int32 version = 2;
-}
-
-message EntityDescriptorProto {
- optional string class_name = 1;
- optional UserPayloadProto user_payload = 2;
- optional bytes history_text = 3;
-}
-
-message IOSpecProto {
- optional string connected_vertex_name = 1;
- optional EntityDescriptorProto io_descriptor = 2;
- optional int32 physical_edge_count = 3;
-}
-
-message GroupInputSpecProto {
- optional string group_name = 1;
- repeated string group_vertices = 2;
- optional EntityDescriptorProto merged_input_descriptor = 3;
-}
-
-
-message FragmentSpecProto {
- optional string fragment_identifier_string = 1;
- optional string dag_name = 2;
- optional int32 dag_id = 11;
- optional string vertex_name = 3;
- optional EntityDescriptorProto processor_descriptor = 4;
- repeated IOSpecProto input_specs = 5;
- repeated IOSpecProto output_specs = 6;
- repeated GroupInputSpecProto grouped_input_specs = 7;
- optional int32 vertex_parallelism = 8;
- optional int32 fragment_number =9;
- optional int32 attempt_number = 10;
-}
-
-message FragmentRuntimeInfo {
- optional int32 num_self_and_upstream_tasks = 1;
- optional int32 num_self_and_upstream_completed_tasks = 2;
- optional int32 within_dag_priority = 3;
- optional int64 dag_start_time = 4;
- optional int64 first_attempt_start_time = 5;
- optional int64 current_attempt_start_time = 6;
-}
-
-enum SourceStateProto {
- S_SUCCEEDED = 1;
- S_RUNNING = 2;
-}
-
-message QueryIdentifierProto {
- optional string app_identifier = 1;
- optional int32 dag_identifier = 2;
-}
-
-message SubmitWorkRequestProto {
- optional string container_id_string = 1;
- optional string am_host = 2;
- optional int32 am_port = 3;
- optional string token_identifier = 4;
- optional bytes credentials_binary = 5;
- optional string user = 6;
- optional string application_id_string = 7;
- optional int32 app_attempt_number = 8;
- optional FragmentSpecProto fragment_spec = 9;
- optional FragmentRuntimeInfo fragment_runtime_info = 10;
-}
-
-enum SubmissionStateProto {
- ACCEPTED = 1;
- REJECTED = 2;
- EVICTED_OTHER = 3;
-}
-
-message SubmitWorkResponseProto {
- optional SubmissionStateProto submission_state = 1;
-}
-
-message SourceStateUpdatedRequestProto {
- optional QueryIdentifierProto query_identifier = 1;
- optional string src_name = 2;
- optional SourceStateProto state = 3;
-}
-
-message SourceStateUpdatedResponseProto {
-}
-
-message QueryCompleteRequestProto {
- optional string query_id = 1;
- optional QueryIdentifierProto query_identifier = 2;
- optional int64 delete_delay = 4 [default = 0];
-}
-
-message QueryCompleteResponseProto {
-}
-
-message TerminateFragmentRequestProto {
- optional QueryIdentifierProto query_identifier = 1;
- optional string fragment_identifier_string = 2;
-}
-
-message TerminateFragmentResponseProto {
-}
-
-message GetTokenRequestProto {
-}
-
-message GetTokenResponseProto {
- optional bytes token = 1;
-}
-
-service LlapDaemonProtocol {
- rpc submitWork(SubmitWorkRequestProto) returns (SubmitWorkResponseProto);
- rpc sourceStateUpdated(SourceStateUpdatedRequestProto) returns (SourceStateUpdatedResponseProto);
- rpc queryComplete(QueryCompleteRequestProto) returns (QueryCompleteResponseProto);
- rpc terminateFragment(TerminateFragmentRequestProto) returns (TerminateFragmentResponseProto);
-}
-
-service LlapManagementProtocol {
- rpc getDelegationToken(GetTokenRequestProto) returns (GetTokenResponseProto);
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
index 44c958d..1cef218 100644
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
+++ b/llap-server/src/test/org/apache/hadoop/hive/llap/daemon/impl/TestLlapDaemonProtocolServerImpl.java
@@ -27,10 +27,11 @@ import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.llap.configuration.LlapConfiguration;
import org.apache.hadoop.hive.llap.daemon.ContainerRunner;
-import org.apache.hadoop.hive.llap.daemon.LlapDaemonProtocolBlockingPB;
+import org.apache.hadoop.hive.llap.protocol.LlapProtocolBlockingPB;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkRequestProto;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmitWorkResponseProto;
import org.apache.hadoop.hive.llap.daemon.rpc.LlapDaemonProtocolProtos.SubmissionStateProto;
+import org.apache.hadoop.hive.llap.impl.LlapProtocolClientImpl;
import org.junit.Test;
public class TestLlapDaemonProtocolServerImpl {
@@ -42,8 +43,8 @@ public class TestLlapDaemonProtocolServerImpl {
int rpcPort = HiveConf.getIntVar(daemonConf, ConfVars.LLAP_DAEMON_RPC_PORT);
int numHandlers = HiveConf.getIntVar(daemonConf, ConfVars.LLAP_DAEMON_RPC_NUM_HANDLERS);
ContainerRunner containerRunnerMock = mock(ContainerRunner.class);
- LlapDaemonProtocolServerImpl server =
- new LlapDaemonProtocolServerImpl(numHandlers, containerRunnerMock,
+ LlapProtocolServerImpl server =
+ new LlapProtocolServerImpl(numHandlers, containerRunnerMock,
new AtomicReference<InetSocketAddress>(), new AtomicReference<InetSocketAddress>(),
rpcPort, rpcPort + 1);
when(containerRunnerMock.submitWork(any(SubmitWorkRequestProto.class))).thenReturn(
@@ -56,8 +57,8 @@ public class TestLlapDaemonProtocolServerImpl {
server.start();
InetSocketAddress serverAddr = server.getBindAddress();
- LlapDaemonProtocolBlockingPB client =
- new LlapDaemonProtocolClientImpl(new Configuration(), serverAddr.getHostName(),
+ LlapProtocolBlockingPB client =
+ new LlapProtocolClientImpl(new Configuration(), serverAddr.getHostName(),
serverAddr.getPort(), null, null);
SubmitWorkResponseProto responseProto = client.submitWork(null,
SubmitWorkRequestProto.newBuilder()
http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapDaemonProtocolClientProxy.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapDaemonProtocolClientProxy.java b/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapDaemonProtocolClientProxy.java
deleted file mode 100644
index a6af8c2..0000000
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapDaemonProtocolClientProxy.java
+++ /dev/null
@@ -1,143 +0,0 @@
-/*
- * Licensed 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.hive.llap.tezplugins;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.protobuf.Message;
-import org.apache.commons.lang3.mutable.MutableInt;
-import org.apache.hadoop.hive.llap.LlapNodeId;
-import org.junit.Test;
-
-public class TestLlapDaemonProtocolClientProxy {
-
- @Test (timeout = 5000)
- public void testMultipleNodes() {
- RequestManagerForTest requestManager = new RequestManagerForTest(1);
-
- LlapNodeId nodeId1 = LlapNodeId.getInstance("host1", 1025);
- LlapNodeId nodeId2 = LlapNodeId.getInstance("host2", 1025);
-
- Message mockMessage = mock(Message.class);
- LlapDaemonProtocolClientProxy.ExecuteRequestCallback mockExecuteRequestCallback = mock(
- LlapDaemonProtocolClientProxy.ExecuteRequestCallback.class);
-
- // Request two messages
- requestManager.queueRequest(
- new CallableRequestForTest(nodeId1, mockMessage, mockExecuteRequestCallback));
- requestManager.queueRequest(
- new CallableRequestForTest(nodeId2, mockMessage, mockExecuteRequestCallback));
-
- // Should go through in a single process call
- requestManager.process();
- assertEquals(2, requestManager.numSubmissionsCounters);
- assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
- assertNotNull(requestManager.numInvocationsPerNode.get(nodeId2));
- assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
- assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId2).getValue().intValue());
- assertEquals(0, requestManager.currentLoopSkippedRequests.size());
- assertEquals(0, requestManager.currentLoopSkippedRequests.size());
- assertEquals(0, requestManager.currentLoopDisabledNodes.size());
- }
-
- @Test(timeout = 5000)
- public void testSingleInvocationPerNode() {
- RequestManagerForTest requestManager = new RequestManagerForTest(1);
-
- LlapNodeId nodeId1 = LlapNodeId.getInstance("host1", 1025);
-
- Message mockMessage = mock(Message.class);
- LlapDaemonProtocolClientProxy.ExecuteRequestCallback mockExecuteRequestCallback = mock(
- LlapDaemonProtocolClientProxy.ExecuteRequestCallback.class);
-
- // First request for host.
- requestManager.queueRequest(
- new CallableRequestForTest(nodeId1, mockMessage, mockExecuteRequestCallback));
- requestManager.process();
- assertEquals(1, requestManager.numSubmissionsCounters);
- assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
- assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
- assertEquals(0, requestManager.currentLoopSkippedRequests.size());
-
- // Second request for host. Single invocation since the last has not completed.
- requestManager.queueRequest(
- new CallableRequestForTest(nodeId1, mockMessage, mockExecuteRequestCallback));
- requestManager.process();
- assertEquals(1, requestManager.numSubmissionsCounters);
- assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
- assertEquals(1, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
- assertEquals(1, requestManager.currentLoopSkippedRequests.size());
- assertEquals(1, requestManager.currentLoopDisabledNodes.size());
- assertTrue(requestManager.currentLoopDisabledNodes.contains(nodeId1));
-
- // Complete first request. Second pending request should go through.
- requestManager.requestFinished(nodeId1);
- requestManager.process();
- assertEquals(2, requestManager.numSubmissionsCounters);
- assertNotNull(requestManager.numInvocationsPerNode.get(nodeId1));
- assertEquals(2, requestManager.numInvocationsPerNode.get(nodeId1).getValue().intValue());
- assertEquals(0, requestManager.currentLoopSkippedRequests.size());
- assertEquals(0, requestManager.currentLoopDisabledNodes.size());
- assertFalse(requestManager.currentLoopDisabledNodes.contains(nodeId1));
- }
-
-
- static class RequestManagerForTest extends LlapDaemonProtocolClientProxy.RequestManager {
-
- int numSubmissionsCounters = 0;
- private Map<LlapNodeId, MutableInt> numInvocationsPerNode = new HashMap<>();
-
- public RequestManagerForTest(int numThreads) {
- super(numThreads);
- }
-
- protected void submitToExecutor(LlapDaemonProtocolClientProxy.CallableRequest request, LlapNodeId nodeId) {
- numSubmissionsCounters++;
- MutableInt nodeCount = numInvocationsPerNode.get(nodeId);
- if (nodeCount == null) {
- nodeCount = new MutableInt(0);
- numInvocationsPerNode.put(nodeId, nodeCount);
- }
- nodeCount.increment();
- }
-
- void reset() {
- numSubmissionsCounters = 0;
- numInvocationsPerNode.clear();
- }
-
- }
-
- static class CallableRequestForTest extends LlapDaemonProtocolClientProxy.CallableRequest<Message, Message> {
-
- protected CallableRequestForTest(LlapNodeId nodeId, Message message,
- LlapDaemonProtocolClientProxy.ExecuteRequestCallback<Message> callback) {
- super(nodeId, message, callback);
- }
-
- @Override
- public Message call() throws Exception {
- return null;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/hive/blob/4185d9b8/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskCommunicator.java
----------------------------------------------------------------------
diff --git a/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskCommunicator.java b/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskCommunicator.java
deleted file mode 100644
index 8f3d104..0000000
--- a/llap-server/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskCommunicator.java
+++ /dev/null
@@ -1,100 +0,0 @@
-/*
- * Licensed 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.hive.llap.tezplugins;
-
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-
-import org.apache.hadoop.hive.llap.LlapNodeId;
-import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.tez.dag.records.TezTaskAttemptID;
-import org.junit.Test;
-
-public class TestLlapTaskCommunicator {
-
- @Test (timeout = 5000)
- public void testEntityTracker1() {
- LlapTaskCommunicator.EntityTracker entityTracker = new LlapTaskCommunicator.EntityTracker();
-
- String host1 = "host1";
- String host2 = "host2";
- String host3 = "host3";
- int port = 1451;
-
-
- // Simple container registration and un-registration without any task attempt being involved.
- ContainerId containerId101 = constructContainerId(101);
- entityTracker.registerContainer(containerId101, host1, port);
- assertEquals(LlapNodeId.getInstance(host1, port), entityTracker.getNodeIdForContainer(containerId101));
-
- entityTracker.unregisterContainer(containerId101);
- assertNull(entityTracker.getContainerAttemptMapForNode(LlapNodeId.getInstance(host1, port)));
- assertNull(entityTracker.getNodeIdForContainer(containerId101));
- assertEquals(0, entityTracker.nodeMap.size());
- assertEquals(0, entityTracker.attemptToNodeMap.size());
- assertEquals(0, entityTracker.containerToNodeMap.size());
-
-
- // Simple task registration and un-registration.
- ContainerId containerId1 = constructContainerId(1);
- TezTaskAttemptID taskAttemptId1 = constructTaskAttemptId(1);
- entityTracker.registerTaskAttempt(containerId1, taskAttemptId1, host1, port);
- assertEquals(LlapNodeId.getInstance(host1, port), entityTracker.getNodeIdForContainer(containerId1));
- assertEquals(LlapNodeId.getInstance(host1, port), entityTracker.getNodeIdForTaskAttempt(taskAttemptId1));
-
- entityTracker.unregisterTaskAttempt(taskAttemptId1);
- assertNull(entityTracker.getContainerAttemptMapForNode(LlapNodeId.getInstance(host1, port)));
- assertNull(entityTracker.getNodeIdForContainer(containerId1));
- assertNull(entityTracker.getNodeIdForTaskAttempt(taskAttemptId1));
- assertEquals(0, entityTracker.nodeMap.size());
- assertEquals(0, entityTracker.attemptToNodeMap.size());
- assertEquals(0, entityTracker.containerToNodeMap.size());
-
- // Register taskAttempt, unregister container. TaskAttempt should also be unregistered
- ContainerId containerId201 = constructContainerId(201);
- TezTaskAttemptID taskAttemptId201 = constructTaskAttemptId(201);
- entityTracker.registerTaskAttempt(containerId201, taskAttemptId201, host1, port);
- assertEquals(LlapNodeId.getInstance(host1, port), entityTracker.getNodeIdForContainer(containerId201));
- assertEquals(LlapNodeId.getInstance(host1, port), entityTracker.getNodeIdForTaskAttempt(taskAttemptId201));
-
- entityTracker.unregisterContainer(containerId201);
- assertNull(entityTracker.getContainerAttemptMapForNode(LlapNodeId.getInstance(host1, port)));
- assertNull(entityTracker.getNodeIdForContainer(containerId201));
- assertNull(entityTracker.getNodeIdForTaskAttempt(taskAttemptId201));
- assertEquals(0, entityTracker.nodeMap.size());
- assertEquals(0, entityTracker.attemptToNodeMap.size());
- assertEquals(0, entityTracker.containerToNodeMap.size());
-
- entityTracker.unregisterTaskAttempt(taskAttemptId201); // No errors
- }
-
-
- private ContainerId constructContainerId(int id) {
- ContainerId containerId = mock(ContainerId.class);
- doReturn(id).when(containerId).getId();
- doReturn((long)id).when(containerId).getContainerId();
- return containerId;
- }
-
- private TezTaskAttemptID constructTaskAttemptId(int id) {
- TezTaskAttemptID taskAttemptId = mock(TezTaskAttemptID.class);
- doReturn(id).when(taskAttemptId).getId();
- return taskAttemptId;
- }
-
-}