hive-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ss...@apache.org
Subject hive git commit: HIVE-10961. LLAP: Fix ShuffleHandler + Submit work init race condition. (Siddharth Seth)
Date Tue, 09 Jun 2015 20:59:20 GMT
Repository: hive
Updated Branches:
  refs/heads/llap 785ec8e75 -> b321c55a3


HIVE-10961. LLAP: Fix ShuffleHandler + Submit work init race condition. (Siddharth Seth)


Project: http://git-wip-us.apache.org/repos/asf/hive/repo
Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/b321c55a
Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/b321c55a
Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/b321c55a

Branch: refs/heads/llap
Commit: b321c55a326093ea554df77e806e2601eab24832
Parents: 785ec8e
Author: Siddharth Seth <sseth@apache.org>
Authored: Tue Jun 9 13:58:49 2015 -0700
Committer: Siddharth Seth <sseth@apache.org>
Committed: Tue Jun 9 13:58:49 2015 -0700

----------------------------------------------------------------------
 .../hive/llap/daemon/impl/AMReporter.java       | 18 +++++++++++------
 .../llap/daemon/impl/ContainerRunnerImpl.java   | 18 +++++------------
 .../hive/llap/daemon/impl/LlapDaemon.java       | 21 ++++++++++++++++----
 3 files changed, 34 insertions(+), 23 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hive/blob/b321c55a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java
b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java
index 7376792..1ba18fc 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/AMReporter.java
@@ -27,6 +27,7 @@ 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.AtomicReference;
 
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
@@ -77,11 +78,12 @@ public class AMReporter extends AbstractService {
 
   private static final Logger LOG = LoggerFactory.getLogger(AMReporter.class);
 
-  private final LlapNodeId nodeId;
+  private volatile LlapNodeId nodeId;
   private final Configuration conf;
   private final ListeningExecutorService queueLookupExecutor;
   private final ListeningExecutorService executor;
   private final DelayQueue<AMNodeInfo> pendingHeartbeatQueeu = new DelayQueue();
+  private final AtomicReference<InetSocketAddress> localAddress;
   private final long heartbeatInterval;
   private final AtomicBoolean isShutdown = new AtomicBoolean(false);
   // Tracks appMasters to which heartbeats are being sent. This should not be used for any
other
@@ -89,9 +91,9 @@ public class AMReporter extends AbstractService {
   private final Map<LlapNodeId, AMNodeInfo> knownAppMasters = new HashMap<>();
   volatile ListenableFuture<Void> queueLookupFuture;
 
-  public AMReporter(LlapNodeId nodeId, Configuration conf) {
+  public AMReporter(AtomicReference<InetSocketAddress> localAddress, Configuration
conf) {
     super(AMReporter.class.getName());
-    this.nodeId = nodeId;
+    this.localAddress = localAddress;
     this.conf = conf;
     ExecutorService rawExecutor = Executors.newCachedThreadPool(
         new ThreadFactoryBuilder().setDaemon(true).setNameFormat("AMReporter %d").build());
@@ -102,7 +104,7 @@ public class AMReporter extends AbstractService {
     this.heartbeatInterval =
         conf.getLong(LlapConfiguration.LLAP_DAEMON_LIVENESS_HEARTBEAT_INTERVAL_MS,
             LlapConfiguration.LLAP_DAEMON_LIVENESS_HEARTBEAT_INTERVAL_MS_DEFAULT);
-    LOG.info("AMReporter running with NodeId: {}", nodeId);
+
   }
 
   @Override
@@ -125,7 +127,8 @@ public class AMReporter extends AbstractService {
         }
       }
     });
-    LOG.info("Started service: " + getName());
+    nodeId = LlapNodeId.getInstance(localAddress.get().getHostName(), localAddress.get().getPort());
+    LOG.info("AMReporter running with NodeId: {}", nodeId);
   }
 
   @Override
