[BugFix] Keep Gloo P2P operations from advancing collective tags

paddle/fluid/distributed/collective/process_group_gloo.cc

@@ -266,10 +266,11 @@ std::shared_ptr<ProcessGroup::Task> ProcessGroupGloo::Send(
     std::vector<DenseTensor>& inputs, int dst_rank) {
   CheckTensorContiguous(inputs);
   std::unique_ptr<SendGlooTask> 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<SendGlooTask>(
-      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<ProcessGroup::Task> ProcessGroupGloo::Recv(DenseTensor* tensor,
 std::shared_ptr<ProcessGroup::Task> ProcessGroupGloo::Recv(
     std::vector<DenseTensor>& outputs, int src_rank) {
   std::unique_ptr<RecvGlooTask> task;
-  auto tag = next_tag();
   auto comm_context = this->GetCommContext();
 
   task = std::make_unique<RecvGlooTask>(
-      comm_context, &outputs, rank_, src_rank, tag);
+      comm_context, &outputs, rank_, src_rank, /*tag=*/0);
   task->Run();
   return task;
 }

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)

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()

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