Python套接字未从C++ Boost asio接收所有数据

Question

简而言之

我正在尝试从 boost::asio 服务器(嵌入式系统)发送编码和压缩的图像数据(~300kB)，并通过 python 套接字客户端(笔记本电脑)接收它。客户端似乎每次只接收 65532 字节，而客户端发送大约 350000 字节。

我的想法是，也许某些东西将 TCP 数据包大小(boost::asio 或 python socket 或其他)限制在 65535 字节左右。

协议(protocol)

1. 客户端发送READY ，长度为5的字符串，发送至服务器。
2. 服务器接收READY来自客户。
3. 服务器传输length消息的 4 字节 uint ，发送给客户端(长度仅为消息的长度；即不包括 length 本身)。
4. 服务器将消息传输给客户端。
5. 客户端收到length以及来自服务器的消息。

描述

服务器将创建一个新的 img_session建立新连接时对象。当读取事件发生时，handle_read()函数将被调用。 img_stream是 zlib 压缩和 base64 编码 std::string通过img_lock“同步” .

客户调用run_image()一旦在一个线程中。该方法将首先发送 READY发出信号并等待服务器响应。服务器的第一个响应将是消息其余部分的长度。然后它会不断接收消息的其余部分并将其保存在变量message中。同时，并发线程将解码和解压缩消息并将图像保存在变量 frame 中。 .

代码

服务器代码 ( img_session.h )

#ifndef _IMG_SESSION_H_
#define _IMG_SESSION_H_

#include <session.h>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <iostream>
#include <utility.h>
#include <boost/thread.hpp>

class img_session : public session
{
public:
    img_session(boost::asio::io_service& io_service, std::string * img_stream, int*  img_lock);
    ~img_session();

    /* Starts the session. This is called in the constructor.*/
    void start();

private:
    void handle_read(const boost::system::error_code& error, size_t bytes_transferred);
    void handle_write(const boost::system::error_code& error);

    std::string * img_stream;
    int * img_lock;
};

#endif /* _IMG_SESSION_H */

服务器代码 ( img_session.cpp )

#include "img_session.h"
#include <session.h>

/* Public Methods */
img_session:: img_session(boost::asio::io_service& io_service, std::string * img_stream, int * img_lock) : session(io_service), img_stream(img_stream), img_lock(img_lock){
}

img_session::~img_session() { }

