boost::asio and multiple client connections using asynch

Question

I need to establish up to three different TCP connections to different servers. All three connections requiring different protocols, different handshakes and different heartbeats. Studying http://www.boost.org/doc/libs/1_61_0/doc/html/boost_asio/example/cpp11/chat/chat_client.cpp, reading stuff here and following Chris Kohlhoffs advices I tried to implement it as below.

The problem is that with this architecture I'm getting a bad_weak_pointer exception at calling shared_from_this() in doConnect() no matter what I'm doing.

Importent These are just snippets of a not running code, which can contain bugs! Importent

I'm having a base class which is containing some basic methods.

Connection.h

class Connection : public std::enable_shared_from_this<Connection>
{
public:
  //! Ctor
  inline Connection();
  //! Dtor
  inline virtual ~Connection();
  inline void setReconnectTime(const long &reconnectAfterMilisec)
  {
    m_reconnectTime = boost::posix_time::milliseconds(reconnectAfterMilisec);
  }
  inline void setHandshakePeriod(const long &periodInMilisec)
  {
    m_handshakePeriod = boost::posix_time::milliseconds(periodInMilisec);
  }
  virtual void doConnect() = 0;
  virtual void stop() = 0;
  //... and some view more...
}

I have then my three classes which are derived from the base class. Here just one (and also the core part) to depict the approach.

ConnectionA.h

//queues which containing also the age of the messages
typedef std::deque<std::pair<handshakeMsg, boost::posix_time::ptime>> handskMsg_queue;
typedef std::deque<std::pair<errorcodeMsg, boost::posix_time::ptime>> ecMsg_queue;
typedef std::deque<std::pair<A_Msg, boost::posix_time::ptime>> A_Msg_queue;

class ConnectionA : public Connection
{
public:
  ConnectionA();
  ConnectionA(const std::string& IP, const int &port);
  ConnectionA& operator=(const ConnectionA &other);
  virtual ~ConnectionA();
  virtual void stop() override;
  virtual void doConnect() override;
  void doPost(std::string &message);
  void doHandshake();
  void sendErrorCode(const int &ec);

  std::shared_ptr<boost::asio::io_service>m_ioS;

private:
  std::shared_ptr<tcp::socket> m_socket;
  std::shared_ptr<boost::asio::deadline_timer> m_deadlineTimer; // for reconnetions
  std::shared_ptr<boost::asio::deadline_timer> m_handshakeTimer; // for heartbeats

  void deadlineTimer_handler(const boost::system::error_code& error);
  void handshakeTimer_handler(const boost::system::error_code& error);
  void doRead();
  void doWrite();

  std::string m_IP;
  int m_port;
  handskMsg_queue m_handskMsgQueue;
  ecMsg_queue m_ecMsgQueue;
  A_Msg_queue m_AMsgQueue;
}

ConnectionA.cpp

ConnectionA::ConnectionA(const std::string &IP, const int &port)
: m_ioS()
, m_socket()
, m_deadlineTimer()
, m_handshakeTimer()
, m_IP(IP)
, m_port(port)
, m_handskMsgQueue(10)
, m_ecMsgQueue(10)
, m_AMsgQueue(10)   
{
  m_ioS = std::make_shared<boost::asio::io_service>();
  m_socket = std::make_shared<tcp::socket>(*m_ioS);
  m_deadlineTimer = std::make_shared<boost::asio::deadline_timer>(*m_ioS);
  m_handshakeTimer = std::make_shared<boost::asio::deadline_timer> (*m_ioS);
  m_deadlineTimer->async_wait(boost::bind(&ConnectionA::deadlineTimer_handler, this, boost::asio::placeholders::error));
  m_handshakeTimer->async_wait(boost::bind(&ConnectionA::handshakeTimer_handler, this, boost::asio::placeholders::error));
}
ConnectionA::~ConnectionA()
{}

void ConnectionA::stop()
{
  m_ioS->post([this]() { m_socket->close(); });
  m_deadlineTimer->cancel();
  m_handshakeTimer->cancel();
}

void ConnectionA::doConnect()
{
  if (m_socket->is_open()){
    return;
  }
  tcp::resolver resolver(*m_ioS);
  std::string portAsString = std::to_string(m_port);
  auto endpoint_iter = resolver.resolve({ m_IP.c_str(), portAsString.c_str() });
  m_deadlineTimer->expires_from_now(m_reconnectTime);
  // this gives me a bad_weak_pointer exception!!!
  auto self = std::static_pointer_cast<ConnectionA>(static_cast<ConnectionA*>(this)->shared_from_this()); 
  boost::asio::async_connect(*m_socket, endpoint_iter, [this, self](boost::system::error_code ec, tcp::resolver::iterator){
    if (!ec)
    {
      doHandshake();
      doRead();
    } 
    else {
      // don't know if async_connect can fail but set the socket to open
      if (m_socket->is_open()){
        m_socket->close();
      }
    }
  });
}

void ConnectionA::doRead()
{
  auto self(shared_from_this());
  boost::asio::async_read(*m_socket, 
    boost::asio::buffer(m_readBuf, m_readBufSize), 
    [this, self](boost::system::error_code ec, std::size_t){
    if(!ec){
        // check server answer for errors
        }
        doRead();
    }
    else {
        stop();
    }
  });
}

