meningkatkan soket server gema iostreams dengan kompresi zlib tidur sampai koneksi akan ditutup

Jan 18 2021

Saya mencoba membuat server gema sederhana dengan kompresi zlib mengikuti ini dan contoh ini .

Ide saya adalah mengirim beberapa string sekarang karena saya dapat mengonversi jenis POD menjadi string ( std::string(reinterpret_cast<const char *>(&pod), sizeof(pod))) sebelum mengirim ketika saya akan memastikan lapisan transport berfungsi.

Dan ada masalah disini. Klien mengompresi data, mengirimkannya dan mengatakan data telah dikirim tetapi server diblokir saat membaca data. Saya tidak mengerti mengapa itu terjadi.

Saya mencoba menggunakan operator<<dengan out.flush(), juga saya mencoba menggunakan boost::iostreams::copy(). Hasilnya sama saja. Contoh kodenya adalah (Saya menggunakan file sumber yang sama untuk server dan klien tergantung pada args):

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

Pertama saya memulai server dan kemudian saya memulai klien. Output berikut dihasilkan:

Server

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

Klien

$ ./example sender
sended: hello world

Program diblokir dengan keluaran di atas. Saya kira server masih menunggu data dari klien dan tidak tahu klien mengirim semua yang dimilikinya.

Jika saya menutup klien dengan Ctrl + Cmaka hasilnya adalah sebagai berikut:

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

dan

$ ./example sender
sended: hello world
^C

Saya kira zlib error-5itu karena server menganggap arsipnya tidak lengkap.

Perilaku yang diharapkan tidak menghalangi. Pesan harus muncul di output program server ketika klien dimulai.

Mengapa program diblokir saat membaca? Bagaimana cara memperbaikinya?

Jawaban

1 sehe Jan 18 2021 at 21:56

iostreams::copy melakukan hal itu: menyalin aliran.

Pujian untuk kode Anda. Sangat mudah dibaca :) Ini mengingatkan saya pada jawaban ini Membaca dan menulis file dengan boost iostream socket . Perbedaan utamanya adalah bahwa jawaban tersebut mengirimkan satu gumpalan terkompresi dan menutup.

Anda "benar" bahwa dekompresor mengetahui saat satu blok terkompresi selesai, tetapi tidak memutuskan bahwa blok lain tidak akan mengikuti.

Jadi, Anda perlu menambahkan pembingkaian. Cara tradisional adalah dengan mengoper long out-of-band. Saya telah menerapkan perubahan sekaligus mengurangi duplikasi kode dengan menggunakan manipulator 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) ;
};

Dan

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

Yang ini terutama merangkum kode yang Anda duplikat antara pengirim / penerima:

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

Saya buat Ttemplated supaya bisa menyimpan std::string&atau std::string const&tergantung kebutuhan.

`LengthPrefixed` manipulator

Manipulator ini tidak peduli apa yang sedang diserialisasi, tetapi hanya akan mengawalnya dengan panjang efektif on-the-wire:

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

Kami menambahkan kehalusan: dengan menyimpan referensi atau nilai tergantung pada apa yang kami buat, kami juga dapat menerima sementara (seperti manipulator ZLIB):

meningkatkan soket server gema iostreams dengan kompresi zlib tidur sampai koneksi akan ditutup

Jawaban

ZLIB manipulator

LengthPrefixed manipulator

PROGRAM DEMO

Daftar Lengkap

`ZLIB` manipulator

`LengthPrefixed` manipulator