Python multithreading for simultaneous input and socket connection

user_16915845 0
Viewed 255

I have started with a friend of mine to dig deeper into network coding. Concurrency and parallelism are a big part of this.

We have created a server and client to connect them and this works fine. Now we want to create a thread in the server for checking for inputs from the keyboard while listing to connections on the socket. Maybe we get something totally wrong but we tried it with this code and a threadpoolexecution but the program get stuck at the first await call

i = await ainput.asyncInput()

We thought that after the await starts the thread wait for an input and the main thread goes on in execution but that seems to be wrong.

Here is the server module:

import socket
import asyncio
import asyncron_Input as ainput

def closeServer():
    exit()    

server_address = ('localhost',6969)

async def main():

   #create TCP Socket
   serverSock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

   # prints the adress and the port of the server
   print("starting up on: ")
   print(server_address[0])
   print("port: ")
   print(server_address[1])

   # binds the socket to the given adress
   serverSock.bind(server_address)

   #listen for connection

   serverSock.listen(1)

   print("End server with 1")

   while True:
       #close server with asynco inputs
       i = await ainput.asyncInput()
       if i == "1":
           closeServer()

       #wait for connection
       print("waiting for conncetion")

       conn,client_address = serverSock.accept()
       try:
           print("connected to",client_address)
        
           while True:
               data = conn.recv(16)
               if data:
                   print("received:",data)
                   data = "succsessful"
               else:
                   print("no data")
                   break
       finally:
           #close connection
           conn.close


asyncio.run(main())

Here is the async input:

import asyncio
from concurrent.futures import ThreadPoolExecutor

async def asyncInput():
    with ThreadPoolExecutor(1,'Async Input') as executor:
        return await asyncio.get_event_loop().run_in_executor(executor,input)

Thanks for your help in advance

python-asyncio threadpoolexecutor multithreading python
1 Answers

There's two problems with your code:

  1. You wait for input before accepting any socket connections, you can't use await if you want code proceeding it to happen concurrently, you need to use a Task.
  2. You're using blocking sockets. sock.accept and sock.recv are blocking by default. They'll halt execution of your event loop, you need to use them in an await expression, which means making your sever socket non-blocking and then using them with special asyncio specific socket methods.

To fix this, you'll need to wrap listening for input in a task, make your server socket non-blocking, get the running event loop and then use the sock_accept and sock_recv methods of the event loop. Putting this all together, your code will look something like this:

import asyncio
import socket
from concurrent.futures import ThreadPoolExecutor

async def asyncInput():
    with ThreadPoolExecutor(1,'Async Input') as executor:
        return await asyncio.get_event_loop().run_in_executor(executor,input)

def closeServer():
    exit()

server_address = ('localhost',8000)

async def loop_for_input():
    while True:
        #close server with asynco inputs
        i = await asyncInput()
        if i == "1":
            closeServer()

async def main():

    #create TCP Socket
    serverSock = socket.socket(socket.AF_INET, socket.SOCK_STREAM, )
    serverSock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

    # prints the adress and the port of the server
    print("starting up on: ")
    print(server_address[0])
    print("port: ")
    print(server_address[1])

    # binds the socket to the given adress
    serverSock.bind(server_address)
    serverSock.setblocking(False) #make your socket non-blocking

    #listen for connection

    serverSock.listen(1)

    print("End server with 1")
    loop = asyncio.get_running_loop() # get the running event loop
    input_task = asyncio.create_task(loop_for_input()) # create an task to run your input loop
    while True:
        #wait for connection
        print("waiting for conncetion")

        conn,client_address = await loop.sock_accept(serverSock) # use the sock_accept coroutine to asynchronously listen for connections
        try:
            print("connected to",client_address)

            while True: # you may also want to create a task for this loop.
                data = await loop.sock_recv(conn, 16) # use the sock_recv coroutine to asynchronously listen for data
                if data:
                    print("received:",data)
                    data = "succsessful"
                else:
                    print("no data")
                    break
        finally:
        #close connection
            conn.close()


asyncio.run(main())

There's potentially a third problem in that your code can only handle one client at a time since you enter an infinite loop for the first connection that comes in. This means any additional clients who connect will be blocked. If you want to solve that problem, any time a client connects, create a new Task to listen for data from the client, similar to what the code above does with asyncInput()

user_29173260
Related