[Bug][Failing Test]: Distributed Comm Ops - distributed/test_shm_broadcast.py

[robertgshaw2-redhat]

Your current environment

pytest -v -x distributed/test_shm_broadcast.py

https://buildkite.com/vllm/ci/builds/20415#0196f100-f85c-4db6-8b50-72d3d5ade137/197-990

🐛 Describe the bug

See above

Before submitting a new issue...

• Make sure you already searched for relevant issues, and asked the chatbot living at the bottom right corner of the documentation page, which can answer lots of frequently asked questions.

[DarkLight1337]

This test has been flaky for a while: https://buildkite.com/organizations/vllm/analytics/suites/ci-1/tests/1ab0ec52-e9c3-8353-86fe-0a6fe0da32ff?period=28days

[markmc]

Test failure:

FAILED distributed/test_shm_broadcast.py::test_shm_broadcast - AssertionError: assert 1 == 0

Logs:

Rank 2 got seed 752
Rank 0 got seed 752
Rank 1 got seed 752
StatelessProcessGroup passed the test!
StatelessProcessGroup passed the test!
StatelessProcessGroup passed the test!
[rank1]:[W520 21:12:22.801158701 TCPStore.cpp:115] [c10d] recvVector failed on SocketImpl(fd=24, addr=[ip-172-16-0-2.us-west-2.compute.internal]:32772, remote=[ip-172-16-0-2.us-west-2.compute.internal]:49587): Connection reset by peer
Exception raised from recvBytes at /pytorch/torch/csrc/distributed/c10d/Utils.hpp:675 (most recent call first):
frame #0: c10::Error::Error(c10::SourceLocation, std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> >) + 0x98 (0x7f06545785e8 in /usr/local/lib/python3.12/dist-packages/torch/lib/libc10.so)
frame #1: <unknown function> + 0x5ba8afe (0x7f063863cafe in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_cpu.so)
frame #2: <unknown function> + 0x5baa15f (0x7f063863e15f in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_cpu.so)
frame #3: <unknown function> + 0x5baa81d (0x7f063863e81d in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_cpu.so)
frame #4: <unknown function> + 0x5bab4a9 (0x7f063863f4a9 in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_cpu.so)
frame #5: c10d::TCPStore::doGet(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) + 0xbd (0x7f0638639f2d in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_cpu.so)
frame #6: c10d::TCPStore::get(std::__cxx11::basic_string<char, std::char_traits<char>, std::allocator<char> > const&) + 0xab (0x7f063863af8b in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_cpu.so)
frame #7: <unknown function> + 0xc0f526 (0x7f064796d526 in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_python.so)
frame #8: <unknown function> + 0x37f17d (0x7f06470dd17d in /usr/local/lib/python3.12/dist-packages/torch/lib/libtorch_python.so)
frame #9: /usr/bin/python3() [0x56ad3d]
frame #10: _PyObject_MakeTpCall + 0x2fb (0x53d40b in /usr/bin/python3)
frame #11: _PyEval_EvalFrameDefault + 0x700 (0x548140 in /usr/bin/python3)
frame #12: _PyObject_FastCallDictTstate + 0x1d8 (0x53fca8 in /usr/bin/python3)
frame #13: /usr/bin/python3() [0x57bf2a]
frame #14: _PyObject_MakeTpCall + 0x2bb (0x53d3cb in /usr/bin/python3)
frame #15: _PyEval_EvalFrameDefault + 0x700 (0x548140 in /usr/bin/python3)
frame #16: _PyObject_FastCallDictTstate + 0x256 (0x53fd26 in /usr/bin/python3)
frame #17: _PyObject_Call_Prepend + 0x59 (0x57c3e9 in /usr/bin/python3)
frame #18: /usr/bin/python3() [0x6673bd]
frame #19: _PyObject_MakeTpCall + 0x2fb (0x53d40b in /usr/bin/python3)
frame #20: _PyEval_EvalFrameDefault + 0x700 (0x548140 in /usr/bin/python3)
frame #21: _PyObject_FastCallDictTstate + 0x256 (0x53fd26 in /usr/bin/python3)
frame #22: _PyObject_Call_Prepend + 0x59 (0x57c3e9 in /usr/bin/python3)
frame #23: /usr/bin/python3() [0x6673bd]
frame #24: _PyObject_Call + 0x93 (0x57f023 in /usr/bin/python3)
frame #25: _PyEval_EvalFrameDefault + 0x4f52 (0x54c992 in /usr/bin/python3)
frame #26: _PyObject_FastCallDictTstate + 0x256 (0x53fd26 in /usr/bin/python3)
frame #27: _PyObject_Call_Prepend + 0x59 (0x57c3e9 in /usr/bin/python3)
frame #28: /usr/bin/python3() [0x6673bd]
frame #29: _PyObject_MakeTpCall + 0x2fb (0x53d40b in /usr/bin/python3)
frame #30: _PyEval_EvalFrameDefault + 0x700 (0x548140 in /usr/bin/python3)
frame #31: _PyObject_FastCallDictTstate + 0x256 (0x53fd26 in /usr/bin/python3)
frame #32: _PyObject_Call_Prepend + 0x59 (0x57c3e9 in /usr/bin/python3)
frame #33: /usr/bin/python3() [0x6673bd]
frame #34: _PyObject_MakeTpCall + 0x2fb (0x53d40b in /usr/bin/python3)
frame #35: _PyEval_EvalFrameDefault + 0x700 (0x548140 in /usr/bin/python3)
frame #36: _PyObject_FastCallDictTstate + 0x256 (0x53fd26 in /usr/bin/python3)
frame #37: _PyObject_Call_Prepend + 0x59 (0x57c3e9 in /usr/bin/python3)
frame #38: /usr/bin/python3() [0x6673bd]
frame #39: _PyObject_MakeTpCall + 0x2fb (0x53d40b in /usr/bin/python3)
frame #40: _PyEval_EvalFrameDefault + 0x700 (0x548140 in /usr/bin/python3)
frame #41: PyEval_EvalCode + 0x99 (0x61d5b9 in /usr/bin/python3)
frame #42: /usr/bin/python3() [0x6591db]
frame #43: /usr/bin/python3() [0x654346]
frame #44: /usr/bin/python3() [0x651295]
frame #45: _PyRun_SimpleFileObject + 0x1a5 (0x650c85 in /usr/bin/python3)
frame #46: _PyRun_AnyFileObject + 0x47 (0x650a87 in /usr/bin/python3)
frame #47: Py_RunMain + 0x375 (0x64d4e5 in /usr/bin/python3)
frame #48: Py_BytesMain + 0x2d (0x6064ad in /usr/bin/python3)
frame #49: <unknown function> + 0x29d90 (0x7f06d5bc7d90 in /lib/x86_64-linux-gnu/libc.so.6)
frame #50: __libc_start_main + 0x80 (0x7f06d5bc7e40 in /lib/x86_64-linux-gnu/libc.so.6)
frame #51: _start + 0x25 (0x606325 in /usr/bin/python3)

