Python TCP Payload Duplication - 同时将数据传递到多个端点

Question

这是我在这里的第一篇文章！

我的目标是复制单向 TCP 流的有效负载并将此有效负载同时发送到多个端点。我有一个用 Python 编写的工作原型，但是我对 Python 和 Socket 编程很陌生。理想情况下，该解决方案能够在 Windows 和 *nix 环境中运行。

这个原型可以工作，但是它会为每个 Buffer 长度（当前设置为 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)

我查看了其他库以尝试实现我的目标，包括使用：

• 扭曲的
• 异步
• 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

如果你不想学习一些更高级的框架，比如 Twisted，有两种方法。

最接近您正在做的事情：使用线程，但每个传出连接需要一个线程 --- 而每个传出数据包不。创建 3 个Queue.Queue 对象，并创建 3 个线程，将Queue 对象和目标之一传递给每个线程。每个线程打开一个套接字，然后在一个循环中，它从自己的Queue 中获取下一个字符串并将其发送到套接字。客户端线程（可以只是主线程，先验）接收数据作为字符串，并将这些字符串中的每一个放入所有队列中。这样一来，发送的数据包就不会乱序，就像您为每个数据包创建一个线程一样。

替代方法是完全避免线程，并使用select()。这有点令人费解。基本上你只有一个以select() 开头的大循环。需要仔细管理才能将正确的套接字列表传递给select()：当有来自入站套接字的传入数据时，您希望对select() 的调用唤醒或者，或者 如果出站套接字既准备好发送更多内容并且还有更多内容要发送。在此模型中，您将有 3 个字符串列表；当您读取传入数据时，您会将其附加到所有三个列表中； select() 调用传递了具有非空列表的出站套接字列表（因此，要发送更多）；并且发送时，在这个模型中不能使用sendall()，而是send()，如果发送的字符串少于完整的字符串，则必须将剩余部分重新添加到相应列表的开头。