最终，我试图将缓冲区从一台机器传输到另一台机器。下面的代码采用 <id><size><data with size bytes> 的流并读取 handleReadHeader 函数中的部分，然后读取 <size>字节数，然后返回并等待另一个 <id><size>一对。
我粘贴了很多代码，但我真正怀疑的唯一功能是:
下行链路::addMsgToQueue
下行链路::writeCallback
下行链路::startWrites()
下行链路::handleReadHeader
下行链路::handleReadFrameDataBGR
using namespace std;
using namespace boost;
using namespace boost::asio;
Downlink::Downlink() :
socket(nIO),
headerSize(sizeof(unsigned int)+1),
connected(false),
isWriting(false),
readHeaderBuffer(headerSize)
{}
Downlink::~Downlink() {
disconnect();
}
bool Downlink::connect(const std::string &robotHost, unsigned int port) {
disconnect();
ip::tcp::resolver resolver(nIO);
ip::tcp::resolver::query query(robotHost, lexical_cast<string>(port));
ip::tcp::resolver::iterator iterator = resolver.resolve(query);
ip::tcp::resolver::iterator end;
boost::system::error_code ec;
for(;iterator!=end;++iterator) {
socket.connect(*iterator, ec);
if(!ec)
break;
socket.close();
}
if(!socket.is_open())
return false;
async_read(socket, buffer(readHeaderBuffer),
bind(&Downlink::handleReadHeader, this, _1, _2));
//start network thread.
lock_guard<mutex> l(msgMutex);
outgoingMessages = queue<vector<char> >();
nIO.reset();
t = thread(bind(&boost::asio::io_service::run, &nIO));
connected = true;
return true;
}
bool Downlink::isConnected() const {
return connected;
}
void Downlink::disconnect() {
nIO.stop();
t.join();
socket.close();
connected = false;
isWriting = false;
nIO.reset();
nIO.run();
}
void Downlink::writeToLogs(const std::string &logMsg) {
vector<char> newMsg(logMsg.length()+headerSize);
newMsg[0] = MSG_WRITE_LOG;
const unsigned int msgLen(logMsg.length());
memcpy(&newMsg[1], &msgLen, sizeof(unsigned int));
vector<char>::iterator dataBegin = newMsg.begin();
advance(dataBegin, headerSize);
copy(logMsg.begin(), logMsg.end(), dataBegin);
assert(newMsg.size()==(headerSize+logMsg.length()));
addMsgToQueue(newMsg);
}
void Downlink::addMsgToQueue(const std::vector<char> &newMsg) {
lock_guard<mutex> l(msgMutex);
outgoingMessages.push(newMsg);
lock_guard<mutex> l2(outMutex);
if(!isWriting) {
nIO.post(bind(&Downlink::startWrites, this));
}
}
void Downlink::writeCallback(const boost::system::error_code& error,
std::size_t bytes_transferred) {
if(error) {
disconnect();
lock_guard<mutex> l(msgMutex);
outgoingMessages = queue<vector<char> >();
return;
}
{
lock_guard<mutex> l2(outMutex);
isWriting = false;
}
startWrites();
}
void Downlink::startWrites() {
lock_guard<mutex> l(msgMutex);
lock_guard<mutex> l2(outMutex);
if(outgoingMessages.empty()) {
isWriting = false;
return;
}
if(!isWriting) {
currentOutgoing = outgoingMessages.front();
outgoingMessages.pop();
async_write(socket, buffer(currentOutgoing),
bind(&Downlink::writeCallback, this, _1, _2));
isWriting = true;
}
}
void Downlink::handleReadHeader(const boost::system::error_code& error,
std::size_t bytes_transferred) {
//TODO: how to handle disconnect on errors?
cout<<"handleReadHeader"<<endl;
if(error) {
return;
}
assert(bytes_transferred==headerSize);
if(bytes_transferred!=headerSize) {
cout<<"got "<<bytes_transferred<<" while waiting for a header."<<endl;
}
currentPacketID = readHeaderBuffer[0];
memcpy(¤tPacketLength, &readHeaderBuffer[1], sizeof(unsigned int));
dataStream.resize(currentPacketLength);
switch(currentPacketID) {
case MSG_FRAME_BGR: {
cout<<"- >> gone to read frame. ("<<currentPacketLength<<")"<<endl;
async_read(socket, asio::buffer(dataStream),
boost::asio::transfer_at_least(currentPacketLength),
bind(&Downlink::handleReadFrameDataBGR, this, _1, _2));
} break;
default: {
cout<<"->>> gone to read other. ("<<currentPacketLength<<")"<<endl;
cout<<" "<<(int)currentPacketID<<endl;
async_read(socket, asio::buffer(dataStream),
boost::asio::transfer_at_least(currentPacketLength),
bind(&Downlink::handleReadData, this, _1, _2));
} break;
}
}
void Downlink::handleReadData(const boost::system::error_code& error,
std::size_t bytes_transferred) {
cout<<"handleReadData"<<endl;
if(error) {
return;
}
if(bytes_transferred!=currentPacketLength) {
cout<<"Got "<<bytes_transferred<<" wanted "<<currentPacketLength<<endl;
}
assert(bytes_transferred==currentPacketLength);
switch(currentPacketID) {
case MSG_ASCII: {
string msg(dataStream.begin(), dataStream.end());
textCallback(&msg);
} break;
case MSG_IMU: {
Eigen::Vector3d a,g,m;
unsigned int stamp;
memcpy(a.data(), &dataStream[0], sizeof(double)*3);
memcpy(m.data(), &dataStream[0]+sizeof(double)*3, sizeof(double)*3);
memcpy(g.data(), &dataStream[0]+sizeof(double)*6, sizeof(double)*3);
memcpy(&stamp, &dataStream[0]+sizeof(double)*9, sizeof(unsigned int));
imuCallback(a,m,g,stamp);
} break;
default:
//TODO: handle this better?
cout<<"Unknown packet ID."<<endl;
}
async_read(socket, buffer(readHeaderBuffer),
boost::asio::transfer_at_least(headerSize),
bind(&Downlink::handleReadHeader, this, _1, _2));
}
void Downlink::handleReadFrameDataBGR(const boost::system::error_code& error,
std::size_t bytes_transferred) {
cout<<"Got a frame"<<endl;
if(error) {
return;
}
if(bytes_transferred!=currentPacketLength) {
cout<<"Got "<<bytes_transferred<<" wanted "<<currentPacketLength<<endl;
}
assert(bytes_transferred==currentPacketLength);
unsigned int imageWidth, imageHeight, cameraID;
unsigned char *readOffset = (unsigned char*)&dataStream[0];
memcpy(&imageWidth, readOffset, sizeof(unsigned int));
readOffset += sizeof(unsigned int);
memcpy(&imageHeight, readOffset, sizeof(unsigned int));
readOffset += sizeof(unsigned int);
memcpy(&cameraID, readOffset, sizeof(unsigned int));
readOffset += sizeof(unsigned int);
cout<<"("<<imageWidth<<"x"<<imageHeight<<") ID = "<<cameraID<<endl;
frameCallback(readOffset, imageWidth, imageHeight, cameraID);
async_read(socket, buffer(readHeaderBuffer),
boost::asio::transfer_at_least(headerSize),
bind(&Downlink::handleReadHeader, this, _1, _2));
}
boost::signals2::connection Downlink::connectTextDataCallback(boost::signals2::signal<void (std::string *)>::slot_type s) {
return textCallback.connect(s);
}
boost::signals2::connection Downlink::connectIMUDataCallback(boost::signals2::signal<void (Eigen::Vector3d, Eigen::Vector3d, Eigen::Vector3d, unsigned int)>::slot_type s) {
return imuCallback.connect(s);
}
boost::signals2::connection Downlink::connectVideoFrameCallback(boost::signals2::signal<void (unsigned char *, unsigned int, unsigned int, unsigned int)>::slot_type s) {
return frameCallback.connect(s);
}
这是另一端的代码。它几乎与其他代码完全相同，但错误可能在两端。
using namespace std;
using namespace boost;
using namespace boost::asio;
Uplink::Uplink(unsigned int port) :
socket(nIO),
acceptor(nIO),
endpoint(ip::tcp::v4(), port),
headerSize(sizeof(unsigned int)+1), //id + data size
headerBuffer(headerSize)
{
//move socket into accept state.
acceptor.open(endpoint.protocol());
acceptor.set_option(ip::tcp::acceptor::reuse_address(true));
acceptor.bind(endpoint);
acceptor.listen(1); //1 means only one client in connect queue.
acceptor.async_accept(socket, bind(&Uplink::accept_handler, this, _1));
//start network thread.
nIO.reset();
t = thread(boost::bind(&boost::asio::io_service::run, &nIO));
}
Uplink::~Uplink() {
nIO.stop(); //tell the network thread to stop.
t.join(); //wait for the network thread to stop.
acceptor.close(); //close listen port.
socket.close(); //close active connections.
nIO.reset();
nIO.run(); //let clients know that we're disconnecting.
}
void Uplink::parse_header(const boost::system::error_code& error,
std::size_t bytes_transferred) {
if(error || bytes_transferred!=headerSize) {
disconnect();
return;
}
currentPacketID = headerBuffer[0];
memcpy(¤tPacketLength, &headerBuffer[1], sizeof(unsigned int));
//move to read data state
//TODO: move to different states to parse various packet types.
async_read(socket, asio::buffer(dataStream), transfer_at_least(currentPacketLength),
bind(&Uplink::parse_data, this, _1, _2));
}
void Uplink::parse_data(const boost::system::error_code& error,
std::size_t bytes_transferred) {
if(error) {
disconnect();
return;
}
if(bytes_transferred != currentPacketLength) {
cout<<"bytes_transferred != currentPacketLength"<<endl;
disconnect();
return;
}
//move back into the header reading state
async_read(socket, buffer(headerBuffer),
bind(&Uplink::parse_header, this, _1, _2));
}
void Uplink::disconnect() {
acceptor.close();
socket.close();
acceptor.open(endpoint.protocol());
acceptor.set_option(ip::tcp::acceptor::reuse_address(true));
acceptor.bind(endpoint);
acceptor.listen(1); //1 means only one client in connect queue.
acceptor.async_accept(socket, bind(&Uplink::accept_handler, this, _1));
}
void Uplink::accept_handler(const boost::system::error_code& error)
{
if (!error) {
//no more clents.
acceptor.close();
//move to read header state.
async_read(socket, buffer(headerBuffer),
bind(&Uplink::parse_header, this, _1, _2));
}
}
void Uplink::sendASCIIMessage(const std::string &m) {
//Format the message
unsigned int msgLength(m.length());
vector<char> outBuffer(msgLength+headerSize);
outBuffer[0] = MSG_ASCII;
memcpy(&outBuffer[1], &msgLength, sizeof(unsigned int));
vector<char>::iterator dataBegin(outBuffer.begin());
advance(dataBegin, headerSize);
copy(m.begin(), m.end(), dataBegin);
//queue the message
addToQueue(outBuffer);
}
void Uplink::sendIMUDataBlock(const nIMUDataBlock *d) {
//Format the message.
//a,g,m, 3 components each plus a stamp
const unsigned int msgLength(3*3*sizeof(double)+sizeof(unsigned int));
vector<char> outBuffer(msgLength+headerSize);
outBuffer[0] = MSG_IMU;
memcpy(&outBuffer[1], &msgLength, sizeof(unsigned int));
const Eigen::Vector3d a(d->getAccel());
const Eigen::Vector3d m(d->getMag());
const Eigen::Vector3d g(d->getGyro());
const unsigned int s(d->getUpdateStamp());
memcpy(&outBuffer[headerSize], a.data(), sizeof(double)*3);
memcpy(&outBuffer[headerSize+3*sizeof(double)], m.data(), sizeof(double)*3);
memcpy(&outBuffer[headerSize+6*sizeof(double)], g.data(), sizeof(double)*3);
memcpy(&outBuffer[headerSize+9*sizeof(double)], &s, sizeof(unsigned int));
/*
cout<<"----------------------------------------"<<endl;
cout<<"Accel = ("<<a[0]<<","<<a[1]<<","<<a[2]<<")"<<endl;
cout<<"Mag = ("<<m[0]<<","<<m[1]<<","<<m[2]<<")"<<endl;
cout<<"Gyro = ("<<g[0]<<","<<g[1]<<","<<g[2]<<")"<<endl;
cout<<"Stamp = "<<s<<endl;
cout<<"----------------------------------------"<<endl;
*/
//queue the message
addToQueue(outBuffer);
}
void Uplink::send_handler(const boost::system::error_code& error,
std::size_t bytes_transferred) {
{
lock_guard<mutex> l(queueLock);
lock_guard<mutex> l2(sendingLock);
if(outQueue.empty()) {
currentlySending = false;
return;
}
}
startSend();
}
void Uplink::addToQueue(const std::vector<char> &out) {
bool needsRestart = false;
{
lock_guard<mutex> l(queueLock);
lock_guard<mutex> l2(sendingLock);
outQueue.push(out);
needsRestart = !currentlySending;
}
if(needsRestart)
nIO.post(bind(&Uplink::startSend, this));
}
void Uplink::startSend() {
lock_guard<mutex> l(queueLock);
lock_guard<mutex> l2(sendingLock);
if(outQueue.empty())
return;
currentlySending = true;
currentWrite = outQueue.front();
outQueue.pop();
async_write(socket, buffer(currentWrite), bind(&Uplink::send_handler,
this, _1, _2));
}
void Uplink::sendVideoFrameBGR(const unsigned int width, const unsigned int height,
const unsigned int cameraID, const unsigned char *frameData) {
// image data image metadata header
const unsigned int packetSize(width*height*3 + sizeof(unsigned int)*3 + headerSize);
const unsigned int dataSize(width*height*3 + sizeof(unsigned int)*3);
vector<char> outgoingBuffer(packetSize);
outgoingBuffer[0] = MSG_FRAME_BGR;
memcpy(&outgoingBuffer[1], &dataSize, sizeof(unsigned int));
char *writePtr = &outgoingBuffer[headerSize];
memcpy(writePtr, &width, sizeof(unsigned int));
writePtr += sizeof(unsigned int);
memcpy(writePtr, &height, sizeof(unsigned int));
writePtr += sizeof(unsigned int);
memcpy(writePtr, &cameraID, sizeof(unsigned int));
writePtr += sizeof(unsigned int);
memcpy(writePtr, frameData, width*height*3*sizeof(char));
//TODO: can we avoid the whole image copy here?
//TODO: should come up with a better packet buffer build system.
//IDEA!: maybe have a "request buffer" funxction so the Uplink
//class can have sole ownership, rather than do the copy in "addtoQueue"
addToQueue(outgoingBuffer);
}
这个程序大部分时间都有效，但很少，当发送大量数据且数据包之间没有延迟时，它会失败。 例如:
sendVideoFrameBGR(...); //occasional fail
sendASCIIMessage("...");
sendVideoFrameBGR(...); //never fails.
sleep(1);
sendASCIIMessage("...");
在下行链路中处理视频帧后，它返回到 hadleHeaderData 并等待长度为几兆字节的数据包和不存在的数据包 ID。流以某种方式被破坏。我不知道为什么。
我不太关心我现在编写的代码，所以如果有人知道一个好的类或库可以为我将 TCP 上的流解析为缓冲区 block ，我宁愿使用它。
编辑:
这是运行数据发送的确切代码:
if(frontImage) {
uplink.sendVideoFrameBGR(frontImage->width, frontImage->height, 0,
(unsigned char*)frontImage->imageData);
cout<<"Sent"<<endl;
//sleep(1); //works fine if this is uncommented !
}
uplink.sendASCIIMessage("Alive...");
sleep(1);
uplink.sendIMUDataBlock(imuDataBlock.get());
cout<<"Loop"<<endl;
sleep(1);
}
最佳答案
问题很可能是你的 ioservice 对象有多个线程处理工作。
当您在第一个发送函数之后立即调用第二个发送函数时，发布到 ioservice 的两个函数对象可能被委托(delegate)给不同的线程。所以基本上，两个写入并行发生在同一个套接字上。这很可能是非法的。将 Winsock2 与非阻塞套接字一起使用，这会导致传出数据损坏。
即使您使用 bool 来检查它当前是否正在发送，在 ioservice 线程之一处理函数之前也不会检查 bool。如果在您发布这两项工作时有两个 ioservice 线程处于事件状态，它可能会同时分派(dispatch)两个发送，导致两个发送函数在不同的线程上异步发生。 “当前正在发送”检查可能在两个调用中都返回 false，因为两个发送是并行运行的。
关于c++ - 尝试非常频繁地发送数据包时，boost::asio 数据包传输失败，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/3444551/