Process Process-2:
Traceback (most recent call last):
  File "/usr/lib/python3.12/multiprocessing/process.py", line 314, in _bootstrap
    self.run()
  File "/usr/lib/python3.12/multiprocessing/process.py", line 108, in run
    self._target(*self._args, **self._kwargs)
  File "/vllm-workspace/tests/distributed/test_shm_broadcast.py", line 52, in wrapped_fn
    fn()
  File "/vllm-workspace/tests/distributed/test_shm_broadcast.py", line 112, in worker_fn
    pg.barrier()
  File "/usr/local/lib/python3.12/dist-packages/vllm/distributed/utils.py", line 218, in barrier
    self.broadcast_obj(None, src=i)
  File "/usr/local/lib/python3.12/dist-packages/vllm/distributed/utils.py", line 199, in broadcast_obj
    recv_obj = pickle.loads(self.store.get(key))
                            ^^^^^^^^^^^^^^^^^^^
torch.distributed.DistNetworkError: Connection reset by peer
FAILED

=================================== FAILURES ===================================
______________________________ test_shm_broadcast ______________________________

    def test_shm_broadcast():
>       distributed_run(worker_fn, 4)

distributed/test_shm_broadcast.py:117: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

fn = <function worker_fn_wrapper.<locals>.wrapped_fn at 0x7f04831bbf60>
world_size = 4

    def distributed_run(fn, world_size):
        number_of_processes = world_size
        processes = []
        for i in range(number_of_processes):
            env = {}
            env['RANK'] = str(i)
            env['LOCAL_RANK'] = str(i)
            env['WORLD_SIZE'] = str(number_of_processes)
            env['LOCAL_WORLD_SIZE'] = str(number_of_processes)
            env['MASTER_ADDR'] = 'localhost'
            env['MASTER_PORT'] = '12345'
            p = multiprocessing.Process(target=fn, args=(env, ))
            processes.append(p)
            p.start()
    
        for p in processes:
            p.join()
    
        for p in processes:
