diff --git a/paddle/fluid/distributed/collective/process_group_gloo.cc b/paddle/fluid/distributed/collective/process_group_gloo.cc index a7037b8a604516..64d7824c9f1f06 100644 --- a/paddle/fluid/distributed/collective/process_group_gloo.cc +++ b/paddle/fluid/distributed/collective/process_group_gloo.cc @@ -266,10 +266,11 @@ std::shared_ptr ProcessGroupGloo::Send( std::vector& inputs, int dst_rank) { CheckTensorContiguous(inputs); std::unique_ptr task; - auto tag = next_tag(); + // Gloo matches point-to-point messages in FIFO order per peer and direction, + // using a separate slot prefix from collectives. auto comm_context = this->GetCommContext(); task = std::make_unique( - comm_context, &inputs, rank_, dst_rank, tag); + comm_context, &inputs, rank_, dst_rank, /*tag=*/0); task->Run(); return task; @@ -311,11 +312,10 @@ std::shared_ptr ProcessGroupGloo::Recv(DenseTensor* tensor, std::shared_ptr ProcessGroupGloo::Recv( std::vector& outputs, int src_rank) { std::unique_ptr task; - auto tag = next_tag(); auto comm_context = this->GetCommContext(); task = std::make_unique( - comm_context, &outputs, rank_, src_rank, tag); + comm_context, &outputs, rank_, src_rank, /*tag=*/0); task->Run(); return task; } diff --git a/test/collective/CMakeLists.txt b/test/collective/CMakeLists.txt index 035838f32893a4..479653d6097d04 100644 --- a/test/collective/CMakeLists.txt +++ b/test/collective/CMakeLists.txt @@ -471,5 +471,20 @@ if((WITH_GPU) AND (LINUX)) "PADDLE_DIST_UT_PORT=22304;http_proxy=;https_proxy=") set_tests_properties(test_comm_group_num PROPERTIES TIMEOUT "120") endif() +if(WITH_DISTRIBUTE) + if(WITH_GLOO) + if(LOCAL_ALL_ARCH AND (LINUX)) + py_test_modules( + test_process_group_gloo_p2p + MODULES + test_process_group_gloo_p2p + ENVS + "PYTHONPATH=..:${PADDLE_BINARY_DIR}/python;http_proxy=;https_proxy=;OMP_NUM_THREADS=1;OPENBLAS_NUM_THREADS=1" + ) + set_tests_properties(test_process_group_gloo_p2p + PROPERTIES TIMEOUT "150" LABELS "RUN_TYPE=DIST") + endif() + endif() +endif() add_subdirectory(fleet) add_subdirectory(multinode) diff --git a/test/collective/test_process_group_gloo_p2p.py b/test/collective/test_process_group_gloo_p2p.py new file mode 100644 index 00000000000000..f015614756261c --- /dev/null +++ b/test/collective/test_process_group_gloo_p2p.py @@ -0,0 +1,118 @@ +# Copyright (c) 2026 PaddlePaddle Authors. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import multiprocessing +import os +import socket +import time +import unittest + +import numpy as np + +import paddle +import paddle.distributed as dist +from paddle.base import core + + +def _run_worker(rank, endpoints, mixed): + os.environ.update( + PADDLE_TRAINER_ID=str(rank), + PADDLE_TRAINERS_NUM=str(len(endpoints)), + PADDLE_CURRENT_ENDPOINT=endpoints[rank], + PADDLE_TRAINER_ENDPOINTS=','.join(endpoints), + PADDLE_DISTRI_BACKEND='gloo', + GLOO_SOCKET_IFNAME='lo', + ) + paddle.set_device('cpu') + dist.init_parallel_env() + + def check_collectives(): + tensor = paddle.to_tensor([rank + 1.0]) + dist.broadcast(tensor, src=0) + np.testing.assert_array_equal(tensor.numpy(), [1.0]) + dist.all_reduce(tensor) + np.testing.assert_array_equal(tensor.numpy(), [4.0]) + + check_collectives() + if mixed: + # Ranks have different peer histories; reversing a pair also exercises + # the independent send and receive queues for that peer. + pairs = [(0, 1), (0, 2), (3, 1), (1, 0), (2, 3), (0, 1)] + for iteration in range(3): + for index, (src, dst) in enumerate(pairs): + expected = np.arange(index + 1, dtype='int64') + iteration * 10 + if rank == src: + dist.send(paddle.to_tensor(expected), dst=dst) + elif rank == dst: + received = paddle.empty([index + 1], dtype='int64') + dist.recv(received, src=src) + np.testing.assert_array_equal(received.numpy(), expected) + check_collectives() + else: + check_collectives() + dist.barrier() + + +@unittest.skipUnless( + hasattr(core, 'ProcessGroupGloo'), 'Paddle is not compiled with Gloo' +) +class TestProcessGroupGlooP2P(unittest.TestCase): + def run_workers(self, mixed): + sockets = [] + processes = [] + try: + for _ in range(4): + sock = socket.socket() + sock.bind(('127.0.0.1', 0)) + sockets.append(sock) + endpoints = [ + f'127.0.0.1:{sock.getsockname()[1]}' for sock in sockets + ] + for sock in sockets: + sock.close() + + ctx = multiprocessing.get_context('spawn') + for rank in range(4): + process = ctx.Process( + target=_run_worker, args=(rank, endpoints, mixed) + ) + process.start() + processes.append(process) + deadline = time.monotonic() + 60 + for process in processes: + process.join(timeout=max(0, deadline - time.monotonic())) + self.assertEqual( + [process.exitcode for process in processes], [0] * 4 + ) + finally: + for sock in sockets: + sock.close() + for process in processes: + if process.is_alive(): + process.terminate() + for process in processes: + process.join(timeout=5) + if process.is_alive(): + process.kill() + process.join() + + def test_collectives(self): + self.run_workers(mixed=False) + + def test_mixed_p2p_collectives(self): + self.run_workers(mixed=True) + + +if __name__ == '__main__': + unittest.main() diff --git a/test/collective/testslist.csv b/test/collective/testslist.csv index a44ee7b3dcb8e9..2bc6bd865c56f6 100644 --- a/test/collective/testslist.csv +++ b/test/collective/testslist.csv @@ -54,3 +54,4 @@ test_strategy_group,linux,rocm;gpu,120,DIST,test_strategy_group.sh,2,,http_proxy test_orthogonal_strategy,linux,rocm;gpu,120,DIST,test_orthogonal_strategy.sh,2,,http_proxy=;https_proxy=, test_new_api_per_op_and_group_intranode,linux,gpu,120,DIST,test_new_api_per_op_and_group_intranode.sh,2,,http_proxy=;https_proxy=, test_comm_group_num,linux,gpu,120,DIST,test_comm_group_num.sh,2,,http_proxy=;https_proxy=, +test_process_group_gloo_p2p,linux,,150,DIST,,,,PYTHONPATH=..;http_proxy=;https_proxy=;OMP_NUM_THREADS=1;OPENBLAS_NUM_THREADS=1,WITH_DISTRIBUTE;WITH_GLOO