WinError 10054：TCP Socket大文件传输异常与效率优化问询

问题描述 #

我正在开发一款基于TCP Socket的文件传输程序，读取文件字节后通过TCP套接字发送至另一台计算机。该程序在传输1~10Kb的小文件时完全正常，但传输大文件时会触发错误：

WinError 10054 An existing connection was forcibly closed by the remote host

添加time.sleep(0.00001)后可正常传输，但传输单张7MB图片的耗时骤增至2分钟（采用500字节数据包）。我想了解是否是协议存在速度限制导致此问题？
目前我使用3个线程分别以500/1000/1500字节数据包传输同一文件，测试不同数据包大小的传输效率。

代码实现 #

import socket, time, threading
from struct import *

def sending(host, port, path_to_file, package_size):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((host, port))
    send_log = open(f"./Logs/send_log_{package_size}.txt", "w")

    # 检查文件是否存在
    try:
        fp = open(path_to_file, "rb")
    except FileNotFoundError:
        sock.close()
        print("文件未找到")
        return

    sock.send((path_to_file.split(".")[-1]).encode())

    fp = open(path_to_file, 'rb')
    start_time = time.time()
    total_bytes = 0
    package_number = 1
    while True:
        msg = fp.read(package_size - 4)
        if len(msg) < package_size - 4:
            final_result = pack(f'I{len(msg)}s', package_number, msg)
            sock.send(final_result)
            total_bytes += len(final_result)
            send_log.write(f'第{package_number}个数据包已发送，大小{len(final_result)}:\n{final_result}\n')
            break
        final_result = pack(f'I{package_size - 4}s', package_number, msg)
        total_bytes += len(final_result)
        send_log.write(f'第{package_number}个数据包已发送，大小{len(final_result)}:\n{final_result}\n')
        sock.send(final_result)
        package_number += 1
    time.sleep(0.0000000000001)
    end_time = time.time()
    sock.close()
    print(f"使用{package_size}字节数据包耗时: {end_time - start_time}s，总发送字节数: {total_bytes}")

    return

def receive(host, port, package_size):

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind((host, port))
    sock.listen(1)
    conn, addr = sock.accept()
    print("已与" + str(addr) + "建立连接")

    extension = conn.recv(5).decode()

    log_file = open(f"./Logs/receive_log_{package_size}.txt", "w")
    fp = open(f'./Files/received_file_{package_size}.{extension}', 'wb')
    total_bytes = 0
    start_time = time.time()
    while True:
        received_package = conn.recv(package_size)
        if len(received_package) < package_size:
            package_number, msg = unpack(f'I{len(received_package) - 4}s', received_package)
            total_bytes += len(msg) + 4
            log_file.write(f"收到第{package_number}个数据包，大小{len(received_package)}:\n{received_package}\n")
            fp.write(msg)
            break
        package_number, msg = unpack(f'I{package_size-4}s', received_package)
        total_bytes += len(msg) + 4
        log_file.write(f"收到第{package_number}个数据包，大小{len(received_package)}:\n{received_package}\n")
        fp.write(msg)


        if not received_package:
            break
    fp.close()
    time.sleep(0.00001)
    log_file.close()
    end_time = time.time()
    print(f"使用{package_size}字节数据包耗时: {end_time - start_time}s，总接收字节数: {total_bytes}")
    sock.close()
    return


def main():

    while True:

        try:
            host = input("输入主机地址: ")
            socket.gethostbyname(host)
            break
        except socket.gaierror:
            print("无效主机地址，请重试")

    port = int(input("输入端口号: "))

    print("将使用以下端口传输文件:")

    for package_size in range(3):

        print(port+package_size)

    while True:

        mode = input("选择模式（s发送/r接收）: ")

        if mode == "s":

            path_to_transfer = input("输入文件路径: ")
            threading.Thread(target=sending, args=(host, port, path_to_transfer, 500)).start()
            threading.Thread(target=sending, args=(host, port+1, path_to_transfer, 1000)).start()
            threading.Thread(target=sending, args=(host, port+2, path_to_transfer, 1500)).start()
            break

        elif mode == "r":
            
            threading.Thread(target=receive, args=(host, port, 500)).start()
            threading.Thread(target=receive, args=(host, port+1, 1000)).start()
            threading.Thread(target=receive, args=(host, port+2, 1500)).start()
            break

        else:

            print("无效模式，请重试")
            continue

    time.sleep(10)



if __name__ == "__main__":
    main()

问题原因与优化建议 #

1. WinError 10054的根本原因 #

这个错误并非TCP协议本身的速度限制导致，而是发送端发送速度远超接收端处理速度，导致接收端TCP缓冲区溢出，操作系统主动关闭连接。

当你不断调用sock.send()发送数据时，TCP会将数据放入发送缓冲区，由系统内核负责发送。但如果接收端的应用层没有及时调用recv()读取数据，接收端的接收缓冲区会被填满，此时发送端的TCP窗口会缩小，最终导致发送端的发送缓冲区也被填满。如果发送端继续强制发送，操作系统就会强制关闭连接，抛出10054错误。

添加time.sleep()后，发送速度被人为降低，给了接收端足够的时间处理缓冲区中的数据，因此不会触发错误，但代价是传输效率暴跌。

2. 代码中的关键问题 #

• 接收端逻辑缺陷：TCP是流式协议，recv()返回的字节数可能小于请求的大小，尤其是大文件传输时。当前代码假设每次recv()都能完整收到一个package_size大小的数据包，中间出现不完整接收时会直接解析错误，进而引发连接异常。
• 缺乏流量控制：发送端没有根据接收端的反馈调整发送速度，无节制发送容易导致缓冲区溢出。
• 数据包大小不合理：500字节的数据包太小，TCP+IP头部（共40字节）的开销占比过高，有效传输效率低下。

3. 优化方案 #

（1）修复接收端流式数据处理

TCP是字节流，不能假设每次recv()都能拿到完整数据包。需要先读取固定长度的包头，再根据包头信息读取对应长度的载荷：

# 接收端改进后的循环逻辑
while True:
    # 先读取4字节的包序号
    header = b''
    while len(header) < 4:
        chunk = conn.recv(4 - len(header))
        if not chunk:
            break
        header += chunk
    if not header:
        break
    package_number = unpack('I', header)[0]
    
    # 读取对应长度的载荷
    payload_len = package_size - 4
    payload = b''
    while len(payload) < payload_len:
        chunk = conn.recv(payload_len - len(payload))
        if not chunk:
            break
        payload += chunk
    if not payload:
        break
    
    total_bytes += 4 + len(payload)
    log_file.write(f"收到第{package_number}个数据包，大小{4+len(payload)}:\n{header+payload}\n")
    fp.write(payload)
    
    # 检查是否是最后一个包
    if len(payload) < package_size -4:
        break

（2）实现简单流量控制

接收端每处理完一定数量的数据包后，向发送端发送确认消息，发送端收到确认后再继续发送下一批，避免发送过快。

（3）调整数据包大小

将数据包大小设置为1460字节（以太网MTU 1500减去TCP和IP头部各20字节），最大化有效载荷占比，提升传输效率。

（4）使用sendall()替代send()

send()可能只发送部分数据，sendall()会确保所有数据都被发送到缓冲区，避免数据遗漏：

sock.sendall(final_result)

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