[Java] Support multiple workers in Java worker process by kfstorm · Pull Request #5505 · ray-project/ray

kfstorm · 2019-08-22T04:18:39Z

Why are these changes needed?

JVM itself costs a significant amount of memory overhead. Using multi-processes means duplicating the overhead. Supporting multiple workers in a single Java worker process can mitigate it.

What do these changes do?

Adds the RayMultiWorkerNativeRuntime to wrap multiple RayNativeRuntime instances, and to act as a proxy to each instance.

Adds the Ray.asyncClosure methods to carry the runtime of current thread to the async Runnable or Callable. Users are required to wrap the Runnable or Callable with Ray.asyncClosure.

Adds a new option --java-num-workers-per-process for ray start.

Changes num_workers_per_process to num_workers_per_process_java in ray_config_def.h to make the config unrelated to Python workers.

NOTE: This is a straight forward solution by making everything related to worker runtime multi-instance. We need more detailed controls about which should be multi-instance and which should be shared. e.g. connections from worker to GCS should be shared to reduce the connection overhead. However, this PR will not address this. By default, a Java worker process will only start one worker unless the num workers per process config is specified, so it should be compatible with current behavior.

Related issue number

#2215

Linter

I've run scripts/format.sh to lint the changes in this PR.

…art` command.

…orker process.

…e same worker process." This reverts commit 223c8c24538ed67a0c5831b7f8631fc4b10d3b2d.

…hread

raulchen · 2019-08-22T06:56:07Z

java/runtime/src/main/java/org/ray/runtime/DefaultRayRuntimeFactory.java

      } else {
-        runtime = new RayNativeRuntime(rayConfig);
+        if (rayConfig.workerMode == WorkerType.DRIVER || rayConfig.numWorkersPerProcess == 1) {
+          runtime = new RayNativeRuntime(rayConfig);


Can we still use RayMultiWorkerNativeRuntime in this case?

I've removed rayConfig.numWorkersPerProcess == 1.

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

raulchen · 2019-08-22T06:58:46Z

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

+    LOGGER.info("Starting {} workers.", workerCount);
+
+    for (int i = 0; i < workerCount; i++) {
+      int finalI = i;


Suggested change

int finalI = i;

final int worker_index = i;

raulchen · 2019-08-22T07:00:52Z

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

+  private final RayNativeRuntime[] runtimes;
+  private final ThreadLocal<RayNativeRuntime> currentThreadRuntime = new ThreadLocal<>();
+
+  private CountDownLatch shutdownCountDownLatch = new CountDownLatch(1);


document these fields

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

raulchen · 2019-08-22T07:23:49Z

python/ray/scripts/scripts.py

+    required=False,
+    default=None,
+    type=int,
+    help="The number of workers per Java worker process.")


I think it's not necessary to add this option to ray start. ray start is already very complex. if one wants to config this option, they can use --internal-config for now. In the future, we should make C++ config easier to set.

The help text of --internal-config says that it's only for debugging and development purposes.

raulchen · 2019-08-22T07:28:05Z

src/ray/common/ray_config_def.h

-/// Number of workers per process
-RAY_CONFIG(int, num_workers_per_process, 1)
+/// Number of workers per Java worker process
+RAY_CONFIG(int, num_workers_per_process_java, 1)


Also add a num_workers_per_process_python and set it to 1?

raulchen · 2019-08-22T07:30:52Z

src/ray/raylet/node_manager.h

-  /// The number of workers per process.
-  int num_workers_per_process;
+  /// The number of workers per process per language.
+  std::unordered_map<Language, int, std::hash<int>> num_workers_per_process_by_lang;


you can use the EnumUnorderedMap util class to get rid of the std::hash<int> part.
But I think it's also fine to just remove this filed from NodeManagerConfig and just let WorkerPool read the config from RayConfig::instance().

Changed to read config in WorkerPool.

raulchen · 2019-08-22T07:39:34Z

src/ray/common/constants.h


 constexpr char kWorkerDynamicOptionPlaceholderPrefix[] = "RAY_WORKER_OPTION_";

+constexpr char kWorkerNumWorkersPlaceholder[] = "RAY_WORKER_NUM_WORKERS";


Can this just be RAY_WORKER_OPTION_0?

It can't. This placeholder is meant to be language independent. The idea is to keep WorkerPool have no idea how the real argument is, but only need to replace the placeholder with a number.

On the other hand, the full java argument is -Dray.raylet.config.num_workers_per_process_java=RAY_WORKER_PLACEHOLDER_NUM_WORKERS. If you replace RAY_WORKER_PLACEHOLDER_NUM_WORKERS with RAY_WORKER_PLACEHOLDER_OPTION_0, then we have to specificly add the -Dray.raylet.config.num_workers_per_process_java= part in C++.

Plus, this argument has nothing to do with dynamic worker options. It's odd to occupy 1 token of dynamic worker options. And we need to make a convention that it would always occupy the first token.

raulchen · 2019-08-22T07:41:30Z

java/api/src/main/java/org/ray/api/Ray.java

+   * @param runnable The runnable to wrap.
+   * @return The wrapped runnable.
+   */
+  public static Runnable asyncClosure(Runnable runnable) {


This name sounds kind of ambiguous to me. But I don't have a good name either. What about just wrapRunnable and wrapCallable?

Changed to wrapRunnable and wrapCallable.

zhijunfu

Thanks, LGTM. Just left a few comments.

zhijunfu · 2019-08-22T07:48:35Z

src/ray/raylet/worker_pool.h

  /// language.
-  WorkerPool(int num_worker_processes, int num_workers_per_process,
+  WorkerPool(int num_worker_processes,
+             const std::unordered_map<Language, int, std::hash<int>> &num_workers_per_process_by_lang,


nit: can use EnumUnorderedMap in src/ray/util/util.h

Thanks. Managed to get rid of it.

zhijunfu · 2019-08-22T07:52:27Z

src/ray/raylet/worker_pool.cc

+        << "Number of workers per process of language " << Language_Name(entry.first)
+        << " must be positive.";
+    state.multiple_for_warning =
+        std::max(num_worker_processes, maximum_startup_concurrency) *


should this be max or min?

This should be max. See line 49 of the original version.

java/runtime/src/main/java/org/ray/runtime/RayNativeRuntime.java

zhijunfu · 2019-08-22T08:03:07Z

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

+    LOGGER.info("Starting {} workers.", workerCount);
+
+    for (int i = 0; i < workerCount; i++) {
+      int finalI = i;


nit: maybe rename finalI ?

something like threadIndex or workerIndex?

AmplabJenkins · 2019-08-22T08:58:40Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16448/
Test PASSed.

AmplabJenkins · 2019-08-22T11:05:57Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16454/
Test PASSed.

AmplabJenkins · 2019-08-22T12:57:57Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16455/
Test PASSed.

jovany-wang · 2019-08-22T12:44:45Z

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

+
+  private static final Logger LOGGER = LoggerFactory.getLogger(RayMultiWorkerNativeRuntime.class);
+
+  private final int workerCount;


workerCount -> numWorkers?

to follow the naming style of numWorkersPerProcess above.

jovany-wang · 2019-08-22T12:54:32Z

java/runtime/src/main/java/org/ray/runtime/runner/worker/DefaultWorker.java

+      }
    } catch (Exception e) {
      LOGGER.error("Failed to start worker.", e);
    }


I think we should invoke Ray.shutdown() here.

FYI, Ray.shutdown() will be called during JVM shutdown because a shutdown hook is set in Ray.init().

jovany-wang · 2019-08-22T13:00:15Z

python/ray/node.py

            include_java=self._ray_params.include_java,
            java_worker_options=self._ray_params.java_worker_options,
+            java_num_workers_per_process=self._ray_params.
+            java_num_workers_per_process,


num_java_workers_per_process ?

Removed this parameter.

jovany-wang · 2019-08-22T13:08:32Z

python/ray/services.py

    command += "-Dray.log-dir={} ".format(os.path.join(session_dir, "logs"))

+    command += ("-Dray.raylet.config.num_workers_per_process_java=" +
+                "RAY_WORKER_NUM_WORKERS ")


Use a prefix like JAVA_WORKER_OPTION_ or JAVA_WORKER_PLACEHOLDER_?

Otherwise it's hard to be read as a placeholder for other developers.(I read it for a while to understand)

Updated both placeholders to RAY_WORKER_PLACEHOLDER_DYNAMIC_OPTION_ and RAY_WORKER_PLACEHOLDER_NUM_WORKERS.

…hread

This reverts commit fab5ae6.

…t#5507)"" This reverts commit dcddef6.

AmplabJenkins · 2019-08-27T14:10:05Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16568/
Test PASSed.

AmplabJenkins · 2019-08-27T14:48:10Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16569/
Test PASSed.

AmplabJenkins · 2019-08-27T15:52:16Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16572/
Test PASSed.

…_thread

AmplabJenkins · 2019-09-06T15:05:57Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16832/
Test FAILed.

raulchen

Thanks. LGTM. Just a few small comments.

java/api/src/main/java/org/ray/api/Ray.java

java/runtime/src/main/java/org/ray/runtime/RayMultiWorkerNativeRuntime.java

raulchen · 2019-09-07T08:21:26Z

java/test/src/main/java/org/ray/api/test/ActorTest.java

    // Wait until the object is deleted, because the above free operation is async.
    while (true) {
-      NativeRayObject result = ((AbstractRayRuntime) Ray.internal()).getObjectStore()
+      NativeRayObject result = TestUtils.getRuntime().getObjectStore()


Nice util function!

src/ray/common/constants.h

AmplabJenkins · 2019-09-07T12:59:46Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/16857/
Test FAILed.

kfstorm added 20 commits August 19, 2019 19:03

Support multiple workers in Java worker process

0f033e1

Fix

ba15466

Refine code

4d72fde

Fix Java glog

2a29c48

Make resetLibraryPath thread-safe

51405b7

ray stop kill processes more carefully

1e4fd5c

Minor fix

5a946e3

Support specify the number of workers per process for Java in `ray st…