void img_session::start()
{
    std::cout << "connection from " << _socket.remote_endpoint() << "\n";
    std::cout << "img_session waiting for READY signal..." << std::endl;
    _socket.async_read_some(boost::asio::buffer(_data, max_length),
        boost::bind(&img_session::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
}

/* Private Methods */

void img_session::handle_read(const boost::system::error_code& error, size_t bytes_transferred)
{
    if (!error)
    {
        char otherString[6];
        std::cout << "[INFO ] IMG READ" << std::endl;
        if (strncmp(_data, "READY", 5) == 0) {
            std::cout << "[INFO ] READY signal received..." << std::endl;
            while (*img_lock == 1) boost::this_thread::yield();
            *img_lock = 1;
            std::string message = std::string(*img_stream);
            unsigned int len = message.length();

            std::cout << "len = " << len << std::endl;
            unsigned char bytes[4];

            bytes[0] = (len >> 24) & 0xFF;
            bytes[1] = (len >> 16) & 0xFF;
            bytes[2] = (len >> 8) & 0xFF;
            bytes[3] = len & 0xFF;
            message = std::string((char* ) bytes, 4) + message;
            std::cout << "img_session.cpp: bytes[] = " << (int) bytes[0] << " " << (int)bytes[1] << " " << (int)bytes[2] << " " << (int)bytes[3] << std::endl;
            std::cout << "img_session.cpp: message.length() = " << message.length() << std::endl;
            std::cout << "img_session.cpp: HEAD: " << message.substr(0, 1024) << std::endl;
            std::cout << "img_session.cpp: TAIL: " << message.substr(message.length() - 1024, message.length() - 1);


            boost::asio::async_write(_socket,
                boost::asio::buffer(message.c_str(), message.length()), 
                boost::bind(&img_session::handle_write, this,
                    boost::asio::placeholders::error));

            *img_lock = 0;
        }
        else {
            std::cout << "[INFO ] READY signal not received..." << std::endl;
            boost::asio::async_write(_socket,
                boost::asio::buffer("", 1),
                boost::bind(&img_session::handle_write, this,
                    boost::asio::placeholders::error));
        }
        clean();
    }
    else
    {
        delete this;
    }
}

void img_session::handle_write(const boost::system::error_code& error)
{
    if (!error)
    {

        std::cout << "img_session waiting for READY signal..." << std::endl;
        _socket.async_read_some(boost::asio::buffer(_data, max_length),
            boost::bind(&img_session::handle_read, this,
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        delete this;
    }
}

客户端代码 ( controller.py )

...

def run_image(hostname, port, interrupt):
    global frame
    sck = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sck.connect((hostname, port))
    sck.settimeout(100)
    message = None
    while(interrupt.is_set()):
        buffer = message
        t = threading.Thread(target=decodeImg, args=(buffer,))
        t.start()
        answer = b" "
        message = b""
        sck.send(b"READY")
        print("Sent READY signal")
        answer = sck.recv(4)
        num = int(answer[0] << 24) + int(answer[1] << 16) + int(answer[2] << 8) + int(answer[3])
        print("num:", num)

        try:
            for i in range(math.ceil(num/1024)):
                print("Receiving... (", i,"of",math.ceil(num/1024),")")
                answer = sck.recv(1024) # row size

                if (i == 0):
                    print("#### HEAD ####")
                print(answer)
                message += answer
        except socket.timeout:
            pass
        print("#### TAIL ####")
        print(answer)
        print("###########################################")
        print("##### Received! message (len=", len(message),") #####")
        print("###########################################")
        t.join()
        #print("Updated")
    sck.close()
    print("run_image exited")

decompress = zlib.decompress
decode = base64.b64decode
toframe = pygame.image.fromstring

def decodeImg(raw):
    if (raw is None or raw == ""):
        return
    global frame
    print("Message length     :", len(raw))
    raw = decode(raw)
    print("Decoded length     :", len(raw))
    raw = decompress(raw)
    print("Decompressed length:", len(raw))
    frame = toframe(raw, image_size, 'RGB')
    if (frame is None):
        print("ERR None frame")

...

(完整客户端代码位于 https://pastebin.com/ut6L70XM )

响应

服务器日志

...

connection from 127.0.0.1:61616
img_session waiting for READY signal...
[DEBUG] server.h: instance of img_session found
[INFO ] IMG READ
[INFO ] READY signal received...
[INFO ] img_stream.length() = 349544
len = 349544
img_session.cpp: bytes[] = 0 5 85 104
img_session.cpp: message.length() = 349548
img_session.cpp: HEAD:  UheJy0vYGZpLgSrIsJMkEmyARMwASZIBNkAiZgQpnQJowJY8Ix4ZUyI
jITqnrPnnvv62V7KBBCUP3xKyJTYtvKdv8p6Td+2u1jeR5gP/W5oaSVtVRuKh8FvhyTV7095VkDz1u+V
ZVrbR8nq79cxfbtFI/Pv5/stmcvpbDJVc0sreC8a1fZarPf7722mddXuBeNLPZTsRfFCiusqrzVdw3vO
1wqjl2rONf6VO0O1Lq1dSCqKmulbLbwp64jqhVchzaUuS3r8Lax2rLpRJWfV2MKm6fL8SuIC6z2x5CLq
VX5o58xrhTNbjjpvu4edjWW8Yt6X8Yqs5ZNK9iCu7eXWm1LteW9/X0D/ZCKG871OByLGmC3ay3vjU3F3
rsaitmdbFar3dL1VduS7nBlDWt7LXu1b8m3V9SwVvZW2/7+3Nr+/r+uXY01vy9lLVZmLdu7YFsre92Hr
b+XrRylHK0eRz1aO+p7qX2vx96wvNd7q/1o63fVeqmjvbfYxrbKj2NtGdV+71qOtKy9+9jXmUdpo+/js
OW9sar8e1dL6+O92Mpuh7dbnR1lViVt/Y6Tro39vYy9t/1drOvjKvn+2HdbWh+rGdj7bsYqbCddB2L7f
vTdVt4fsfKuoR8dy7vCVc8x5rGaus57jHasSxv9XWx9fP9+r8+DW+b7nH28D+32+73MPrE+bOO0LcOW9
8f35mOtzPf2MdZGlZ/Vtmt5l8PhOHDiqLXBtpza+F7KmPvgRhRGtUOHHKp2tyYNtootPHBdOOMx38t5j
HqM4te789rfy/sGHrxXuF3vjb2tjbh16+69b/tm6/sqj9u4bjWL2eGo7X2WceC8645V+43t3bbbxzn89
rK221XUdRPsRg0rOXiXum3B3XivzHHyhhxz1dPfV3mui8WucU5bxvsujmvYlvf6NVeBc66P/b1MFFjHv
rfs65avwnNOHD77+yzn+irnuXcWu/p52oroVjLmHjR5bKm3YjcA/bKdp/hCKVX4laoJrL9C7yuUP8vc1
1lz/Ydj
img_session.cpp: TAIL: sC5fp569IQ/fnp9NsY4EfMdjx8cR77ftC8426c2vPM27KFV/q+Crrf1tH
/fnYt+KoafkL+S9PaSs96I4qAmH7fHK/p2qcY6IFrOdR14uAs87ssSvGTw+rx/ib0ci+kXMpRGd2D2Sx
cHWl4rCb42a3MHfFyUzlPUTwhm5rA+DWF1Z6JPUtpwUqX90KiWpYMzOer1mRRv5K1yOUPHB36F/H6PjY
Hw4NkqwQb0WHrJn2ws85PB3owhbaZ4iECNkp2D0aylzFBMLVOK4USwzpKFWQZpFXd2lOm8eicRyLAuKL
y3Gou2FvyGELfezceNR8I31QOXroJiuOOH9OMhSpxVgmPGlQ4XMXw5by49xfWOSrSIlVVOoBlUxAnp9u
xwSTPras50W+LMZAE6iSe26qbDbAPcyfeyQpIQ+91BTfkKdYbcaJBWH35RQ8qVLKKuoMaAvQlnIBms4l
XNeiFJesperkO000isv5NQ48FkQZ1Ua/w7AqRI6M1EEGuc4sSw7GdFmibbpNMGUnyjCxtkGQzjnaouUr
2o+ajPq3JSM2pt03zroAyzDs1axEu05eHWd5WDo5ZC0rvEd8RR496IsuxuzWCyeNszWWM9lQOGMAmtI4
Dm1I/egQrD6knzxq3sY2oXHNbJk1i5At0YpbLrOntb1AgHKcuyxZ/pXBvIAMpDac/LVXtNhXvA6Hel2Z
PvwOBj2GaXgpt7Evaj/Vm6WeePzRrQ+nQfzYDPUHkr5/B0rGvgQKrgi/Hz+ehT4zLPgu9GCHvmNIYStZ
fC7L7luR1NVhtBwKmSuQNip9DjfAe9/0a0k2fkx0Vko4HKijDkh62YmXF5YTBgG3jSgmRv/qvbeIZwP/
NsG+HIdYWGoqmxX0C5PamF936VxCd0bYz8rtjnf1v35pj8soTpHasDdfTFsvGuuRdMtFLycccHIVgK3I
gX9UnpZIjQnaeD2HpK2vubAyVc8CrIXAzdemYjOXHTo3EeI3/7RCR3yDA7qDsZIkNWj7vKl84XsQeH/A
X1TLNQ=

...

客户日志

Sent READY signal
num: 349544
Receiving... ( 0 of 342 )
#### HEAD ####
b'eJy0vYGZpLgSrIsJMkEmyARMwASZIBNkAiZgQpnQJowJY8Ix4ZUyIjITqnrPnnvv62V7KBBCUP3xKyJTYtvKdv8p6Td+2u1jeR5gP/W5oaSVtVRuKh8FvhyTV7095VkDz1u+VZVrbR8nq79cxfbtFI/Pv5/stmcvpbDJVc0sreC8a1fZarPf7722mddXuBeNLPZTsRfFCiusqrzVdw3vO1wqjl2rONf6VO0O1Lq1dSCqKmulbLbwp64jqhVchzaUuS3r8Lax2rLpRJWfV2MKm6fL8SuIC6z2x5CLqVX5o58xrhTNbjjpvu4edjWW8Yt6X8Yqs5ZNK9iCu7eXWm1LteW9/X0D/ZCKG871OByLGmC3ay3vjU3F3rsaitmdbFar3dL1VduS7nBlDWt7LXu1b8m3V9SwVvZW2/7+3Nr+/r+uXY01vy9lLVZmLdu7YFsre92Hrb+XrRylHK0eRz1aO+p7qX2vx96wvNd7q/1o63fVeqmjvbfYxrbKj2NtGdV+71qOtKy9+9jXmUdpo+/jsOW9sar8e1dL6+O92Mpuh7dbnR1lViVt/Y6Tro39vYy9t/1drOvjKvn+2HdbWh+rGdj7bsYqbCddB2L7fvTdVt4fsfKuoR8dy7vCVc8x5rGaus57jHasSxv9XWx9fP9+r8+DW+b7nH28D+32+73MPrE+bOO0LcOW98f35mOtzPf2MdZGlZ/Vtmt5l8PhOHDiqLXBtpza+F7KmPvgRhRGtUOHHKp2tyYNtootPHBdOOMx38t5jHqM4te789rfy/sGHrxXuF3vjb2tjbh16+69b/tm6/sqj9u4bjWL2eGo7X2WceC8645V+43t3bbbxzn89rK221XUdRPsRg0rOXiXum3B3XivzHHyhhxz1dPfV3mui8WucU5bxvsujmvYlvf6NVeBc66P/b1MFFjHvrfs65avwnNOHD77+yzn+irnuXcWu/p52oroVjLmHjR5bKm3YjcA/bKdp/hCKVX4laoJrL9C7yuUP8vc11lz/Ydjfr/M'

... 

Receiving... ( 63 of 342 )
b'JGPgXARwxzUdiKKMzTRc4OaHzidOcNJrBQHbD+M6nZ6PKBrUlsRiq9KhIQY/JTOVqSd5ufKNbatfImARAtARU6pbAa2mh0YHVQrR04zzmEVA5Pnrc4Gt4kUiaWzVqu/Nv7tWw63102hqqZHgKyrXxsBq8emZeBQCriWU18xHfJZ4Fwq6+8cs0rMdIeCKitrmHTO6gFcZ1W3Q1zTDLQ6rbpMNOhSTBjIQrOwv+7xLiykV3pxd4SpDFC+JxcWaX5mUXWVBI7PLjG5EkKM0GAHgkfm8ynnGLITQs97JCiW/KwLEc+ZVW0IOrwJxiw3ZDnr4z7kRtOnjoX9ZiDSCvgwHP6mCn09UB+9AsCVnhSlt/TpQo0T+ShQnmXDwNpXLPSAQFCxT+6CDTYROLxKRWI6eCQlz57ipCo7OrIuZYFLQdzaATzI62KStlL8tv/CWbwmLUdsnul0T5hyyB/h7nFeiodJy7E5rStiJfCJXxwLl7zv2y/cDI0Apa37CfmjK4SO9ns7qcPgeoYt9zZcGLbeMULcaNMAYL3653y4Wy3ZO2dHKiPv6v8vg/Z+j7yDSy8IWHuC4bYB35eHfEei08vZDUpexFVfB8HEsyDJSIN4/7/uINS/0ptkjI+vi8DJiyijSO/xHfvInih/DS9aEvBv99pLhk7+RV/5GFD0pgcQPBYm0nwmLgE0PG9b/h8LQ+FvlPwf7Sp346Adq5nN6/nNXrhSN2RmpNfDXeUvX3mqRz/3DA6TnvZ16VGOCEgrag9DZHG7hZAZwE3brJ+Wl+BLRRMzvGpNq2mHqGU0cmeSdfJyAvciWWd1XQ2J2/JlGMu4AN7VfjqRiE9ctjQ2kp1sl2npXPKIqbhsDJ30cM4E7yOjIjjNs8yOOIZfb859t/y5sk2GueLHypprjHk2nlfzcKbQrQ95qdVVVOIyuGqz89XqiCqAui8K+CBP3pgjyhGwo9MXSqyCZrf8zrWtMaTQhGPJ5EWcfaf5BK/hlEZJiwTEX0oNuNFUw/Ge50BvJ'
Receiving... ( 64 of 342 )
b''
Receiving... ( 65 of 342 )
b''

...

Receiving... ( 340 of 342 )
b''
Receiving... ( 341 of 342 )
b''
#### TAIL ####
b''
###########################################
##### Received! message (len= 65532 ) #####
###########################################
Message length     : 65532
Decoded length     : 49149

Exception in thread Thread-4:
Traceback (most recent call last):
  File "C:\Python35\lib\threading.py", line 914, in _bootstrap_inner
    self.run()
  File "C:\Python35\lib\threading.py", line 862, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Python35\controller.py", line 204, in decodeImg
    raw = decompress(raw)
zlib.error: Error -5 while decompressing data: incomplete or truncated stream

Answer 1

您正在从局部变量发送数据，但在写入完成之前它超出了范围。

将message设为成员变量。

NOTE: The img_lock is not thread safe

这是一个工作演示:

Live On Coliru

#ifndef _IMG_SESSION_H_
#define _IMG_SESSION_H_

#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <iostream>
#include <boost/thread.hpp>

struct session {
    session(boost::asio::io_service& svc) : svc(svc) {
    }

    void clean() {}
    boost::asio::io_service& svc;
    boost::asio::ip::tcp::socket _socket{svc};
    int max_length = 1024;
    //std::array<char, 1024> _data;
    char _data[1024];
};

class img_session : public session
{
public:
    img_session(boost::asio::io_service& io_service, std::string * img_stream, int*  img_lock);
    ~img_session();

    /* Starts the session. This is called in the constructor.*/
    void start();

private:
    void handle_read(const boost::system::error_code& error, size_t bytes_transferred);
    void handle_write(const boost::system::error_code& error, size_t bytes_transferred);

    std::string * img_stream;
    int * img_lock;
    std::string message;
};

#endif /* _IMG_SESSION_H */

/* Public Methods */
img_session:: img_session(boost::asio::io_service& io_service, std::string * img_stream, int * img_lock) : session(io_service), img_stream(img_stream), img_lock(img_lock){
}

img_session::~img_session() { }

void img_session::start()
{
    std::cout << "connection from " << _socket.remote_endpoint() << "\n";
    std::cout << "img_session waiting for READY signal..." << std::endl;
    _socket.async_read_some(boost::asio::buffer(_data, max_length),
        boost::bind(&img_session::handle_read, this,
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
}

/* Private Methods */

void img_session::handle_read(const boost::system::error_code& error, size_t /*bytes_transferred*/)
{
    if (!error)
    {
        //char otherString[6];
        std::cout << "[INFO ] IMG READ" << std::endl;
        if (strncmp(_data, "READY", 5) == 0) {
            std::cout << "[INFO ] READY signal received..." << std::endl;
            while (*img_lock == 1) boost::this_thread::yield();
            *img_lock = 1;
            message = "0000" + *img_stream;
            size_t len = message.length();

            std::cout << "len = " << len << std::endl;

            message[0] = (len >> 24) & 0xFF;
            message[1] = (len >> 16) & 0xFF;
            message[2] = (len >> 8) & 0xFF;
            message[3] = len & 0xFF;
            std::cout << "img_session.cpp: bytes[] = " << (int) message[0] << " " << (int)message[1] << " " << (int)message[2] << " " << (int)message[3] << std::endl;
            std::cout << "img_session.cpp: message.length() = " << message.length() << std::endl;
            std::cout << "img_session.cpp: HEAD: " << message.substr(4, 1024) << std::endl;
            std::cout << "img_session.cpp: TAIL: " << message.substr(message.length() - 1024, message.length() - 1);


            boost::asio::async_write(_socket,
                boost::asio::buffer(message), 
                boost::bind(&img_session::handle_write, this,
                    boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));

            *img_lock = 0;
        }
        else {
            std::cout << "[INFO ] READY signal not received..." << std::endl;
            /*
             *boost::asio::async_write(_socket,
             *    boost::asio::buffer("", 1),
             *    boost::bind(&img_session::handle_write, this,
             *        boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
             */
        }
        clean();
    }
    else
    {
        delete this;
    }
}

void img_session::handle_write(const boost::system::error_code& error, size_t bytes_transferred)
{
    if (!error)
    {
        std::cout << "img_session waiting for READY signal..." << std::endl;
        _socket.async_read_some(boost::asio::buffer(_data, max_length),
            boost::bind(&img_session::handle_read, this,
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
    }
    else
    {
        std::cout << __FUNCTION__ << ":" << error.message() << "\n";
        delete this;
    }
}

int main() {
    boost::asio::io_service svc;
    std::string img_stream(300<<10, '*');
    int img_lock = 0;
    auto s = new img_session(svc, &img_stream, &img_lock);

    using boost::asio::ip::tcp;

    tcp::acceptor a(svc);
    a.open(tcp::v4());
    a.set_option(tcp::acceptor::reuse_address(true));
    a.bind({{}, 6767});
    a.listen(1);
    a.accept(s->_socket);

    s->start();

    svc.run();
}

使用客户端时的情况

netcat localhost 6767 | wc 
READY

打印

  0       1  307204