Do more iteration at a time python

Question

I have a blocking for-loop where I try to connect with socket to a server. I want to make a non-blocking loop because it takes a lot of time, like overlap connections.
Here's my example code:

import socket

for _ in range(100):
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
        s.connect(('', 4444))

If you search in your browser for "Python multiprocessing", you'll find references that can explain this much better than we can manage here. — Prune, Aug 28 '19 at 14:25
i know threading, multiprocessing and asyncio but i need an example on my code, if you can. Thank you again — Gianla, Aug 28 '19 at 14:32

Pynchia · Answer 1 · 2019-08-28T15:13:11.227

As suggested by MLAlex, it makes sense to use the asyncio library.

What about this, inspired from the asyncio docs:

import asyncio

async def tcp_echo_client(message):
    reader, writer = await asyncio.open_connection(
        '127.0.0.1', 8888)

    print(f'Send: {message!r}')
    writer.write(message.encode())

    await asyncio.sleep(randint(1,3))

    data = await reader.read(100)
    print(f'Received: {data.decode()!r}')

    print('Close the connection')
    writer.close()
    await writer.wait_closed()

    return data.decode()


async def main():
    requests = [
            tcp_echo_client(f"msg {i}") for i in range(10)
    ]
    echoes = await asyncio.gather(*requests)
    print(echoes)


asyncio.run(main())

e.g. coupled with this echo server (which takes three seconds to echo back one message):

import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')

    print(f"Received {message!r} from {addr!r}")

    await asyncio.sleep(3)

    print(f"Send: {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()

async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addr = server.sockets[0].getsockname()
    print(f'Serving on {addr}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

produces:

Send: 'msg 0'
Send: 'msg 1'
Send: 'msg 2'
Send: 'msg 3'
Send: 'msg 4'
Send: 'msg 5'
Send: 'msg 6'
Send: 'msg 7'
Send: 'msg 8'
Send: 'msg 9'
Received: 'msg 0'
Close the connection
Received: 'msg 1'
Close the connection
Received: 'msg 2'
Close the connection
Received: 'msg 3'
Close the connection
Received: 'msg 4'
Close the connection
Received: 'msg 5'
Close the connection
Received: 'msg 6'
Close the connection
Received: 'msg 7'
Close the connection
Received: 'msg 8'
Close the connection
Received: 'msg 9'
Close the connection
['msg 0', 'msg 1', 'msg 2', 'msg 3', 'msg 4', 'msg 5', 'msg 6', 'msg 7', 'msg 8', 'msg 9']

in three seconds or so, i.e. the time it takes to satisfy one request only

real    0m3,169s
user    0m0,044s
sys     0m0,084s

If we introduce some variance in the time it takes the server to reply, we can see the clients receiving results out of order. e.g. on line 10:

await asyncio.sleep(randint(3,4))

then the output of the client becomes:

(tih1) SO $ time python aio_cnx.py 
Send: 'msg 0'
Send: 'msg 1'
Send: 'msg 2'
Send: 'msg 3'
Send: 'msg 4'
Send: 'msg 5'
Send: 'msg 6'
Send: 'msg 7'
Send: 'msg 8'
Send: 'msg 9'
Received: 'msg 1'
Close the connection
Received: 'msg 2'
Close the connection
Received: 'msg 3'
Close the connection
Received: 'msg 7'
Close the connection
Received: 'msg 8'
Close the connection
Received: 'msg 9'
Close the connection
Received: 'msg 0'
Close the connection
Received: 'msg 4'
Close the connection
Received: 'msg 5'
Close the connection
Received: 'msg 6'
Close the connection
['msg 0', 'msg 1', 'msg 2', 'msg 3', 'msg 4', 'msg 5', 'msg 6', 'msg 7', 'msg 8', 'msg 9']

real    0m4,135s
user    0m0,059s
sys     0m0,013s

I really appreciate your comment and code, but my code is more complicated than the example i put in the question, and yours will never work with mine. Sorry — Gianla, Aug 28 '19 at 15:17
I see. No problem, this is one way to do it, based on the info provided. If it doesn't fit your constraints it might help others as an answer — Pynchia, Aug 28 '19 at 15:19
yeah, it's becouse I don't want to make heavy my code or upset it totally. I'm sorry again — Gianla, Sep 03 '19 at 11:25

score 1 · Answer 2 · answered Aug 28 '19 at 16:54

give a try on selectors, base on the fact that you are trying to do non-block I/O on sockets

my sample code to play around localhost ssh server with 10 concurrent connection attempts

import sys
import socket
import selectors

sel = selectors.EpollSelector()

def read_conn(obj, mask):
    try:
        data = obj.recv(1024)
        if data :
            print(data.decode("utf-8"))
        else:
            sel.unregister(obj)
            obj.close()
    except Exception as  ex:
        print( ex )
        sel.unregister(obj)
        obj.close()


def write_conn(obj, mask):
    msg=b"flooding\n\n"
    obj.send(msg)
    sel.unregister(obj)
    sel.register(obj,selectors.EVENT_READ,read_conn)
    return

for i in range(10) :

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    addr = ('127.0.0.1',22)
    try:
        sock.connect(addr)
        print(f"tried {i}")
        sock.setblocking(False)
        sel.register(sock,selectors.EVENT_WRITE, write_conn)
    except socket.error as ex:
        print("fail to connect error {}".format(ex))
        sys.exit(1)

while True:
    events = sel.select(1)
    if len(events) == 0:
        print("all done")
        break
    for key, mask in events:
        callback= key.data
        callback(key.fileobj,mask)

on my vm its output

tried 0
tried 1
tried 2
tried 3
tried 4
tried 5
tried 6
tried 7
tried 8
tried 9
SSH-2.0-OpenSSH_7.4
Protocol mismatch.

SSH-2.0-OpenSSH_7.4
Protocol mismatch.

SSH-2.0-OpenSSH_7.4
Protocol mismatch.

[Errno 104] Connection reset by peer
[Errno 104] Connection reset by peer
[Errno 104] Connection reset by peer
SSH-2.0-OpenSSH_7.4
Protocol mismatch.

  cut out duplicate outputs

all done

i did the non-block setting after socket.connect(), because connect() raises exceptions easily in non-block mode, error no 115 — James Li, Aug 28 '19 at 17:02
if your want to do non-block connect replace the sock.connect() call with sock.connect_ex() and move sock.setblocking() before sock.connect_ex() see also https://stackoverflow.com/questions/10204134/tcp-connect-error-115-operation-in-progress-what-is-the-cause — James Li, Aug 28 '19 at 17:13

Do more iteration at a time python

2 Answers2