I am working on an application where the user will have flexibility to create and destroy a server instance at any port he/she wishes. User can create multi servers operating simultaneously - each dedicated for a particular client and independent of other servers.I have been working on this for quite sometime now, and finally reached a state where the code is working satisfactorily.I have referred below examples:Chat.HTTP: Single Threaded Implementation.I had real trouble maintaining a list of active servers and destroying any one at will, was throwing exceptions weirdly. Finally I got it working, Could some one please review it?I have concerns mainly for:How to use io_service::run()? The way I am using in my code, is it satisfactory?How to ensure clean destruction of server when calling commManager::dropServer()?Clean destruction of connection object with respect to ongoing operations and sockets in open state.Here is my code: Compiled with Visual Studio 2013.Is this code OK to be operated in multithreaded environment?buffer.hxx:#pragma once#include <string>#include <iostream>#include <iomanip> //==========================================================================================================================!namespace http{//#define DEBUG ON#ifdef DEBUG#define DEBUG_MSG(str) do {std::cout << std::setw(75) << std::left  << __FUNCTION__ \    << std::setw(3) << std::left << : << std::setw(5) << std::left << __LINE__ \    << std::setw(5) << std::left << :\    << std::left  << str \    << std::endl;} while( false )#else#define DEBUG_MSG(str) do { } while ( false )#endif    class CBuffer    {    public:        enum { buffsize = 32 };        CBuffer();        CBuffer(const std::string str);        bool empty() const;        const std::string getString() const;        const char* getReceived() const;    private:        char received_[buffsize];    };}buffer.cpp:#include buffer.hxx#include <string>#include <mutex>using namespace http;CBuffer::CBuffer(){    received_[0] = '\0';}CBuffer::CBuffer(const std::string str){    //Truncate if Overflow    auto len = str.size();    if (len >= buffsize)    {        len = buffsize - 1;    }    std::copy(str.begin(), str.begin() + len, received_);    received_[len] = '\0';}bool CBuffer::empty() const{    auto i = 0;    //No need for critical sections as this is read only fn    //and no method to update shared member data except the construction    while (received_[i] != '\0')        i++;    if (i == 0)        return true;    else        return false;}const std::string CBuffer::getString() const{    //No need for critical sections as this is read only fn    //and no method to update shared member data except the construction    return std::string(received_);}const char* CBuffer::getReceived() const{    //No need for critical sections as this is read only fn    //and no method to update shared member data except the construction    return received_;}server.hpp:#ifndef HTTP_SERVER_HPP#define HTTP_SERVER_HPP#include <asio.hpp>#include <string>#include <boost/noncopyable.hpp>#include connection.hpp#include connection_manager.hppnamespace http {    class connection;    typedef boost::shared_ptr<connection> connection_ptr;    using weakPtrConnection = boost::weak_ptr < connection > ;    namespace server {        /// The top-level class of the HTTP server.        class server            : public boost::enable_shared_from_this<server>, private boost::noncopyable        {        public:            /// Construct the server to listen on the specified TCP address and port, and            /// serve up files from the given directory.            explicit server(/*asio::io_service& io,*/const std::string& address, const std::string& port);            ~server();            /// Run the server's io_service loop.            void run();            void stop();            void Init();        private:            /// Initiate an asynchronous accept operation.            void start_accept();            /// Handle completion of an asynchronous accept operation.            void handle_accept(const asio::error_code& e);            /// Handle a request to stop the server.            void handle_stop();            /// The io_service used to perform asynchronous operations.            asio::io_service io_service_;            /// The signal_set is used to register for process termination notifications.            asio::signal_set signals_;            /// Acceptor used to listen for incoming connections.            asio::ip::tcp::acceptor acceptor_;            /// The connection manager which owns all live connections.            connection_manager connection_manager_;            /// The next connection to be accepted.            connection_ptr new_connection_;            std::string IP_;            std::string port_;        };    } // namespace server} // namespace http#endif // HTTP_SERVER_HPPserver.cpp:#include server.hpp#include <boost/bind.hpp>#include <signal.h>#include <boost/thread.hpp>#include <boost/bind.hpp>namespace http {    namespace server {        server::server(/*asio::io_service& io,*/const std::string& address, const std::string& port)            : io_service_(/*io*/),            signals_(io_service_),            acceptor_(io_service_),            connection_manager_(),            new_connection_(), IP_(address), port_(port)        {            // Register to handle the signals that indicate when the server should exit.            // It is safe to register for the same signal multiple times in a program,            // provided all registration for the specified signal is made through Asio.            DEBUG_MSG(Created);            signals_.add(SIGINT);            signals_.add(SIGTERM);#if defined(SIGQUIT)            signals_.add(SIGQUIT);#endif // defined(SIGQUIT)            //signals_.async_wait(boost::bind(&server::handle_stop, this));        }        void server::Init()        {            // Open the acceptor with the option to reuse the address (i.e. SO_REUSEADDR).            asio::ip::tcp::resolver resolver(io_service_);            asio::ip::tcp::resolver::query query(IP_, port_);            asio::ip::tcp::endpoint endpoint = *resolver.resolve(query);            acceptor_.open(endpoint.protocol());            acceptor_.set_option(asio::ip::tcp::acceptor::reuse_address(true));            acceptor_.bind(endpoint);            acceptor_.listen();            start_accept();        }        server::~server()        {            DEBUG_MSG(Destroyed);            io_service_.stop();        }        void server::run()        {            // The io_service::run() call will block until all asynchronous operations            // have finished. While the server is running, there is always at least one            // asynchronous operation outstanding: the asynchronous accept call waiting            // for new incoming connections.            //io_service_.run();            DEBUG_MSG(Run);            //io_service_.run();            //asio::thread t(boost::bind(&asio::io_service::run, &(io_service_)));            boost::thread t(boost::bind(&asio::io_service::run, &io_service_));        }        void server::start_accept()        {            DEBUG_MSG(Accepting connection);            new_connection_.reset(new connection(io_service_,                connection_manager_));            acceptor_.async_accept(new_connection_->socket(),                boost::bind(&server::handle_accept, shared_from_this(),                 asio::placeholders::error));        }        void server::handle_accept(const asio::error_code& e)        {            // Check whether the server was stopped by a signal before this completion            // handler had a chance to run.            if (!acceptor_.is_open())            {                DEBUG_MSG(Acceptor Not Open);                return;            }            if (!e)            {                DEBUG_MSG(Accepted);                connection_manager_.start(new_connection_);            }            start_accept();        }        void server::handle_stop()        {            // The server is stopped by cancelling all outstanding asynchronous            // operations. Once all operations have finished the io_service::run() call            // will exit.            DEBUG_MSG(Stopping Acceptor);            connection_manager_.stop_all();            acceptor_.close();        }        void server::stop()        {            DEBUG_MSG(Request posted to stop server);            io_service_.post(boost::bind(&server::handle_stop, shared_from_this()));        }    } // namespace server} // namespace httpconnection_manager.hpp:#ifndef HTTP_CONNECTION_MANAGER_HPP#define HTTP_CONNECTION_MANAGER_HPP#include <set>#include <boost/noncopyable.hpp>#include connection.hppnamespace http {    namespace server {        /// Manages open connections so that they may be cleanly stopped when the server        /// needs to shut down.        class connection_manager            : private boost::noncopyable        {        public:            connection_manager();            ~connection_manager();            /// Add the specified connection to the manager and start it.            void start(connection_ptr c);            /// Stop the specified connection.            void stop(connection_ptr c);            /// Stop all connections.            void stop_all();        private:            /// The managed connections.            std::set<connection_ptr> connections_;        };    } // namespace server} // namespace http#endif // HTTP_CONNECTION_MANAGER_HPPconnection_manager.cpp:#include connection_manager.hpp#include <algorithm>#include <boost/bind.hpp>namespace http {    namespace server {        connection_manager::connection_manager()        {            DEBUG_MSG(Created);        }        connection_manager::~connection_manager()        {            DEBUG_MSG(Destroyed);        }        void connection_manager::start(connection_ptr c)        {            DEBUG_MSG(Session Added);            connections_.insert(c);            c->start();        }        void connection_manager::stop(connection_ptr c)        {            DEBUG_MSG(Session Deleted);            connections_.erase(c);            c->stop();        }        void connection_manager::stop_all()        {            DEBUG_MSG(Request to stop all);            std::for_each(connections_.begin(), connections_.end(),                boost::bind(&connection::stop, _1));            connections_.clear();        }    } // namespace server} // namespace httpconnection.hpp:#include connection_manager.hpp#include <algorithm>#include <boost/bind.hpp>namespace http {    namespace server {        connection_manager::connection_manager()        {            DEBUG_MSG(Created);        }        connection_manager::~connection_manager()        {            DEBUG_MSG(Destroyed);        }        void connection_manager::start(connection_ptr c)        {            DEBUG_MSG(Session Added);            connections_.insert(c);            c->start();        }        void connection_manager::stop(connection_ptr c)        {            DEBUG_MSG(Session Deleted);            connections_.erase(c);            c->stop();        }        void connection_manager::stop_all()        {            DEBUG_MSG(Request to stop all);            std::for_each(connections_.begin(), connections_.end(),                boost::bind(&connection::stop, _1));            connections_.clear();        }    } // namespace server} // namespace httpconnection.cpp:#include connection.hpp#include <vector>#include <boost/bind.hpp>#include connection_manager.hppnamespace http {    namespace server {        connection::connection(asio::io_service& io_service,            connection_manager& manager)            : socket_(io_service),            connection_manager_(manager)        {            DEBUG_MSG(Session Created);        }        connection::~connection()        {            DEBUG_MSG(Session Destroyed);        }        asio::ip::tcp::socket& connection::socket()        {            return socket_;        }        void connection::start()        {            DEBUG_MSG(Reading);            socket_.async_read_some(asio::buffer(const_cast<char *> (buffer_.getReceived()), buffer_.buffsize),                boost::bind(&connection::handle_read, shared_from_this(),                asio::placeholders::error,                asio::placeholders::bytes_transferred));        }        void connection::stop()        {            DEBUG_MSG(Closed);            socket_.close();        }        void connection::handle_read(const asio::error_code& e,            std::size_t bytes_transferred)        {            if (!e)            {                //DEBUG_MSG(buffer_.getString());                std::cout << buffer_.getString() << std::endl;                start();            }            else if (e != asio::error::operation_aborted)            {                DEBUG_MSG(Read Error);                connection_manager_.stop(shared_from_this());            }        }        void connection::handle_write(const asio::error_code& e)        {            if (!e)            {                // Initiate graceful connection closure.                asio::error_code ignored_ec;                socket_.shutdown(asio::ip::tcp::socket::shutdown_both, ignored_ec);            }            if (e != asio::error::operation_aborted)            {                connection_manager_.stop(shared_from_this());            }        }    } // namespace server} // namespace httpmain.cpp:#include <iostream>#include <string>#include <asio.hpp>#include <boost/bind.hpp>#include server.hpp#include <boost/thread.hpp>#include <set>using sharedPtr = boost::shared_ptr < http::server::server > ;using weakPtr = boost::weak_ptr < http::server::server > ;class commManager{public:    commManager()    {        DEBUG_MSG(Created);    }    ~commManager()    {        DEBUG_MSG(Destroyed);    }    weakPtr createServer(std::string IP, std::string port)    {        DEBUG_MSG(Creating Server);        auto ptr = boost::shared_ptr<http::server::server>(new http::server::server(IP, port));        //auto ptr = std::make_shared<http::server::server>(IP, port);        servers_.insert(ptr);        ptr->Init();        ptr->run();        return weakPtr(ptr);    }    void dropServer(weakPtr ptr)    {        DEBUG_MSG(Dropping Server);        ptr.lock()->stop();        servers_.erase(ptr.lock());    }private:    std::set<sharedPtr> servers_;};int main(){    try    {        commManager mng;        auto s1 = mng.createServer(/*io,*/127.0.0.1, 8973);        auto s2 = mng.createServer(/*io,*/127.0.0.1, 8974);        // Run the server until stopped.        boost::this_thread::sleep_for(boost::chrono::seconds(300));        mng.dropServer(s1);        mng.dropServer(s2);        boost::this_thread::sleep_for(boost::chrono::seconds(15));    }    catch (std::exception& e)    {        std::cerr << exception:  << e.what() << \n;    }    system(Pause);    return 0;}Client Code to Test the server:ioservice.hxx:#include <asio.hpp>using asio::ip::tcp;namespace channel{    class CIOService{    public:        CIOService(const CIOService & rhs) = delete;        CIOService & operator = (const CIOService & rhs) = delete;        static CIOService & fetchIOService();        std::shared_ptr<asio::io_service> getIO();    private:        CIOService();        ~CIOService();    private:        std::shared_ptr<asio::io_service> pIO_;    };}ioservice.cpp:#include ioservice.hxx#include <iostream>using namespace channel;CIOService & CIOService::fetchIOService(){    static CIOService IO;    return IO;}CIOService::CIOService(){    pIO_ = std::make_shared<asio::io_service>();    std::cout << IOService Created << std::endl;}std::shared_ptr<asio::io_service> CIOService::getIO(){    return pIO_;}CIOService::~CIOService(){}session.hxx:#pragma once#include <iostream>#include <asio.hpp>#include <memory>#include <deque>using asio::ip::tcp;namespace channel {#define defaultIP 127.0.0.1   #define defaultPort 8000    struct Buffer    {        enum { buffsize = 32 };        char received_[buffsize];        Buffer(){}        Buffer(std::string str)        {            //Truncate if Overflow            auto len = str.size();            if (len >= buffsize)            {                len = buffsize - 1;            }            std::copy(str.begin(), str.begin() + len, received_);            received_[str.size()] = '\0';        }    };    class CSession{    public:        CSession(std::string ip = defaultIP, std::string port = defaultPort);        ~CSession();        void initSession();        void handle_connect(const asio::error_code& error /*error*/, tcp::resolver::iterator i);        void handle_write(const asio::error_code& error /*error*/, size_t bytes_transferred /*bytes_transferred*/);        void handle_read(const asio::error_code& error /*error*/, size_t bytes_transferred /*bytes_transferred*/);        void write(const std::string & message);        void read(Buffer & buff);        void close();        tcp::socket& socket();        bool getState();    private:        void beginWrite(const Buffer & message);        void doClose();    private:        mutable tcp::socket socket_;   // client connection        std::string connetIP_;        std::string connectPort_;        std::string identifier_;        Buffer msg_;        std::deque<Buffer> writeQueue_;        bool state_;    };}session.cpp:#include ioservice.hxx#include <string>#include <mutex>#include session.hxx#include <sstream> using namespace channel;CSession::CSession(std::string ip, std::string port) : connetIP_(ip), connectPort_(port), state_(true),socket_(*CIOService::fetchIOService().getIO()){    std::cout << CSession Created << std::endl;}CSession::~CSession(){    close();}void CSession::initSession(){    //std::cout << CSession::initSession() Called << std::endl;    tcp::resolver resolver(*CIOService::fetchIOService().getIO());    tcp::resolver::query query(connetIP_, connectPort_);    tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);    asio::async_connect(socket_, endpoint_iterator, std::bind(&CSession::handle_connect, this, /*std::ref(error)*/        std::placeholders::_1, std::placeholders::_2));}void CSession::read(Buffer & buff){    //std::cout << CSession::read() Called << std::endl;    //Check if the socket is open or not?    asio::async_read(socket_, asio::buffer(buff.received_, buff.buffsize),        std::bind(&CSession::handle_read, this,        std::placeholders::_1, std::placeholders::_2));}void CSession::write(const std::string& message){    //std::cout << CSession::write() Called << std::endl;    (*CIOService::fetchIOService().getIO()).post(std::bind(&CSession::beginWrite, this, Buffer(message)));}void CSession::beginWrite(const Buffer & message){    //Check if the socket is open or not?    bool writeInProgress = !writeQueue_.empty();    writeQueue_.push_back(message);    if (!writeInProgress)    {        asio::async_write(socket_, asio::buffer(writeQueue_.front().received_, writeQueue_.front().buffsize),            std::bind(&CSession::handle_write, this,            std::placeholders::_1, std::placeholders::_2));    }}void CSession::close(){    std::cout << CSession::close() Called << std::endl;    (*CIOService::fetchIOService().getIO()).post(std::bind(&CSession::doClose, this));}void CSession::doClose(){    socket_.close();}tcp::socket& CSession::socket(){    return socket_;}bool CSession::getState(){    return state_;}void CSession::handle_connect(const asio::error_code& error /*error*/, tcp::resolver::iterator i){    //std::cout << CSession::handle_connect() Called << std::endl;    if (error){        std::cout << error.message() << std::endl;        state_ = false;        doClose();        return;    }    static std::mutex m;    //Must send the identifier: Think if send failed how should failed identifier be handled    std::stringstream ss;    ss << socket_.local_endpoint();    std::lock_guard<std::mutex> lock(m);    identifier_ = ss.str();    std::cout << identifier_ <<  is my identity. << std::endl;    read(msg_);}void CSession::handle_write(const asio::error_code& error /*error*/, size_t bytes_transferred /*bytes_transferred*/){    //std::cout << CSession::handle_write() Called << ( << __FILE__ <<  :  << __LINE__ << ) << std::endl;    if (!error)    {        //std::cout << bytes_transferred <<  bytes written to the socket. << std::endl;        writeQueue_.pop_front();        if (!writeQueue_.empty())        {            asio::async_write(socket_, asio::buffer(writeQueue_.front().received_, writeQueue_.front().buffsize),                std::bind(&CSession::handle_write, this,                std::placeholders::_1, std::placeholders::_2));        }    }    else    {        std::cout << Write Error Detected << std::endl;        std::cout << error.message() << std::endl;        state_ = false;        doClose();        return;    }}void CSession::handle_read(const asio::error_code& error /*error*/, size_t bytes_transferred /*bytes_transferred*/){    if (!error)    {        std::cout.write(msg_.received_, bytes_transferred);        std::cout << \n;        asio::async_read(socket_,            asio::buffer(msg_.received_, msg_.buffsize),            std::bind(&CSession::handle_read, this,            std::placeholders::_1, std::placeholders::_2));    }    else    {        std::cout << Read Error Detected << std::endl;        std::cout << error.message() << std::endl;        state_ = false;        doClose();        return;    }}main.cpp:#include session.hxx#include <thread>#include <boost/thread/thread.hpp> #include <asio.hpp>#include <boost/bind.hpp>#include ioservice.hxx#include <boost/date_time/posix_time/posix_time.hpp>using namespace channel;bool flag = false;void setFlag(const asio::error_code& /*e*/){    flag = true;}void Client(std::string IP, std::string port){    CSession Session(IP, port);    Session.initSession();    asio::thread t(boost::bind(&asio::io_service::run, &(*CIOService::fetchIOService().getIO())));    asio::deadline_timer timer(*CIOService::fetchIOService().getIO(), boost::posix_time::seconds(375));    timer.async_wait(&setFlag);    while (!flag)    {        Session.write(Client 1);    }    //char line[128];    //while (std::cin.getline(line, 128))    //{    //  char msg[128];    //  std::cout << Entered Line size:  << strlen(line) << std::endl;    //  memcpy(msg, line, strlen(line) + 1);    //  std::cout << Message to be written is :  << msg << std::endl;    //  Session.write(msg);    //}    boost::this_thread::sleep_for(boost::chrono::seconds(60));     Session.close();    t.join();}void main(){    Client(localhost, 8973);    system(Pause);}
Multi Server Architecture: ASIO C++
c++;networking;boost