我试图在python中实现Reactor pattern。我想我已经开始使用multiprocessingselect.select了。然而,我正试图对我的服务器进行压力测试,所以我编写了一个简单的DoS客户机来让它充满连接但我有个有趣的错误:
[Winerror 10061]无法建立连接,因为目标计算机主动拒绝了它
有趣的是,我正在服务器上为backlog amount执行socket.listen(5)当我把select.select中的读者准备好后,我会显示计数,而且我只有1或2个——而不是我期望的5个。
对于少量的线程(~20),我没有注意到它会阻塞,但对于更多的线程(50+),它确实会拒绝连接。
我的问题是在服务器端还是客户端(或者只是在操作系统/套接字级别)这是我能解决的问题吗如果是,怎么做?
这是我的代码:
顾客

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))

服务器
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()

最佳答案

我试着测试你的代码,但是在import queue上失败了。
不过,可能是
您的操作系统充当指定的listen()函数:“实现可能会对backlog施加限制,并静默地减少指定的值。”
一旦有足够的未完成连接,您的操作系统就会停止接受该连接,这可能无法根据请求显示。
这些只是猜测可能的原因;也许我完全错了。

07-24 09:52
查看更多