[SPARK-10001] Consolidate Signaling and SignalLogger.

core/src/main/scala/org/apache/spark/util/Signaling.scala → core/src/main/scala/org/apache/spark/util/SignalUtils.scala

@@ -17,28 +17,69 @@
 
 package org.apache.spark.util
 
-import java.util.{Collections, LinkedList}
+import java.util.Collections
 
 import scala.collection.JavaConverters._
-import scala.collection.mutable.HashMap
 
 import org.apache.commons.lang3.SystemUtils
+import org.slf4j.Logger
 import sun.misc.{Signal, SignalHandler}
 
 import org.apache.spark.internal.Logging
 
-
 /**
  * Contains utilities for working with posix signals.
  */
-private[spark] object Signaling extends Logging {
+private[spark] object SignalUtils extends Logging {
+
+  /** A flag to make sure we only register the logger once. */
+  private var loggerRegistered = false
+
+  /** Register a signal handler to log signals on UNIX-like systems. */
+  def registerLogger(log: Logger): Unit = synchronized {
+    if (!loggerRegistered) {
+      Seq("TERM", "HUP", "INT").foreach { sig =>
+        SignalUtils.register(sig) {
+          log.error("RECEIVED SIGNAL " + sig)
+          false
+        }
+      }
+      loggerRegistered = true
+    }
+  }
+
+  /**
+   * Adds an action to be run when a given signal is received by this process.
+   *
+   * Note that signals are only supported on unix-like operating systems and work on a best-effort
+   * basis: if a signal is not available or cannot be intercepted, only a warning is emitted.
+   *
+   * All actions for a given signal are run in a separate thread.
+   */
+  def register(signal: String)(action: => Boolean): Unit = synchronized {
+    if (SystemUtils.IS_OS_UNIX) {
+      try {
+        val handler = handlers.getOrElseUpdate(signal, {
+          logInfo("Registered signal handler for " + signal)
+          new ActionHandler(new Signal(signal))
+        })
+        handler.register(action)
+      } catch {
+        case ex: Exception => logWarning(s"Failed to register signal handler for " + signal, ex)
+      }
+    }
+  }
 
   /**
    * A handler for the given signal that runs a collection of actions.
    */
   private class ActionHandler(signal: Signal) extends SignalHandler {
 
-    private val actions = Collections.synchronizedList(new LinkedList[() => Boolean])
+    /**
+     * List of actions upon the signal; the callbacks should return true if the signal is "handled",
+     * i.e. should not escalate to the next callback.
+     */
+    private val actions = Collections.synchronizedList(new java.util.LinkedList[() => Boolean])
 
     // original signal handler, before this handler was attached
     private val prevHandler: SignalHandler = Signal.handle(signal, this)
@@ -51,11 +92,10 @@ private[spark] object Signaling extends Logging {
       // register old handler, will receive incoming signals while this handler is running
       Signal.handle(signal, prevHandler)
 
-      val escalate = actions.asScala forall { action =>
-        !action()
-      }
-
-      if(escalate) {
+      // run all actions, escalate to parent handler if no action catches the signal
+      // (i.e. all actions return false)
+      val escalate = actions.asScala.forall { action => !action() }
+      if (escalate) {
         prevHandler.handle(sig)
       }
 
@@ -64,36 +104,13 @@ private[spark] object Signaling extends Logging {
     }
 
     /**
-     * Add an action to be run by this handler.
+     * Adds an action to be run by this handler.
      * @param action An action to be run when a signal is received. Return true if the signal
-     * should be stopped with this handler, false if it should be escalated.
+     *               should be stopped with this handler, false if it should be escalated.
      */
     def register(action: => Boolean): Unit = actions.add(() => action)
-
-  }
-
-  // contains association of signals to their respective handlers
-  private val handlers = new HashMap[String, ActionHandler]
-
-  /**
-   * Adds an action to be run when a given signal is received by this process.
-   *
-   * Note that signals are only supported on unix-like operating systems and work on a best-effort
-   * basis: if a signal is not available or cannot be intercepted, only a warning is emitted.
-   *
-   * All actions for a given signal are run in a separate thread.
-   */
-  def register(signal: String)(action: => Boolean): Unit = synchronized {
-    if (SystemUtils.IS_OS_UNIX) try {
-      val handler = handlers.getOrElseUpdate(signal, {
-        val h = new ActionHandler(new Signal(signal))
-        logInfo("Registered signal handler for " + signal)
-        h
-      })
-      handler.register(action)
-    } catch {
-      case ex: Exception => logWarning(s"Failed to register signal handler for " + signal, ex)
-    }
   }
 
+  /** Mapping from signal to their respective handlers. */
+  private val handlers = new scala.collection.mutable.HashMap[String, ActionHandler]
 }

core/src/main/scala/org/apache/spark/util/Utils.scala

@@ -2284,7 +2284,7 @@ private[spark] object Utils extends Logging {
    */
   def initDaemon(log: Logger): Unit = {
     log.info(s"Started daemon with process name: ${Utils.getProcessName()}")
-    SignalLogger.register(log)
+    SignalUtils.registerLogger(log)
   }
 }
 

repl/src/main/scala/org/apache/spark/repl/Signaling.scala

@@ -19,7 +19,7 @@ package org.apache.spark.repl
 
 import org.apache.spark.SparkContext
 import org.apache.spark.internal.Logging
-import org.apache.spark.util.{Signaling => USignaling}
+import org.apache.spark.util.SignalUtils
 
 private[repl] object Signaling extends Logging {
 
@@ -28,7 +28,7 @@ private[repl] object Signaling extends Logging {
    * when no jobs are currently running.
    * This makes it possible to interrupt a running shell job by pressing Ctrl+C.
    */
-  def cancelOnInterrupt(ctx: SparkContext): Unit = USignaling.register("INT") {
+  def cancelOnInterrupt(ctx: SparkContext): Unit = SignalUtils.register("INT") {
     if (!ctx.statusTracker.getActiveJobIds().isEmpty) {
       logWarning("Cancelling all active jobs, this can take a while. " +
         "Press Ctrl+C again to exit now.")

yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala

@@ -716,7 +716,7 @@ object ApplicationMaster extends Logging {
   private var master: ApplicationMaster = _
 
   def main(args: Array[String]): Unit = {
-    SignalLogger.register(log)
+    SignalUtils.registerLogger(log)
     val amArgs = new ApplicationMasterArguments(args)
     SparkHadoopUtil.get.runAsSparkUser { () =>
       master = new ApplicationMaster(amArgs, new YarnRMClient)