如何允許更多的同時套接字連接？

Question

我試圖在Python中實現Reactor pattern。我想我有一個非常體面的開始使用multiprocessing和select.select。但是，我試圖對我的服務器進行壓力測試，所以我編寫了一個簡單的DoS客戶端，用連接對其進行洪泛。但是，我得到一個有趣的錯誤：

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

關於這個有趣的是，我做socket.listen(5)爲backlog amount在服務器上。在我從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() 

Answer 1

我想測試你的代碼，但未能上import queue。

然而，它可能是

• 您的操作系統充當listen()函數指定：「實現可能對積壓的限制，靜靜減少指定的值。」
• 只要有足夠的不完整連接，您的操作系統就會停止接受連接，這可能無法根據請求顯示。

這些只是猜測可能是什麼原因;也許我完全錯了。