From 28f08823879dd0ca72679104a741e7464f788a9b Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Tue, 16 May 2017 20:06:13 -0700 Subject: [PATCH] Expose function table to python global control state API (#542) * expose function table to python global control state API * fix * fix linting * add test for function table --- python/ray/experimental/state.py | 20 ++++++++++++++++++++ test/runtest.py | 19 +++++++++++++------ 2 files changed, 33 insertions(+), 6 deletions(-) diff --git a/python/ray/experimental/state.py b/python/ray/experimental/state.py index 3de63ebaf..e7c649f28 100644 --- a/python/ray/experimental/state.py +++ b/python/ray/experimental/state.py @@ -20,6 +20,7 @@ OBJECT_INFO_PREFIX = "OI:" OBJECT_LOCATION_PREFIX = "OL:" OBJECT_SUBSCRIBE_PREFIX = "OS:" TASK_PREFIX = "TT:" +FUNCTION_PREFIX = "RemoteFunction:" OBJECT_CHANNEL_PREFIX = "OC:" # This mapping from integer to task state string must be kept up-to-date with @@ -194,6 +195,25 @@ class GlobalState(object): task_id_binary) return results + def function_table(self, function_id=None): + """Fetch and parse the function table. + + Returns: + A dictionary that maps function IDs to information about the function. + """ + self._check_connected() + function_table_keys = self.redis_client.keys(FUNCTION_PREFIX + "*") + results = {} + for key in function_table_keys: + info = self.redis_client.hgetall(key) + function_info_parsed = { + "DriverID": binary_to_hex(info[b"driver_id"]), + "Module": decode(info[b"module"]), + "Name": decode(info[b"name"]) + } + results[binary_to_hex(info[b"function_id"])] = function_info_parsed + return results + def client_table(self): """Fetch and parse the Redis DB client table. diff --git a/test/runtest.py b/test/runtest.py index d785353ea..46178a880 100644 --- a/test/runtest.py +++ b/test/runtest.py @@ -1600,6 +1600,9 @@ class GlobalStateAPI(unittest.TestCase): with self.assertRaises(Exception): ray.global_state.client_table() + with self.assertRaises(Exception): + ray.global_state.function_table() + ray.init() self.assertEqual(ray.global_state.object_table(), dict()) @@ -1654,12 +1657,16 @@ class GlobalStateAPI(unittest.TestCase): if task_table[task_id]["State"] == "DONE": break time.sleep(0.1) - self.assertEqual(task_table[task_id]["TaskSpec"]["ActorID"], - ID_SIZE * "ff") - self.assertEqual(task_table[task_id]["TaskSpec"]["Args"], [1, "hi", x_id]) - self.assertEqual(task_table[task_id]["TaskSpec"]["DriverID"], driver_id) - self.assertEqual(task_table[task_id]["TaskSpec"]["ReturnObjectIDs"], - [result_id]) + function_table = ray.global_state.function_table() + task_spec = task_table[task_id]["TaskSpec"] + self.assertEqual(task_spec["ActorID"], ID_SIZE * "ff") + self.assertEqual(task_spec["Args"], [1, "hi", x_id]) + self.assertEqual(task_spec["DriverID"], driver_id) + self.assertEqual(task_spec["ReturnObjectIDs"], [result_id]) + function_table_entry = function_table[task_spec["FunctionID"]] + self.assertEqual(function_table_entry["Name"], "__main__.f") + self.assertEqual(function_table_entry["DriverID"], driver_id) + self.assertEqual(function_table_entry["Module"], "__main__") self.assertEqual(task_table[task_id], ray.global_state.task_table(task_id))