Implement launching cluster with shell

scripts/cluster.py

@@ -65,7 +65,7 @@ def start_ray_multi_node(node_ip_addresses, username, key_file, worker_path, ins
   shell_script_path = os.path.join(args.installation_directory, "ray/scripts/shell.py")
   print """
     source "{}";
-    python "{}" --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001
+    python "{}" --scheduler-address={}:10001 --objstore-address={}:20001 --worker-address={}:30001 --attach
   """.format(setup_env_path, shell_script_path, node_ip_addresses[0], node_ip_addresses[0], node_ip_addresses[0])
 
 def stop_ray_multi_node(node_ip_addresses, username, key_file):

scripts/default_worker.py

@@ -0,0 +1,28 @@
+import sys
+import argparse
+import numpy as np
+
+import ray.arrays.remote as ra
+import ray.arrays.distributed as da
+
+import ray
+import ray.services as services
+import ray.worker as worker
+
+parser = argparse.ArgumentParser(description="Parse addresses for the worker to connect to.")
+parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address")
+parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address")
+parser.add_argument("--worker-address", default="127.0.0.1:40001", type=str, help="the worker's address")
+
+if __name__ == "__main__":
+  args = parser.parse_args()
+  worker.connect(args.scheduler_address, args.objstore_address, args.worker_address)
+
+  ray.register_module(ra)
+  ray.register_module(ra.random)
+  ray.register_module(ra.linalg)
+  ray.register_module(da)
+  ray.register_module(da.random)
+  ray.register_module(da.linalg)
+
+  worker.main_loop()

scripts/shell.py

@@ -1,3 +1,4 @@
+import os
 import argparse
 import numpy as np
 
@@ -8,14 +9,29 @@ import ray.worker as worker
 import ray.array.remote as ra
 import ray.array.distributed as da
 
-parser = argparse.ArgumentParser(description='Parse addresses for the worker to connect to.')
+DEFAULT_NUM_WORKERS = 10
+DEFAULT_WORKER_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "default_worker.py")
+
+parser = argparse.ArgumentParser(description="Parse shell options")
 parser.add_argument("--scheduler-address", default="127.0.0.1:10001", type=str, help="the scheduler's address")
 parser.add_argument("--objstore-address", default="127.0.0.1:20001", type=str, help="the objstore's address")
 parser.add_argument("--worker-address", default="127.0.0.1:30001", type=str, help="the worker's address")
+parser.add_argument("--attach", action="store_true", help="If true, attach the shell to an already running cluster. If false, start a new cluster.")
+parser.add_argument("--worker-path", help="Path to the worker script")
+parser.add_argument("--num-workers", help="Number of workers to start")
 
 if __name__ == "__main__":
   args = parser.parse_args()
-  worker.connect(args.scheduler_address, args.objstore_address, args.worker_address)
+  if args.attach:
+    assert args.worker_path is None, "when attaching, no new worker can be started"
+    assert args.num_workers is None, "when attaching, no new worker can be started"
+    worker.connect(args.scheduler_address, args.objstore_address, args.worker_address)
+  else:
+    services.start_ray_local(num_workers=args.num_workers if not args.num_workers is None else DEFAULT_NUM_WORKERS,
+                             worker_path=args.worker_path if not args.num_workers is None else DEFAULT_WORKER_PATH)
 
   import IPython
   IPython.embed()
+
+  if not args.attach:
+    services.cleanup()