66dab94

…art` command.

Loosen ray stop filter of python processes.

42e63b7

rename

94f00c8

Warmup cluster in test case 'testGetThrowsQuicklyWhenFoundException'

8792eb9

Update worker process count warning

8c5d387

Ignore dynamic worker options if start multiple workers in the same w…

838cf86

…orker process.

Revert "Ignore dynamic worker options if start multiple workers in th…

6eb6d12

…e same worker process." This reverts commit 223c8c24538ed67a0c5831b7f8631fc4b10d3b2d.

Make num workers per process argument generated by raylet with template

75b1657

Fix bug

1019619

lint

d45c5be

Fix test due to dynamic worker options bug

03db792

Fix worker pool test

02516bf

Merge remote-tracking branch 'origin/master' into java_worker_multi_t…

c804212

…hread

kfstorm requested review from jovany-wang, pcmoritz, raulchen, robertnishihara and zhijunfu August 22, 2019 04:18

kfstorm changed the title ~~Support multiple workers in Java worker process~~ [Java] Support multiple workers in Java worker process Aug 22, 2019

raulchen reviewed Aug 22, 2019

View reviewed changes

lint

4834567

kfstorm force-pushed the java_worker_multi_thread branch from 38b1e07 to 4834567 Compare August 22, 2019 08:06

zhijunfu reviewed Aug 22, 2019

View reviewed changes

jovany-wang reviewed Aug 22, 2019

View reviewed changes

kfstorm added 9 commits August 27, 2019 15:06

Merge remote-tracking branch 'origin/master' into java_worker_multi_t…

3dc96b3

…hread

Fix build after merge

6517ed5

Address comments

82d6637

Address comments

986e654

Revert "[Java] Automatically clean up temp files. (ray-project#5507)"

dcddef6

This reverts commit fab5ae6.

Fix tests

5d44535

Revert "Revert "[Java] Automatically clean up temp files. (ray-projec…

6c48611

…t#5507)"" This reverts commit dcddef6.

Minor fix

65aff73

Fix test

de9548d

kfstorm added 2 commits September 6, 2019 19:19

Merge remote-tracking branch 'upstream/master' into java_worker_multi…

18a5dd2

…_thread

Address comments

30a9f56

raulchen approved these changes Sep 7, 2019

View reviewed changes

Address comments

f591d9d

raulchen merged commit 732336f into ray-project:master Sep 7, 2019

raulchen deleted the java_worker_multi_thread branch September 7, 2019 14:52

kfstorm mentioned this pull request Mar 16, 2020

Support multiple core workers in one process #7623

Merged

10 tasks


		constexpr char kWorkerDynamicOptionPlaceholderPrefix[] = "RAY_WORKER_OPTION_";

		constexpr char kWorkerNumWorkersPlaceholder[] = "RAY_WORKER_NUM_WORKERS";


		private static final Logger LOGGER = LoggerFactory.getLogger(RayMultiWorkerNativeRuntime.class);

		private final int workerCount;

Conversation

kfstorm commented Aug 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why are these changes needed?

What do these changes do?

Related issue number

Linter

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

raulchen Aug 22, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

zhijunfu left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AmplabJenkins commented Aug 22, 2019

Uh oh!

AmplabJenkins commented Aug 22, 2019

Uh oh!

AmplabJenkins commented Aug 22, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kfstorm commented Aug 22, 2019 •

edited

Loading

raulchen Aug 22, 2019 •

edited

Loading

jovany-wang Aug 22, 2019 •

edited

Loading