Boost socket iostreams echo server со сжатием zlib засыпает, пока соединение не будет закрыто
Я пытаюсь создать простой эхо-сервер со сжатием zlib, следуя этому и этому примерам.
Моя идея состоит в том, чтобы отправить некоторую строку сейчас, потому что я могу преобразовать типы POD в string ( std::string(reinterpret_cast<const char *>(&pod), sizeof(pod))
) перед отправкой, когда я буду уверен, что транспортный уровень работает.
И здесь есть проблема. Клиент сжимает данные, отправляет их и сообщает, что данные были отправлены, но сервер заблокирован при чтении данных. Я не могу понять, почему это происходит.
Пробовал использовать operator<<
с out.flush()
, тоже пробовал использовать boost::iostreams::copy()
. Результат тот же. Пример кода (я использую один и тот же исходный файл для сервера и клиента в зависимости от аргументов):
#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/filter/zlib.hpp>
#include <boost/iostreams/copy.hpp>
#include <boost/asio.hpp>
#include <iostream>
#include <sstream>
namespace ip = boost::asio::ip;
using ip::tcp;
const unsigned short port = 9999;
const char host[] = "127.0.0.1";
void receive()
{
boost::asio::io_context ctx;
tcp::endpoint ep(ip::address::from_string(host), port);
tcp::acceptor a(ctx, ep);
tcp::iostream stream;
a.accept(stream.socket());
std::stringstream buffer;
std::cout << "start session" << std::endl;
try
{
for (;;)
{
{
boost::iostreams::filtering_istream in;
in.push(boost::iostreams::zlib_decompressor());
in.push(stream);
std::cout << "start reading" << std::endl;
// looks like server is blocked here
boost::iostreams::copy(in, buffer);
}
std::cout << "data: " << buffer.str() << std::endl;
{
boost::iostreams::filtering_ostream out;
out.push(boost::iostreams::zlib_compressor());
out.push(stream);
boost::iostreams::copy(buffer, out);
}
std::cout << "Reply is sended" << std::endl;
}
}
catch(const boost::iostreams::zlib_error &e)
{
std::cerr << e.what() << e.error() << '\n';
stream.close();
}
}
void send(const std::string &data)
{
tcp::endpoint ep(ip::address::from_string(host), port);
tcp::iostream stream;
stream.connect(ep);
std::stringstream buffer;
buffer << data;
if (!stream)
{
std::cerr << "Cannot connect to " << host << ":" << port << std::endl;
return;
}
try
{
{
boost::iostreams::filtering_ostream out;
out.push(boost::iostreams::zlib_compressor());
out.push(stream);
out << buffer.str();
out.flush();
}
std::cout << "sended: " << data << std::endl;
buffer.str("");
{
boost::iostreams::filtering_istream in;
in.push(boost::iostreams::zlib_decompressor());
in.push(stream);
// looks like client is blocked here
boost::iostreams::copy(in, buffer);
}
std::cout << "result: " << buffer.str() << std::endl;
}
catch(const boost::iostreams::zlib_error &e)
{
std::cerr << e.what() << '\n';
}
}
int main(int argc, const char *argv[])
{
if (argc > 1 && argv[1] == std::string("sender"))
send("hello world");
else
receive();
return 0;
}
Сначала я запускаю сервер, а затем запускаю клиента. Будет произведен следующий вывод:
Сервер
$ ./example
# now it waits while client will be accepted
start session
start reading
Клиент
$ ./example sender
sended: hello world
Программы заблокированы с выводом выше. Я предполагаю, что сервер все еще ждет данных от клиента, и он не знает, что клиент отправил все, что у него было.
Если я закрою клиент с помощью, Ctrl + C
то вывод будет следующим:
$ ./example
# now it waits while client will be accepted
start session
start reading
# now it is blocked until I press Ctrl + C
data: hello world
Reply is sended
start reading
zlib error-5
а также
$ ./example sender
sended: hello world
^C
Я предполагаю, что zlib error-5
это потому, что сервер считает, что архив неполный.
Ожидаемое поведение - отсутствие блокировки. Сообщение должно появиться в выходных данных серверной программы при запуске клиента.
Почему программа блокируется при чтении? Как я могу это исправить?
Ответы
iostreams::copy
делает именно это: он копирует поток.
Комплименты вашему коду. Он очень удобочитаемый :) Он напоминает мне этот ответ Чтение и запись файлов с помощью сокета boost iostream . Основное отличие состоит в том, что этот ответ отправляет один сжатый двоичный объект и закрывается.
Вы «правы» в том, что декомпрессор знает, когда один сжатый блок завершен, но не решает, что другой не последует.
Итак, вам нужно добавить обрамление. Традиционный способ - передать длительность вне диапазона. Я внес изменения, а также уменьшил дублирование кода с помощью манипуляторов ввода-вывода.
template <typename T> struct LengthPrefixed {
T _wrapped;
friend std::ostream& operator<<(std::ostream& os, LengthPrefixed lp) ;
friend std::istream& operator>>(std::istream& is, LengthPrefixed lp) ;
};
А также
template <typename T> struct ZLIB {
T& data;
ZLIB(T& ref) : data(ref){}
friend std::ostream& operator<<(std::ostream& os, ZLIB z) ;
friend std::istream& operator>>(std::istream& is, ZLIB z) ;
};
ZLIB
манипулятор
Он в основном инкапсулирует код, который вы дублировали между отправителем / получателем:
template <typename T> struct ZLIB {
T& data;
ZLIB(T& ref) : data(ref){}
friend std::ostream& operator<<(std::ostream& os, ZLIB z) {
{
boost::iostreams::filtering_ostream out;
out.push(boost::iostreams::zlib_compressor());
out.push(os);
out << z.data << std::flush;
}
return os.flush();
}
friend std::istream& operator>>(std::istream& is, ZLIB z) {
boost::iostreams::filtering_istream in;
in.push(boost::iostreams::zlib_decompressor());
in.push(is);
std::ostringstream oss;
copy(in, oss);
z.data = oss.str();
return is;
}
};
Я сделал
T
шаблон, чтобы его можно было хранитьstd::string&
или вstd::string const&
зависимости от необходимости.
LengthPrefixed
манипулятор
Этот манипулятор не заботится о том, что сериализуется, он просто префикс его эффективной длины по сети:
template <typename T> struct LengthPrefixed {
T _wrapped;
friend std::ostream& operator<<(std::ostream& os, LengthPrefixed lp) {
std::ostringstream oss;
oss << lp._wrapped;
auto on_the_wire = std::move(oss).str();
debug << "Writing length " << on_the_wire.length() << std::endl;
return os << on_the_wire.length() << "\n" << on_the_wire << std::flush;
}
friend std::istream& operator>>(std::istream& is, LengthPrefixed lp) {
size_t len;
if (is >> std::noskipws >> len && is.ignore(1, '\n')) {
debug << "Reading length " << len << std::endl;
std::string on_the_wire(len, '\0');
if (is.read(on_the_wire.data(), on_the_wire.size())) {
std::istringstream iss(on_the_wire);
iss >> lp._wrapped;
}
}
return is;
}
};
Мы добавляем тонкость: сохраняя ссылку или значение в зависимости от того, из чего мы созданы, мы также можем принимать временные файлы (например, манипулятор ZLIB):
template <typename T> LengthPrefixed(T&&) -> LengthPrefixed<T>;
template <typename T> LengthPrefixed(T&) -> LengthPrefixed<T&>;
Я не подумал сделать
ZLIB
манипулятор столь же универсальным. Так что я оставляю это как экзорцизм для читателя
ДЕМО ПРОГРАММА
Комбинируя эти два, вы можете написать отправителя / получателя просто как:
void server() {
boost::asio::io_context ctx;
tcp::endpoint ep(ip::address::from_string(host), port);
tcp::acceptor a(ctx, ep);
tcp::iostream stream;
a.accept(stream.socket());
std::cout << "start session" << std::endl;
for (std::string data; stream >> LengthPrefixed{ZLIB{data}};) {
std::cout << "data: " << std::quoted(data) << std::endl;
stream << LengthPrefixed{ZLIB{data}} << std::flush;
}
}
void client(std::string data) {
tcp::endpoint ep(ip::address::from_string(host), port);
tcp::iostream stream(ep);
stream << LengthPrefixed{ZLIB{data}} << std::flush;
std::cout << "sent: " << std::quoted(data) << std::endl;
stream >> LengthPrefixed{ZLIB{data}};
std::cout << "result: " << std::quoted(data) << std::endl;
}
Действительно, он печатает:
reader: start session
sender: Writing length 19
reader: Reading length 19
sender: sent: "hello world"
reader: data: "hello world"
reader: Writing length 19
sender: Reading length 19
sender: result: "hello world"
Полный список
#include <boost/iostreams/filtering_stream.hpp>
#include <boost/iostreams/filter/zlib.hpp>
#include <boost/iostreams/copy.hpp>
#include <boost/asio.hpp>
#include <iostream>
#include <iomanip>
#include <sstream>
namespace ip = boost::asio::ip;
using ip::tcp;
const unsigned short port = 9999;
const char host[] = "127.0.0.1";
#ifdef DEBUG
std::ostream debug(std::cerr.rdbuf());
#else
std::ostream debug(nullptr);
#endif
template <typename T> struct LengthPrefixed {
T _wrapped;
friend std::ostream& operator<<(std::ostream& os, LengthPrefixed lp) {
std::ostringstream oss;
oss << lp._wrapped;
auto on_the_wire = std::move(oss).str();
debug << "Writing length " << on_the_wire.length() << std::endl;
return os << on_the_wire.length() << "\n" << on_the_wire << std::flush;
}
friend std::istream& operator>>(std::istream& is, LengthPrefixed lp) {
size_t len;
if (is >> std::noskipws >> len && is.ignore(1, '\n')) {
debug << "Reading length " << len << std::endl;
std::string on_the_wire(len, '\0');
if (is.read(on_the_wire.data(), on_the_wire.size())) {
std::istringstream iss(on_the_wire);
iss >> lp._wrapped;
}
}
return is;
}
};
template <typename T> LengthPrefixed(T&&) -> LengthPrefixed<T>;
template <typename T> LengthPrefixed(T&) -> LengthPrefixed<T&>;
template <typename T> struct ZLIB {
T& data;
ZLIB(T& ref) : data(ref){}
friend std::ostream& operator<<(std::ostream& os, ZLIB z) {
{
boost::iostreams::filtering_ostream out;
out.push(boost::iostreams::zlib_compressor());
out.push(os);
out << z.data << std::flush;
}
return os.flush();
}
friend std::istream& operator>>(std::istream& is, ZLIB z) {
boost::iostreams::filtering_istream in;
in.push(boost::iostreams::zlib_decompressor());
in.push(is);
std::ostringstream oss;
copy(in, oss);
z.data = oss.str();
return is;
}
};
void server() {
boost::asio::io_context ctx;
tcp::endpoint ep(ip::address::from_string(host), port);
tcp::acceptor a(ctx, ep);
tcp::iostream stream;
a.accept(stream.socket());
std::cout << "start session" << std::endl;
for (std::string data; stream >> LengthPrefixed{ZLIB{data}};) {
std::cout << "data: " << std::quoted(data) << std::endl;
stream << LengthPrefixed{ZLIB{data}} << std::flush;
}
}
void client(std::string data) {
tcp::endpoint ep(ip::address::from_string(host), port);
tcp::iostream stream(ep);
stream << LengthPrefixed{ZLIB{data}} << std::flush;
std::cout << "sent: " << std::quoted(data) << std::endl;
stream >> LengthPrefixed{ZLIB{data}};
std::cout << "result: " << std::quoted(data) << std::endl;
}
int main(int argc, const char**) {
try {
if (argc > 1)
client("hello world");
else
server();
} catch (const std::exception& e) {
std::cerr << e.what() << '\n';
}
}
Проблема решается с помощью boost::serialization
следующих шагов:
- Прежде всего, я переместил архивирование на такие функции:
namespace io = boost::iostreams;
namespace my {
std::string compress(const std::string &data)
{
std::stringstream input, output;
input << data;
io::filtering_ostream io_out;
io_out.push(io::zlib_compressor());
io_out.push(output);
io::copy(input, io_out);
return output.str();
}
std::string decompress(const std::string &data)
{
std::stringstream input, output;
input << data;
io::filtering_istream io_in;
io_in.push(io::zlib_decompressor());
io_in.push(input);
io::copy(io_in, output);
return output.str();
}
} // namespace my
- Затем я создал такую оболочку для строкового буфера (следуя руководству из документации ):
class Package
{
public:
Package(const std::string &buffer) : buffer(buffer) {}
private:
std::string buffer;
friend class boost::serialization::access;
template<class Archive>
void serialize(Archive &ar, const unsigned int)
{
ar & buffer;
}
};
- И, наконец, я добавил сериализацию после прочтения и перед отправкой.
/**
* receiver
*/
Package request;
{
boost::archive::text_iarchive ia(*stream);
ia >> request;
}
std::string data = my::decompress(request.buffer);
// do something with data
Package response(my::compress(data));
{
boost::archive::text_oarchive oa(*stream);
oa << response;
}
/**
* sender
*/
std::string data = "hello world";
Package package(my::compress(data));
// send request
{
boost::archive::text_oarchive oa(*m_stream);
oa << package;
}
// waiting for a response
{
boost::archive::text_iarchive ia(*m_stream);
ia >> package;
}
// decompress response buffer
result = my::decompress(package.get_buffer());