From 687581c3ec2b6b8310bd5be9f2d15b25b9051aac Mon Sep 17 00:00:00 2001 From: Charles Reiss Date: Tue, 19 Feb 2013 11:52:35 -0800 Subject: Paranoid uncaught exception handling for exceptions during shutdown --- core/src/main/scala/spark/executor/Executor.scala | 29 ++++++++++++++++++----- 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/core/src/main/scala/spark/executor/Executor.scala b/core/src/main/scala/spark/executor/Executor.scala index bd21ba719a..b63bec11ad 100644 --- a/core/src/main/scala/spark/executor/Executor.scala +++ b/core/src/main/scala/spark/executor/Executor.scala @@ -50,14 +50,31 @@ private[spark] class Executor extends Logging { override def uncaughtException(thread: Thread, exception: Throwable) { try { logError("Uncaught exception in thread " + thread, exception) - if (exception.isInstanceOf[OutOfMemoryError]) { - System.exit(ExecutorExitCode.OOM) - } else { - System.exit(ExecutorExitCode.UNCAUGHT_EXCEPTION) + + // We may have been called from a shutdown hook. If so, we must not call System.exit(). + // (If we do, we will deadlock.) Runtime#addShutdownHook should fail if we are shutting + // down, which would either occur if we were called from a shutdown hook or if + // a System.exit() occured concurrently. + var shuttingDown = false + try { + val hook = new Thread { + override def run() {} + } + Runtime.getRuntime.addShutdownHook(hook) + Runtime.getRuntime.removeShutdownHook(hook) + } catch { + case ise: IllegalStateException => shuttingDown = true + } + if (!shuttingDown) { + if (exception.isInstanceOf[OutOfMemoryError]) { + System.exit(ExecutorExitCode.OOM) + } else { + System.exit(ExecutorExitCode.UNCAUGHT_EXCEPTION) + } } } catch { - case oom: OutOfMemoryError => System.exit(ExecutorExitCode.OOM) - case t: Throwable => System.exit(ExecutorExitCode.UNCAUGHT_EXCEPTION_TWICE) + case oom: OutOfMemoryError => Runtime.getRuntime.halt(ExecutorExitCode.OOM) + case t: Throwable => Runtime.getRuntime.halt(ExecutorExitCode.UNCAUGHT_EXCEPTION_TWICE) } } } -- cgit v1.2.3 From d0588bd6d7da3ba5adaba24303ad8616bdc2484f Mon Sep 17 00:00:00 2001 From: Charles Reiss Date: Tue, 19 Feb 2013 11:53:01 -0800 Subject: Catch/log errors deleting temp dirs --- core/src/main/scala/spark/storage/DiskStore.scala | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/spark/storage/DiskStore.scala b/core/src/main/scala/spark/storage/DiskStore.scala index 7e5b820cbb..ddbf8821ad 100644 --- a/core/src/main/scala/spark/storage/DiskStore.scala +++ b/core/src/main/scala/spark/storage/DiskStore.scala @@ -178,7 +178,11 @@ private class DiskStore(blockManager: BlockManager, rootDirs: String) Runtime.getRuntime.addShutdownHook(new Thread("delete Spark local dirs") { override def run() { logDebug("Shutdown hook called") - localDirs.foreach(localDir => Utils.deleteRecursively(localDir)) + try { + localDirs.foreach(localDir => Utils.deleteRecursively(localDir)) + } catch { + case t: Throwable => logError("Exception while deleting local spark dirs", t) + } } }) } -- cgit v1.2.3 From 092c631fa8da6381b814f4d262c884ba08629b39 Mon Sep 17 00:00:00 2001 From: Charles Reiss Date: Tue, 19 Feb 2013 17:49:55 -0800 Subject: Pull detection of being in a shutdown hook into utility function. --- core/src/main/scala/spark/Utils.scala | 21 +++++++++++++++++++++ core/src/main/scala/spark/executor/Executor.scala | 16 ++-------------- 2 files changed, 23 insertions(+), 14 deletions(-) diff --git a/core/src/main/scala/spark/Utils.scala b/core/src/main/scala/spark/Utils.scala index 28d643abca..81daacf958 100644 --- a/core/src/main/scala/spark/Utils.scala +++ b/core/src/main/scala/spark/Utils.scala @@ -454,4 +454,25 @@ private object Utils extends Logging { def clone[T](value: T, serializer: SerializerInstance): T = { serializer.deserialize[T](serializer.serialize(value)) } + + /** + * Detect whether this thread might be executing a shutdown hook. Will always return true if + * the current thread is a running a shutdown hook but may spuriously return true otherwise (e.g. + * if System.exit was just called by a concurrent thread). + * + * Currently, this detects whether the JVM is shutting down by Runtime#addShutdownHook throwing + * an IllegalStateException. + */ + def inShutdown(): Boolean = { + try { + val hook = new Thread { + override def run() {} + } + Runtime.getRuntime.addShutdownHook(hook) + Runtime.getRuntime.removeShutdownHook(hook) + } catch { + case ise: IllegalStateException => return true + } + return false + } } diff --git a/core/src/main/scala/spark/executor/Executor.scala b/core/src/main/scala/spark/executor/Executor.scala index b63bec11ad..5de09030aa 100644 --- a/core/src/main/scala/spark/executor/Executor.scala +++ b/core/src/main/scala/spark/executor/Executor.scala @@ -52,20 +52,8 @@ private[spark] class Executor extends Logging { logError("Uncaught exception in thread " + thread, exception) // We may have been called from a shutdown hook. If so, we must not call System.exit(). - // (If we do, we will deadlock.) Runtime#addShutdownHook should fail if we are shutting - // down, which would either occur if we were called from a shutdown hook or if - // a System.exit() occured concurrently. - var shuttingDown = false - try { - val hook = new Thread { - override def run() {} - } - Runtime.getRuntime.addShutdownHook(hook) - Runtime.getRuntime.removeShutdownHook(hook) - } catch { - case ise: IllegalStateException => shuttingDown = true - } - if (!shuttingDown) { + // (If we do, we will deadlock.) + if (!Utils.inShutdown()) { if (exception.isInstanceOf[OutOfMemoryError]) { System.exit(ExecutorExitCode.OOM) } else { -- cgit v1.2.3