You are viewing a plain text version of this content. The canonical link for it is here.

Posted to notifications@asterixdb.apache.org by "Michael Blow (Code Review)" <do...@asterixdb.incubator.apache.org> on 2017/10/09 21:04:58 UTC

Change in asterixdb[master]: [ASTERIXDB-1076][HYR] Generate heartbeats in their own thread

Michael Blow has uploaded a new change for review.

  https://asterix-gerrit.ics.uci.edu/2060

Change subject: [ASTERIXDB-1076][HYR] Generate heartbeats in their own thread
......................................................................

[ASTERIXDB-1076][HYR] Generate heartbeats in their own thread

Generate & send NC heartbeats in their own thread to prevent starvation
/ scheduling issues

Change-Id: Ieae21b1596013a699f27975fb21894244c536395
---
M hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
1 file changed, 22 insertions(+), 17 deletions(-)


  git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/60/2060/1

diff --git a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
index 350343b..c34abc8 100644
--- a/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
+++ b/hyracks-fullstack/hyracks/hyracks-control/hyracks-control-nc/src/main/java/org/apache/hyracks/control/nc/NodeControllerService.java
@@ -27,7 +27,6 @@
 import java.lang.management.OperatingSystemMXBean;
 import java.lang.management.RuntimeMXBean;
 import java.lang.management.ThreadMXBean;
-import java.lang.reflect.Field;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
 import java.util.Hashtable;
@@ -133,7 +132,7 @@
 
     private NodeParameters nodeParameters;
 
-    private HeartbeatTask heartbeatTask;
+    private Thread heartbeatThread;
 
     private final ServerContext serverCtx;
 
@@ -308,15 +307,12 @@
 
         workQueue.start();
 
-        heartbeatTask = new HeartbeatTask(ccs);
+        // Start heartbeat generator.
+        heartbeatThread = new Thread(new HeartbeatTask(ccs, nodeParameters.getHeartbeatPeriod()), id + "-Heartbeat");
+        heartbeatThread.setPriority(Thread.MAX_PRIORITY);
+        heartbeatThread.setDaemon(true);
+        heartbeatThread.start();
 
-        // Use reflection to set the priority of the timer thread.
-        Field threadField = timer.getClass().getDeclaredField("thread");
-        threadField.setAccessible(true);
-        Thread timerThread = (Thread) threadField.get(timer); // The internal timer thread of the Timer object.
-        timerThread.setPriority(Thread.MAX_PRIORITY);
-        // Schedule heartbeat generator.
-        timer.schedule(heartbeatTask, 0, nodeParameters.getHeartbeatPeriod());
         // Schedule tracing a human-readable datetime
         timer.schedule(new TraceCurrentTimeTask(serviceCtx.getTracer()), 0, 60000);
 
@@ -401,7 +397,8 @@
              * Stop heartbeat after NC has stopped to avoid false node failure detection
              * on CC if an NC takes a long time to stop.
              */
-            heartbeatTask.cancel();
+            heartbeatThread.interrupt();
+            heartbeatThread.join();
             LOGGER.log(Level.INFO, "Stopped NodeControllerService");
         } else {
             LOGGER.log(Level.SEVERE, "Duplicate shutdown call; original: " + Arrays.toString(shutdownCallStack),
@@ -478,17 +475,15 @@
         return workQueue;
     }
 
-    public ThreadMXBean getThreadMXBean() {
-        return threadMXBean;
-    }
-
-    private class HeartbeatTask extends TimerTask {
+    private class HeartbeatTask implements Runnable {
         private IClusterController cc;
+        private final int heartbeatPeriod;
 
         private final HeartbeatData hbData;
 
-        public HeartbeatTask(IClusterController cc) {
+        HeartbeatTask(IClusterController cc, int heartbeatPeriod) {
             this.cc = cc;
+            this.heartbeatPeriod = heartbeatPeriod;
             hbData = new HeartbeatData();
             hbData.gcCollectionCounts = new long[gcMXBeans.size()];
             hbData.gcCollectionTimes = new long[gcMXBeans.size()];
@@ -496,6 +491,13 @@
 
         @Override
         public void run() {
+            while (!Thread.currentThread().isInterrupted()) {
+                execute();
+            }
+            LOGGER.log(Level.INFO, "Heartbeat thread interrupted; shutting down");
+        }
+
+        private void execute() {
             MemoryUsage heapUsage = memoryMXBean.getHeapMemoryUsage();
             hbData.heapInitSize = heapUsage.getInit();
             hbData.heapUsedSize = heapUsage.getUsed();
@@ -541,6 +543,9 @@
 
             try {
                 cc.nodeHeartbeat(id, hbData);
+                Thread.sleep(heartbeatPeriod);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
             } catch (Exception e) {
                 LOGGER.log(Level.SEVERE, "Exception sending heartbeat", e);
             }

-- 
To view, visit https://asterix-gerrit.ics.uci.edu/2060
To unsubscribe, visit https://asterix-gerrit.ics.uci.edu/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: Ieae21b1596013a699f27975fb21894244c536395
Gerrit-PatchSet: 1
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Owner: Michael Blow <mb...@apache.org>