dotnet
/
spark
зеркало из https://github.com/dotnet/spark.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

DotnetBackend handles .NET process in a more robust way (#424)

This commit is contained in:
Tylor Lake 2020-02-12 11:47:13 -08:00 коммит произвёл GitHub
Родитель 4dbfb34960
Коммит f740e07e6d
Не найден ключ, соответствующий данной подписи
Идентификатор ключа GPG: 4AEE18F83AFDEB23
3 изменённых файлов: 102 добавлений и 12 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

38
src/scala/microsoft-spark-2.3.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala
Просмотреть файл

@ -110,6 +110,7 @@ object DotnetRunner extends Logging {
if (initialized.tryAcquire(backendTimeout, TimeUnit.SECONDS)) {
if (!runInDebugMode) {
var returnCode = -1
var process: Process = null
try {
val builder = new ProcessBuilder(processParameters)
val env = builder.environment()
@ -120,7 +121,7 @@ object DotnetRunner extends Logging {
logInfo(s"Adding key=$key and value=$value to environment")
}
builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
val process = builder.start()
process = builder.start()
// Redirect stdin of JVM process to stdin of .NET process.
new RedirectThread(System.in, process.getOutputStream, "redirect JVM input").start()
@ -128,11 +129,13 @@ object DotnetRunner extends Logging {
new RedirectThread(process.getInputStream, System.out, "redirect .NET stdout").start()
new RedirectThread(process.getErrorStream, System.out, "redirect .NET stderr").start()
returnCode = process.waitFor()
closeBackend(dotnetBackend)
process.waitFor()
} catch {
case t: Throwable =>
logError(s"${t.getMessage} \n ${t.getStackTrace}")
logThrowable(t)
} finally {
returnCode = closeDotnetProcess(process)
closeBackend(dotnetBackend)
}
if (returnCode != 0) {
@ -232,6 +235,30 @@ object DotnetRunner extends Logging {
dotnetBackend.close()
}
private def closeDotnetProcess(dotnetProcess: Process): Int = {
if (dotnetProcess == null) {
return -1
} else if (!dotnetProcess.isAlive) {
return dotnetProcess.exitValue()
}
// Try to (gracefully on Linux) kill the process and resort to force if interrupted
var returnCode = -1
logInfo("Closing .NET process")
try {
dotnetProcess.destroy()
returnCode = dotnetProcess.waitFor()
} catch {
case _: InterruptedException =>
logInfo("Thread interrupted while waiting for graceful close. Forcefully closing .NET process")
returnCode = dotnetProcess.destroyForcibly().waitFor()
case t: Throwable =>
logThrowable(t)
}
returnCode
}
private def initializeSettings(args: Array[String]): (Boolean, Int) = {
val runInDebugMode = (args.length == 1 || args.length == 2) && args(0).equalsIgnoreCase(
"debug")
@ -246,4 +273,7 @@ object DotnetRunner extends Logging {
(runInDebugMode, portNumber)
}
private def logThrowable(throwable: Throwable): Unit =
logError(s"${throwable.getMessage} \n ${throwable.getStackTrace.mkString("\n")}")
}

38
src/scala/microsoft-spark-2.4.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala
Просмотреть файл

@ -110,6 +110,7 @@ object DotnetRunner extends Logging {
if (initialized.tryAcquire(backendTimeout, TimeUnit.SECONDS)) {
if (!runInDebugMode) {
var returnCode = -1
var process: Process = null
try {
val builder = new ProcessBuilder(processParameters)
val env = builder.environment()
@ -120,7 +121,7 @@ object DotnetRunner extends Logging {
logInfo(s"Adding key=$key and value=$value to environment")
}
builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
val process = builder.start()
process = builder.start()
// Redirect stdin of JVM process to stdin of .NET process.
new RedirectThread(System.in, process.getOutputStream, "redirect JVM input").start()
@ -128,11 +129,13 @@ object DotnetRunner extends Logging {
new RedirectThread(process.getInputStream, System.out, "redirect .NET stdout").start()
new RedirectThread(process.getErrorStream, System.out, "redirect .NET stderr").start()
returnCode = process.waitFor()
closeBackend(dotnetBackend)
process.waitFor()
} catch {
case t: Throwable =>
logError(s"${t.getMessage} \n ${t.getStackTrace}")
logThrowable(t)
} finally {
returnCode = closeDotnetProcess(process)
closeBackend(dotnetBackend)
}
if (returnCode != 0) {
@ -232,6 +235,30 @@ object DotnetRunner extends Logging {
dotnetBackend.close()
}
private def closeDotnetProcess(dotnetProcess: Process): Int = {
if (dotnetProcess == null) {
return -1
} else if (!dotnetProcess.isAlive) {
return dotnetProcess.exitValue()
}
// Try to (gracefully on Linux) kill the process and resort to force if interrupted
var returnCode = -1
logInfo("Closing .NET process")
try {
dotnetProcess.destroy()
returnCode = dotnetProcess.waitFor()
} catch {
case _: InterruptedException =>
logInfo("Thread interrupted while waiting for graceful close. Forcefully closing .NET process")
returnCode = dotnetProcess.destroyForcibly().waitFor()
case t: Throwable =>
logThrowable(t)
}
returnCode
}
private def initializeSettings(args: Array[String]): (Boolean, Int) = {
val runInDebugMode = (args.length == 1 || args.length == 2) && args(0).equalsIgnoreCase(
"debug")
@ -246,4 +273,7 @@ object DotnetRunner extends Logging {
(runInDebugMode, portNumber)
}
private def logThrowable(throwable: Throwable): Unit =
logError(s"${throwable.getMessage} \n ${throwable.getStackTrace.mkString("\n")}")
}

38
src/scala/microsoft-spark-3.0.x/src/main/scala/org/apache/spark/deploy/dotnet/DotnetRunner.scala
Просмотреть файл

@ -110,6 +110,7 @@ object DotnetRunner extends Logging {
if (initialized.tryAcquire(backendTimeout, TimeUnit.SECONDS)) {
if (!runInDebugMode) {
var returnCode = -1
var process: Process = null
try {
val builder = new ProcessBuilder(processParameters)
val env = builder.environment()
@ -120,7 +121,7 @@ object DotnetRunner extends Logging {
logInfo(s"Adding key=$key and value=$value to environment")
}
builder.redirectErrorStream(true) // Ugly but needed for stdout and stderr to synchronize
val process = builder.start()
process = builder.start()
// Redirect stdin of JVM process to stdin of .NET process.
new RedirectThread(System.in, process.getOutputStream, "redirect JVM input").start()
@ -128,11 +129,13 @@ object DotnetRunner extends Logging {
new RedirectThread(process.getInputStream, System.out, "redirect .NET stdout").start()
new RedirectThread(process.getErrorStream, System.out, "redirect .NET stderr").start()
returnCode = process.waitFor()
closeBackend(dotnetBackend)
process.waitFor()
} catch {
case t: Throwable =>
logError(s"${t.getMessage} \n ${t.getStackTrace}")
logThrowable(t)
} finally {
returnCode = closeDotnetProcess(process)
closeBackend(dotnetBackend)
}
if (returnCode != 0) {
@ -232,6 +235,30 @@ object DotnetRunner extends Logging {
dotnetBackend.close()
}
private def closeDotnetProcess(dotnetProcess: Process): Int = {
if (dotnetProcess == null) {
return -1
} else if (!dotnetProcess.isAlive) {
return dotnetProcess.exitValue()
}
// Try to (gracefully on Linux) kill the process and resort to force if interrupted
var returnCode = -1
logInfo("Closing .NET process")
try {
dotnetProcess.destroy()
returnCode = dotnetProcess.waitFor()
} catch {
case _: InterruptedException =>
logInfo("Thread interrupted while waiting for graceful close. Forcefully closing .NET process")
returnCode = dotnetProcess.destroyForcibly().waitFor()
case t: Throwable =>
logThrowable(t)
}
returnCode
}
private def initializeSettings(args: Array[String]): (Boolean, Int) = {
val runInDebugMode = (args.length == 1 || args.length == 2) && args(0).equalsIgnoreCase(
"debug")
@ -246,4 +273,7 @@ object DotnetRunner extends Logging {
(runInDebugMode, portNumber)
}
private def logThrowable(throwable: Throwable): Unit =
logError(s"${throwable.getMessage} \n ${throwable.getStackTrace.mkString("\n")}")
}