[asio] Async (or sync) in threads

24 Jul 2013

      Hello,

I am having a hard time getting my Asio usage off the ground.

I started with what I thought was a nominal (knock on wood)
asynchronous server, which runs in a worker thread, but was running
into seg fault issues.

In the interest of "simplifying" the issue, I decided to simplify to a
"blocking" server, hopefully also hosted in a thread, but am also
running into seg fault issues.

First I am trying to capture the server instance in a lambda passing
it to a thread as a runner. Then I am trying to capture a shared
pointer to the server thinking there is some move semantic or
something. Seg fault in either case.

Most of the examples are one main being the sole main thread hosting
the io_service. Would that this were the case with me. This has to be
one worker among several other workers running in the same process.

Wouldn't mind if there were a couple of ideas recommending what I
might explore next.

Thank you...

In the diagnostic routine I've got wired up:

    boost::shared_ptrdchem::asio::ip::tcp::server2 server_
    = boost::make_shareddchem::asio::ip::tcp::server2(17017);

    boost::thread runner_([&server_]() {
        server_->start_server();
    });
    runner_.start_thread();

    while (!timer_.has_elapsed_query_wait(wait_timeout_,
elapsed_milliseconds_)) {
        //...
    }

    runner_.interrupt();
    runner_.join();

And the server itself, header first:

class server2 {
public:

    /**
     * \brief Endpoint type definition.
     */
    typedef boost::asio::ip::tcp::endpoint endpoint_type;

    /**
     * \brief Acceptor type definition.
     */
    typedef boost::asio::ip::tcp::acceptor acceptor_type;

    /**
     * \brief Service type definition.
     */
    typedef boost::asio::io_service service_type;

    /**
     * \brief Socket type definition.
     */
    typedef boost::asio::ip::tcp::socket socket_type;

    /**
     * \brief Socket pointer type definition.
     */
    typedef boost::shared_ptr socket_pointer;

    /**
     * \brief Port type definition.
     */
    typedef unsigned int port_type;

protected:

    /**
     * \brief Returns whether bytes are available.
     * \param s A socket reference.
     */
    static bool are_bytes_readable(socket_type& s);

    /**
     * \brief Starts a new session.
     * \param s Starts running a new session.
     */
    void session(socket_pointer s);

public:

    /**
     * \brief Constructor
     * \param port A TCP port.
     */
    server2(port_type port);

    /**
     * \brief Destructor
     */
    virtual ~server2();

    /**
     * \brief Starts the server.
     */
    void start_server();

private:

    /**
     * \brief Service.
     */
    service_type m_service;

    /**
     * \brief Port.
     */
    port_type m_port;
};

/////////////////////////////////////////////////////////////////////
server2::server2(port_type port)
    : m_service(),
      m_port(port) {
}

/////////////////////////////////////////////////////////////////////
server2::~server2() {
}

/////////////////////////////////////////////////////////////////////
bool server2::are_bytes_readable(socket_type& s) {

    boost::asio::socket_base::bytes_readable cmd_(true);
    s.io_control(cmd_);
    std::size_t cmd_result_ = cmd_.get();
    return cmd_result_ > 0;
}

/////////////////////////////////////////////////////////////////////
void server2::start_server() {

    try {

        endpoint_type ep_(boost::asio::ip::tcp::v4(), m_port);
        acceptor_type a_(m_service, ep_);

        std::cout << "accepting connection" << std::endl;

        boost::system::error_code ec_;

        auto s_ = boost::make_shared(m_service);
        a_.accept((*s_), ec_);

        if (!ec_) return;

        std::cout << "starting session" << std::endl;

        session(s_);
    }
    catch (boost::thread_interrupted& tiex) {
    }
    catch (...) {
    }
}

/////////////////////////////////////////////////////////////////////
void server2::session(socket_pointer s) {

    auto& s_ = (*s);

    while (true) {

        {
            utils::sleep_for_milliseconds(100);

            threading::thread_interruption_disabler tid;

            std::cout << "are bytes readable" << std::endl;

            if (!are_bytes_readable(s_)) continue;

            std::cout << "reading some" << std::endl;

            boost::system::error_code ec_;
            std::vector<byte> buffer_;

            size_t length_ = s_.read_some(boost::asio::buffer(buffer_), ec_);

            //Connection reset cleanly by peer.
            if (ec_ == boost::asio::error::eof)
                break;
            else if (ec_) //Some other error.
                throw boost::system::system_error(ec_);

            boost::asio::write(s_, boost::asio::buffer(buffer_, length_));
        }
    }
}

Regards,

Michael Powell

Michael Powell

Michael Powell

Anthony Foiani

Michael Powell

Michael Powell

Michael Powell

Anthony Foiani

Michael Powell

Anthony Foiani

Michael Powell

tags

participants (2)