>           assert p.exitcode == 0
E           AssertionError: assert 1 == 0
E            +  where 1 = <Process name='Process-2' pid=34731 parent=19 stopped exitcode=1>.exitcode

distributed/test_shm_broadcast.py:42: AssertionError
=============================== warnings summary ===============================
../../usr/local/lib/python3.12/dist-packages/schemathesis/generation/coverage.py:305
  /usr/local/lib/python3.12/dist-packages/schemathesis/generation/coverage.py:305: DeprecationWarning: jsonschema.exceptions.RefResolutionError is deprecated as of version 4.18.0. If you wish to catch potential reference resolution errors, directly catch referencing.exceptions.Unresolvable.
    ref_error: type[Exception] = jsonschema.RefResolutionError,

tests/distributed/test_shm_broadcast.py::test_shm_broadcast
tests/distributed/test_shm_broadcast.py::test_shm_broadcast
tests/distributed/test_shm_broadcast.py::test_shm_broadcast
tests/distributed/test_shm_broadcast.py::test_shm_broadcast
  /usr/lib/python3.12/multiprocessing/popen_fork.py:66: DeprecationWarning: This process (pid=19) is multi-threaded, use of fork() may lead to deadlocks in the child.
    self.pid = os.fork()

-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html
=========================== short test summary info ============================
FAILED distributed/test_shm_broadcast.py::test_shm_broadcast - AssertionError: assert 1 == 0
 +  where 1 = <Process name='Process-2' pid=34731 parent=19 stopped exitcode=1>.exitcode
======================== 1 failed, 5 warnings in 15.52s ========================
^^^ +++
🚨 Error: The command exited with status 1
^^^ +++
user command error: The plugin docker command hook exited with status 1
~~~ Running global pre-exit hook
$ /etc/buildkite-agent/hooks/pre-exit
~~~ Running plugin docker pre-exit hook
$ /var/lib/buildkite-agent/plugins/bk-gpu-4-queue-ci-i-041889359d2c9cb88-1/github-com-buildkite-plugins-docker-buildkite-plugin-v5-2-0/hooks/pre-exit

[russellb]

I'll take a look at this one. Last time I was looking at this, it seemed kind of racy. It has some sleeps in it, which are always a red flag ...

[russellb]

I believe I have identified the root cause. The implementation of barrier() in StatelessProcessGroup is not sufficient. It is allowing the process running the server side of the TCPStore to exit before all others are finished using it. I've got a fix in progress ....

vllm/vllm/distributed/utils.py

Lines 215 to 218 in 94d8ec8

	def barrier(self):
	"""A barrier to synchronize all ranks."""
	for i in range(self.world_size):
	self.broadcast_obj(None, src=i)

[russellb]

proposed fix here: #18506