Python 选择器库 - 如何在同一个套接字上发送两种方式？

Question

我正在尝试使用 Python selectors库 ( docs ) 以两种方式发送消息。我使用此资源对 selectors 有一个基本的了解。图书馆:
https://realpython.com/python-sockets/
这是带有代码的 GitHub:
https://github.com/realpython/materials/tree/master/python-sockets-tutorial
在有人将其标记为重复之前，它不是，我特别询问使用 selectors 的 2 路套接字通信。图书馆，它比 select 更现代和不同图书馆，其他答案都是针对的。
如果上述链接中断并遵循不只提供链接的 SO 政策，这里是多连接服务器和客户端，并由我进行了一些小的重新排列和添加评论(所有通过上述链接归功于原作者):
MulticonnServer.py:

# MulticonnServer.py

import sys
import socket
import selectors
import types


def main():
    # instantiate a selector
    sel = selectors.DefaultSelector()

    # define host, port, and number of connections
    host = '127.0.0.1'
    port = 65432
    num_conns = 2

    # instantiate a socket
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    # bind and listen
    sock.bind((host, port))
    sock.listen()
    print('listening on', (host, port))
    # set non-blocking
    sock.setblocking(False)
    # register the socket with the selector
    sel.register(sock, selectors.EVENT_READ, data=None)

    try:
        # while True:
        # ToDo: figure out why have to multiply x2 here ??
        for i in range(num_conns * 2):
            # execution waits here until ??
            events = sel.select(timeout=None)
            for key, mask in events:
                if key.data is None:
                    accept_wrapper(sel, key.fileobj)
                else:
                    service_connection(sel, key, mask)
                # end if
            # end for
        # end for
    except KeyboardInterrupt:
        print('caught keyboard interrupt, exiting')
    finally:
        sel.close()
    # end try
# end function

def accept_wrapper(sel, sock):
    conn, addr = sock.accept()  # Should be ready to read

    print('accepted connection from', addr)
    conn.setblocking(False)
    data = types.SimpleNamespace(addr=addr, inb=b"", outb=b"")
    events = selectors.EVENT_READ | selectors.EVENT_WRITE
    sel.register(conn, events, data=data)
# end function

def service_connection(sel, key, mask):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        if recv_data:
            data.outb += recv_data
        else:
            print("closing connection to", data.addr)
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if data.outb:
            print("echoing", repr(data.outb), "to", data.addr)
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]
        # end if
    # end if
# end function

if __name__ == '__main__':
    main()

MulticonnClient.py:

# MulticonnClient.py

import sys
import socket
import selectors
import types


def main():
    sel = selectors.DefaultSelector()
    messages = [b"Message 1 from client.", b"Message 2 from client."]

    host = '127.0.0.1'
    port = 65432
    num_conns = 2

    start_connections(sel, host, int(port), int(num_conns), messages)

    try:
        while True:
            events = sel.select(timeout=1)
            if events:
                for key, mask in events:
                    service_connection(key, mask, sel)
            # Check for a socket being monitored to continue.
            if not sel.get_map():
                break
            # end if
        # end while
    except KeyboardInterrupt:
        print('caught keyboard interrupt, exiting')
    finally:
        sel.close()
    # end try
# end function

def start_connections(sel, host, port, num_conns, messages):
    server_addr = (host, port)
    for i in range(0, num_conns):
        connid = i + 1
        print("starting connection", connid, "to", server_addr)
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setblocking(False)
        sock.connect_ex(server_addr)
        events = selectors.EVENT_READ | selectors.EVENT_WRITE
        data = types.SimpleNamespace(
            connid=connid,
            msg_total=sum(len(m) for m in messages),
            recv_total=0,
            messages=list(messages),
            outb=b"",
        )
        sel.register(sock, events, data=data)
    # end for
# end function

def service_connection(key, mask, sel):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        if recv_data:
            print("received", repr(recv_data), "from connection", data.connid)
            data.recv_total += len(recv_data)
        if not recv_data or data.recv_total == data.msg_total:
            print("closing connection", data.connid)
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if not data.outb and data.messages:
            data.outb = data.messages.pop(0)
        if data.outb:
            print("sending", repr(data.outb), "to connection", data.connid)
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]
        # end if
    # end if
# end function

if __name__ == '__main__':
    main()

这些程序工作得很好，但我接下来想做的是在同一个套接字上以一种方式发送一条消息(例如，客户端到服务器，就像上面已经做的那样)和另一种方式(例如，服务器到客户端)的下一条消息.
我不是 100% 确定这是可能的。特别是这行:

sel.register(sock, selectors.EVENT_READ, data=None)

在服务器是有关。如何重新设计此示例以在同一套接字上提供 2 路通信？
- - 编辑 - -
我发现更改 sel.register输入 MulticonnServer.py到:

sel.register(sock, selectors.EVENT_READ | selectors.EVENT_WRITE, data=None)

不会导致崩溃，所以也许有希望？但是，我仍然不确定如何在同一个套接字上设置 2-way comm。

Answer 1

我也在做这件事并让它发挥作用。我必须做的是:

创建一个列表以捕获任何未决消息

并且，修改 service_connection() 如下

selectors.EVENT_WRITE 其实没必要设置.这似乎无关紧要。

第一部分是:

#
# capture pending messages: (fd,message) where fd is the socket fd, a numeric integer value. For example:

pending_messages = []

...
pending_messages.append( (key.fd, message) )
...

然后在 service_connection 内部(见底部，这是唯一的 mod):

def service_connection(sel, key, mask):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        if recv_data:
            data.outb += recv_data
        else:
            print("closing connection to", data.addr)
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if data.outb:
            print("echoing", repr(data.outb), "to", data.addr)
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]
        elif len(pending_messages) > 0:
            if pending_messages[0][0] == key.fd:
                data.outb = pending_messages[0][1]
                pending_messages.pop(0)
# end function

系统准备好写入似乎很重要。您不能随时随地发送消息。