Как синхронизировать два совместных упражнения трио?

Question

Я просматриваю Trio tutorial , и я создал эхо-клиент , который отправляет сообщение на эхо-сервер в течение 10 секунд:

async def sender(client_stream, flag):
    print("sender: started!")
    end_time = time.time() + 10
    while time.time() < end_time:
        data = b"async can sometimes be confusing, but I believe in you!"
        print("sender: sending {!r}".format(data))
        await client_stream.send_all(data)
        await trio.sleep(0)
    flag = False
    print("Left the while 10 seconds loops")

и дождитесь ответов, пока flag будет `True.

async def receiver(client_stream, flag):
    print("receiver: started!")
    while(flag):
        data = await client_stream.receive_some()
        print("receiver: got data {!r}".format(data))
    print("receiver: connection closed")
    sys.exit()

Проблема в том, что иногда программа зависает на строке data = await client_stream.receive_some() из-за проблем с параллелизмом в отношении переменной flag.

Как мне послать сигнал из подпрограммы sender в подпрограмму receiver?

Вот вся программа , которая ты можешь бежать.

Answer 1

Он не просто иногда зависает, он все время зависает, потому что переменная flag в receiver() никогда не изменяется. Я думаю, у вас сложилось впечатление, что он каким-то образом разделен между receiver() и sender(). Это не так.

Самый простой способ исправить это - передать его в контейнер:

async def sender(client_stream, flag):
    print("sender: started!")
    end_time = time.time() + 10
    while time.time() < end_time:
        data = b"async can sometimes be confusing, but I believe in you!"
        print("sender: sending {!r}".format(data))
        await client_stream.send_all(data)
        await trio.sleep(0)
    flag[0] = False
    print("Left the while 10 seconds loops")

async def receiver(client_stream, flag):
    print("receiver: started!")
    while flag[0]:
        data = await client_stream.receive_some()
        print("receiver: got data {!r}".format(data))
    print("receiver: connection closed")
    sys.exit()

async def start_server():
    print("parent: connecting to 127.0.0.1:{}".format(PORT))
    client_stream = await trio.open_tcp_stream("127.0.0.1", PORT)
    flag = [False]
    async with client_stream:
        async with trio.open_nursery() as nursery:
            print("parent: spawning sender...")
            nursery.start_soon(sender, client_stream, flag)

            print("parent: spawning receiver...")
            nursery.start_soon(receiver, client_stream, flag)

Более элегантным решением было бы закрыть поток в sender() и поймать ClosedResourceError in receiver():

async def sender(client_stream):
    print("sender: started!")
    data = b"async can sometimes be confusing, but I believe in you!"
    with trio.move_on_after(10):
        print("sender: sending {!r}".format(data))
        await client_stream.send_all(data)

    await client_stream.aclose()
    print("Left the while 10 seconds loops")

async def receiver(client_stream):
    print("receiver: started!")
    try:
        async for data in client_stream:
            print("receiver: got data {!r}".format(data))
    except trio.ClosedResourceError:
        print("receiver: connection closed")

Обратите внимание, что вам даже не нужно sys.exit() для завершения программы.