You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by go...@apache.org on 2020/06/04 17:19:51 UTC
[hive] branch master updated: HIVE-23596: LLAP: Encode initial
guaranteed task information in containerId (Mustafa Iman,
reviewed by Gopal V)
This is an automated email from the ASF dual-hosted git repository.
gopalv pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new b5e3bcd HIVE-23596: LLAP: Encode initial guaranteed task information in containerId (Mustafa Iman, reviewed by Gopal V)
b5e3bcd is described below
commit b5e3bcd575aa7f904ec0955793916d0b11f38215
Author: Mustafa Iman <mu...@gmail.com>
AuthorDate: Thu Jun 4 10:18:29 2020 -0700
HIVE-23596: LLAP: Encode initial guaranteed task information in containerId (Mustafa Iman, reviewed by Gopal V)
Signed-off-by: Gopal V <go...@apache.org>
---
.../hive/llap/tezplugins/ContainerFactory.java | 41 ++++++++++++++++++++--
.../hive/llap/tezplugins/LlapTaskCommunicator.java | 5 +--
.../llap/tezplugins/LlapTaskSchedulerService.java | 39 +++++---------------
.../tezplugins/TestLlapTaskSchedulerService.java | 27 ++++++++++++++
4 files changed, 76 insertions(+), 36 deletions(-)
diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/ContainerFactory.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/ContainerFactory.java
index f1feec7..d340c22 100644
--- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/ContainerFactory.java
+++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/ContainerFactory.java
@@ -28,6 +28,29 @@ class ContainerFactory {
final ApplicationAttemptId customAppAttemptId;
AtomicLong nextId;
+ private static final int GUARANTEED_WIDTH = 1;
+
+ private static final long GUARANTEED_BIT_MASK = (1L << GUARANTEED_WIDTH) - 1;
+
+ /**
+ * This is a hack to pass initial guaranteed information from {@link LlapTaskSchedulerService}
+ * to {@link LlapTaskCommunicator}. Otherwise, we ended up synchronizing communication and
+ * scheduling. This workaround can be removed after TEZ-4192 and HIVE-23589 are merged.
+ *
+ * Note: This method is reliable only for initial allocation of the task. Guaranteed status
+ * may change via separate requests later. Therefore, do not rely on this method other
+ * than creating initial submit work request.
+ *
+ * Even containerId -> guaranteed
+ * Odd containerId -> not guaranteed
+ *
+ * @param containerId
+ * @return {@code true} if the task associated with container was guaranteed initially.
+ */
+ public static boolean isContainerInitializedAsGuaranteed(ContainerId containerId) {
+ return (containerId.getContainerId() & GUARANTEED_BIT_MASK) == 0;
+ }
+
public ContainerFactory(ApplicationAttemptId appAttemptId, long appIdLong) {
this.nextId = new AtomicLong(1);
ApplicationId appId =
@@ -37,9 +60,9 @@ class ContainerFactory {
}
public Container createContainer(Resource capability, Priority priority, String hostname,
- int port, String nodeHttpAddress) {
+ int port, String nodeHttpAddress, boolean isGuaranteed) {
ContainerId containerId =
- ContainerId.newContainerId(customAppAttemptId, nextId.getAndIncrement());
+ ContainerId.newContainerId(customAppAttemptId, nextContainerId(isGuaranteed));
NodeId nodeId = NodeId.newInstance(hostname, port);
Container container =
@@ -47,4 +70,18 @@ class ContainerFactory {
return container;
}
+
+ /**
+ * See {@link #isContainerInitializedAsGuaranteed(ContainerId)}
+ * @param isInitialGuaranteed
+ * @return
+ */
+ private long nextContainerId(boolean isInitialGuaranteed) {
+ long candidate = nextId.getAndIncrement();
+ candidate <<= GUARANTEED_WIDTH;
+ if (!isInitialGuaranteed) {
+ candidate |= 1;
+ }
+ return candidate;
+ }
}
diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
index 36a2d6b..73ae191 100644
--- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
+++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
@@ -858,10 +858,7 @@ public class LlapTaskCommunicator extends TezTaskCommunicatorImpl {
taskSpec, currentQueryIdentifierProto, getTokenIdentifier(), user, hiveQueryId)).build());
// Don't call builder.setWorkSpecSignature() - Tez doesn't sign fragments
builder.setFragmentRuntimeInfo(fragmentRuntimeInfo);
- if (scheduler != null) { // May be null in tests
- // TODO: see javadoc
- builder.setIsGuaranteed(scheduler.isInitialGuaranteed(taskSpec.getTaskAttemptID()));
- }
+ builder.setIsGuaranteed(ContainerFactory.isContainerInitializedAsGuaranteed(containerId));
return builder.build();
}
diff --git a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
index 4e5e43c..d4cc7a3 100644
--- a/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
+++ b/llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
@@ -1334,35 +1334,6 @@ public class LlapTaskSchedulerService extends TaskScheduler {
handleUpdateResult(info, true);
}
- /**
- * A hacky way for communicator and scheduler to share per-task info. Scheduler should be able
- * to include this with task allocation to be passed to the communicator, instead. TEZ-3866.
- * @param attemptId Task attempt ID.
- * @return The initial value of the guaranteed flag to send with the task.
- */
- boolean isInitialGuaranteed(TezTaskAttemptID attemptId) {
- if (!workloadManagementEnabled) {
- return false;
- }
- TaskInfo info = null;
- readLock.lock();
- try {
- info = tasksById.get(attemptId);
- } finally {
- readLock.unlock();
- }
- if (info == null) {
- WM_LOG.warn("Status requested for an unknown task " + attemptId);
- return false;
- }
- synchronized (info) {
- if (info.isGuaranteed == null) return false; // TODO: should never happen?
- assert info.lastSetGuaranteed == null;
- info.requestedValue = info.isGuaranteed;
- return info.isGuaranteed;
- }
- }
-
// Must be called under the epic lock.
private TaskInfo distributeGuaranteedOnTaskCompletion() {
List<TaskInfo> toUpdate = new ArrayList<>(1);
@@ -2005,12 +1976,14 @@ public class LlapTaskSchedulerService extends TaskScheduler {
if (selectHostResult.scheduleResult != ScheduleResult.SCHEDULED) {
return selectHostResult.scheduleResult;
}
+ boolean isGuaranteed = false;
if (unusedGuaranteed > 0) {
boolean wasGuaranteed = false;
synchronized (taskInfo) {
assert !taskInfo.isPendingUpdate; // No updates before it's running.
wasGuaranteed = taskInfo.isGuaranteed;
taskInfo.isGuaranteed = true;
+ isGuaranteed = true;
}
if (wasGuaranteed) {
// This should never happen - we only schedule one attempt once.
@@ -2032,18 +2005,24 @@ public class LlapTaskSchedulerService extends TaskScheduler {
synchronized (taskInfo) {
assert !taskInfo.isPendingUpdate; // No updates before it's running.
taskInfo.isGuaranteed = true;
+ isGuaranteed = true;
}
// Note: after this, the caller MUST send the downgrade message to downgradedTask
// (outside of the writeLock, preferably), before exiting.
}
}
+ // in LLAP, the containers are used exactly once for exactly one attempt
+ // the container ids are entirely arbitrary and have 64 bit space per application
+ // this is why the isGuaranteed (& any other initial information) can be encoded into the
+ // bits indicating container id
NodeInfo nodeInfo = selectHostResult.nodeInfo;
Container container =
containerFactory.createContainer(nodeInfo.getResourcePerExecutor(), taskInfo.priority,
nodeInfo.getHost(),
nodeInfo.getRpcPort(),
- nodeInfo.getServiceAddress());
+ nodeInfo.getServiceAddress(),
+ isGuaranteed);
writeLock.lock(); // While updating local structures
// Note: this is actually called under the epic writeLock in schedulePendingTasks
try {
diff --git a/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java b/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
index 7070123..14820cb 100644
--- a/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
+++ b/llap-tez/src/test/org/apache/hadoop/hive/llap/tezplugins/TestLlapTaskSchedulerService.java
@@ -1925,6 +1925,33 @@ public class TestLlapTaskSchedulerService {
}
}
+ @Test
+ public void testInitialGuaranteedInfoIsEncodedInContainerId() {
+ ApplicationAttemptId appAttemptId = ApplicationAttemptId.newInstance(ApplicationId.newInstance(1000, 1), 1);
+ ContainerFactory containerFactory = new ContainerFactory(appAttemptId, 15);
+ Container c1 = createDummyContainer(containerFactory, false);
+ Container c2 = createDummyContainer(containerFactory, true);
+ Container c3 = createDummyContainer(containerFactory, true);
+ Container c4 = createDummyContainer(containerFactory, false);
+ Container c5 = createDummyContainer(containerFactory, true);
+ Container c6 = createDummyContainer(containerFactory, false);
+ Container c7 = createDummyContainer(containerFactory, false);
+ Container c8 = createDummyContainer(containerFactory, true);
+
+ assertFalse(ContainerFactory.isContainerInitializedAsGuaranteed(c1.getId()));
+ assertTrue(ContainerFactory.isContainerInitializedAsGuaranteed(c2.getId()));
+ assertTrue(ContainerFactory.isContainerInitializedAsGuaranteed(c3.getId()));
+ assertFalse(ContainerFactory.isContainerInitializedAsGuaranteed(c4.getId()));
+ assertTrue(ContainerFactory.isContainerInitializedAsGuaranteed(c5.getId()));
+ assertFalse(ContainerFactory.isContainerInitializedAsGuaranteed(c6.getId()));
+ assertFalse(ContainerFactory.isContainerInitializedAsGuaranteed(c7.getId()));
+ assertTrue(ContainerFactory.isContainerInitializedAsGuaranteed(c8.getId()));
+ }
+
+ private Container createDummyContainer(ContainerFactory containerFactory, boolean isGuaranteed) {
+ return containerFactory.createContainer(mock(Resource.class), mock(Priority.class), "hostname", 0, null, isGuaranteed);
+ }
+
private static class TestTaskSchedulerServiceWrapper {
static final Resource resource = Resource.newInstance(1024, 1);
Configuration conf;