Python中UDP吞吐量远低于TCP的原因排查求助

问题背景 #

我用Python开发了客户端和服务端应用，两者除主线程外都额外创建两个线程，分别处理TCP、UDP的数据收发操作。

客户端实现 #

clientTCP模块 #

class ClientTCP:
    def __init__(self, host_address: str, port: int):
        self.client_socket = None
        self.host_address = host_address
        self.port = port

    def connect_to_server(self, is_nagle_disabled: bool):
        try:
            self.client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            if is_nagle_disabled:
                self.client_socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, True)
            self.client_socket.connect((self.host_address, self.port))
        except socket.error:
            print('Error: probably wrong port passed for connection')
            sys.exit(1)

    def send_message_to_server(self, user_input: str):
        try:
            #print(f'MESSAGE TO SEND AS CLIENT TCP: {user_input}')
            self.client_socket.sendall(bytes(user_input, 'utf-8'))
        except socket.error as e:
            print('OVER HERE SEND MESSAGE TCP ERROR')
            print(f'Error: {e}')
            self.client_socket.close()
            sys.exit(1)

    def receive_message_from_server(self, buffer_size: int):
        data_from_server = []
        while True:
            byte_portion_of_data = self.client_socket.recv(buffer_size)
            if byte_portion_of_data.decode("utf-8") in ['BUSY', 'READY']:
                return data_from_server
            elif byte_portion_of_data:
                #print(f'TCP -> byte portion of data: {byte_portion_of_data.decode("utf-8")}')
                data_from_server.append(byte_portion_of_data)
            else:
                print('Entirety of message from server received')
                break
        return data_from_server

    def get_client_socket(self):
        return self.client_socket

    def __del__(self):
        self.client_socket.close()


def tcp_send_data_to_server(client: ClientTCP, data_to_send: list[int], size_of_buffer: int, is_nagle_disabled: bool, stop):
    try:
        #print('OVER HERE TCP!!!')
        client.connect_to_server(is_nagle_disabled)
        client_connection_list = client.receive_message_from_server(10)
        client_connection_message = ''.join([x.decode("utf-8") for x in client_connection_list])
        if client_connection_message == 'BUSY':
            return
        elif client_connection_message == 'READY':
            client.send_message_to_server(f"SIZE:{str(size_of_buffer)}")
        while True:
            #print("TCP!!!")
            if stop():
                break
            message_to_send = ''.join([str(x) for x in data_to_send])
            client.send_message_to_server(message_to_send)
            time.sleep(1)
    except ConnectionResetError:
        print("Socket was closed due to some unknown reasons. Sorry. :(")

senderUDP模块 #

class SenderUDP:
    def __init__(self, host_address: str, port: int):
        self.client_socket = None
        self.host_address = host_address
        self.port = port

    def connect_to_server(self):
        try:
            self.client_socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
        except socket.error as e:
            print(str(e))
            sys.exit(1)

    def send_message_to_server(self, input_to_send: str):
        try:
            #print(input_to_send)
            #print('OVER HERE 1 UDP send')
            self.client_socket.sendto(input_to_send.encode(), (self.host_address, self.port))
            #print('OVER HERE 2 UDP send')
        except Exception as e:
            print('Error: ' + str(e))
            self.client_socket.close()
            sys.exit(1)

    def get_client_socket(self):
        return self.client_socket

    def __del__(self):
        self.client_socket.close()


def sending(sender: SenderUDP, data_to_send: list[int], size_of_buffer: int, stop):
    try:
        sender.connect_to_server()
        sender.send_message_to_server(f"SIZE:{size_of_buffer}")
        while True:
            #print("UDP!!!")
            if stop():
                sender.send_message_to_server('END')
                break
            message_to_send = ''.join([str(x) for x in data_to_send])
            sender.send_message_to_server(message_to_send)
            sleep(1)
    except ConnectionResetError:
        print("Socket was closed due to some unknown reasons. Sorry. :(")

服务端实现 #

serverTCP模块 #

import socket
import sys
import re
import time
from datetime import datetime


