TCP Socket跨网传输JPG字节丢失致图片损坏问题求助

我搭建了一套简单的客户端-服务器架构用于传输JPG字节，本地运行完全正常，但跨互联网传输时JPG文件损坏，解码后出现严重视觉伪影。

客户端（Rust Tokio实现） #

客户端从摄像头获取JPEG流并将JPG字节推送到TCP Socket：

// Async application to run on the edge (Raspberry Pi) 
// Reads MJPEG HTTP stream from provided URL and sends it to the CWS server over TCP

#![warn(rust_2018_idioms)] 

use std::convert::TryFrom;

use futures::StreamExt;

use tokio::signal;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpStream;

use std::error::Error;

use std::path::PathBuf;
use structopt::StructOpt;

//DEBUG
use std::fs;

#[derive(StructOpt, Debug)]
#[structopt(name = "basic")]

struct Opt {
    #[structopt(long = "stream", required(true))]
    stream: String,

    #[structopt(long = "server", required(true))]
    server: String,
}

async fn acquire_tcp_connection(server: &str) -> Result<TcpStream, Box<dyn Error>> {
    // "127.0.0.1:6142"
    loop {
        match TcpStream::connect(server).await {
            Ok(stream) => {
                println!("Connected to server");
                return Ok(stream);
            }
            Err(e) => {
                println!("Failed to connect to server: {}", e);
                tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
            }
        }
    }

}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let opt = Opt::from_args();
    
    let url = http::Uri::try_from(opt.stream).unwrap();

    loop {
        let mut tcp_conn = acquire_tcp_connection(&opt.server).await?;

        // hyper client
        let client = hyper::Client::new();
        // Do the request
        let res = client.get(url.clone()).await.unwrap();
        // Check the status
        if !res.status().is_success() {
            eprintln!("HTTP request failed with status {}", res.status());
            std::process::exit(1);
        }
        // Basically HTTP response content 
        let content_type: mime::Mime = res
            .headers()
            .get(http::header::CONTENT_TYPE)
            .unwrap()
            .to_str()
            .unwrap()
            .parse()
            .unwrap();

        assert_eq!(content_type.type_(), "multipart");

        let boundary = content_type.get_param(mime::BOUNDARY).unwrap();
        let stream = res.into_body();
        let mut stream = multipart_stream::parse(stream, boundary.as_str());

    
        'outer: while let Some(p) = stream.next().await {
            let p = p.unwrap();
            // Split the jpeg bytes into chunks of 2048 bytes
            for slice in p.body.chunks(2048) {
                // DEBUG capture bytes to a file just for debugging
                fs::write("tcp_debug.txt", &slice).expect("Unable to write file"); 

                let tcp_result = tcp_conn.write(&slice).await;

                match tcp_result {
                    Ok(_) => {
                        println!("Sent {} bytes", slice.len());
                    }
                    Err(e) => {
                        println!("Failed to send data: {}", e);
                        break 'outer;
                    }
                }
            }
        }
    }

    Ok(())
}

服务器（Python实现） #

服务器负责接收TCP连接并解码收到的字节：

import asyncio
from threading import Thread

import cv2
import numpy as np


class SingleFrameReader:
    """
    Simple API for reading a single frame from a video source
    """
    def __init__(self, video_source, ..., tcp=False):
        self._video_source = video_source
        ...
        elif tcp:
            self._stop_tcp_server = False
            self._tcp_image = None
            self._tcp_addr = video_source
            self._tcp_thread = Thread(target=asyncio.run, args=(self._start_tcp_server(),)).start()
            self.read = lambda: self._tcp_image

    async def _start_tcp_server(self):
        uri, port = self._tcp_addr.split(':')
        port = int(port)
        print(f"Starting TCP server on {uri}:{port}")
        server = await asyncio.start_server(
            self.handle_client, uri, port
        )

        addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
        print(f'Serving on {addrs}')

        async with server:
            await server.serve_forever()

    async def handle_client(self, reader, writer):
        """
        Handle a client connection. Receive JPEGs from the client and have them ready to be ready
        """
        client_addr = writer.get_extra_info('peername')
        print(f"New connection from {client_addr}")
        jpg_bytes = b''
        while True:
            if self._stop_tcp_server:
                break
            data = await reader.read(2**16)

            #DEBUG
            print(f"Received {len(data)} bytes from {client_addr}")
            byes_file = open('tcp_bytes_server.txt', 'wb')

            if not data:
                print("Client disconnected")
                break

            #DEBUG the corruption issue
            byes_file.write(data)

            jpg_bytes += data
            
            start_idx = jpg_bytes.find(b'\xff\xd8')
            end_idx = jpg_bytes.find(b'\xff\xd9')

            if start_idx != -1 and end_idx != -1:
                nparr  = np.frombuffer(jpg_bytes[start_idx:end_idx+2], np.uint8)
                img_np = cv2.imdecode(nparr, cv2.IMREAD_COLOR)
                if img_np is None:
                    continue
                self._tcp_image = img_np
                jpg_bytes = jpg_bytes[end_idx+2:]
                
    ...

已尝试的排查步骤 #

• 调整不同分片大小，一次性发送整张图片时伪影更严重
• 对比客户端发送和服务器接收的字节文件：服务器接收的字节约1.6KB，远小于客户端发送的2KB，且开头字节不匹配
• 本地运行时可实时流畅显示摄像头流，仅跨互联网传输时出现字节损坏

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