Модераторы: Daevaorn
  

Поиск:

Ответ в темуСоздание новой темы Создание опроса
> boost async read some 
:(
    Опции темы
Randajad
Дата 30.7.2012, 22:45 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 295
Регистрация: 15.3.2012

Репутация: 8
Всего: 8
Всем доброго времени суток. Имеем:
Код
 
char buf[...];

void read_handler(const boost::system::error_code &ec, size_t readed_len) {

    ...

    auto func = bind(&tcp_socket::read_handler, this, _1, _2);
    sock.async_read_some(boost::asio::buffer(buf, sizeof buf), func);
}


Корректно ли вызывать async_read_some внутри самого хендлера? Гугл ничего не подсказывает, так что вроде как да. Только проблема в том, что когда отправляем пакет на N байт, то приходит два раза по N байт. А в чем ошибка понять не могу. Может это происходит из-за того, что параллельно с чтением идет асинхронная запись тоже?

Это сообщение отредактировал(а) Randajad - 30.7.2012, 22:47
PM MAIL   Вверх
bsa
Дата 30.7.2012, 22:49 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Эксперт
****


Профиль
Группа: Модератор
Сообщений: 9185
Регистрация: 6.4.2006
Где: Москва, Россия

Репутация: 63
Всего: 196
вообще-то, методы async_* и нужно вызывать из хендлеров. Так что ты сделал правильно. Ошибку искать нужно где-то в другом месте.


--------------------
Правильно заданный вопрос - половина ответа
Отзывы о работе модератора.
PM   Вверх
Randajad
Дата 31.7.2012, 01:04 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 295
Регистрация: 15.3.2012

Репутация: 8
Всего: 8
Спасибо за ответ. Но я не знаю уже где ошибку искать. :(
PM MAIL   Вверх
boostcoder
Дата 31.7.2012, 06:26 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


pattern`щик
****


Профиль
Группа: Завсегдатай
Сообщений: 5458
Регистрация: 1.4.2010

Репутация: 49
Всего: 110
Цитата(Randajad @  30.7.2012,  22:45 Найти цитируемый пост)
когда отправляем пакет на N байт, то приходит два раза по N байт.

"отправляем" это async_write?
что значит "приходит" ? читающая сторона дважды прочитывает одни и те же данные?

Цитата(Randajad @  30.7.2012,  22:45 Найти цитируемый пост)
Может это происходит из-за того, что параллельно с чтением идет асинхронная запись тоже?

нет.
нельзя одновременно иметь две операции чтения или записи.

Цитата(Randajad @  31.7.2012,  01:04 Найти цитируемый пост)
не знаю уже где ошибку искать

показывай больше кода.



--------------------
компилятор онлайн с поддержкой C++11
сборки компиляторов MinGW
PM WWW   Вверх
Randajad
Дата 31.7.2012, 10:26 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 295
Регистрация: 15.3.2012

Репутация: 8
Всего: 8
Нет. Есть opensource сервер, проверенный временем. Я пишу к нему клиент. Сервер отправляет пакет на 31 байт. Эти 31 байта у меня читаются, все хорошо. Только следом за этими 31-ю байтами мне приходят еще 31 байт. Я не знаю откуда они берутся.

У меня не две операции чтения/записи. У меня чтение и запись одновременно. По 1 операции того и того.

Код

class tcp_socket {
public:
    inline tcp_socket() : is_valid(false), sock(io), resolver(io) { reset(); }
    inline ~tcp_socket() { close(); }

    inline tcp_socket &connect(cstr host, u16 port, void_func on_connect = 0) {
        close();
        io.reset();

        using boost::asio::ip::tcp;
        tcp::resolver::query query(host, "http");

        auto conn_func = [this, on_connect](const boost::system::error_code &ec) {
            LIB_CH_TE(ec.value() == 0, "can't connect to host: %s", ec.message().c_str());
            is_valid = true;

            auto func = bind(&tcp_socket::read_handler, this, _1, _2);
            printf("first start async\n");
            sock.async_read_some(boost::asio::buffer(rbuf, sizeof rbuf), func);

            if(on_connect)
                on_connect();
        };

        auto res_func = [this, port, conn_func](const boost::system::error_code &ec, tcp::resolver::iterator dest) {
            LIB_CH_TE(ec.value() == 0, "can't resolve host: %s", ec.message().c_str());

            boost::asio::ip::tcp::endpoint ep(dest->endpoint().address(), port);
            sock.async_connect(ep, conn_func);
        };

        resolver.async_resolve(query, res_func);
        return *this;
    }

    inline tcp_socket &close() {
        if(is_valid) {
            reset();
            is_valid = false;

            printf("closing\n");
            sock.shutdown(boost::asio::ip::tcp::socket::shutdown_both);
            sock.close();
            printf("closed\n");
        }

        return *this;
    }

    inline tcp_socket &poll() {
        io.poll();
        return *this;
    }

    inline tcp_socket &write(cvoid data, u16 write_len) {
        LIB_CH_TE(wlen + write_len <= sizeof wbuf, "too many data to write to socket(add lenght is %u)", write_len);

        memcpy(wbuf + wlen, data, write_len);
        wlen += write_len;

        if(wpos)
            return *this;

        auto func = bind(&tcp_socket::write_handler, this, _1, _2);
        boost::asio::async_write(sock, boost::asio::buffer(wbuf, write_len), func);
        return *this;
    }

    inline tcp_socket &skip(u16 i) {
        p += i;

        len -= i;
        rpos += i;
        return *this;
    }

    inline bool valid() const {
        return is_valid;
    }

    u08 *p;
    u16 len;
private:
    inline void reset() {
        p = 0;

        len = 0;
        wlen = 0;

        rpos = 0;
        wpos = 0;
    }

    inline void write_handler(const boost::system::error_code &ec, size_t) {
        if(ec) {
            if(ec == boost::asio::error::operation_aborted)
                return;

            close();
            LIB_TE("can't write data to socket: %s", ec.message().c_str());
        }

        wlen -= wpos;
        memmove(wbuf, wbuf + wpos, wlen);

        wpos = wlen;

        if(!wlen)
            return;

        auto func = bind(&tcp_socket::write_handler, this, _1, _2);
        boost::asio::async_write(sock, boost::asio::buffer(wbuf, wlen), func);
    }

    inline void read_handler(const boost::system::error_code &ec, size_t readed_len) {
        printf("READ LEN %d\n", readed_len);
        if(ec) {
            printf("error %d\n", ec.value());
            if(ec == boost::asio::error::operation_aborted)
                return;

            close();
            LIB_TE("can't read data from socket: %s", ec.message().c_str());
        } else if(readed_len == 0) {
            printf("CASE\n");
            return;
        }

        len += readed_len;
        memmove(rbuf, rbuf + rpos, len);

        rpos = 0;
        p = rbuf;

        u16 alen = sizeof rbuf - len;
        LIB_CH_TE(alen != 0, "it's need a place to read to socket");

        printf("start async\n");
        auto func = bind(&tcp_socket::read_handler, this, _1, _2);
        sock.async_read_some(boost::asio::buffer(rbuf, alen), func);
    }

    u16 wlen;

    u16 rpos;
    u16 wpos;

    u08 rbuf[8192];
    u08 wbuf[8192];

    bool is_valid;

    boost::asio::io_service io;

    boost::asio::ip::tcp::socket sock;
    boost::asio::ip::tcp::resolver resolver;
};



error.h
Код

/********************************************//**
 * \brief Throws with formatted string exception only when statement == false.
 *
 * \param st Statement.
 * \param format String format.
 * \param ... Format arguments.
 * \return Nothing.
 *
 ***********************************************/
#define LIB_CH_TE(st, format, ...) (st) || LIB_TE(format, ## __VA_ARGS__)

/********************************************//**
 * \brief Throws exception with formatted string.
 *
 * \param format String format.
 * \param ... Format arguments.
 * \return Nothing.
 *
 ***********************************************/
#define LIB_TE(format, ...) lib::throw_exception(__FILE__, __LINE__, GetLastError(), errno, format, ## __VA_ARGS__)


/********************************************//**
 * \brief Prints fortatted string to stderr.
 *
 * \param format Format.
 * \param a Format args.
 * \return Number of printed bytes.
 *
 ***********************************************/
template<typename... A>
inline u32 eprintf(cstr format, A... a) {
    return fprintf(stderr, format, a...);
}

/********************************************//**
 * \brief Custom error class. Used for internal actions.
 ***********************************************/
class error : public std::exception {
public:
    inline error(cstr s, u16 len) { memcpy(str, s, len + 1); }
    inline virtual const char *what() const throw() { return str; }
private:
    char str[384];
};

/********************************************//**
 * \brief Function that throws exception. Used for internal actions.
 ***********************************************/
template<typename... A>
inline bool throw_exception(cstr file, u16 line, u32 code, u32 ecode, cstr format, A... a) {
    char buf[128];
    sprintf(buf, "%s%s", format, "(%d, %d, %s:%d)");

    char str[384];
    u16 len = sprintf(str, buf, a..., code, ecode, base_name(file), line);

    throw error(str, len);
}



tcp socket предоставляет асинхронный сокет, который выполняет свои действия при вызове метода poll, также он предоставляет const u08* pointer, где находятся прочтенные данные. Ну и len - их длина.

Добавлено через 2 минуты и 26 секунд
В частности выводится READ LEN 31 два раза. :(

Это сообщение отредактировал(а) Randajad - 31.7.2012, 10:27
PM MAIL   Вверх
Randajad
Дата 31.7.2012, 10:58 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 295
Регистрация: 15.3.2012

Репутация: 8
Всего: 8
Код

#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <boost/asio.hpp>

#define LIB_CH_TE(a, ...) { if(!(a)) { printf(__VA_ARGS__); abort(); } }

class tcp_socket {
public:
    inline tcp_socket() : is_valid(false), sock(io), resolver(io) { reset(); }
    inline ~tcp_socket() { close(); }

    inline tcp_socket &connect(const char* host, int port, boost::function<void()> on_connect = 0) {
        close();
        io.reset();

        using boost::asio::ip::tcp;
        tcp::resolver::query query(host, "http");

        auto conn_func = [this, on_connect](const boost::system::error_code &ec) {
            LIB_CH_TE(ec.value() == 0, "can't connect to host: %s", ec.message().c_str());
            is_valid = true;

            printf("first start of async\n");
            auto func = bind(&tcp_socket::read_handler, this, _1, _2);
            sock.async_read_some(boost::asio::buffer(rbuf, sizeof rbuf), func);

            if(on_connect)
                on_connect();
        };

        auto res_func = [this, port, conn_func](const boost::system::error_code &ec, tcp::resolver::iterator dest) {
            LIB_CH_TE(ec.value() == 0, "can't resolve host: %s", ec.message().c_str());

            boost::asio::ip::tcp::endpoint ep(dest->endpoint().address(), port);
            sock.async_connect(ep, conn_func);
        };

        resolver.async_resolve(query, res_func);
        return *this;
    }

    inline tcp_socket &close() {
        if(is_valid) {
            reset();
            is_valid = false;

            printf("closing\n");
            sock.shutdown(boost::asio::ip::tcp::socket::shutdown_both);
            sock.close();
            printf("closed\n");
        }

        return *this;
    }

    inline tcp_socket &poll() {
        io.poll();
        return *this;
    }

    inline tcp_socket &write(const void *data, int write_len) {
        LIB_CH_TE(wlen + write_len <= (int)sizeof wbuf, "too many data to write to socket(add lenght is %u)", write_len);

        memcpy(wbuf + wlen, data, write_len);
        wlen += write_len;

        if(wpos)
            return *this;

        auto func = bind(&tcp_socket::write_handler, this, _1, _2);
        boost::asio::async_write(sock, boost::asio::buffer(wbuf, write_len), func);
        return *this;
    }

    inline tcp_socket &skip(int i) {
        p += i;

        len -= i;
        rpos += i;
        return *this;
    }

    inline bool valid() const {
        return is_valid;
    }

    const char *p;
    int len;
private:
    inline void reset() {
        p = 0;

        len = 0;
        wlen = 0;

        rpos = 0;
        wpos = 0;
    }

    inline void write_handler(const boost::system::error_code &ec, size_t) {
        if(ec) {
            if(ec == boost::asio::error::operation_aborted)
                return;

            close();
            LIB_CH_TE(0, "can't write data to socket: %s", ec.message().c_str());
        }

        wlen -= wpos;
        memmove(wbuf, wbuf + wpos, wlen);

        wpos = wlen;

        if(!wlen)
            return;

        auto func = bind(&tcp_socket::write_handler, this, _1, _2);
        boost::asio::async_write(sock, boost::asio::buffer(wbuf, wlen), func);
    }

    inline void read_handler(const boost::system::error_code &ec, size_t readed_len) {
        printf("READED LEN %d\n", readed_len);

        if(ec) {
            printf("error %d\n", ec.value());
            if(ec == boost::asio::error::operation_aborted)
                return;

            close();
            LIB_CH_TE(0, "can't read data from socket: %s", ec.message().c_str());
        }

        len += readed_len;
        memmove(rbuf, rbuf + rpos, len);

        rpos = 0;
        p = rbuf;

        int alen = sizeof rbuf - len;
        LIB_CH_TE(alen != 0, "it's need a place to read to socket");

        printf("start of async\n");
        auto func = bind(&tcp_socket::read_handler, this, _1, _2);
        sock.async_read_some(boost::asio::buffer(rbuf, alen), func);
    }

    int wlen;

    int rpos;
    int wpos;

    char rbuf[8192];
    char wbuf[8192];

    bool is_valid;

    boost::asio::io_service io;

    boost::asio::ip::tcp::socket sock;
    boost::asio::ip::tcp::resolver resolver;
};

tcp_socket s;

void on_connect() {
    char buf[55];

    *(short *)(buf + 0) = 0x64; /// packet id
    *(int *)(buf + 2) = 0; /// client version

    memcpy(buf + 6, "test", 5); /// username
    memcpy(buf + 30, "test", 5); /// password

    *(buf + 54) = 0; /// client type

    s.write(buf, 55);
}

int main() {
    s.connect("sacred-ro.ru", 9000, on_connect);

    while(1)
        s.poll();

    return 0;
}


Тесткейс. Выведет 2 раза readed len 23. А должен один раз.
PM MAIL   Вверх
boostcoder
Дата 31.7.2012, 11:26 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


pattern`щик
****


Профиль
Группа: Завсегдатай
Сообщений: 5458
Регистрация: 1.4.2010

Репутация: 49
Всего: 110
Цитата(Randajad @  31.7.2012,  10:58 Найти цитируемый пост)
Выведет 2 раза readed len 23.


Цитата(Randajad @  31.7.2012,  10:58 Найти цитируемый пост)
    inline void read_handler(const boost::system::error_code &ec, size_t readed_len) {
        printf("READED LEN %d\n", readed_len);
        if(ec) {
            printf("error %d\n", ec.value());
            if(ec == boost::asio::error::operation_aborted)
                return;
            close();
            LIB_CH_TE(0, "can't read data from socket: %s", ec.message().c_str());
        }
        len += readed_len;
        memmove(rbuf, rbuf + rpos, len);
        rpos = 0;
        p = rbuf;
        int alen = sizeof rbuf - len;
        LIB_CH_TE(alen != 0, "it's need a place to read to socket");
        printf("start of async\n");
        auto func = bind(&tcp_socket::read_handler, this, _1, _2);
        sock.async_read_some(boost::asio::buffer(rbuf, alen), func);
    }

измени этот метод так:
Код

    inline void read_handler(const boost::system::error_code &ec, size_t readed_len) {
        if(ec) {
            printf("error %d\n", ec.value());
//            if(ec == boost::asio::error::operation_aborted)
                return;
            close();
            LIB_CH_TE(0, "can't read data from socket: %s", ec.message().c_str());
        }
        printf("READED LEN %d\n", readed_len);
        len += readed_len;
        memmove(rbuf, rbuf + rpos, len);
        rpos = 0;
        p = rbuf;
        int alen = sizeof rbuf - len;
        LIB_CH_TE(alen != 0, "it's need a place to read to socket");
        printf("start of async\n");
        auto func = bind(&tcp_socket::read_handler, this, _1, _2);
        sock.async_read_some(boost::asio::buffer(rbuf, alen), func);
    }



--------------------
компилятор онлайн с поддержкой C++11
сборки компиляторов MinGW
PM WWW   Вверх
Randajad
Дата 31.7.2012, 11:31 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 295
Регистрация: 15.3.2012

Репутация: 8
Всего: 8
Ошибка там не появляется, это для другого было. smile
Я решил еще более урезать test case. Ошибка пропала.

Код

#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <boost/asio.hpp>

#define LIB_CH_TE(a, ...) { if(!(a)) { printf(__VA_ARGS__); abort(); } }

class tcp_socket {
public:
    inline tcp_socket() : sock(io), resolver(io) { }

    inline tcp_socket &connect(const char* host, int port, boost::function<void()> on_connect = 0) {
        using boost::asio::ip::tcp;
        tcp::resolver::query query(host, "http");

        auto conn_func = [this, on_connect](const boost::system::error_code &ec) {
            LIB_CH_TE(ec.value() == 0, "can't connect to host: %s", ec.message().c_str());

            printf("first start of async\n");
            auto func = bind(&tcp_socket::read_handler, this, _1, _2);
            sock.async_read_some(boost::asio::buffer(rbuf, sizeof rbuf), func);

            if(on_connect)
                on_connect();
        };

        auto res_func = [this, port, conn_func](const boost::system::error_code &ec, tcp::resolver::iterator dest) {
            LIB_CH_TE(ec.value() == 0, "can't resolve host: %s", ec.message().c_str());

            boost::asio::ip::tcp::endpoint ep(dest->endpoint().address(), port);
            sock.async_connect(ep, conn_func);
        };

        resolver.async_resolve(query, res_func);
        return *this;
    }

    inline tcp_socket &poll() {
        io.poll();
        return *this;
    }

    inline tcp_socket &write(const void *data, int write_len) {
        memcpy(wbuf, data, write_len);

        auto func = bind(&tcp_socket::write_handler, this, _1, _2);
        boost::asio::async_write(sock, boost::asio::buffer(wbuf, write_len), func);
        return *this;
    }

private:
    inline void write_handler(const boost::system::error_code &ec, size_t) {
        if(ec) {
            if(ec == boost::asio::error::operation_aborted)
                return;

            LIB_CH_TE(0, "can't write data to socket: %s", ec.message().c_str());
        }

        printf("sended\n");
    }

    inline void read_handler(const boost::system::error_code &ec, size_t readed_len) {
        printf("readed lenght %d\n", readed_len);

        if(ec) {
            printf("error %d\n", ec.value());
            if(ec == boost::asio::error::operation_aborted)
                return;

            LIB_CH_TE(0, "can't read data from socket: %s", ec.message().c_str());
        }

        printf("start of async\n");
        auto func = bind(&tcp_socket::read_handler, this, _1, _2);
        sock.async_read_some(boost::asio::buffer(rbuf, sizeof rbuf), func);
    }

    char rbuf[8192];
    char wbuf[8192];

    boost::asio::io_service io;

    boost::asio::ip::tcp::socket sock;
    boost::asio::ip::tcp::resolver resolver;
};

tcp_socket s;

void on_connect() {
    char buf[55];

    *(short *)(buf + 0) = 0x64; /// packet id
    *(int *)(buf + 2) = 0; /// client version

    memcpy(buf + 6, "test", 5); /// username
    memcpy(buf + 30, "test", 5); /// password

    *(buf + 54) = 0; /// client type

    s.write(buf, 55);
}

int main() {
    s.connect("sacred-ro.ru", 9000, on_connect);

    while(1)
        s.poll();

    return 0;
}


Правда непонятно из-за чего это все. Сейчас попробую медленно урезать.

Это сообщение отредактировал(а) Randajad - 31.7.2012, 11:32
PM MAIL   Вверх
Randajad
Дата 31.7.2012, 11:48 (ссылка) | (нет голосов) Загрузка ... Загрузка ... Быстрая цитата Цитата


Опытный
**


Профиль
Группа: Участник
Сообщений: 295
Регистрация: 15.3.2012

Репутация: 8
Всего: 8
Всем спасибо! Кажется, разобрался с ошибкой. В write забыл wpos += len. Из-за чего пакет отправлялся 2 раза, а сервер 2 раза отвечал, что странно.
Всем большое спасибо!

Это сообщение отредактировал(а) Randajad - 31.7.2012, 11:48
PM MAIL   Вверх
  
 Ответ в темуСоздание новой темы Создание опроса
Правила форума "С++:Общие вопросы"
Earnest Daevaorn

Добро пожаловать!

  • Черновик стандарта C++ (за октябрь 2005) можно скачать с этого сайта. Прямая ссылка на файл черновика(4.4мб).
  • Черновик стандарта C (за сентябрь 2005) можно скачать с этого сайта. Прямая ссылка на файл черновика (3.4мб).
  • Прежде чем задать вопрос, прочтите это и/или это!
  • Здесь хранится весь мировой запас ссылок на документы, связанные с C++ :)
  • Не брезгуйте пользоваться тегами [code=cpp][/code].
  • Пожалуйста, не просите написать за вас программы в этом разделе - для этого существует "Центр Помощи".
  • C++ FAQ

Если Вам понравилась атмосфера форума, заходите к нам чаще! С уважением, Earnest Daevaorn
 
0 Пользователей читают эту тему (0 Гостей и 0 Скрытых Пользователей)
0 Пользователей:
« Предыдущая тема | C/C++: Общие вопросы | Следующая тема »

Подписаться на тему | Подписка на этот форум | Скачать/Распечатать тему


 




[ Время генерации скрипта: 0.0551 ]   [ Использовано запросов: 22 ]   [ GZIP включён ]


Реклама на сайте     Информационное спонсорство

 
 По вопросам размещения рекламы пишите на vladimir(sobaka)vingrad.ru
Отказ от ответственности     Powered by Invision Power Board(R) 1.3 © 2003  IPS, Inc.