c++ - boost::asio 不会触发读取处理程序,而 wireshark 会看到数据进入
问题描述
我正在尝试发送一些数据并根据回复采取行动。我看到(使用wireshark)系统发送和接收数据,但boost::asio
不会触发我的回调。有人知道我做错了什么吗?
#include <asio.hpp>
#include <bits/stdint-uintn.h>
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <system_error>
#include <thread>
static const int polynomial = 0x1021; // represents x^16+x^12+x^5+1
uint16_t calc(uint8_t* bytes, std::size_t length)
{
uint16_t new_crc = 0x0000;
// bytes part
for (std::size_t j = 0; j < length; ++j)
{
for (int i = 0; i < 8; ++i)
{
bool bit = ((bytes[j] >> (7 - i) & 1) == 1);
bool c15 = ((new_crc >> 15 & 1) == 1);
new_crc <<= 1;
// If coefficient of bit and remainder polynomial = 1 xor crc with polynomial
if (c15 ^ bit) new_crc ^= polynomial;
}
}
return new_crc;
}
int main(int argc, const char* argv[])
{
asio::io_service main_io_service;
std::string ip = "192.168.100.155";
int portP = 4001, portS = 4002;
auto sPrimary = std::shared_ptr<asio::ip::tcp::socket>(new asio::ip::tcp::socket(main_io_service));
auto sSecondary = std::shared_ptr<asio::ip::tcp::socket>(new asio::ip::tcp::socket(main_io_service));
auto epPrimary = asio::ip::tcp::endpoint(asio::ip::address::from_string(ip), portP);
auto epSecondary = asio::ip::tcp::endpoint(asio::ip::address::from_string(ip), portS);
std::error_code ec;
sPrimary->connect(epPrimary, ec);
if (ec || !sPrimary->is_open())
{
std::cerr << "primary failed to connect" << std::endl;
}
ec.clear();
sSecondary->connect(epSecondary, ec);
if (ec || !sSecondary->is_open())
{
std::cerr << "secondary failed to connect" << std::endl;
}
std::mutex mutex;
std::unique_lock<std::mutex> lock(mutex);
std::condition_variable cv;
const std::size_t msgSize = 9;
uint8_t msg[msgSize];
int i = 0;
msg[i++] = 0x02;
msg[i++] = 0xFF;
msg[i++] = 0x00;
msg[i++] = 0x00;
msg[i++] = 0x00;
msg[i++] = 0x00;
uint16_t crc = calc(msg, i);
msg[i++] = (uint8_t) (crc & 0xFF);
msg[i++] = (uint8_t) (crc >> 8);
msg[i++] = 0x03;
const std::size_t buffSize = 1024;
uint8_t buff[buffSize];
std::size_t bytesRead = 0;
asio::async_write((*sPrimary.get()), asio::buffer(msg, msgSize), [&sPrimary, &cv, &buff, &buffSize, &bytesRead](const std::error_code &ec, std::size_t bytesWritten)
{
asio::async_read((*sPrimary.get()), asio::buffer(buff, buffSize), [&cv, &bytesRead](const std::error_code &ec, std::size_t currentBytesRead)
{
bytesRead += currentBytesRead;
cv.notify_one();
});
});
main_io_service.run();
cv.wait(lock);
for (std::size_t i = 0; i < bytesRead; ++i)
std::cout << std::hex << buff[i];
main_io_service.stop();
return 0;
}
只需添加将编译的整个测试代码。尽管您需要一个可以回答的设备。此代码与一个串行服务器通信,该服务器具有一个回复发送数据包的硬件。
谢谢!
解决方案
@rafix07 提出的问题是您的问题。
即使您通过io_service::run()
在另一个线程上运行来“伪造”它,从技术上讲,您仍然有相同竞争条件的时间窗口。
通常,锁定同步原语不会与基于任务的并行性混合。在这种情况下,您很可能只想
- 读取完成后向服务发布另一个任务
- 使您可以响应的计时器过期
在您的代码非常简单的情况下,您甚至可以使用其他更简单的选项:
- 利用
run()
阻塞直到所有处理程序完成的事实。也就是说,您可以将run()
返回的事实作为读取完成的指示: - 不要使用异步处理程序,因为它没有任何用途(这可能归结为此处过于简化的示例代码)
4.使用同步IO
这是迄今为止最简单的。在整个程序中进行的许多其他简化
#include <cstdint>
#include <iostream>
#include <string>
#ifndef NOBOOST
#include <boost/asio.hpp>
namespace asio = boost::asio;
using boost::system::error_code;
#else
#include <asio.hpp>
namespace asio = boost::asio;
using std::error_code;
#endif
static const int polynomial = 0x1021; // represents x^16+x^12+x^5+1
uint16_t calc_crc(uint8_t* bytes, std::size_t length) {
uint16_t new_crc = 0x0000;
// bytes part
for (std::size_t j = 0; j < length; ++j) {
for (int i = 0; i < 8; ++i) {
bool bit = ((bytes[j] >> (7 - i) & 1) == 1);
bool c15 = ((new_crc >> 15 & 1) == 1);
new_crc <<= 1;
// If coefficient of bit and remainder polynomial = 1 xor crc with polynomial
if (c15 ^ bit)
new_crc ^= polynomial;
}
}
return new_crc;
}
int main() {
static const std::string ip = "127.0.0.1";
static const unsigned short portP = 4001, portS = 4002;
using asio::ip::address;
asio::io_service io;
asio::ip::tcp::socket sPrimary(io), sSecondary(io);
sPrimary.connect({ address::from_string(ip), portP });
sSecondary.connect({ address::from_string(ip), portS });
uint8_t msg[] {
0x02, 0xFF, 0x00, 0x00, 0x00, 0x00,
0x00, 0x00, //crc
0x03
};
{ // set crc
uint16_t const crc = calc_crc(msg, sizeof(msg)-3);
msg[sizeof(msg)-3] = (uint8_t)(crc & 0xFF);
msg[sizeof(msg)-2] = (uint8_t)(crc >> 8);
}
std::string buff;
auto bytesWritten = asio::write(sPrimary, asio::buffer(msg));
std::cout << bytesWritten << " sent" << std::endl;
auto bytesRead = asio::read(sPrimary, asio::dynamic_buffer(buff), asio::transfer_exactly(32));
std::cout << bytesRead << " received" << std::endl;
for (uint8_t ch : buff)
std::cout << std::hex << static_cast<int>(ch);
std::cout << std::endl;
}
印刷
9 sent
32 received
23696e636c756465203c63737464696e743ea23696e636c756465203c696f73
事实上,这就是前 32 个字节的十六进制编码main.cpp
3.使用隐式完成
相信处理程序在run()
返回时运行(需要错误处理)。代码本质上是相同的,但对 lambda 捕获和对象生命周期有更详细的关注。
注意:所有其他简化仍然适用
asio::async_write(sPrimary, asio::buffer(msg), [&sPrimary, &buff](error_code ec, size_t bytesWritten) {
std::cout << "async_write: " << ec.message() << ", " << bytesWritten << " sent" << std::endl;
asio::async_read(sPrimary, asio::dynamic_buffer(buff), asio::transfer_exactly(32), [](error_code ec, size_t bytesRead) {
std::cout << "async_read: " << ec.message() << ", " << bytesRead << " received" << std::endl;
});
});
io.run();
for (uint8_t ch : buff)
std::cout << std::hex << static_cast<int>(ch);
std::cout << std::endl;
印刷:
async_write: Success, 9 sent
async_read: Success, 32 received
23696e636c756465203c63737464696e743ea23696e636c756465203c696f73
2.使用定时器信号
通过使用计时器对象来表示条件,这与您拥有的 CV 方法最“相似”。
- 值得注意的是,这比上面的“3”更好地处理错误。代码
- 另请注意,它保证调用的完成处理程序
signal_complete
(除非程序过早终止) - 因此,信息在
expiry()
计时器中,而不是在错误代码中(时间将始终显示为已取消)
std::string buff;
asio::high_resolution_timer signal_complete(io, std::chrono::high_resolution_clock::time_point::max());
signal_complete.async_wait([&signal_complete, &buff](error_code ec) {
std::cout << "signal_complete: " << ec.message() << std::endl;
if (signal_complete.expiry() < std::chrono::high_resolution_clock::now()) {
for (uint8_t ch : buff)
std::cout << std::hex << static_cast<int>(ch);
std::cout << std::endl;
}
});
asio::async_write(sPrimary, asio::buffer(msg), [&sPrimary, &buff, &signal_complete](error_code ec, size_t bytesWritten) {
std::cout << "async_write: " << ec.message() << ", " << bytesWritten << " sent" << std::endl;
asio::async_read(sPrimary, asio::dynamic_buffer(buff), asio::transfer_exactly(32), [&signal_complete](error_code ec, size_t bytesRead) {
std::cout << "async_read: " << ec.message() << ", " << bytesRead << " received" << std::endl;
if (!ec) {
signal_complete.expires_at(std::chrono::high_resolution_clock::time_point::min());
} else {
signal_complete.cancel();
}
});
});
io.run();
印刷:
async_write: Success, 9 sent
async_read: Success, 32 received
signal_complete: Operation canceled
23696e636c756465203c63737464696e743ea23696e636c756465203c696f73
1. 读取完成后发布另一个任务
这最适合大多数异步 IO 场景,因为它将所有任务放在同一个队列中。
唯一更复杂的部分是正确设置(共享)对象的生命周期。
std::string buff;
asio::async_write(sPrimary, asio::buffer(msg), [&io, &sPrimary, &buff](error_code ec, size_t bytesWritten) {
std::cout << "async_write: " << ec.message() << ", " << bytesWritten << " sent" << std::endl;
asio::async_read(sPrimary, asio::dynamic_buffer(buff), asio::transfer_exactly(32), [&io, &buff](error_code ec, size_t bytesRead) {
std::cout << "async_read: " << ec.message() << ", " << bytesRead << " received" << std::endl;
if (!ec) {
post(io, [&buff] {
for (uint8_t ch : buff)
std::cout << std::hex << static_cast<int>(ch);
std::cout << std::endl;
});
}
});
});
io.run();
再次打印:
async_write: Success, 9 sent
async_read: Success, 32 received
23696e636c756465203c63737464696e743ea23696e636c756465203c696f73
推荐阅读
- command-line - 如何使用 jq 命令从文件内容创建表
- python - 谷歌分析 api 每次都崩溃
- sql - 连续编号的值组(包括重复)
- java - Java 编译器只考虑第一个数字
- angular - 设置多个验证器 - Angular Reactive Form
- python-3.x - Python:pd.date_range,返回的内容不一致
- c# - 基于 DateTime.Now 和预定义小时比较检测工作班次。C#
- arrays - 如果时间复杂度为 O(x*y) 其中 x+y = n,则时间复杂度为 n
- c# - C# 远程桌面遗留控制台应用程序
- java - 当写入大量数据时,部分数据会丢失/当每个数据都存在时,写入过程非常慢