-
Notifications
You must be signed in to change notification settings - Fork 28.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[SPARK-11035][CORE] Add in-process Spark app launcher.
This change adds a new launcher that allows applications to be run in a separate thread in the same process as the calling code. To achieve that, some code from the child process implementation was moved to abstract classes that implement the common functionality, and the new launcher inherits from those. The new launcher was added as a new class, instead of implemented as a new option to the existing SparkLauncher, to avoid ambigous APIs. For example, SparkLauncher has ways to set the child app's environment, modify SPARK_HOME, or control the logging of the child process, none of which apply to in-process apps. The in-process launcher has limitations: it needs Spark in the context class loader of the calling thread, and it's bound by Spark's current limitation of a single client-mode application per JVM. It also relies on the recently added SparkApplication trait to make sure different apps don't mess up each other's configuration, so config isolation is currently limited to cluster mode. I also chose to keep the same socket-based communication for in-process apps, even though it might be possible to avoid it for in-process mode. That helps both implementations share more code. Tested with new and existing unit tests, and with a simple app that uses the launcher; also made sure the app ran fine with older launcher jar to check binary compatibility. Author: Marcelo Vanzin <[email protected]> Closes #19591 from vanzin/SPARK-11035.
- Loading branch information
Showing
21 changed files
with
1,139 additions
and
505 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
129 changes: 129 additions & 0 deletions
129
launcher/src/main/java/org/apache/spark/launcher/AbstractAppHandle.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,129 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one or more | ||
| * contributor license agreements. See the NOTICE file distributed with | ||
| * this work for additional information regarding copyright ownership. | ||
| * The ASF licenses this file to You under the Apache License, Version 2.0 | ||
| * (the "License"); you may not use this file except in compliance with | ||
| * the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package org.apache.spark.launcher; | ||
|
|
||
| import java.io.IOException; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.logging.Level; | ||
| import java.util.logging.Logger; | ||
|
|
||
| abstract class AbstractAppHandle implements SparkAppHandle { | ||
|
|
||
| private static final Logger LOG = Logger.getLogger(ChildProcAppHandle.class.getName()); | ||
|
|
||
| private final LauncherServer server; | ||
|
|
||
| private LauncherConnection connection; | ||
| private List<Listener> listeners; | ||
| private State state; | ||
| private String appId; | ||
| private boolean disposed; | ||
|
|
||
| protected AbstractAppHandle(LauncherServer server) { | ||
| this.server = server; | ||
| this.state = State.UNKNOWN; | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void addListener(Listener l) { | ||
| if (listeners == null) { | ||
| listeners = new ArrayList<>(); | ||
| } | ||
| listeners.add(l); | ||
| } | ||
|
|
||
| @Override | ||
| public State getState() { | ||
| return state; | ||
| } | ||
|
|
||
| @Override | ||
| public String getAppId() { | ||
| return appId; | ||
| } | ||
|
|
||
| @Override | ||
| public void stop() { | ||
| CommandBuilderUtils.checkState(connection != null, "Application is still not connected."); | ||
| try { | ||
| connection.send(new LauncherProtocol.Stop()); | ||
| } catch (IOException ioe) { | ||
| throw new RuntimeException(ioe); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized void disconnect() { | ||
| if (!disposed) { | ||
| disposed = true; | ||
| if (connection != null) { | ||
| try { | ||
| connection.close(); | ||
| } catch (IOException ioe) { | ||
| // no-op. | ||
| } | ||
| } | ||
| server.unregister(this); | ||
| } | ||
| } | ||
|
|
||
| void setConnection(LauncherConnection connection) { | ||
| this.connection = connection; | ||
| } | ||
|
|
||
| LauncherConnection getConnection() { | ||
| return connection; | ||
| } | ||
|
|
||
| boolean isDisposed() { | ||
| return disposed; | ||
| } | ||
|
|
||
| void setState(State s) { | ||
| setState(s, false); | ||
| } | ||
|
|
||
| synchronized void setState(State s, boolean force) { | ||
| if (force || !state.isFinal()) { | ||
| state = s; | ||
| fireEvent(false); | ||
| } else { | ||
| LOG.log(Level.WARNING, "Backend requested transition from final state {0} to {1}.", | ||
| new Object[] { state, s }); | ||
| } | ||
| } | ||
|
|
||
| synchronized void setAppId(String appId) { | ||
| this.appId = appId; | ||
| fireEvent(true); | ||
| } | ||
|
|
||
| private void fireEvent(boolean isInfoChanged) { | ||
| if (listeners != null) { | ||
| for (Listener l : listeners) { | ||
| if (isInfoChanged) { | ||
| l.infoChanged(this); | ||
| } else { | ||
| l.stateChanged(this); | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.