Global scheduler skeleton #45

robertnishihara · 2016-11-17T08:04:41Z

Things done in this PR:

Enable subscribing to the "local scheduler" table.
Bare bones scheduler that subscribes to the local scheduler table and the task table, so it receives updates about all tasks that are scheduled and all new local schedulers that connect to Redis. When it receives a task, it immediately assigns it to a local scheduler (if no local schedulers have connected yet, we currently just fail).
Allow the local scheduler to start without connecting to Redis, in which case everything is done locally. If a Redis address is provided, then the local scheduler subscribes to updates to the task table that are assigned to it.

Things that can be done in a subsequent PR:

For calls to PUBLISH, check that the number of clients that received the message is what we expect
Replace concept of node_id with client_id or perhaps db_client_id.
Subscribe to updates from the object table.
Create some state managed by the global scheduling algorithm.
Allow subscriptions to the various tables get everything that has already been placed in those tables.

Notes:

This PR makes the microbenchmark of just submitting as many tasks as quickly as possible slower (20us to 30us on my laptop (using 1 worker)). This may be because the load is increased on the local scheduler because most tasks get submitted to the local scheduler once by the driver and then again by the global scheduler. This could potentially be solved by more clever local/global scheduling algorithms.
-We should have an option for always forwarding tasks to the global scheduler for benchmarking purposes.

…that we can test it without a global scheduler.

…ves a task.

… task came from the global scheduler or not.

stephanie-wang · 2016-11-18T21:56:24Z

src/global_scheduler/global_scheduler.c

