Same function assigned different IDs on driver and worker

[AdamGleave]

System information

• OS Platform and Distribution (e.g., Linux Ubuntu 16.04): Linux Ubuntu 17.10
• Ray installed from (source or binary): Binary
• Ray version: 0.40
• Python version: Python 3.6.5
• Exact command to reproduce:

Describe the problem

ray/worker.py:compute_function_id checks whether main has the file attribute to decide whether to include the source code in the function hash. When starting a Python executable using python -m mypkg.executable, dependencies in mypkg can be imported before main is loaded, resulting in this check failing (and source code being excluded) in the driver. When it comes to running the code, however, main exists. This can lead to function IDs differing between the driver and worker in nested parallelism settings.

The error I get is similar to that of #1446 so it may have a common cause. After diagnosing this I think I've become convinced that it is best for executables to be in the top-level and not nested in the package structure; I'm posting this mostly as a warning to others as it is far from obvious what caused this error.

I think we can make this more robust by using a try-except statement rather than guessing when the source code is present; I'll submit a PR.

Source code / logs

This bug is only triggered with a particular package structure, so this needs a few files. First, create a directory foo containing:

# __init__.py
from foo import bar

# main.py
import ray                                                                                                                                                                                            

from foo import bar

if __name__ == '__main__':
    ray.init(redirect_worker_output=True)
    bar.run()

# bar.py
import ray                                                                                                                                                                                            

@ray.remote
def f(x):
    return x

def g(x):
    return f.remote(x)

@ray.remote
def h(x):
    return ray.get(g(x))

def run():
    print(ray.get(h.remote(42)))

Running this produces:

python -m foo.main
Process STDOUT and STDERR is being redirected to /tmp/raylogs/.
Waiting for redis server at 127.0.0.1:39255 to respond...
Waiting for redis server at 127.0.0.1:20998 to respond...
Starting local scheduler with the following resources: {'CPU': 12, 'GPU': 2}.

======================================================================
View the web UI at http://localhost:8889/notebooks/ray_ui70409.ipynb?token=d6d894f5459ef32b1986d4cdf348574421d1787c71960fef
======================================================================

Traceback (most recent call last):
  File "/home/adam/bin/anaconda3/envs/mypirl/lib/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/home/adam/bin/anaconda3/envs/mypirl/lib/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/adam/dev/ray/foo/main.py", line 7, in <module>
    bar.run()
  File "/home/adam/dev/ray/foo/bar.py", line 15, in run
    print(ray.get(h.remote(42)))
Remote function foo.bar.h failed with:

Traceback (most recent call last):
  File "/home/adam/dev/ray/foo/bar.py", line 12, in h
    return ray.get(g(x))
  File "/home/adam/dev/ray/foo/bar.py", line 8, in g
    return f.remote(x)
  File "/home/adam/bin/anaconda3/envs/mypirl/lib/python3.6/site-packages/ray/worker.py", line 2602, in func_call
    return _submit(args=args, kwargs=kwargs)
  File "/home/adam/bin/anaconda3/envs/mypirl/lib/python3.6/site-packages/ray/worker.py", line 2622, in _submit
    resources=resources)
  File "/home/adam/bin/anaconda3/envs/mypirl/lib/python3.6/site-packages/ray/worker.py", line 2430, in _submit_task
    return global_worker.submit_task(function_id, *args, **kwargs)
  File "/home/adam/bin/anaconda3/envs/mypirl/lib/python3.6/site-packages/ray/worker.py", line 580, in submit_task
    self.task_driver_id.id()][function_id.id()]
KeyError: b'\x8d\x04\x11\xd9\xea\xd4\xbe\xad\x90\x0c\x0b\x10\x0f^\x0cq\xcc\xe3\xb1\x07'

[robertnishihara]

I can reproduce this. On the current master, the error I get is

22:14 $ python -m foo.main
Process STDOUT and STDERR is being redirected to /tmp/raylogs/.
Waiting for redis server at 127.0.0.1:60793 to respond...
Waiting for redis server at 127.0.0.1:55793 to respond...
Starting local scheduler with the following resources: {'CPU': 8, 'GPU': 0}.

======================================================================
View the web UI at http://localhost:8889/notebooks/ray_ui68917.ipynb?token=0994393722132acb1df5b8f144620fdb0bd8a5ef269d0e05
======================================================================

This worker was asked to execute a function that it does not have registered. You may have to restart Ray.

  You can inspect errors by running

      ray.error_info()

  If this driver is hanging, start a new one with

      ray.init(redis_address="192.168.1.103:60793")

So I think the underlying issue is that compute_function_id is run on the "driver" when the remote function is defined as well as on the workers when bar is imported. Because we're run with python -m, the IDs are computed differently on the driver and on the worker. When the h task is executed on a worker, it submits an f task using the ID computed by the worker, which in this case has never been exported to the workers.

Your fix in #2090 basically works by expanding the set of cases where compute_function_id finds the source code (to include the python -m case and maybe others), and so increases the likelihood of avoiding this issue. However, any time where compute_function_id on the driver doesn't find the source, we could have this issue (e.g., if the driver is an interactive python interpreter).

Does that sound right to you?

I would have expected to be able to still cause the error (even with your fix) by doing

import ray                                                                                                                                                                                            

from foo import bar

ray.init(num_cpus=1)
bar.run()

in python interpreter (but not IPython). However, it seems to run correctly.

[AdamGleave]

Your description sounds correct to me. I think the issue with the Python interpreter is only with functions defined in the interpreter itself, it should still be able to read source code of functions defined in imported modules fine.

I'm not sure what other reasons could cause inspect.getsource to fail on the driver but not the worker. If interactively defined functions in the Python interpreter are the only problem, I think it's probably OK to leave that case broken and document it as a caveat. But if it's more wide-ranging it may be worth making broader changes.