boost socket iostreams echo server z kompresją zlib jest w stanie uśpienia, dopóki połączenie nie zostanie zamknięte

Jan 18 2021

Próbuję stworzyć prosty serwer echa z kompresją zlib, postępując zgodnie z tym i tym przykładem.

Moim pomysłem jest wysłanie teraz jakiegoś ciągu, ponieważ mogę przekonwertować typy POD na string ( std::string(reinterpret_cast<const char *>(&pod), sizeof(pod))) przed wysłaniem, kiedy upewnię się, że warstwa transportowa działa.

I tu jest problem. Klient kompresuje dane, wysyła je i informuje, że dane zostały wysłane, ale serwer jest zablokowany podczas odczytu danych. Nie mogę zrozumieć, dlaczego tak się dzieje.

Próbowałem używać operator<<z out.flush(), również próbowałem używać boost::iostreams::copy(). Wynik jest taki sam. Przykładowy kod to (używam tego samego pliku źródłowego dla serwera i klienta w zależności od argumentów):

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

Najpierw uruchamiam serwer, a potem klienta. Generowane są następujące dane wyjściowe:

serwer

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

Klient

$ ./example sender
sended: hello world

Programy są blokowane powyższym wyjściem. Wydaje mi się, że serwer nadal czeka na dane od klienta i nie wie, że klient wysłał wszystko, co miał.

Jeśli zamknę klienta za Ctrl + Cpomocą, wynik będzie następujący:

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

Myślę, że dzieje się zlib error-5tak, ponieważ serwer uważa, że archiwum jest niekompletne.

Oczekiwane zachowanie to brak blokowania. Komunikat musi pojawić się w danych wyjściowych programu serwera podczas uruchamiania klienta.

Dlaczego program jest zablokowany podczas czytania? Jak mogę to naprawić?

Odpowiedzi

1 sehe Jan 18 2021 at 21:56

iostreams::copy robi właśnie to: kopiuje strumień.

Komplementy dla twojego kodu. Jest bardzo czytelny :) Przypomina mi tę odpowiedź Czytanie i zapisywanie plików z gniazdem boost iostream . Główna różnica polega na tym, że ta odpowiedź wysyła pojedynczy skompresowany obiekt blob i zamyka.

Masz "rację", że dekompresor wie, kiedy jeden skompresowany blok jest kompletny, ale nie decyduje, czy następny nie nastąpi.

Musisz więc dodać kadrowanie. Tradycyjnym sposobem jest przepuszczenie długości poza pasmem. Zaimplementowałem zmiany, jednocześnie zmniejszając powielanie kodu za pomocą manipulatorów IO.

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

Ten zawiera głównie kod, który skopiowałeś między nadawcą / odbiorcą:

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

Zrobiłem Tszablon tak, aby można go było przechowywać std::string&lub w std::string const&zależności od potrzeb.

`LengthPrefixed` manipulator

Ten manipulator nie dba o to, co jest serializowane, ale po prostu doda przed nim efektywną długość na kablu:

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

Dodajemy subtelność: przechowując odniesienie lub wartość w zależności od tego, z czym jesteśmy skonstruowani, możemy również zaakceptować tymczasowe (takie jak manipulator ZLIB):

boost socket iostreams echo server z kompresją zlib jest w stanie uśpienia, dopóki połączenie nie zostanie zamknięte

Odpowiedzi

ZLIB manipulator

LengthPrefixed manipulator

PROGRAM DEMO

Kompletna lista

`ZLIB` manipulator

`LengthPrefixed` manipulator