mark all remaining placeable tasks pending with task dependency manager

[atumanov]

What do these changes do?

This PR registers all remaining placeable tasks with the task dependency manager, to ensure reconstruction policy is not triggered on them.

Related issue number

#2527

[stephanie-wang]

Thanks! Do we expect there to be many tasks in the placeable queue? Just wondering if there is some way we can avoid iterating through the queue every time we call ScheduleTasks().

[robertnishihara]

Can you add a note here saying that it's ok to call TaskPending multiple times on the same task?

[robertnishihara]

Maybe the right time to do this is when we put something into the Placeable queue?

[atumanov]

hmm, good point. Other options I can think of are :

1. mark all new tasks pending with the task dependency manager on SubmitTask (so there's a tradeoff between the overhead of that and iterating over remaining placeable queue).
2. piggyback the policy, which already iterates over the queue. (This is undesirable, because it breaks the abstraction, as we'd need the policy to have access to /handle on the task dependency manager).
3. since it's a fifo queue , we can keep a pointer to the last placeable task that was registered. This could be kept as an extra private field in the SchedulingQueue class.

Depending on the policy, this queue may be large.

[atumanov]

@robertnishihara , I think it might be worth trying both and comparing performance. I don't have a good feel (yet) for the overhead of task_dependency_manager_.TaskPending().

[stephanie-wang]

The overhead of calling task_dependency_manager_.TaskPending() is in allocating a timer and a GCS operation to acquire the task lease. @atumanov, we originally used option 1, but it ended up being very inefficient because many of the tasks got forwarded to another node immediately afterward. But if this isn't happening too often anymore with #2420, then maybe option 1 is now okay.

[robertnishihara]

@stephanie-wang there's no additional timer if we call the method twice on the same task, right?

[stephanie-wang]

@robertnishihara, that's right, the method will just exit if it gets called twice on the same task.

The main concern is about calling task_dependency_manager_.TaskPending on a task that will get task_dependency_manager_.TaskCanceled (e.g., due to getting forwarded to another node) soon. Calling task_dependency_manager_.TaskPending on the same task twice is probably okay.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/7069/
Test PASSed.

[AmplabJenkins]

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/7072/
Test FAILed.

[stephanie-wang]

Okay, I think we should just merge this to unblock #2420. The iteration through the placeable queue is probably okay, since we're doing it anyway in ScheduleTasks(). We can come back to it if it becomes an issue.

[stephanie-wang]

jenkins, retest this please

[robertnishihara]

@atumanov we can merge it in its current state, but please add a comment describing the potential performance problem.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/7104/
Test PASSed.

[atumanov]

Added the TODO, this should be ready to merge when tests pass.

[AmplabJenkins]

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/Ray-PRB/7157/
Test PASSed.

[robertnishihara]

To summarize some points from a conversation that @atumanov and I had in person, the main confusion I have with this PR is that the developer has to have a lot of knowledge to determine that all tasks are correctly registered with the task_dependency_manager_. It would be conceptually much simpler if we could maintain an invariant like "a task is registered with the task_dependency_manager_ if and only if it is in local_queues_. However, there is a performance issue in this case when we queue something and then immediately forward it to another machine.

I'm merging this to unblock the scheduling PR, but we should redesign this to reduce the cognitive load.