aboutsummaryrefslogtreecommitdiff
path: root/core
diff options
context:
space:
mode:
authorTerence Yim <terence@cask.co>2016-04-12 13:46:39 -0700
committerReynold Xin <rxin@databricks.com>2016-04-12 13:46:39 -0700
commit3e53de4bdd6d7b6de1fe3e5bfbdc53180aa9a737 (patch)
treee02718d2767554e986f3765dff9c5241c8a5d319 /core
parentbcd2076274b1a95f74616d0ceacb0696e38b5f4c (diff)
downloadspark-3e53de4bdd6d7b6de1fe3e5bfbdc53180aa9a737.tar.gz
spark-3e53de4bdd6d7b6de1fe3e5bfbdc53180aa9a737.tar.bz2
spark-3e53de4bdd6d7b6de1fe3e5bfbdc53180aa9a737.zip
[SPARK-14513][CORE] Fix threads left behind after stopping SparkContext
## What changes were proposed in this pull request? Shutting down `QueuedThreadPool` used by Jetty `Server` to avoid threads leakage after SparkContext is stopped. Note: If this fix is going to apply to the `branch-1.6`, one more patch on the `NettyRpcEnv` class is needed so that the `NettyRpcEnv._fileServer.shutdown` is called in the `NettyRpcEnv.cleanup` method. This is due to the removal of `_fileServer` field in the `NettyRpcEnv` class in the master branch. Please advice if a second PR is necessary for bring this fix back to `branch-1.6` ## How was this patch tested? Ran the ./dev/run-tests locally Author: Terence Yim <terence@cask.co> Closes #12318 from chtyim/fixes/SPARK-14513-thread-leak.
Diffstat (limited to 'core')
-rw-r--r--core/src/main/scala/org/apache/spark/HttpServer.scala7
-rw-r--r--core/src/main/scala/org/apache/spark/ui/JettyUtils.scala14
-rw-r--r--core/src/main/scala/org/apache/spark/ui/WebUI.scala2
3 files changed, 21 insertions, 2 deletions
diff --git a/core/src/main/scala/org/apache/spark/HttpServer.scala b/core/src/main/scala/org/apache/spark/HttpServer.scala
index 9fad1f6786..982b6d6b61 100644
--- a/core/src/main/scala/org/apache/spark/HttpServer.scala
+++ b/core/src/main/scala/org/apache/spark/HttpServer.scala
@@ -25,6 +25,7 @@ import org.eclipse.jetty.server.Server
import org.eclipse.jetty.server.bio.SocketConnector
import org.eclipse.jetty.server.ssl.SslSocketConnector
import org.eclipse.jetty.servlet.{DefaultServlet, ServletContextHandler, ServletHolder}
+import org.eclipse.jetty.util.component.LifeCycle
import org.eclipse.jetty.util.security.{Constraint, Password}
import org.eclipse.jetty.util.thread.QueuedThreadPool
@@ -155,6 +156,12 @@ private[spark] class HttpServer(
throw new ServerStateException("Server is already stopped")
} else {
server.stop()
+ // Stop the ThreadPool if it supports stop() method (through LifeCycle).
+ // It is needed because stopping the Server won't stop the ThreadPool it uses.
+ val threadPool = server.getThreadPool
+ if (threadPool != null && threadPool.isInstanceOf[LifeCycle]) {
+ threadPool.asInstanceOf[LifeCycle].stop
+ }
port = -1
server = null
}
diff --git a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
index c3c59f857d..119165f724 100644
--- a/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
+++ b/core/src/main/scala/org/apache/spark/ui/JettyUtils.scala
@@ -30,6 +30,7 @@ import org.eclipse.jetty.server.handler._
import org.eclipse.jetty.server.nio.SelectChannelConnector
import org.eclipse.jetty.server.ssl.SslSelectChannelConnector
import org.eclipse.jetty.servlet._
+import org.eclipse.jetty.util.component.LifeCycle
import org.eclipse.jetty.util.thread.QueuedThreadPool
import org.json4s.JValue
import org.json4s.jackson.JsonMethods.{pretty, render}
@@ -350,4 +351,15 @@ private[spark] object JettyUtils extends Logging {
private[spark] case class ServerInfo(
server: Server,
boundPort: Int,
- rootHandler: ContextHandlerCollection)
+ rootHandler: ContextHandlerCollection) {
+
+ def stop(): Unit = {
+ server.stop()
+ // Stop the ThreadPool if it supports stop() method (through LifeCycle).
+ // It is needed because stopping the Server won't stop the ThreadPool it uses.
+ val threadPool = server.getThreadPool
+ if (threadPool != null && threadPool.isInstanceOf[LifeCycle]) {
+ threadPool.asInstanceOf[LifeCycle].stop
+ }
+ }
+}
diff --git a/core/src/main/scala/org/apache/spark/ui/WebUI.scala b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
index 250b7f2e5f..3939b111b5 100644
--- a/core/src/main/scala/org/apache/spark/ui/WebUI.scala
+++ b/core/src/main/scala/org/apache/spark/ui/WebUI.scala
@@ -153,7 +153,7 @@ private[spark] abstract class WebUI(
def stop() {
assert(serverInfo.isDefined,
"Attempted to stop %s before binding to a server!".format(className))
- serverInfo.get.server.stop()
+ serverInfo.get.stop()
}
}