Deprecate num_workers argument to ray.init and ray start. (#3114)

* Remove num_workers argument.

* Fix

* Fix

python/benchmarks/benchmark_actor.py

@@ -9,7 +9,7 @@ NUM_WORKERS = 4
 
 def setup():
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=NUM_WORKERS, num_cpus=4)
+        ray.init(num_cpus=4)
         setup.is_initialized = True
 
 

python/benchmarks/benchmark_get.py

@@ -9,7 +9,7 @@ import ray
 
 def setup():
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=4, num_cpus=4)
+        ray.init(num_cpus=4)
         setup.is_initialized = True
 
 

python/benchmarks/benchmark_put.py

@@ -9,7 +9,7 @@ import ray
 
 def setup():
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=4, num_cpus=4)
+        ray.init(num_cpus=0)
         setup.is_initialized = True
 
 

python/benchmarks/benchmark_queue.py

@@ -8,7 +8,7 @@ from ray.experimental.queue import Queue
 
 def setup():
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=4, num_cpus=4)
+        ray.init(num_cpus=4)
         setup.is_initialized = True
 
 

python/benchmarks/benchmark_task.py

@@ -7,7 +7,7 @@ import ray
 
 def setup():
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=10, num_cpus=10, resources={"foo": 1})
+        ray.init(num_cpus=10, resources={"foo": 1})
         setup.is_initialized = True
 
 

python/benchmarks/benchmark_wait.py

@@ -9,7 +9,7 @@ import ray
 
 def setup(*args):
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=4, num_cpus=4)
+        ray.init(num_cpus=4)
         setup.is_initialized = True
 
 

python/benchmarks/benchmarks.py

@@ -7,7 +7,7 @@ import ray
 
 def setup():
     if not hasattr(setup, "is_initialized"):
-        ray.init(num_workers=4, num_cpus=4)
+        ray.init(num_cpus=4)
         setup.is_initialized = True
 
 

python/ray/services.py

@@ -1328,7 +1328,8 @@ def start_ray_processes(address_info=None,
         resources = num_local_schedulers * [resources]
 
     if num_workers is not None:
-        workers_per_local_scheduler = num_local_schedulers * [num_workers]
+        raise Exception("The 'num_workers' argument is deprecated. Please use "
+                        "'num_cpus' instead.")
     else:
         workers_per_local_scheduler = []
         for resource_dict in resources:
@@ -1479,7 +1480,7 @@ def start_ray_node(node_ip_address,
                    redis_address,
                    object_manager_ports=None,
                    node_manager_ports=None,
-                   num_workers=0,
+                   num_workers=None,
                    num_local_schedulers=1,
                    object_store_memory=None,
                    redis_password=None,
@@ -1572,7 +1573,7 @@ def start_ray_head(address_info=None,
                    node_ip_address="127.0.0.1",
                    redis_port=None,
                    redis_shard_ports=None,
-                   num_workers=0,
+                   num_workers=None,
                    num_local_schedulers=1,
                    object_store_memory=None,
                    worker_path=None,

python/ray/worker.py

@@ -1303,8 +1303,6 @@ def _init(address_info=None,
             object IDs. The same value can be used across multiple runs of the
             same job in order to generate the object IDs in a consistent
             manner. However, the same ID should not be used for different jobs.
-        num_workers (int): The number of workers to start. This is only
-            provided if start_ray_local is True.
         num_local_schedulers (int): The number of local schedulers to start.
             This is only provided if start_ray_local is True.
         object_store_memory: The maximum amount of memory (in bytes) to
@@ -1554,8 +1552,6 @@ def init(redis_address=None,
             object IDs. The same value can be used across multiple runs of the
             same job in order to generate the object IDs in a consistent
             manner. However, the same ID should not be used for different jobs.
-        num_workers (int): The number of workers to start. This is only
-            provided if redis_address is not provided.
         local_mode (bool): True if the code should be executed serially
             without Ray. This is useful for debugging.
         redirect_worker_output: True if the stdout and stderr of worker