Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use flatbuffers for some messages from Redis. #341

Merged
merged 14 commits into from
Mar 11, 2017
Merged

Use flatbuffers for some messages from Redis. #341

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
14 commits
Select commit Hold shift + click to select a range
c5703d2
Compile the Ray redis module with C++.
robertnishihara Mar 5, 2017
03bfe3c
Redo parsing of object table notifications with flatbuffers.
robertnishihara Mar 5, 2017
382f501
Update redis module python tests.
robertnishihara Mar 6, 2017
24ffe3b
Redo parsing of task table notifications with flatbuffers.
robertnishihara Mar 6, 2017
105e2ce
Fix linting.
robertnishihara Mar 7, 2017
5c7b1a7
Redo parsing of db client notifications with flatbuffers.
robertnishihara Mar 7, 2017
949779e
Redo publishing of local scheduler heartbeats with flatbuffers.
robertnishihara Mar 7, 2017
8e056bd
Fix linting.
robertnishihara Mar 7, 2017
a99a0ed
Remove usage of fixed-width formatting of scheduling state in channel…
robertnishihara Mar 8, 2017
43a8982
Reply with flatbuffer object to task table queries, also simplify red…
robertnishihara Mar 8, 2017
754b69f
Fix linting and tests.
robertnishihara Mar 9, 2017
167d2f2
fix
pcmoritz Mar 9, 2017
e7f3a14
cleanup
pcmoritz Mar 9, 2017
bc3761d
simplify logic in ReplyWithTask
pcmoritz Mar 9, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Fix linting and tests.
  • Loading branch information
@robertnishihara
robertnishihara committed Mar 9, 2017
commit 754b69f5bf8f68224a965215bd0518fa74cb53e8
18 changes: 9 additions & 9 deletions src/common/state/redis.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -862,17 +862,17 @@ void redis_task_table_subscribe(TableCallbackData *callback_data) {
if (IS_NIL_ID(data->local_scheduler_id)) {
/* TODO(swang): Implement the state_filter by translating the bitmask into
* a Redis key-matching pattern. */
status = redisAsyncCommand(
db->sub_context, redis_task_table_subscribe_callback,
(void *) callback_data->timer_id, "PSUBSCRIBE %s*:%d",
TASK_CHANNEL_PREFIX, data->state_filter);
status =
redisAsyncCommand(db->sub_context, redis_task_table_subscribe_callback,
(void *) callback_data->timer_id, "PSUBSCRIBE %s*:%d",
TASK_CHANNEL_PREFIX, data->state_filter);
} else {
DBClientID local_scheduler_id = data->local_scheduler_id;
status = redisAsyncCommand(
db->sub_context, redis_task_table_subscribe_callback,
(void *) callback_data->timer_id, "SUBSCRIBE %s%b:%d",
TASK_CHANNEL_PREFIX, (char *) local_scheduler_id.id,
sizeof(local_scheduler_id.id), data->state_filter);
status =
redisAsyncCommand(db->sub_context, redis_task_table_subscribe_callback,
(void *) callback_data->timer_id, "SUBSCRIBE %s%b:%d",
TASK_CHANNEL_PREFIX, (char *) local_scheduler_id.id,
sizeof(local_scheduler_id.id), data->state_filter);
}
if ((status == REDIS_ERR) || db->sub_context->err) {
LOG_REDIS_DEBUG(db->sub_context, "error in redis_task_table_subscribe");
Expand Down
13 changes: 10 additions & 3 deletions test/stress_tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,9 @@
import time
import redis

# Import flatbuffer bindings.
from ray.core.generated.TaskReply import TaskReply

class TaskTests(unittest.TestCase):

def testSubmittingTasks(self):
Expand Down Expand Up @@ -164,9 +167,13 @@ def tearDown(self):
r = redis.StrictRedis(port=self.redis_port)
task_ids = r.keys("TT:*")
task_ids = [task_id[3:] for task_id in task_ids]
node_ids = [r.execute_command("ray.task_table_get", task_id)[1] for task_id
in task_ids]
self.assertEqual(len(set(node_ids)), self.num_local_schedulers)
local_scheduler_ids = []
for task_id in task_ids:
message = r.execute_command("ray.task_table_get", task_id)
task_reply_object = TaskReply.GetRootAsTaskReply(message, 0)
local_scheduler_ids.append(task_reply_object.LocalSchedulerId())

self.assertEqual(len(set(local_scheduler_ids)), self.num_local_schedulers)

# Clean up the Ray cluster.
ray.worker.cleanup()
Expand Down