Warn on resource deadlock; improve object store error messages

[ericl]

Why are these changes needed?

• Improve several object store error messages
• Warn if the node is full of actors and hence pending tasks or actors cannot be executed ("resource deadlock")

The detection of resource deadlock is somewhat subtle. We periodically check if there are no active running tasks. If this is the case, and there also happen to be tasks queued on this node in READY state, then it means that those tasks are likely to be queued indefinitely since the resources on the local node must be occupied by lifetime-resource actors.

There might be some false positives, e.g., if the actors are going to be destroyed soon, or the cluster is scaling up, so also change the messages from ERROR to WARNING level.

Related issue number

Closes #5468

Linter

• I've run scripts/format.sh to lint the changes in this PR.

[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/16582/
Test PASSed.

[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/16581/
Test PASSed.

[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/16653/
Test PASSed.

[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/16654/
Test PASSed.

[stephanie-wang]

Let's call this in DispatchTasks instead, and only call it if we weren't able to dispatch any tasks. Otherwise, we'll end up pushing an error every heartbeat for as long as the deadlock is happening (which is probably forever).

[ericl]

I tried doing this, but it ended up printing out too many false positives. The issue is that you only want to fire the warning if there has been a significant delay, and right after DispatchTasks is not it (though if you wait even a tiny bit of time resources could immediately free up, like if a task returns right after creating an actor).

I think printing forever is fine -- it is deadlocked after all.

[stephanie-wang]

Hmm, maybe I'm misunderstanding how DispatchTasks works, but if DispatchTasks doesn't succeed in scheduling anything, and the conditions that you check are true (no running tasks), then doesn't that mean it'll never succeed again? I think when this happens, it can only be because there are no available workers or if all the cores are taken up by actors. We can make sure it's the second case if we also check that there are no resources available.

[ericl]

Not necessarily, a block/unblock could free up resources, here is the example I was testing:

import ray

@ray.remote(num_cpus=1)
class A:
    def f(self): pass


@ray.remote
def f():
    a = A.remote()
    b = A.remote()
    c = A.remote()
    print("get 1")
    ray.get(a.f.remote())
    print("get 2")
    ray.get(b.f.remote())
    print("get 3")
    ray.get(c.f.remote())

ray.init(num_cpus=2)
ray.get(f.remote())

[stephanie-wang]

Maybe add an additional check that there are no resources available on the local node. The RUNNING queue can also be empty if there are no workers available (because they haven't started yet or are all died).

[ericl]

I thought about that but it could be the actor is not placeable due to its size of resources. We could try checking for that too, but it seems simpler to rely on the periodic delay to avoid the warning firing too often.

[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/16668/
Test PASSed.