@@ -170,7 +173,7 @@ public class AMReporter extends AbstractService {
     synchronized (knownAppMasters) {
       amNodeInfo = knownAppMasters.get(amNodeId);
       if (amNodeInfo == null) {
-        LOG.info(("Ignoring duplocate unregisterRequest for am at: " + amLocation + ":" +
port));
+        LOG.info(("Ignoring duplicate unregisterRequest for am at: " + amLocation + ":" +
port));
       }
       amNodeInfo.decrementAndGetTaskCount();
       // Not removing this here. Will be removed when taken off the queue and discovered
to have 0
@@ -184,6 +187,9 @@ public class AMReporter extends AbstractService {
     // knownAppMasters is used for sending heartbeats for queued tasks. Killed messages use
a new connection.
     LlapNodeId amNodeId = LlapNodeId.getInstance(amLocation, port);
     AMNodeInfo amNodeInfo = new AMNodeInfo(amNodeId, user, jobToken, conf);
+
+    // Even if the service hasn't started up. It's OK to make this invocation since this
will
+    // only happen after the AtomicReference address has been populated. Not adding an additional
check.
     ListenableFuture<Void> future =
         executor.submit(new KillTaskCallable(taskAttemptId, amNodeInfo));
     Futures.addCallback(future, new FutureCallback<Void>() {

http://git-wip-us.apache.org/repos/asf/hive/blob/b321c55a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
index 6f9f429..10e192e 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/ContainerRunnerImpl.java
@@ -25,7 +25,6 @@ import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.hive.llap.LlapNodeId;
 import org.apache.hadoop.hive.llap.daemon.ContainerRunner;
 import org.apache.hadoop.hive.llap.daemon.FragmentCompletionHandler;
 import org.apache.hadoop.hive.llap.daemon.HistoryLogger;
@@ -67,7 +66,7 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
   private static final Logger LOG = LoggerFactory.getLogger(ContainerRunnerImpl.class);
   public static final String THREAD_NAME_FORMAT_PREFIX = "ContainerExecutor ";
 
-  private volatile AMReporter amReporter;
+  private final AMReporter amReporter;
   private final QueryTracker queryTracker;
   private final Scheduler<TaskRunnerCallable> executorService;
   private final AtomicReference<InetSocketAddress> localAddress;
@@ -81,12 +80,14 @@ public class ContainerRunnerImpl extends CompositeService implements ContainerRu
   public ContainerRunnerImpl(Configuration conf, int numExecutors, int waitQueueSize,
       boolean enablePreemption, String[] localDirsBase, int localShufflePort,
       AtomicReference<InetSocketAddress> localAddress,
-      long totalMemoryAvailableBytes, LlapDaemonExecutorMetrics metrics) {
+      long totalMemoryAvailableBytes, LlapDaemonExecutorMetrics metrics,
+      AMReporter amReporter) {
     super("ContainerRunnerImpl");
     this.conf = conf;
     Preconditions.checkState(numExecutors > 0,
         "Invalid number of executors: " + numExecutors + ". Must be > 0");
     this.localAddress = localAddress;
+    this.amReporter = amReporter;
 
     this.queryTracker = new QueryTracker(conf, localDirsBase);
     addIfService(queryTracker);
@@ -122,22 +123,13 @@ public class ContainerRunnerImpl extends CompositeService implements
ContainerRu
 
   @Override
   public void serviceStart() throws Exception {
-    // The node id will only be available at this point, since the server has been started
in LlapDaemon
     super.serviceStart();
-    LlapNodeId llapNodeId = LlapNodeId.getInstance(localAddress.get().getHostName(),
-        localAddress.get().getPort());
-    this.amReporter = new AMReporter(llapNodeId, conf);
-    amReporter.init(conf);
-    amReporter.start();
+
   }
 
   @Override
   protected void serviceStop() throws Exception {
     super.serviceStop();
-    if (amReporter != null) {
-      amReporter.stop();
-      amReporter = null;
-    }
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/hive/blob/b321c55a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
----------------------------------------------------------------------
diff --git a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
index 75377d4..1801212 100644
--- a/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
+++ b/llap-server/src/java/org/apache/hadoop/hive/llap/daemon/impl/LlapDaemon.java
@@ -63,6 +63,7 @@ public class LlapDaemon extends CompositeService implements ContainerRunner,
Lla
   private final Configuration shuffleHandlerConf;
   private final LlapDaemonProtocolServerImpl server;
   private final ContainerRunnerImpl containerRunner;
+  private final AMReporter amReporter;
   private final LlapRegistryService registry;
   private final LlapWebServices webServices;
   private final AtomicLong numSubmissions = new AtomicLong(0);
@@ -155,8 +156,12 @@ public class LlapDaemon extends CompositeService implements ContainerRunner,
Lla
     LOG.info("Started LlapMetricsSystem with displayName: " + displayName +
         " sessionId: " + sessionId);
 
+
+    this.amReporter = new AMReporter(address, daemonConf);
+
+
     this.server = new LlapDaemonProtocolServerImpl(numHandlers, this, address, rpcPort);
-    addIfService(server);
+
 
     this.containerRunner = new ContainerRunnerImpl(daemonConf,
         numExecutors,
@@ -166,13 +171,19 @@ public class LlapDaemon extends CompositeService implements ContainerRunner,
Lla
         shufflePort,
         address,
         executorMemoryBytes,
-        metrics);
+        metrics,
+        amReporter);
     addIfService(containerRunner);
 
     this.registry = new LlapRegistryService(true);
     addIfService(registry);
     this.webServices = new LlapWebServices();
     addIfService(webServices);
+    // Bring up the server only after all other components have started.
+    addIfService(server);
+    // AMReporter after the server so that it gets the correct address. It knows how to deal
with
+    // requests before it is started.
+    addIfService(amReporter);
   }
 
   private long getTotalHeapSize() {
@@ -220,14 +231,16 @@ public class LlapDaemon extends CompositeService implements ContainerRunner,
Lla
 
   @Override
   public void serviceStart() throws Exception {
-    super.serviceStart();
+    // Start the Shuffle service before the listener - until it's a service as well.
     ShuffleHandler.initializeAndStart(shuffleHandlerConf);
+    super.serviceStart();
+    LOG.info("LlapDaemon serviceStart complete");
   }
 
   public void serviceStop() throws Exception {
     super.serviceStop();
-    shutdown();
     ShuffleHandler.shutdown();
+    shutdown();
     LOG.info("LlapDaemon shutdown complete");
   }
 


Mime
View raw message