How do I allow more simultaneous socket connections?

Question

I'm attempting to implement the Reactor pattern in Python. I think I have a pretty decent start using multiprocessing and select.select. However, I'm trying to stress-test my server, so I wrote a simple DoS client to flood it with connections. But I get an interesting error:

[WinError 10061] No connection could be made because the target machine actively refused it

The interesting thing about this is that I'm doing socket.listen(5) for the backlog amount on the server. After I get readers ready from select.select I display the count, and I only ever have 1 or 2 - not the 5 that I would expect.

For a small number of threads (~20) I haven't noticed it choke, but for a larger number (50+) it does tend to refuse connections.

Is my problem on the server or client side (or just at the OS/socket level)? Is this a problem that I can fix? And if so, how?

Here is my code:

Client

import threading
import time
import socket
from contextlib import contextmanager

IP = '127.0.0.1'
PORT = 4200

@contextmanager
def open_socket(ip, port):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        sock.connect((ip, port))
        yield sock
    finally:
        sock.close()


class Flood(threading.Thread):
    def __init__(self, id):
        super(Flood, self).__init__()
        self.id = id
        self.failed = False


    def run(self):
        try:
            with open_socket(IP, PORT) as sock:
                msg = "Hello this is some data from %d" % self.id
                sock.send(msg.encode())
        except Exception as e:
            print(e)
            self.failed = True


def make_threads(count):
    return [Flood(_) for _ in range(count)]


threads = make_threads(5000)

start = time.time()
for t in threads:
    t.start()

for t in threads:
    t.join()

print("Failed: ", sum(1 if x.failed else 0 for x in threads))
print("Done in %f seconds" % (time.time() - start))

Server

import sys
import logging
import socket
import select
import time
import queue
from multiprocessing import Process, Queue, Value
log = logging.getLogger(__name__)
log.setLevel(logging.DEBUG)
log.addHandler(logging.StreamHandler())

IP = '127.0.0.1'
PORT = 4200

keep_running = True

def dispatcher(q, keeprunning):
    try:
        while keeprunning:
            val = None
            try:
                val = q.get(True, 5)
                if val:
                    log.debug(val[0].recv(1024).decode())
                    val[0].shutdown(socket.SHUT_RDWR)
                    val[0].close()
            except queue.Empty:
                pass
        log.debug("Dispatcher quitting")
    except KeyboardInterrupt:
        log.debug("^C caught, dispatcher quitting")


def mainloop(sock):
    readers, writers, errors = [sock], [], []
    timeout = 5
    while True:
        readers, writers, errors = select.select(readers,
                                                 writers,
                                                 errors,
                                                 timeout)
        incoming = yield readers, writers, errors
        if incoming and len(incoming) == 3:
            readers, writers, errors = incoming
        if not readers:
            readers.append(sock)


def run_server():
    keeprunning = Value('b', True)
    q = Queue()
    p = Process(target=dispatcher, args=(q, keep_running))
    try:
        p.start()
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind((IP, PORT))
        sock.listen(50)
        sock.setblocking(0)
        loop = mainloop(sock)
        for readers, writers, errors in loop:
            if readers:
                client, addr = readers[0].accept()
                q.put((client, addr))
            log.debug('*'*50)
            log.debug('%d Readers', len(readers))
            log.debug('%d Writers', len(writers))
            log.debug('%d Errors', len(errors))


    except KeyboardInterrupt:
        log.info("^C caught, shutting down...")
    finally:
        keeprunning.value = False
        sock.close()
        p.join()

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: test.py (client|server)")
    elif sys.argv[1] == 'client':
        run_client()
    elif sys.argv[1] == 'server':
        run_server()

Answer 1

I tried to test your code, but failed on import queue.

Nevertheless, it might be that

• your OS acts as the listen() function is specified: "Implementations may impose a limit on backlog and silently reduce the specified value."
• your OS stops accepting the connection as soon there are enough incomplete connections, which might not be displayed upon request.

These are just guesses about what might be the reason; maybe I am completely wrong.