void ConnectionA::doPost(std::string &message)
{
    A_Msg newMsg (message);
    auto self(shared_from_this());
    m_ioS->post([this, self, newMsg](){
    bool writeInProgress = false;
    if (!m_A_MsgQueue.empty()){
        writeInProgress = true;
    }
    boost::posix_time::ptime currentTime = time_traits_t::now();
    m_AMsgQueue.push_back(std::make_pair(newMsg,currentTime));
    if (!writeInProgress)
    { 
        doWrite();
    }       
  });   
}

void ConnectionA::doWrite()
{
  while (!m_AMsgQueue.empty())
  {
    if (m_AMsgQueue.front().second + m_maxMsgAge < time_traits_t::now()){
            m_AMsgQueue.pop_front();
            continue;
    }
    if (!m_socket->is_open()){
        continue;
    }
    auto self(shared_from_this());
    boost::asio::async_write(*m_socket,
      boost::asio::buffer(m_AMsgQueue.front().first.data(),
      m_AMsgQueue.front().first.A_lenght),
      [this, self](boost::system::error_code ec, std::size_t /*length*/)
    {
        if (!ec) // successful
      {
        m_handshakeTimer->expires_from_now(m_handshakePeriod); // reset timer
        m_AMsgQueue.pop_front();
        doWrite();
      }
      else {
        if (m_socket->is_open()){
          m_socket->close();
        }
      }
    });
  }
} 

void ConnectionA::deadlineTimer_handler(const boost::system::error_code& error){
  if (m_stopped){
    return;
  }
  m_deadlineTimer->async_wait(boost::bind(&ConnectionA::deadlineTimer_handler, this, boost::asio::placeholders::error));
  if (!error && !m_socket->is_open()) // timer expired and no connection was established
  {     
    doConnect();
  }
  else if (!error && m_socket->is_open()){ // timer expired and connection was established
    m_deadlineTimer->expires_at(boost::posix_time::pos_infin); // to reactivate timer call doConnect()
  }
}

And finally there is also another class which encapsulate these classes make it more comfortable to use:

TcpConnect.h

class CTcpConnect
{
public:
  /*! Ctor
  */
  CTcpConnect();
  //! Dtor
  ~CTcpConnect();

  void initConnectionA(std::string &IP, const int &port);
  void initConnectionB(std::string &IP, const int &port);
  void initConnectionC(std::string &IP, const int &port);

  void postMessageA(std::string &message);

  void run();
  void stop();
private:
  ConnectionA m_AConnection;
  ConnectionB m_BConnection;
  ConnectionC m_CConnection;
}

TcpConnect.cpp

CTcpConnect::CTcpConnect()
: m_AConnection()
, m_BConnection()
, m_CConnection()
{}

CTcpConnect::~CTcpConnect()
{}

void CTcpConnect::run(){
  [this](){ m_AConnection.m_ioS->run(); };
  [this](){ m_BConnection.m_ioS->run(); };
  [this](){ m_CConnection.m_ioS->run(); };
}

void CTcpConnect::stop(){
  m_AConnection.stop();
  m_BConnection.stop();
  m_CConnection.stop();
}

void CTcpConnect::initConnectionA(std::string &IP, const int &port)
{
  m_AConnection = ConnectionA(IP, port);
  m_AConnection.setMaxMsgAge(30000);
  //... set some view parameter more
  m_AConnection.doConnect();
}
// initConnectionB & initConnectionC are quite the same

void CTcpConnect::postMessageA(std::string &message)
{
  m_AConnection.doWrite(message);
}

In the beginning I tried also to have only one io_service (for my approach this would be fine), but holding the service just as reference gave some headache, because my implementation requires also a default constructor for the connections. Now each connection has its own io-service.

Any ideas how I can bring this code to run? Feel free to make suggestion for other architectures. If you could came up this some snippets would be even the better. I'm struggling with this implementation for weeks already. I'm grateful for every hint.

BTW I'm using boost 1.61 with VS12.

Answer 1

This is the problem:

m_AConnection = ConnectionA(IP, port);

That is, ConnectionA derives from Connection which derives from enable_shared_from_this. That means that ConnectionA must be instantiated as a shared pointer for shared_from_this to work.

Try this:

void CTcpConnect::initConnectionA(std::string &IP, const int &port)
{
  m_AConnection = std::make_shared<ConnectionA>(IP, port);
  m_AConnection->setMaxMsgAge(30000);
  //... set some view parameter more
  m_AConnection->doConnect();
}

EDIT1:

You are right. That was the issue. Now I realised that the way I'm calling io-service.run() is total crap.

It is very uncommon to use more than one io_service, and extremely uncommon to use one per connection :)

However, do you know if I need the cast then calling shared_from_this()? I noticed the asynch_connect() works fine with and without the cast.

Many Asio examples use shared_from_this() for convenience, I for example don't use it in my projects at all. There are certain rules that you need to be careful when working with Asio. For example, one is that the reading and writing buffers must not be destructed before the corresponding callback is executed, if the lambda function captures a shared pointer to the object that holds the buffers, this condition holds trivially.

You could for example do something like this as well:

auto data = std::make_shared<std::vector<uint8_t>>(10);
async_read(socket,
           boost::asio::const_buffer(*data),
           [data](boost::system::error_code, size_t) {});

It would be valid but would have the performance drawback that you'd be allocating a new data inside std::vector on each read.

Another reason why shared_from_this() is useful can be seen when you look at some some of your lambdas, they often have the form:

[this, self,...](...) {...}

That is, you very often want to use this inside them. If you did not capture self as well, you'd need to use other measures to make sure this has not been destroyed when the handler is invoked.