如何使用asyncio数据报端点实现UDP广播？

我基于asyncio的UDP回显客户端示例构建了一个Wake On LAN（WOL）广播程序（精简了部分细节），但代码始终发送失败，哪怕用非广播IP也不行。把allow_broadcast设为False时代码能正常执行。我用的是Windows 11、Python 3.10。另外我注释了两行代码避免套接字过早关闭（否则会触发其他错误，这个后续再处理）。

出错的代码 #

import asyncio
from typing import Optional

BROADCAST_IP = "255.255.255.255"
DEFAULT_PORT = 9


class _WOLProtocol:
    def __init__(self, *messages):
        self.packets = messages
        self.done = asyncio.get_running_loop().create_future()
        self.transport = None

    def connection_made(self, transport):
        for p in self.packets:
            transport.sendto(p.encode())

        #transport.close()

    def error_received(self, exc):
        self.done.set_exception(exc)

    def connection_lost(self, exc):
        print('closing')
        #self.done.set_result(None)


async def send_magic_packet(
    *macs: str,
    ip_address: str = BROADCAST_IP,
    port: int = DEFAULT_PORT,
    interface: Optional[str] = None
) -> None:
    loop = asyncio.get_running_loop()
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: _WOLProtocol(*macs),
        remote_addr=(ip_address, port),
        allow_broadcast = True,
        local_addr=(interface, 0) if interface else None
        )

    try:
        await protocol.done
    finally:
        transport.close()

if __name__ == "__main__":
    asyncio.run(send_magic_packet('test'))

报错信息 #

Exception in callback _ProactorDatagramTransport._loop_reading()
handle: <Handle _ProactorDatagramTransport._loop_reading()>
Traceback (most recent call last):
  File "C:\Program Files\Python310\lib\asyncio\proactor_events.py", line 570, in _loop_reading
    self._read_fut = self._loop._proactor.recv(self._sock,
  File "C:\Program Files\Python310\lib\asyncio\windows_events.py", line 458, in recv
    ov.WSARecv(conn.fileno(), nbytes, flags)
OSError: [WinError 10022] An invalid argument was supplied

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "C:\Program Files\Python310\lib\asyncio\events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Program Files\Python310\lib\asyncio\proactor_events.py", line 576, in _loop_reading
    self._protocol.error_received(exc)
  File ".....\code.py", line 23, in error_received
    self.done.set_exception(exc)
asyncio.exceptions.InvalidStateError: invalid state
closing
Traceback (most recent call last):
  File ".....\code.py", line 50, in <module>
    asyncio.run(send_magic_packet('test'))
  File "C:\Program Files\Python310\lib\asyncio\runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "C:\Program Files\Python310\lib\asyncio\base_events.py", line 649, in run_until_complete
    return future.result()
  File ".....\code.py", line 45, in send_magic_packet
    await protocol.done
  File "C:\Program Files\Python310\lib\asyncio\proactor_events.py", line 530, in _loop_writing
    self._write_fut = self._loop._proactor.send(self._sock,
  File "C:\Program Files\Python310\lib\asyncio\windows_events.py", line 541, in send
    ov.WSASend(conn.fileno(), buf, flags)
OSError: [WinError 10057] A request to send or receive data was disallowed because the socket is not connected and (when sending on a datagram socket using a sendto call) no address was supplied

可正常运行的同步版本参考 #

# credit: pywakeonlan项目
import socket
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
    if interface is not None:
        sock.bind((interface, 0))
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.connect((ip_address, port))
    for packet in macs:
        sock.send(packet.encode())

解决方案 #

问题根源 #

1. Windows Proactor事件循环兼容性问题：Python 3.10在Windows上默认使用Proactor事件循环，它处理UDP广播套接字的recv操作时会抛出WinError 10022，因为广播套接字未连接时，Proactor的读取逻辑存在缺陷。
2. Future状态错误：error_received回调多次尝试给已完成的done Future设置异常，触发InvalidStateError。
3. 发送方式错误：使用transport.sendto却未指定目标地址，而同步版本是先connect再用send；asyncio中指定remote_addr后，应使用transport.send而非sendto。

修复后的代码 #

import asyncio
from typing import Optional
import socket

BROADCAST_IP = "255.255.255.255"
DEFAULT_PORT = 9


class _WOLProtocol:
    def __init__(self, *messages):
        self.packets = messages
        self.done = asyncio.get_running_loop().create_future()

    def connection_made(self, transport):
        try:
            # 已指定remote_addr，直接用send而非sendto
            for p in self.packets:
                transport.send(p.encode())
            # 发送完成后标记Future完成
            self.done.set_result(None)
        except Exception as exc:
            self.done.set_exception(exc)

    def error_received(self, exc):
        # 仅当Future未完成时设置异常
        if not self.done.done():
            self.done.set_exception(exc)

    def connection_lost(self, exc):
        print('closing')
        if not self.done.done():
            if exc:
                self.done.set_exception(exc)
            else:
                self.done.set_result(None)


async def send_magic_packet(
    *macs: str,
    ip_address: str = BROADCAST_IP,
    port: int = DEFAULT_PORT,
    interface: Optional[str] = None
) -> None:
    loop = asyncio.get_running_loop()
    # 手动创建并配置套接字，规避Proactor的广播配置问题
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    if interface:
        sock.bind((interface, 0))
    
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: _WOLProtocol(*macs),
        remote_addr=(ip_address, port),
        sock=sock  # 使用手动配置的套接字
    )

    try:
        await protocol.done
    finally:
        transport.close()

if __name__ == "__main__":
    asyncio.run(send_magic_packet('test'))

关键修复点 #

• 手动配置套接字：直接设置SO_BROADCAST选项后传给create_datagram_endpoint，避免asyncio内部在Proactor下的配置异常。
• 替换发送方法：用transport.send替代sendto，因为已通过remote_addr指定目标地址。
• 保护Future状态：在error_received和connection_lost中先检查done是否未完成，再设置结果或异常，避免状态错误。
• 主动标记任务完成：发送完所有包后立即设置done为完成，无需等待连接关闭。

内容的提问来源于stack exchange，提问作者Diederick C. Niehorster