wynot12 commented on a change in pull request #73: [Nemo-62] Support Multiple
Jobs Submission in a Single User Program
URL: https://github.com/apache/incubator-nemo/pull/73#discussion_r201420338
##########
File path: client/src/main/java/edu/snu/nemo/client/JobLauncher.java
##########
@@ -118,22 +153,39 @@ public static void main(final String[] args) throws
Exception {
*/
// When modifying the signature of this method, see
CompilerTestUtil#compileDAG and make corresponding changes
public static void launchDAG(final DAG dag) {
- try {
- if (jobAndDriverConf == null || deployModeConf == null || builtJobConf
== null) {
- throw new RuntimeException("Configuration for launching driver is not
ready");
- }
- serializedDAG =
Base64.getEncoder().encodeToString(SerializationUtils.serialize(dag));
- // Launch and wait indefinitely for the job to finish
- final LauncherStatus launcherStatus =
DriverLauncher.getLauncher(deployModeConf)
- .run(jobAndDriverConf);
- final Optional<Throwable> possibleError = launcherStatus.getError();
- if (possibleError.isPresent()) {
- throw new RuntimeException(possibleError.get());
- } else {
- LOG.info("Job successfully completed");
+ serializedDAG =
Base64.getEncoder().encodeToString(SerializationUtils.serialize(dag));
+ jobDoneLatch = new CountDownLatch(1);
+
+ if (!driverActive) {
+ LOG.info("LaunchDAG: Launching driver");
+ try {
+ if (jobAndDriverConf == null || deployModeConf == null || builtJobConf
== null) {
+ throw new RuntimeException("Configuration for launching driver is
not ready");
+ }
+
+ driverLauncher = DriverLauncher.getLauncher(deployModeConf);
+ driverLauncher.submit(jobAndDriverConf, 500);
+ // When the driver is up and the resource is ready, the ResourceReady
message is delivered,
+ // triggering the DRIVER_RPC_SERVER to send the LaunchDAG message to
the driver.
+ } catch (final InjectionException e) {
+ throw new RuntimeException(e);
}
- } catch (final InjectionException e) {
- throw new RuntimeException(e);
+ } else {
+ LOG.info("LaunchDAG: Using the existing driver");
+ driverRPCServer.send(ControlMessage.ClientToDriverMessage.newBuilder()
+ .setType(ControlMessage.ClientToDriverMessageType.LaunchDAG)
+
.setLaunchDAG(ControlMessage.LaunchDAGMessage.newBuilder().setDag(serializedDAG).build())
+ .build());
+ }
+
+ // Wait for the JobDone message from the driver
Review comment:
Are _Job_ and _DAG_ exactly same term?
If they are, isn't it better to use only one of them for minimizing
confusion and for consistency?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
