Boost Asio and asynchronous I/O

I'm just going to check if anyone has any experience with Boost Asio here...
I can't get asynchronous I/O to work properly...
This code may be a little long, but it is simple:

Code:

#include <memory>
#include <algorithm>
#include <windows.h>
#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/cstdint.hpp>
#include <boost/scope_exit.hpp>
#include <boost/bind.hpp>

using namespace boost;
using asio::ip::tcp;
namespace ip = asio::ip;

class tcp_connection
{
protected:
	std::shared_ptr<tcp::socket> m_socket;

public:
	tcp_connection(asio::io_service& io_service): m_socket(new tcp::socket(io_service)) {}
	tcp::socket& get_socket() { return *m_socket; }
};

void handle_request(tcp_connection& connection, const boost::system::error_code& err);

int APIENTRY _tWinMain(HINSTANCE /*hInstance*/,
	HINSTANCE /*hPrevInstance*/,
	LPTSTR    /*lpCmdLine*/,
	int       /*nCmdShow*/)
{
	asio::io_service io_service;
	tcp::acceptor acceptor(io_service, tcp::endpoint(tcp::v4(), 1));

	for (;;)
	{
		tcp_connection connection(io_service);
		boost::system::error_code err;
		acceptor.accept(connection.get_socket(), err);
		boost::thread thread(&handle_request, connection, err);
	}
}

std::vector<tcp::endpoint> g_SmsGenerators;

const char MsgSmsNotAvailable = 1;
const char MsgReplyOk = 5;			// Data was recieved; OK to send more now

class XSocket
{
public:
	XSocket(tcp::socket& socket): m_socket(socket) 
	{
		m_Event = CreateEvent(nullptr, FALSE, FALSE, L"");
	}

	template<typename T> void send(const T& data, boost::uint32_t Timeout)
	{
		const T _data[1] = { data };

		//m_socket.get_io_service().run();
		asio::async_write(m_socket, asio::buffer(_data), boost::bind(&XSocket::handle_send_complete, this, _1, _2));
		m_socket.get_io_service().run();
		unsigned long status = WaitForSingleObject(m_Event, Timeout * 1000);
		m_socket.cancel();
	}

private:
	void handle_send_complete(const boost::system::system_error& err, std::size_t bytes)
	{
		SetEvent(m_Event);
	}

public:
	template<typename T> T get()
	{
		std::vector<char> buf(1024);
		m_socket.receive(asio::buffer(buf));

		char reply[1] = { MsgReplyOk };
		m_socket.send(asio::buffer(reply));

		return boost::lexical_cast<T>(&buf[0]);
	}

protected:
	tcp::socket& m_socket;
	HANDLE m_Event;
};

void handle_request(tcp_connection& connection, const boost::system::error_code& err)
{
	XSocket socket(connection.get_socket());
	auto data = socket.get<int>();

	switch (data)
	{
		case MsgRequestSms:
			socket.send(MsgSmsNotAvailable, 10);
	}
}

The problem is basically that at the SECOND client connection, the send complete handler is never run (handle_send_complete). Instead, the wait times out. See XSocket::send.
What is worse, on the THIRD client connection, it sees that the handler for the second write is complete and calls handle_send_complete. However, by then, the associated instance of XSocket is long gone, and it crashes horribly.

What I don't understand is how to make sure that bloody library calls handle_send_complete EVERY time the send finishes. It's obviously important.

Any insight whatsoever would be appreciated.

I assume you realize that your code creates a new thread for each new client connection. [1] That's generally not the best way to use Boost Asio. You can, if you wish, use Boost Asio like other socket libraries in a synchronous manner (call available() to check for data, read() to read it, etc.) and maintain your own event loop, which can be convenient in e.g. a single-threaded game (I did this once quite successfully). You can also make use of the asynchronous I/O capabilities and handle all of the socket events in one thread (or a pool of threads if you want to be really fancy). This is harder to set up, but more convenient.

So as to your problem: I can only guess that your accept() loop is fishy. Are the other threads actually being created? Print something at the beginning of handle_request(). Try using async_accept() and then calling run() on the io_service.

Another potential issue: with Boost Asio you have to re-register a handler each time one is called, if you want it to be called again. Notice how here handle_accept re-registers itself, for example: doc/html/boost_asio/example/echo/async_tcp_echo_server.cpp - Boost 1.46.1
Just something to keep in mind when you're moving on to larger Asio programs.

Also, having send() itself call io_service.run() is very strange. I don't have Boost Asio on this computer so I can't test anything, unfortunately, but a general format you could follow is this:

Code:

create io_service to be used throughout the code
create server socket
register async_accept_handler on the server socket
call io_service.run()

async_accept_handler() {
    get new socket and add it to a list somewhere
    register async_read_handler for new socket
    reregister async_accept_handler
}

async_read_handler() {
    do whatever with the data
    reregister async_read_handler
}

You see, run() doesn't return until it has no more work to do. Work includes listening on a server socket. You only have to put run() in a loop if you're using run_once() or if the io_service might run out of work, e.g. you're not always listening on a port or something. And you just pile more and more async callbacks into the same io_service. A read handler for each new socket that you accept(), and so on.

I assume you've found the other Boost Asio examples. They're a lot more complicated than they need to be, but combining them with looking up all the functionality in the reference docs was pretty much how I got started . . . .

Let me know if it would be helpful for you to see a working example -- I can write one up based on my own projects. I'm not sure how helpful it would be because I find it hard to read other peoples' Asio code, or maybe it's just the tutorials. But do let me know. Sorry for rambling, and good luck!

[1] Quick side note about Boost threads: if you start using a class which overloads operator()() instead of a bare function, you have to be careful. When you construct the object for the boost thread, the constructor you originally call will be in the original thread context. The object will then be copied somewhere (I believe the other thread's stack) with the object's copy constructor. Usually the original object was created inline and then gets destroyed. So don't, like me, go "new X" in the default constructor and then "delete X" in the destructor because then X will be freed by the time the new thread tries to access it.

OK, so first... I am not very experienced with Boost Asio, or networking in general. I simply had the need to use a server and client model, so I figured Boost Asio would help here (since it's portable, it's boost, and all). All of my code are just based on experiments, whatever documentation and examples I could find. If there is anything wrong, then I would gladly accept advice on how it's properly done.
I'll shortly explain my model here, just in case.

Let's say there are three parts: the producer, the consumer and the server. The producer produces what I need, and the consumers need this item. The server is basically the middle hand, as it is.
So the basic model is this: producer starts, notifies server that it has started. Consumer requests an item from server, which checks if there are any producers, then requests an item from one of the producers and sends it to the consumer.
So it's like a web server. It needs to be able to handle a lot of connections simultaneously. I generally find Asio lacking in the async department. Either it's very tricky to use, or it's useless. Such as the async_accept, which requires me to bind to a socket, making it difficult for me to pass it along to the thread that will handle the request. I tried allocating it on the heap, yet to no avail. An example is welcome here, if forking off a thread like this is bad.

But what do you mean by "the other threads being created"? I know for sure that incoming connections are accepted and spawned off to threads. Everything works fine until the second connection. Using synchronous I/O, all is fine. I tested this, but I need to be able to terminate the connection with some specified timeout (which, again, the async calls doesn't support).

How does one "register" a accept handler?

Also, from my experience, io_service.run() finishes up all events, then exits. So if I put it at the beginning of the program, nothing happens. In order to actually get any notification at all to the read handler, I have to put a run inside the send method. Perhaps it's wrong, I don't know. I just know that it's the only way I've gotten it to work.

You can probably understand that having a good example with proper explanations of how asio works would probably be a good thing. The documentation is poor as always when it comes to boost.

ZeroMQ might also fit the bill here.

Sorry, it's built by Linux people, and they know, as usual, nothing about Windows. Not touching that.

Originally Posted by Elysia

Sorry, it's built by Linux people, and they know, as usual, nothing about Windows. Not touching that.

So you have tried and/or inspected that library and found that the authors know nothing about Windows? If not, such a dismissal of a library that appears to cater to Windows is unwarranted bias that does not serve you well.

That might have sounded a little harsh. For that, I'm sorry.
But yes, I inspected the library and downloaded it. And it's your typical Linux library. Configure, make, install. Which works great on Linux, I bet. But on Windows, it's a pain. And the instructions are your generic Linux instructions for all versions. Ie, cd here, do ./configure, ./make, etc.
And the fact that you have to compile the source also. This is just a pain and why I hate "Linux" libraries.

Also, many Linux people tend to take for granted that posix is available and tend to use GNU extensions, and of course, take for granted that GCC is installed, configured, etc, etc. But the point is... I don't use make. I don't use configure. I don't use posix. I don't use GNU extensions. And I don't use GCC. So it will take hours of pain and annoyance to even get it to compile properly.

So yeah, thanks for the suggestion. Unfortunately, I'm not going to use it.

But yes, I inspected the library and downloaded it. And it's your typical Linux library. Configure, make, install. Which works great on Linux, I bet. But on Windows, it's a pain. And the instructions are your generic Linux instructions for all versions. Ie, cd here, do ./configure, ./make, etc.
And the fact that you have to compile the source also. This is just a pain and why I hate "Linux" libraries.

WTF are you talking about, Elysia?

To build on Windows

You need Microsoft Visual C++ 2008 or newer.
Unpack the .zip source archive.
In Visual C++ open the solution builds\msvc\msvc.sln.
Build the solution.
ØMQ libraries will be in the lib subdirectory.

From here.

Thank Og I don't have to work with you, for as bright and talented as you may be, your irrational hatred of things you feel are "typical Linux" or "not Windows" make you a shortsighted bigot.

That was very nice of them. So why didn't they put clear instructions in the install file that said:

"Open X.sln. Build. Done."

Instead of having all the configure and make crap?
My hatred or dislike typically come from the fact that I don't find it to work too well with my style. Linux just isn't for me. I tried that, and I know. I would rather not have to spend hours trying to get something to work because I'm not good at it or don't like it, if I don't have any obligations.
You don't need to use software X if you don't like it. Likewise, I do not have to use Linux if I don't like it.
It's such a common thing in our lives, and we all make choices about how we do stuff and what stuff we use.

Sorry to resurrect this thread, but Elysia requested a Boost.Asio example, and I think that it's related enough to this thread to continue with here. If a moderator disagrees please feel free to split it.

client.cpp:

Code:

#include <iostream>
#include <sstream>
#include <string>

#include "boost/asio/io_service.hpp"
#include "boost/asio/read.hpp"
#include "boost/asio/ip/tcp.hpp"

#include "boost/shared_ptr.hpp"

#define PORT 4423  // a macro so you can change this at compile time

using namespace boost::asio::ip;

class ClientSocket {
private:
    // pointer to io_service object, which should be shared between sockets
    boost::shared_ptr<boost::asio::io_service> io_service;
    boost::shared_ptr<tcp::socket> socket;
public:
    ClientSocket(boost::shared_ptr<boost::asio::io_service> io_service)
        : io_service(io_service) {}
    
    /** Connects this client socket to a server socket at the given
        @a host and @a port.
        
        @throws boost::system::system_error
    */
    void open(const std::string &host, unsigned short port);
    
    bool read(std::string &data, int size);
};

void ClientSocket::open(const std::string &host, unsigned short port) {
    // convert port number to string (I believe named ports work too, e.g. "ssh")
    std::ostringstream portStream;
    portStream << port;
    
    tcp::resolver resolver(*io_service);
    tcp::resolver::query query(host, portStream.str());
    tcp::resolver::iterator it = resolver.resolve(query);  // throws boost::system::system_error
    tcp::resolver::iterator end;
    
    socket = boost::shared_ptr<tcp::socket>(new tcp::socket(*io_service));
    
    // Loop through all the possible ways of resolving the given host and port,
    // trying each one until one address can be successfully connected to.
    boost::system::error_code error = boost::asio::error::host_not_found;
    while(error && it != end) {
        if(socket->is_open()) socket->close();
        socket->connect(*it++, error);  // sets error instead of throwing
    }
    
    // Most boost::asio functions have two variants, one which throws an
    // exception upon error, and one that sets a boost::system::error_code
    // parameter.
    // 
    // Here we've used an error code because we might have multiple endpoints
    // to try connecting to. But by this point if there is still an error we
    // could not connect, so mimic the boost::asio behaviour by throwing an
    // exception.
    if(error) {
        throw boost::system::system_error(error);
    }
}

bool ClientSocket::read(std::string &data, int size) {
    std::vector<char> buffer(size);
    
    boost::system::error_code error;
    boost::asio::read(*socket,
        boost::asio::buffer(buffer),
        boost::asio::transfer_all(), error);
    
    if(!error) {
        // There's no easy way to read into an std::string directly, but it's
        // possible to read into an std::vector and then use the fact that
        // vectors are guaranteed to be contiguous to construct a string from
        // the underlying raw string. It's also possible to use a boost::array
        // instead.
        data = std::string(&buffer[0], buffer.size());
        
        return true;
    }
    
    return false;
}

int main() {
    try {
        // One io_service should be used to control all sockets. This allows
        // e.g. async events for multiple sockets to be handled in one place,
        // which can be more efficient if boost is using an operating system
        // mechanism like select() under the hood.
        boost::shared_ptr<boost::asio::io_service> io_service(
            new boost::asio::io_service());
        
        ClientSocket socket(io_service);
        socket.open("localhost", PORT);  // throws on error
        
        std::string data;
        if(socket.read(data, 10)) {
            std::cout << "client: Read 10 bytes: <" << data << ">\n";
        }
        else {
            std::cout << "client: Error reading data\n";
        }
    }
    catch(std::exception &e) {
        std::cerr << "client: Exception: " << e.what() << std::endl;
    }
    
    return 0;
}

server.cpp:

Code:

#include <iostream>
#include <string>

#include "boost/asio/io_service.hpp"
#include "boost/asio/ip/tcp.hpp"
#include "boost/asio/write.hpp"

#define PORT 4423  // a macro so you can change this at compile time

using namespace boost::asio::ip;

class ServerSocket {
private:
    // pointer to io_service object, which should be shared between sockets
    // (newly accepted connections use this io_service)
    boost::shared_ptr<boost::asio::io_service> io_service;
    boost::shared_ptr<tcp::acceptor> acceptor;
public:
    ServerSocket(boost::shared_ptr<boost::asio::io_service> io_service)
        : io_service(io_service) {}
    
    /** Begin listening on the given port.
    */
    void listen(unsigned short port);
    
    /** Accept a single connection on the server socket.
        
        Must be called after listen().
    */
    void accept();
};

void ServerSocket::listen(unsigned short port) {
    acceptor = boost::shared_ptr<tcp::acceptor>(
        new tcp::acceptor(
            *io_service,
            tcp::endpoint(tcp::v4(), port)));
}

void ServerSocket::accept() {
    tcp::socket socket(*io_service);
    
    // This will block until a client connection is detected, at which point
    // the socket variable will be initialized and ready to use at this end.
    acceptor->accept(socket);
    
    std::string message = "Greetings.";  // 10 characters
    boost::asio::write(socket, boost::asio::buffer(message));
    
    std::cout << "server: Sent message \"" << message << "\"\n";
    
    // automatically close the socket when the variable goes out of scope
    // (use a boost::shared_ptr to prevent this from happening)
}

int main() {
    try {
        boost::shared_ptr<boost::asio::io_service> io_service(
            new boost::asio::io_service());
        
        ServerSocket socket(io_service);
        
        socket.listen(PORT);  // throws on error
        
        std::cout << "server: Listening on port " << PORT << " ...\n";
        
        // This function blocks until a connection is received on the port
        // we're listening on. If this was a real server we could do this in
        // a loop, but here we just want to deal with one client.
        socket.accept();
    }
    catch(std::exception &e) {
        std::cerr << "server: Exception: " << e.what() << std::endl;
    }
    
    return 0;
}

Compile with

Code:

$ g++ client.cpp -o client -lboost_system
$ g++ server.cpp -o server -lboost_system

Boost.Asio is for the most part (maybe entirely?) a headers-only library, so there's nothing to link with for it. I used some code from Boost.System in the examples; actually maybe only the client requires the library to be linked in, I didn't try the server without it.

This dependency is not required: you can call almost any Boost.Asio function with a variant that throws exceptions, or a variant that takes a boost::system::error_code. If you want to avoid Boost.System dependencies I believe you can use only the exception variants.

Make sure port 4423 is open in your firewall, launch the server in one terminal and launch the client in another. The server will send a 10-byte message to the client and both programs will exit.

This is just a quick example to get you started. You'll soon run into problems if you try to build a larger program: for example, what if you want to read from multiple sockets at once? One way to do it is to call the .available() function to see if enough data is ready, and so not use blocking function calls. Unfortunately you can't use this method to detect if a socket has been disconnected (I did some research into this at one point and I believe it's an operating system limitation).

At some point you'll start using the asynchronous boost asio functions, for these reasons or others. The idea is simple: there are versions of functions called e.g. async_accept() to replace accept(). You pass in a function (which can be a method, if you use boost::bind) to be called when the operation completes. Your code goes on its merry way and whenever you call the run() method of the associated io_service, any waiting events will have their handlers called.

You can get away with using just synchronous I/O for a first example however, and I recommend it or else you may find Asio too confusing. I made the networking for a fairly large multiplayer game in this way without too much difficulty, although I had to write some code to e.g. detect disconnects with ping packets. In more recent projects I've found asynchronous I/O to be more useful, but it took quite some time to be understandable.

Good luck in your Asio adventures.