# 异步通信实现
class AsyncCommunication:
    def __init__(self):
        self.communication_queue = asyncio.Queue()
        self.computation_tasks = []
    
    async def async_all_reduce(self, tensor, op='sum'):
        """异步All-Reduce操作"""
        # 启动通信任务
        comm_task = asyncio.create_task(self._all_reduce_async(tensor, op))
        
        # 继续其他计算任务
        while not comm_task.done():
            # 处理其他计算任务
            await self._process_other_tasks()
            await asyncio.sleep(0.001)  # 短暂让出控制权
        
        return await comm_task
    
    async def _all_reduce_async(self, tensor, op):
        """异步执行All-Reduce"""
        # 发送数据到其他节点
        send_tasks = []
        for rank in range(self.world_size):
            if rank != self.rank:
                task = asyncio.create_task(
                    self._send_tensor(tensor, rank)
                )
                send_tasks.append(task)
        
        # 接收其他节点的数据
        recv_tasks = []
        for rank in range(self.world_size):
            if rank != self.rank:
                task = asyncio.create_task(
                    self._recv_tensor(rank)
                )
                recv_tasks.append(task)
        
        # 等待所有通信完成
        await asyncio.gather(*send_tasks, *recv_tasks)
        
        # 聚合结果
        return self._aggregate_results(tensor, recv_tasks, op)