Sending messages correctly with Async Server Python

Question

I am running into an issue with an asyncio server I have set up. I've done alot of research on the topic but I can't seem to find anything helpful. Essentially my server is listening on a port, but when I send messages from a "client" It seems to be sending a large amount of messages. Previously I had been using a Multi Threaded socket server, but after some research it seemed that an async server would be more applicable in my project.
I'm not sure the best way to setup the server, so any constructive feedback on best practices and what not would be greatly appreciated.

lia_server.py

import logging
import asyncio
from datetime import datetime
from os import path
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s %(name)-12s %(levelname)-8s %(message)s',
                    datefmt='%Y-%m-%d_%H:%M:%S',
                    filename=f'{path.abspath("../log")}'
                             f'\\srv_{datetime.strftime(datetime.today(), "%m_%d_%Y_%H_%M_%S")}',
                    filemode='w')
header = 1024
msg_format = 'utf-8'
disconnect_message = "BYE"


async def handle_client(reader, writer):
    request = None
    while request != disconnect_message:
        request = (await reader.read()).decode(msg_format)
        logging.info(f'[Message Received] {request}')
        response = str(request)
        writer.write(response.encode(msg_format))
        await writer.drain()
    writer.close()


class LiaServer:
    def __init__(self, host, port):
        self.server = host
        self.port = port

        logging.info(f'LiaServer Class object has been created with values of {self.server}:{self.port}')

    async def run_server(self):
        logging.info(f'[Server Start] Attempting to start the server now')
        self.server_instance = await asyncio.start_server(handle_client, self.server, self.port)
        async with self.server_instance:
            logging.info(f'[Server Start] Server is now LISTENING on {self.server}:{self.port}')
            await self.server_instance.serve_forever()

    def stop_server(self):
        logging.info(f'[Server Stop] Attempting to stop the server now')
        pass


async def main():
    srv_cls = LiaServer('localhost', '1338')
    taskserver = asyncio.create_task(srv_cls.run_server())
    await taskserver


if __name__ == "__main__":
    asyncio.run(main())

client_test.py

import socket
import lia_server
import asyncio
msg_format = 'utf-8'
header = lia_server.header
disconnect_message = lia_server.disconnect_message


async def receive_cfg_and_send_msg(cls, msg: str):
    address = (cls.server, cls.port)
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect(address)
    await send(client, msg)


async def send(client, msg):
    message = msg.encode(msg_format)
    msg_length = len(message)
    send_length = str(msg_length).encode(msg_format)
    send_length += b' ' * (header - len(send_length))
    client.send(message)


def main():
    server = lia_server.LiaServer('localhost', 1338)
    asyncio.run(receive_cfg_and_send_msg(server, "Hello World"))
    asyncio.run(receive_cfg_and_send_msg(server, disconnect_message))
    print("Disconnected")


if __name__ == "__main__":
    main()

Then Whenever running the server, then the client, this happens

log

2021-07-14_00:10:21 root         INFO     LiaServer Class object has been created with values of localhost:1338
2021-07-14_00:10:21 root         INFO     [Server Start] Attempting to start the server now
2021-07-14_00:10:21 root         INFO     [Server Start] Server is now LISTENING on localhost:1338
2021-07-14_00:10:33 root         INFO     [Message Received] Hello World
2021-07-14_00:10:33 root         INFO     [Message Received] 
2021-07-14_00:10:33 root         INFO     [Message Received] 
2021-07-14_00:10:33 root         INFO     [Message Received] 
2021-07-14_00:10:33 root         INFO     [Message Received] 
The [Message Received] Lines keep repeating for a very long time.

The behavior of the server makes me feel that it is looping constantly, but I am not sure as to exactly why.
I am looking to have the server receive the message, then parse the message and check if it is the disconnect message. Afterwards the server to send a response to the client saying the disconnect message back.

Thank you in advance!

Answer 1

There are multiple problems with your code. The one that is causing the immediate issue you are seeing is that you are using await reader.read() to read a "message" from the client. TCP is a streaming protocol, so it doesn't have a concept of messages, only of a stream of bytes that arrive in the order in which they were sent by the client. The read() method will read all data until EOF. This is almost certainly not what you want. Since you have a "header" defined, you should probably be calling reader.readexactly(header). Then you should strip the trailing whitespace off the message, and only then should you try to match it against known strings.

The next problem is in the condition while request != disconnect_message. If the end-of-file condition is encountered, StreamReader.read will return an empty bytes object, b''. This will be not compare equal to the disconnect message, so your loop will continue looping, with each subsequent read() again returning b'' to again indicate EOF. This causes the infinite loop you are seeing, and you can fix it by checking for either disconnect_message or b''.

Your client also has an issue: it creates two completely separate connections to the server - notice how receive_cfg_and_send_msg opens a new connection. The first connection is closed by the garbage collector on function exit which causes the infinite loop on the server. The second connection tries to reach the server, but the server is now completely stuck in the infinite loop of the first connection. (See e.g. this answer for explanation why it's stuck despite apparently awaiting.)

Finally, the client contains multiple calls to asyncio.run(). While you can do that, it's not how asyncio.run is supposed to be used - you should probably have a single async def main() function and call it through asyncio.run(main()). Inside that function you should await the async functions you need to run (or combine them using asyncio.gather(), etc.)