2
\$\begingroup\$

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:

  1. Chat.

  2. 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:

  1. How to use io_service::run()? The way I am using in my code, is it satisfactory?

  2. How to ensure clean destruction of server when calling commManager::dropServer()?

  3. 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.hpp"

namespace 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_HPP

server.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 http

connection_manager.hpp:

#ifndef HTTP_CONNECTION_MANAGER_HPP
#define HTTP_CONNECTION_MANAGER_HPP

#include <set>
#include <boost/noncopyable.hpp>
#include "connection.hpp"

namespace 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_HPP

connection_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 http

connection.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 http

connection.cpp:

#include "connection.hpp"
#include <vector>
#include <boost/bind.hpp>
#include "connection_manager.hpp"


namespace 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 http

main.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");
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.