boost::asio::streambuf::consume - Injects garbage

Question

When I lose connection, in my server code, I try to reconnect in a loop forever. Once I reconnect, I send a login message to the component I am connected to. That component then sends back a login response that looks like "MyResponse"

The initial connection works fine. After I reconnect, however, I get garbage before the expected message, that looks like: "ýMyResponse"

After googling this. I see many questions on Stack Overflow about the boost::asio::streambuf that is used for async sockets in boost::asio. Particularly about reusing he buffer. I have followed the advice there and called consume upon disconnecting. In other words, I call boost::asio::streambuf::consume after I call shutdown and close on my socket, after being called back with an error on recv in response to a call to recv_until.

I have also used wireshark to make sure that the garbage character is not being sent and it is not.

After much debugging, it appears as though the calls to consume are injecting a character rather than clearing out all characters.

Here is a minimal example:

#include <boost/asio.hpp>

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

int main()
{
    boost::asio::streambuf buffer;

    std::cout << "buffer size " << buffer.size() << std::endl;

    buffer.consume(std::numeric_limits<size_t>::max());

    std::cout << "buffer size " << buffer.size() << std::endl;

    std::istream is(&buffer);
    std::string contents;
    is >> contents;

    std::cout << "Contents: "  << contents << std::endl;

    std::cout << "buffer size " << buffer.size() << std::endl;

    return 0;
}

Output:

buffer size 0
buffer size 1
Contents: ²
buffer size 0

Expected Output:

buffer size 0
buffer size 0
Contents:
buffer size 0

If I do not use consume, I get some of the message, previous to disconnect, before the first message after reconnection, in my server code.

If I do use consume, I get a garbage character.

See:

Working with boost::asio::streambuf

Read until a string delimiter in boost::asio::streambuf

boost asio async_read: the read message adds to itself

Answer 1

boost::asio::streambuf::consume overflows.

Use

buffer.consume(buffer.consume(buffer.size());

Rather than

buffer.consume(std::numeric_limits<size_t>::max());

Reporting bug to boost mailing list.

Answer 2

The ý character indicates a debug heap memory fence:

This means that the streambuf pointers are corrupted.

I don't have the full picture for your program, but I do note that you're not properly synchronizing writes/reads on the socket.

The Send Side

For example, looking at MyClass::Send, it appends to the existing m_sendBuffer but sends all of it again using async_write. This is

• redundant, because of anything was already in m_sendBuffer that would mean an async_write had already been started on it

• undefined behaviour, because the docs say:

  This operation is implemented in terms of zero or more calls to the stream's async_write_some function, and is known as a composed operation. The program must ensure that the stream performs no other write operations (such as async_write, the stream's async_write_some function, or any other composed operations that perform writes) until this operation completes.

• a data race: m_sendBuffer is being modified while a previous write operation could be flight

The usual approach to fixing this is to have a queue of outgoing buffers, instead of a single one, and send them in succession, e.g. boost asio async_write : how to not interleaving async_write calls?

The Receive Side

Here I don't actively spot any additional problems. The same problems from the send-side still apply though.

Once you have udefined behaviour, you can expect any behaviour, so also affecting the receive behaviour.

Most importantly, the read-loop causes a data-race to exist on m_socket: Send is triggered from the ProcessingThreadProc while the async_read loop is running on the service thread(s).

Thread Safety Of m_socket

The tcp::socket class is not thread-safe. Because you have a worker thread as well as the thread that runs the io_service (and hence the completion handlers) you cannot access m_socket without synchronization.

You should use a strand to serialize the operations on the shared object m_socket. In that case access to m_socket is already safe inside completion handlers running on that strand. All other accesses should be posted to the strand, like:

m_strand.post([this] {
    auto callback = boost::bind(&MyClass::OnSend, this, boost::asio::placeholders::error);
    async_write(m_socket, m_sendBuffer, m_strand.wrap(callback));
});

In fact, you could use the strand to synchronize access to the m_numPostedSocketIO variable, but that would remove full-duplex ability of read/write operations. Besides, you don't need the m_numPostedSocketIO counter:

Closing the socket

Note that closing the socket does already cancel all pending asynchronous IO operations. They will complete with the error code boost::asio::error::operation_aborted.

This means that you can simplify the part where you wait for pending IO operations to complete. You can simply close the socket. Notes:

• if you have multiple service threads, have a strand to serialize operations

• in that case to prevent unsynchronized access to m_socket, post on the strand:

  m_strand.post([this] { m_socket.close(); }); // e.g.
  

---