Asynchronous processing in c++

Question

A Server which will run forever and processes requests needs an asynchronous part of code in it which will execute some database queries and update only when there are any new changes to it. The server has to run forever and this function for executing a db function again and again has to run asynchronously, so that there is not hindrance the server because of update once in every 'x' minutes.

How best this can be processed asynchronously in c++ ? How can I set that function alone to run on a daemon so that it doesn't block the server at all ?

Answer 1

I'd strongly recommend using Boost's ASIO library

You'd need a class to accept new requests and another to periodically check for updates. Both could do their work asynchronously and use the same boost::asio::io_service to schedule work.

The setup would be

• A network asynchronous boost::asio::ip::tcp::acceptor listening for new requests.
• A boost::asio::deadline_time do an asynchronous wait do check for updates to the database.

Pseudo code for what I understand you are describing is below:

#include <iostream>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <string>

class DatabaseUpdateChecker{
    public:
    DatabaseUpdateChecker(boost::asio::io_service& io, const int& sleepTimeSeconds)
    :timer_(io,boost::posix_time::seconds(sleepTimeSeconds)),sleepSeconds_(sleepTimeSeconds){
        this->timer_.async_wait(boost::bind(&DatabaseUpdateChecker::doDBUpdateCheck,this,boost::asio::placeholders::error));
    };

    protected:
    void doDBUpdateCheck(const boost::system::error_code& error){
        if(!error){
            std::cout << " Checking Database for updates" << std::endl;
            //Reschdule ourself
            this->timer_.expires_at(timer_.expires_at() + boost::posix_time::seconds(this->sleepSeconds_));
            this->timer_.async_wait(boost::bind(&DatabaseUpdateChecker::doDBUpdateCheck,this,boost::asio::placeholders::error));
        }
    };
    private:
    boost::asio::deadline_timer timer_;
    int sleepSeconds_;  
};

typedef boost::shared_ptr<boost::asio::ip::tcp::socket> TcpSocketPtr;

class NetworkRequest{
    public: 
    NetworkRequest(boost::asio::io_service& io, const int& port)
    :acceptor_(io,boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(),port)){
        this->start_accept();   
    };  
    protected:
    void start_accept(){
        TcpSocketPtr socketPtr(new boost::asio::ip::tcp::socket(acceptor_.get_io_service()));
        std::cout << "About to accept new connection" << std::endl;
        acceptor_.async_accept(*socketPtr,boost::bind(&NetworkRequest::handle_accept,this,socketPtr,boost::asio::placeholders::error));
    };  
    void handle_accept(TcpSocketPtr socketPtr,const boost::system::error_code& error){
        std::cout << "Accepted new network connection" << std::endl;
        if(!error){
            std::string response("This is a response\n");
            boost::asio::async_write(*socketPtr,boost::asio::buffer(response),
                boost::bind(&NetworkRequest::handle_write,this,boost::asio::placeholders::error,boost::asio::placeholders::bytes_transferred));
        }
        //Start listeing for a new connection
        this->start_accept();
    }   
    void handle_write(const boost::system::error_code& error,size_t size){
        if(!error){
            std::cout << "Wrote out " << size << " bytes to the network connection" << std::endl;
        }

    }   
    private:
    boost::asio::ip::tcp::acceptor acceptor_;
};

int main(int argc, char *argv[]) {
    static const int DB_TIMER_SECONDS=5;
    static const int LISTENING_TCP_PORT=4444;

    std::cout << "About to start" << std::endl;
    boost::asio::io_service io;

    DatabaseUpdateChecker dbChecker(io,DB_TIMER_SECONDS);
    NetworkRequest networkRequestAcceptor(io,LISTENING_TCP_PORT);

    io.run();

    std::cout << "This won't be printed" << std::endl;  
    return 0;
}

Compiling the above and running it will show that the Database Update Checker will check for updates every 5 seconds while listening for connections on TCP port 4444. To see the code accept a new connection you can use telnet/netcat/your favorite network client tool....

telnet 127.0.0.1 4444
Trying 127.0.0.1...
Connected to localhost.
Escape character is '^]'.
This is a response
Connection closed by foreign host.

If you find that the processing of updates and/or requests takes a significant amount of time then I'd look into threading your application and running each task in it's own thread. io_service will schedule what work it has to do and not complete until there is no more work. The trick is to have the classes doing work reschedule themselves when they are done.

Of course you have to take into account the comments of others on your question. I dont' know how a CORBA interface might complicate this but I would think boost::asio as an asynchronous C++ library would be a good decision and flexible enough for what you describe.

Answer 2

It sounds like what it means is that while the system processes network requests continuously it asynchronously communicates with the DB.

What it means is that when it needs to talk to DB it sends a query but does not wait for response.

When it gets response from DB it processes it.

The asynchronous part could be implemented by having a separate thread that talks to DB and when it gets response it posts an even in the server queue for processing.

Or the server could be listening on many sockets for data and one of them could be a database connection where it gets responses from DB.

Answer 3

So basically (if I understand you correctly) you need to poll a database periodically to see if it's changed. And when it has changed, you need to notify a CORBA-based request processor that there has been a change.

What I would do is add a new request type to your CORBA server. The request would be "The database has updated.". Then you can write a small, completely different program who's only job is to poll the database and send the CORBA request when the database has been updated.

That way, you can fold the database update message into the main request stream for the CORBA server.

No threads, no asynchronous anything. Just two processes each doing their own thing.