class ServerTCP:
    def __init__(self, address, port, buffer: int):
        self.server_socket = None
        self.host_address = address
        self.port = port
        self.number_of_clients = 0
        self.buffer = buffer
        self.client_socket = None
        self.count_bytes_from_client = 0
        self.count_time_in_seconds = 0

    def create_socket(self):
        try:
            socket.inet_aton(self.host_address)
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.server_socket.bind((self.host_address, self.port))
        except socket.error:
            print('Error: ' + str(socket.error))
            sys.exit(1)

    def start_listening(self):
        try:
            self.server_socket.listen(1)
        except socket.error as msg:
            print('Error: ' + str(socket.error))
            self.server_socket.close()
            sys.exit(1)

    def set_client_socket(self, client_socket):
        self.client_socket = client_socket
        self.count_bytes_from_client = 0
        self.count_time_in_seconds = 0
        self.client_socket.send('READY'.encode())

    def handle_another_client(self, another_client_socket):
        another_client_socket.send('BUSY'.encode())
        another_client_socket.close()

    def remove_client_socket(self):
        self.client_socket.shutdown(socket.SHUT_WR)
        self.client_socket.close()
        self.buffer = 100
        self.client_socket = None

    def get_server_socket(self):
        return self.server_socket

    def get_client_socket(self):
        return self.client_socket

    def get_transmission_data(self):
        return self.count_bytes_from_client / 1000, self.count_time_in_seconds

    def __del__(self):
        self.server_socket.shutdown(socket.SHUT_RDWR)
        self.server_socket.close()
        print("Server socket closed")

    def __str__(self):
        kbytes, time = self.get_transmission_data()
        return f"TCP: {kbytes} per {time} seconds"

def tcp_get_data_from_client(server: ServerTCP):
    # data_from_client = []
    try:
        start = datetime.now()
        while True:
            byte_portion_of_data = server.client_socket.recv(server.buffer)
            if not byte_portion_of_data:
                break
            # data_from_client.append(byte_portion_of_data)
            string_data_from_client = byte_portion_of_data.decode('utf-8')
            if string_data_from_client.startswith('SIZE:'):
                temp = re.findall(r'\d+', string_data_from_client[5:])
                res = list(map(int, temp))
                if len(res) != 0:
                    server.buffer = res[0]
                    start = datetime.now()
            else:
                server.count_bytes_from_client += len(string_data_from_client)
            #print('End of the ServerTCP loop')
            time.sleep(0)
        end = datetime.now()
        server.count_time_in_seconds += (end - start).total_seconds()
        print(server)
        server.remove_client_socket()
    except ConnectionResetError:
        print("Socket was closed due to some unknown reasons. Sorry. :(")

receiverUDP模块 #

import socket
import struct
import sys
import re
import time
from datetime import datetime


class ReceiverUDP:
    def __init__(self, group, port: int, buffer: int):
        self.group = group
        self.port = port
        self.socket = None
        self.buffer = buffer
        self.count_bytes_from_client = 0
        self.count_time_in_seconds = 0
        self.start = None

    def starting(self):
        try:
            self.socket = socket.socket(family=socket.AF_INET, type=socket.SOCK_DGRAM)
            self.socket.bind((self.group, self.port))
        except socket.error as e:
            print(f'Error: {str(e)}')

    def get_transmission_data(self):
        return self.count_bytes_from_client / 1000, self.count_time_in_seconds

    def clean_transmission_data(self):
        self.count_bytes_from_client = 0
        self.count_time_in_seconds = 0

    def close_of_receiver(self):
        try:
            self.socket.close()
        except socket.error as e:
            print(f'Error: {str(e)}')

    def __str__(self):
        kbytes, time = self.get_transmission_data()
        return f"UDP: {kbytes} per {time}"


def receiving(receiver: ReceiverUDP, stop_thread):
    is_first_client = True
    try:
        #print('Start of ReceiverUDP')
        while True:
            #print('Before recvfrom, ReceiverUDP')
            #print(f'Start buffer: {receiver.buffer}')
            data = receiver.socket.recvfrom(receiver.buffer)
            message = data[0].decode("utf-8")
            print(f"Server UDP, message: {message}")
            if message.startswith("SIZE:"):
                temp = re.findall(r'\d+', message[5:])
                res = list(map(int, temp))
                print(f'New buffer: {res[0]}')
                if len(res) != 0:
                    if is_first_client:
                        receiver.start = datetime.now()
                        is_first_client = False
                    receiver.buffer = res[0]
                    print(f'New buffer assigned: {receiver.buffer}')
            elif message.__contains__("END"):
                receiver.count_time_in_seconds += (datetime.now() - receiver.start).total_seconds()
                receiver.start = datetime.now()
                print(receiver)
            else:
                receiver.count_bytes_from_client += len(message)
            #print(f"Message from Server: ")
            time.sleep(0)
    except ConnectionResetError:
        print(receiver)
        print("Socket was closed due to some unknown reasons. Sorry. :(")

测试异常情况 #

上述模块中的静态方法均作为线程方法使用。根据常识，UDP传输速度应该远快于TCP，但实际测试结果完全相反：无论本地运行还是将服务端部署在Docker容器（客户端连接Docker网关地址，如172.16.0.1），UDP吞吐量均低于TCP。

服务端输出的测试数据如下：

TCP: 2.3kB per 15.004 sec
   UDP: 1.5kB per 15.009 sec

显然UDP吞吐量远低于TCP，请问这是什么原因？我哪里出错了？恳请各位提供建议。

内容的提问来源于stack exchange，提问作者Hubertius