boost socket iostreams echo server con compresión zlib duerme hasta que se cierre la conexión

Jan 18 2021

Intento crear un servidor de eco simple con compresión zlib siguiendo este y este ejemplo.

Mi idea es enviar algo de cadena ahora porque puedo convertir los tipos de POD a cadena ( std::string(reinterpret_cast<const char *>(&pod), sizeof(pod))) antes de enviar cuando me aseguraré de que la capa de transporte funcione.

Y aquí hay un problema. El cliente comprime los datos, los envía y dice que los datos se enviaron pero que el servidor está bloqueado en la lectura de datos. No puedo entender por qué sucede.

Intenté usar operator<<con out.flush(), también intenté usar boost::iostreams::copy(). El resultado es el mismo. El ejemplo de código es (uso el mismo archivo fuente para el servidor y el cliente según los argumentos):

#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;
}

Primero inicio el servidor y luego inicio el cliente. Se produce el siguiente resultado:

Servidor

$ ./example
# now it waits while client will be accepted
start session
start reading

Cliente

$ ./example sender
sended: hello world

Los programas están bloqueados con la salida anterior. Supongo que el servidor todavía espera los datos del cliente y no sabe que el cliente envió todo lo que tenía.

Si cierro el cliente con Ctrl + C, el resultado es el siguiente:

$ ./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

Supongo que zlib error-5se debe a que el servidor cree que el archivo está incompleto.

El comportamiento esperado es sin bloqueo. El mensaje debe aparecer en la salida del programa del servidor cuando se inició el cliente.

¿Por qué el programa está bloqueado en lectura? ¿Cómo puedo arreglarlo?

Respuestas

1 sehe Jan 18 2021 at 21:56

iostreams::copy hace precisamente eso: copia el flujo.

Felicitaciones a su código. Es muy legible :) Me recuerda esta respuesta Lectura y escritura de archivos con boost iostream socket . La principal diferencia es que esa respuesta envía un único blob comprimido y se cierra.

Tiene "razón" en que el descompresor sabe cuándo se completa un bloque comprimido, pero no decide que no le siga otro.

Entonces necesitas agregar encuadre. La forma tradicional es pasar un tramo fuera de banda. Implementé los cambios al mismo tiempo que reduje la duplicación de código mediante el uso de manipuladores de E / S.

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` manipulador

Este encapsula principalmente el código que duplicó entre el remitente / receptor:

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;
    }
};

Hice la Tplantilla para que se pueda almacenar std::string&o std::string const&según la necesidad.

`LengthPrefixed` manipulador

A este manipulador no le importa lo que se está serializando, sino que simplemente lo antepondrá con la longitud efectiva en el cable:

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;
    }
};

Agregamos una sutileza: al almacenar una referencia o valor dependiendo de con qué estemos construidos también podemos aceptar temporales (como el manipulador ZLIB):

boost socket iostreams echo server con compresión zlib duerme hasta que se cierre la conexión

Respuestas

ZLIB manipulador

LengthPrefixed manipulador

PROGRAMA DE DEMOSTRACIÓN

Listado completo

`ZLIB` manipulador

`LengthPrefixed` manipulador