Clean up random name generation.

Author: robertnishihara

lib/python/ray/services.py

@@ -28,6 +28,9 @@ def address(host, port):
 def new_port():
   return random.randint(10000, 65535)
 
+def random_name():
+  return str(random.randint(0, 99999999))
+
 def cleanup():
   """When running in local mode, shutdown the Ray processes.
 
@@ -65,7 +68,7 @@ def start_redis(port):
 
 def start_local_scheduler(redis_address, plasma_store_name):
   local_scheduler_filepath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "../photon/build/photon_scheduler")
-  local_scheduler_name = "/tmp/scheduler{}".format(random.randint(0, 10000))
+  local_scheduler_name = "/tmp/scheduler{}".format(random_name())
   p = subprocess.Popen([local_scheduler_filepath, "-s", local_scheduler_name, "-r", redis_address, "-p", plasma_store_name])
   if cleanup:
     all_processes.append(p)
@@ -81,12 +84,12 @@ def start_objstore(node_ip_address, redis_address, cleanup):
       that imported services exits.
   """
   plasma_store_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../plasma/build/plasma_store")
-  store_name = "/tmp/ray_plasma_store{}".format(random.randint(0, 10000))
+  store_name = "/tmp/ray_plasma_store{}".format(random_name())
   p1 = subprocess.Popen([plasma_store_executable, "-s", store_name])
 
   plasma_manager_executable = os.path.join(os.path.abspath(os.path.dirname(__file__)), "../plasma/build/plasma_manager")
-  manager_name = "/tmp/ray_plasma_manager{}".format(random.randint(0, 10000))
-  manager_port = random.randint(10000, 50000)
+  manager_name = "/tmp/ray_plasma_manager{}".format(random_name())
+  manager_port = new_port()
   p2 = subprocess.Popen([plasma_manager_executable,
                          "-s", store_name,
                          "-m", manager_name,

lib/python/ray/worker.py

@@ -32,6 +32,9 @@
 def random_object_id():
   return photon.ObjectID("".join([chr(random.randint(0, 255)) for _ in range(20)]))
 
+def random_string():
+  return "".join([chr(random.randint(0, 255)) for _ in range(20)])
+
 class FunctionID(object):
   def __init__(self, function_id):
     self.function_id = function_id
@@ -516,7 +519,7 @@ def run_function_on_all_workers(self, function):
     if self.mode is None:
       self.cached_functions_to_run.append(function)
     else:
-      function_to_run_id = np.random.randint(0, 1000)
+      function_to_run_id = random_string()
       key = "FunctionsToRun:{}".format(function_to_run_id)
       self.redis_client.hmset(key, {"function_id": function_to_run_id,
                                     "function": pickling.dumps(function)})
@@ -747,7 +750,7 @@ def fetch_and_process_reusable_variable(key, worker=global_worker):
     # record the traceback and notify the scheduler of the failure.
     traceback_str = format_error_message(traceback.format_exc())
     # Log the error message.
-    error_key = "ReusableVariableImportError:{}".format("".join([chr(random.randint(0, 255)) for _ in range(20)]))
+    error_key = "ReusableVariableImportError:{}".format(random_string())
     worker.redis_client.hmset(error_key, {"name": reusable_variable_name,
                                           "message": traceback_str})
     worker.redis_client.rpush("ErrorKeys", error_key)
@@ -766,7 +769,7 @@ def fetch_and_process_function_to_run(key, worker=global_worker):
     traceback_str = traceback.format_exc()
     # Log the error message.
     name = function.__name__  if "function" in locals() and hasattr(function, "__name__") else ""
-    error_key = "FunctionToRunError:{}".format("".join([chr(random.randint(0, 255)) for _ in range(20)]))
+    error_key = "FunctionToRunError:{}".format(random_string())
     worker.redis_client.hmset(error_key, {"name": name,
                                           "message": traceback_str})
     worker.redis_client.rpush("ErrorKeys", error_key)
@@ -832,7 +835,7 @@ def connect(address_info, mode=WORKER_MODE, worker=global_worker):
     mode: The mode of the worker. One of SCRIPT_MODE, WORKER_MODE, PYTHON_MODE,
       and SILENT_MODE.
   """
-  worker.worker_id = np.random.randint(0, 1000000)
+  worker.worker_id = random_string()
   worker.connected = True
   worker.set_mode(mode)
   # If running Ray in PYTHON_MODE, there is no need to create call create_worker
@@ -1069,7 +1072,7 @@ def process_task(task): # wrapping these lines in a function should cause the lo
       failure_objects = [failure_object for _ in range(len(return_object_ids))]
       store_outputs_in_objstore(return_object_ids, failure_objects, worker)
       # Log the error message.
-      error_key = "TaskError:{}".format("".join([chr(random.randint(0, 255)) for _ in range(20)]))
+      error_key = "TaskError:{}".format(random_string())
       worker.redis_client.hmset(error_key, {"function_id": function_id.id(),
                                             "function_name": function_name,
                                             "message": traceback_str})
@@ -1082,7 +1085,7 @@ def process_task(task): # wrapping these lines in a function should cause the lo
       # The attempt to reinitialize the reusable variables threw an exception.
       # We record the traceback and notify the scheduler.
       traceback_str = format_error_message(traceback.format_exc())
-      error_key = "ReusableVariableReinitializeError:{}".format("".join([chr(random.randint(0, 255)) for _ in range(20)]))
+      error_key = "ReusableVariableReinitializeError:{}".format(random_string())
       worker.redis_client.hmset(error_key, {"task_instance_id": "NOTIMPLEMENTED",
                                             "task_id": "NOTIMPLEMENTED",
                                             "function_id": function_id.id(),