Python TCP有效載荷複製 - 同時將數據傳遞到多個端點

Question

這是我的第一篇文章！

我的目標是複製單向TCP流的有效負載，並將此有效負載併發發送到多個端點。我有一個用Python編寫的工作原型，但是我對Python和Socket編程都很陌生。理想的解決方案可以在Windows和* nix環境中運行。

這個原型可以工作，但它爲每個緩衝區長度（當前設置爲4096字節）創建一個新的發送TCP連接。這樣做的主要問題是我最終會用盡本地端口發送，理想情況下，我希望數據從每個傳入的TCP流傳遞到單個TCP流（對於每個端點）。傳入數據可以從小於1024字節到數百兆字節不等。

此時，每4096個字節啓動一個新的傳出TCP流。我不確定問題出在我的線程實現上，還是我錯過了其他非常明顯的東西。

在我的研究中，我發現select（）可以提供幫助，但是我不確定它是否合適，因爲我可能需要處理一些傳入數據並在將來對某些情況作出響應。

這是我迄今（有些我已經嘗試了代碼的變化被註釋掉）的代碼：我已研究過其他圖書館努力實現我的目標，包括使用

#!/usr/bin/python 
#One way TCP payload duplication 
import sys 
import threading 
from socket import * 
bufsize = 4096 
host= '' 

# Methods: 
#handles sending the data to the endpoints 
def send(endpoint,port,data): 
    sendSocket = socket(AF_INET, SOCK_STREAM) 
    #sendSocket.setblocking(1) 
    sendSocket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) 
    #sendport = sendSocket.getsockname 
    #print sendport 
    try: 
     sendSocket.connect((endpoint, port)) 
     sendSocket.sendall(data) 
    except IOError as msg: 
     print "Send Failed. Error Code: " + str(msg[0]) + ' Message: ' + msg[1] 
     sys.exit() 

#handles threading for sending data to endpoints 
def forward(service, ENDPOINT_LIST, port, data): 
    #for each endpoint in the endpoint list start a new send thread 
    for endpoint in ENDPOINT_LIST: 
     print "Forwarding data for %s from %s:%s to %s:%s" % (service,host,port,endpoint,port) 
     #send(endpoint,port,data) 
     ethread = threading.Thread(target=send, args=(endpoint,port,data)) 
     ethread.start() 

#handles threading for incoming clients 
def clientthread(conn,service,ENDPOINT_LIST,port): 
    while True: 
     #receive data form client 
     data = conn.recv(bufsize) 
     if not data: 
      break 
     cthread = threading.Thread(target=forward, args=(service, ENDPOINT_LIST, port, data)) 
     cthread.start() 
    #no data? then close the connection 
    conn.close() 

#handles listening to sockets for incoming connections 
def listen(service, ENDPOINT_LIST, port): 
    #create the socket 
    listenSocket = socket(AF_INET, SOCK_STREAM) 
    #Allow reusing addresses - I think this is important to stop local ports getting eaten up by never-ending tcp streams that don't close 
    listenSocket.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1) 
    #try to bind the socket to host and port 
    try: 
     listenSocket.bind((host, port)) 
    #display an error message if you can't 
    except IOError as msg: 
     print "Bind Failed. Error Code: " + str(msg[0]) + ' Message: ' + msg[1] 
     sys.exit() 
    #start listening on the socket 
    listenSocket.listen(10) 
    print "Service %s on port %s is listening" %(service,port) 
    while True: 
     #wait to accept a connection 
     conn, addr = listenSocket.accept() 
     print 'Connected to ' + addr[0] + ':' + str(addr[1]) + ' on port ' + str(port) 
     #start new thread for each connection 
     lthread = threading.Thread(target=clientthread , args=(conn,service,ENDPOINT_LIST,port)) 
     lthread.start() 
    #If no data close the connection 
    listenSocket.close() 

service = "Dumb-one-way-tcp-service-name1" 
ENDPOINT_LIST = ["192.168.1.100","192.168.1.200"] 
port = 55551  
listen(service,ENDPOINT_LIST,port) 

：

• 扭曲
• Asyncore
• Scapy的

但是，我發現它們對於我的適度需求和編程技巧水平而言非常複雜。

如果任何人有任何建議，我可以改進我的方法或任何其他方式可以實現此目標，請讓我知道！

Answer 1

總之，你的問題是沒有足夠的端口，對不對？看起來您發送後沒有關閉套接字。在send()試試這個：

... 
except IOError as msg: 
    print "Send Failed. Error Code: " + str(msg[0]) + ' Message: ' + msg[1] 
    sys.exit() 
finally: 
    sendSocket.close() 

Answer 2

有兩種方式，如果你不想學像扭曲的一些更高級的框架。

最接近你在做什麼：使用線程，但你需要有一個線程每個傳出連接---和而不是每個傳出數據包。創建3個Queue.Queue對象，並創建3個線程，傳遞給Queue對象之一和目標之一。每個線程打開一個套接字，然後在一個循環中，它從它自己的Queue獲得下一個字符串並將其發送到套接字。 clientthread（可以只是主線程，先驗）以字符串形式接收數據，並將每個字符串放入所有隊列中。這樣，發送的數據包就不會出現亂序，因爲如果您爲每個數據包創建一個線程，就可能發生這種情況。

另一種方法是完全避免線程，並使用select()。這是更多的思想彎曲。基本上你只有一個以select()開頭的大循環。它需要精心管理，插座的正確列表傳遞給select()：你要調用select()醒來要麼時，有從入站插座輸入數據，或如果出站插槽既可以發送更多還有更多要發送的東西。在這個模型中，你會有3個字符串列表;當你讀取傳入的數據時，你將它附加到所有三個列表中; select()調用被傳遞到具有非空列表的出站套接字列表（因此，要發送更多）;並且在發送時，您不得在此模型中使用sendall()，但在send()中使用sendall()，並且如果發送的字符串少於完整字符串，則必須將其餘部分重新添加到相應列表的開頭。