[core] Tasks with duplicate args sometimes don't get scheduled

[stephanie-wang]

What is the problem?

Ray version and other system information (Python version, TensorFlow version, OS): 1.4 and earlier

The raylet overestimates the number of missing args for tasks that have duplicate args. This can lead to the task never being scheduled.

Reproduction (REQUIRED)

Please provide a short code snippet (less than 50 lines if possible) that can be copy-pasted to reproduce the issue. The snippet should have no external library dependencies (i.e., use fake or mock data / environments):

If the code snippet cannot be run by itself, the issue will be closed with "needs-repro-script".

def test_many_args(ray_start_cluster):
    # This test ensures that a task will run where its task dependencies are
    # located, even when those objects are borrowed.
    cluster = ray_start_cluster
    object_size = int(1e6)

    # Disable worker caching so worker leases are not reused, and disable
    # inlining of return objects so return objects are always put into Plasma.
    for _ in range(4):
        cluster.add_node(
            num_cpus=1, object_store_memory=(4 * object_size * 25))
    ray.init(address=cluster.address)

    @ray.remote
    def f(i, *args):
        print(i)
        return

    @ray.remote
    def put():
        return np.zeros(object_size, dtype=np.uint8)

    xs = [put.remote() for _ in range(100)]
    ray.wait(xs, num_returns=len(xs), fetch_local=False)
    tasks = []
    for i in range(100):
        args = [np.random.choice(xs) for _ in range(25)]
        tasks.append(f.remote(i, *args))
    ray.get(tasks)

• I have verified my script runs in a clean environment and reproduces the issue.
• I have verified the issue also occurs with the latest wheels.