+ task *original_task,
+ node_id node_id) {
+ task *updated_task =
+ alloc_task(task_task_spec(original_task), TASK_STATUS_SCHEDULED, node_id);


Do we need to allocate a new task here? I think we agreed earlier that the scheduling_state and node_id fields of a task instance should be mutable.

Good point.

stephanie-wang · 2016-11-18T21:58:57Z

src/global_scheduler/global_scheduler.c

+ db_client_table_subscribe(g_state->db, process_new_db_client,
+ (void *) g_state, &retry, NULL, NULL);
+ /* Subscribe to notifications about waiting tasks. */
+ task_table_subscribe(g_state->db, NIL_ID, TASK_STATUS_WAITING,


Should we also do the initial read of all the tasks that are in state TASK_STATUS_WAITING? Or at least, a TODO to come back to this.

I'll add a TODO.

stephanie-wang · 2016-11-18T22:00:21Z

src/global_scheduler/global_scheduler_algorithm.h

+ *
+ */
+
+void handle_task_waiting(global_scheduler_state *state, task *original_task);


Documentation for these methods.

stephanie-wang · 2016-11-18T22:00:45Z

src/global_scheduler/global_scheduler_algorithm.c

+
+#include "global_scheduler_algorithm.h"
+
+void handle_task_waiting(global_scheduler_state *state, task *original_task) {


I'm assuming that this code is throwaway, but can we document what the current algorithm is anyway?

stephanie-wang · 2016-11-18T22:02:19Z

src/global_scheduler/test/test.py

+ task_contents = self.redis_client.hgetall(task_entries[0])
+ task_status = int(task_contents["state"])
+ self.assertTrue(task_status in [TASK_STATUS_WAITING, TASK_STATUS_SCHEDULED])
+ if task_status == TASK_STATUS_SCHEDULED:


Add a check to make sure that the task is scheduled on the right node?

stephanie-wang · 2016-11-18T22:02:36Z

src/global_scheduler/test/test.py

+ task_contents = [self.redis_client.hgetall(task_entries[i]) for i in range(len(task_entries))]
+ task_statuses = [int(contents["state"]) for contents in task_contents]
+ self.assertTrue(all([status in [TASK_STATUS_WAITING, TASK_STATUS_SCHEDULED] for status in task_statuses]))
+ if all([status == TASK_STATUS_SCHEDULED for status in task_statuses]):


Same check here.

mehrdadn

Dammit, just learned I need to click "Submit Review" for anyone to see these now :( I hate GitHub's new interface...

Anyway, most of the comments were superficial and not really important... but a few of them really should be addressed I think (especially the signal handling one) so please do take a look at all of them...

mehrdadn · 2016-11-18T10:03:22Z

src/common/logging.c

- event_type, message, utstring_body(timestamp));
+ /* Fill in the client ID and send the message to Redis. */
+ redisAsyncCommand(db->context, NULL, NULL, utstring_body(formatted_message),
+ (char *) db->client.id, sizeof(db_client_id));


Do these calls not have return values? If they do can you check them?

Good point, I'll fix that.

mehrdadn · 2016-11-18T10:15:39Z

src/common/state/redis.c

 CHECK(reply->element[j]->type == REDIS_REPLY_STRING);
- managers[j] = atoi(reply->element[j]->str);
- redis_get_cached_service(db, managers[j], manager_vector + j);
+ memcpy(managers[j].id, reply->element[j]->str, sizeof(db_client_id));


You're only copying 1 field of the struct, but you're taking the size of the whole struct. There is no guarantee the two will match in size, at least if you happen to add a field later (though I'm not even sure if you can assume a lack of padding as you are here).

Even if the above wasn't the case, you should always use sizeof(field) instead of sizeof(type) when you can. That way when you change the field type later, the code won't break.

So this should be changed to memcpy(managers[j].id, reply->element[j]->str, sizeof(managers[j].id));
Perhaps worthy of a macro at some point, though not necessarily now.

Actually, scratch that. Don't use memcpy if you can avoid it. Just do something like
managers[j] = *(db_client_id const *)reply->element[j]->str
and let the compiler take care of the assignment.

I think I've tried stuff like that and remember getting errors like this.

state/redis.c:372:22: error: array type 'unsigned char [20]' is not assignable managers[j].id = *(db_client_id const *) reply->element[j]->str; ~~~~~~~~~~~~~~ ^

Oh, that's because you were assigning to id instead of the whole struct.
Just assign to the whole struct.

mehrdadn · 2016-11-18T10:20:39Z

src/common/state/redis.c

+void redis_db_client_table_subscribe_callback(redisAsyncContext *c,
+ void *r,
+ void *privdata) {
+ REDIS_CALLBACK_HEADER(db, callback_data, r)


Put semicolons after macros that are statements (and ideally, write the macros such that these are required).
It makes it clear that these macros are statements and not declarations of some sort.
(Indeed some declarations also require semicolons, but not all of them do. However, all statements do.)
Bonus points: Some editors/IDEs choke on their indentations when you don't do this.

mehrdadn · 2016-11-18T10:23:32Z

src/common/state/redis.c

+ /* Otherwise, parse the payload and call the callback. */
+ db_client_table_subscribe_data *data = callback_data->data;
+ db_client_id client;
+ memcpy(client.id, payload->str, sizeof(db_client_id));


Again, same thing here as before. Fix this everywhere.

mehrdadn · 2016-11-18T10:27:13Z

src/common/state/redis.h

- * Should only be used very rarely, it is not asynchronous. */
+ /** Cache for the IP addresses of db clients. This is a hash table mapping
+ * client IDs to addresses. */
+ db_client_cache_entry *db_client_cache;


I'm confused... if this is a hashtable then why is it a pointer to an entry? Am I misreading it?

If this is just how you're representing it, use a typedef to abstract it away properly. This should point to a db_client_cache or something. It doesn't make sense as it is written now, and it makes the code more brittle.

That's the way the uthash library is used. https://troydhanson.github.io/uthash/userguide.html

Oh I see :\ okay if it's idiomatic use of the library then never mind.

mehrdadn · 2016-11-18T10:45:16Z

src/photon/photon_algorithm.c

+ /* Copy the spec and add it to the task queue. The allocated spec will be
+ * freed when it is assigned to a worker. */
+ task_queue_entry *elt = malloc(sizeof(task_queue_entry));
+ elt->spec = malloc(task_spec_size(spec));


Why not cast the return value of malloc but cast the return value of others like utarray_back?
Be consistent. I'd say cast all of them so people can compile the code with a C++ compiler too.

(not terribly important given this is already merged, but worth keeping in mind for new code)

mehrdadn · 2016-11-18T10:47:08Z

src/photon/photon_algorithm.c

+ DCHECK(!from_global_scheduler);
+ task *task = alloc_task(spec, TASK_STATUS_WAITING, NIL_ID);
+ DCHECK(info->db != NULL);
+ task_table_add_task(info->db, task, (retry_info *) &photon_retry, NULL, NULL);


Is this mutating photon_retry? Or is casting away const only to make the pointer types match...?

Shouldn't mutate anything, probably just get to get rid of a compiler warning.

mehrdadn · 2016-11-18T10:47:54Z

src/photon/photon_algorithm.c

+ /* If this task's dependencies are available locally, and if there is an
+ * available worker, then assign this task to an available worker. If we
+ * cannot assign the task to a worker immediately, queue the task locally. */
+ if ((utarray_len(s->available_workers) > 0) && can_run(s, spec)) {


Remove extra parentheses; they make code harder to read.

I often include them to save myself time when debugging. E.g., if there is a bug, I'll start to wonder if I got the order of operations wrong, and so I'll go and add in the parentheses just to be sure..

Haha, okay fair enough.

mehrdadn · 2016-11-18T10:48:42Z

src/photon/photon_scheduler.c

+ /* Update the global task table. */
+ if (info->db != NULL) {
+ retry_info retry = {
+ .num_retries = 0, .timeout = 100, .fail_callback = NULL,


Again, avoid designated initializers. (I'll stop commenting but there are more.)

mehrdadn · 2016-11-18T10:50:39Z

src/photon/photon_scheduler.c

+ char redis_addr[16] = {0};
+ char redis_port[6] = {0};
+ if (sscanf(redis_addr_port, "%15[0-9.]:%5[0-9]", redis_addr, redis_port) !=
+ 2) {


This was painful to read. Maybe store the output of sscanf in a variable nassigned and then test nassigned != 2 on a single line.

robertnishihara force-pushed the scheduler branch 5 times, most recently from 7b68d8d to 2e2c6c1 Compare November 17, 2016 23:49

pcmoritz and others added 11 commits November 17, 2016 22:36

Initial scheduler commit

6b2bb25

global scheduler

9a8818f

add global scheduler

6e89496

Implement global scheduler skeleton.

4ffcd81

Formatting.

c105692

Allow local scheduler to be started without a connection to redis so …

dc5d529

…that we can test it without a global scheduler.

Fail if there are no local schedulers when the global scheduler recei…

a74f20a

…ves a task.

Initialize uninitialized value and formatting fix.

49c41d6

Generalize local scheduler table to db client table.

d6399f6

Remove code duplication in local scheduler and add flag for whether a…

0f82311

… task came from the global scheduler or not.

Queue task specs in the local scheduler instead of tasks.

cab197b

robertnishihara force-pushed the scheduler branch 6 times, most recently from 0802d56 to f476e14 Compare November 18, 2016 08:25

Simple global scheduler tests, including valgrind.

49e76a5

robertnishihara force-pushed the scheduler branch from f476e14 to 49e76a5 Compare November 18, 2016 08:34

stephanie-wang reviewed Nov 18, 2016

View reviewed changes

robertnishihara force-pushed the scheduler branch 2 times, most recently from d35f0c6 to 04b5b79 Compare November 19, 2016 00:46

Factor out functions for starting processes.

f63a549

robertnishihara force-pushed the scheduler branch from 04b5b79 to f63a549 Compare November 19, 2016 02:06

Fixes.

6ea0905

robertnishihara force-pushed the scheduler branch from 2f7db53 to 6ea0905 Compare November 19, 2016 03:27

stephanie-wang merged commit d77b685 into master Nov 19, 2016

stephanie-wang deleted the scheduler branch November 19, 2016 03:58

mehrdadn reviewed Nov 19, 2016

View reviewed changes

franklsf95 mentioned this pull request Apr 14, 2021

Ray client crashes on connecting #15289

Closed

2 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Global scheduler skeleton #45

Global scheduler skeleton #45

robertnishihara commented Nov 17, 2016 •

edited

Loading

stephanie-wang Nov 18, 2016

robertnishihara Nov 19, 2016

stephanie-wang Nov 18, 2016

robertnishihara Nov 19, 2016

stephanie-wang Nov 18, 2016

stephanie-wang Nov 18, 2016

stephanie-wang Nov 18, 2016

stephanie-wang Nov 18, 2016

mehrdadn left a comment •

edited

Loading

mehrdadn Nov 18, 2016

robertnishihara Nov 19, 2016 •

edited

Loading

mehrdadn Nov 18, 2016

mehrdadn Nov 18, 2016

robertnishihara Nov 19, 2016

mehrdadn Nov 19, 2016

mehrdadn Nov 18, 2016

robertnishihara Nov 19, 2016

mehrdadn Nov 18, 2016

mehrdadn Nov 18, 2016

robertnishihara Nov 19, 2016

mehrdadn Nov 19, 2016

mehrdadn Nov 18, 2016

mehrdadn Nov 19, 2016 •

edited

Loading

mehrdadn Nov 18, 2016

robertnishihara Nov 19, 2016 •

edited

Loading

mehrdadn Nov 18, 2016

robertnishihara Nov 19, 2016

mehrdadn Nov 19, 2016

mehrdadn Nov 18, 2016

mehrdadn Nov 18, 2016


		#include "global_scheduler_algorithm.h"

		void handle_task_waiting(global_scheduler_state state, task original_task) {

Global scheduler skeleton #45

Global scheduler skeleton #45

Conversation

robertnishihara commented Nov 17, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mehrdadn left a comment • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertnishihara Nov 19, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mehrdadn Nov 19, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertnishihara Nov 19, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

robertnishihara commented Nov 17, 2016 •

edited

Loading

mehrdadn left a comment •

edited

Loading

robertnishihara Nov 19, 2016 •

edited

Loading

mehrdadn Nov 19, 2016 •

edited

Loading

robertnishihara Nov 19, 2016 •

edited

Loading