From 7c1fd19fd951524a617a43044fcdf810bbb65e34 Mon Sep 17 00:00:00 2001
From: Hanwei Jin <chuxi@users.noreply.github.com>
Date: Wed, 24 Oct 2018 20:43:39 +0800
Subject: [PATCH] [Java] support python worker command in raylet (#3092)

<!--
Thank you for your contribution!

Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request.
-->

## What do these changes do?

support raylet, which is started by java runManager, to start python default_worker.py .

So when doing local test of java call python task, it helps auto start python worker.

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this change? -->
---
 .../org/ray/runtime/config/RayConfig.java     |  7 +++++++
 .../org/ray/runtime/runner/RunManager.java    | 20 ++++++++++++++++++-
 2 files changed, 26 insertions(+), 1 deletion(-)

diff --git a/java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java b/java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java
index b172a4111..1688fa6db 100644
--- a/java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java
+++ b/java/runtime/src/main/java/org/ray/runtime/config/RayConfig.java
@@ -54,6 +54,7 @@ public class RayConfig {
   public final String plasmaStoreExecutablePath;
   public final String rayletExecutablePath;
   public final String driverResourcePath;
+  public final String pythonWorkerCommand;
 
   private void validate() {
     if (workerMode == WorkerMode.WORKER) {
@@ -136,6 +137,12 @@ public class RayConfig {
       jvmParameters = ImmutableList.of();
     }
 
+    if (config.hasPath("ray.worker.python-command")) {
+      pythonWorkerCommand = config.getString("ray.worker.python-command");
+    } else {
+      pythonWorkerCommand = null;
+    }
+
     // redis configurations
     String redisAddress = config.getString("ray.redis.address");
     if (!redisAddress.isEmpty()) {
diff --git a/java/runtime/src/main/java/org/ray/runtime/runner/RunManager.java b/java/runtime/src/main/java/org/ray/runtime/runner/RunManager.java
index 3be219dca..940d0e78b 100644
--- a/java/runtime/src/main/java/org/ray/runtime/runner/RunManager.java
+++ b/java/runtime/src/main/java/org/ray/runtime/runner/RunManager.java
@@ -185,7 +185,7 @@ public class RunManager {
         "0", // number of initial workers
         String.valueOf(maximumStartupConcurrency),
         ResourceUtil.getResourcesStringFromMap(rayConfig.resources),
-        "", // python worker command
+        buildPythonWorkerCommand(), // python worker command
         buildWorkerCommandRaylet() // java worker command
     );
 
@@ -247,4 +247,22 @@ public class RunManager {
     startProcess(command, null, "plasma_store");
   }
 
+  private String buildPythonWorkerCommand() {
+    // disable python worker start from raylet, which starts from java
+    if (rayConfig.pythonWorkerCommand == null) {
+      return "";
+    }
+
+    List<String> cmd = new ArrayList<>();
+    cmd.add(rayConfig.pythonWorkerCommand);
+    cmd.add("--node-ip-address=" + rayConfig.nodeIp);
+    cmd.add("--object-store-name=" + rayConfig.objectStoreSocketName);
+    cmd.add("--raylet-name=" + rayConfig.rayletSocketName);
+    cmd.add("--redis-address=" + rayConfig.getRedisAddress());
+
+    String command = cmd.stream().collect(Collectors.joining(" "));
+    LOGGER.debug("python worker command: {}", command);
+    return command;
+  }
+
 }