CODEWITHSUNDEEP
c++socketsboosttcpboost-asio
sachinsaini
sachinsaini

Reputation: 73

How to pass a boost asio tcp socket to a thread for sending heartbeat to client or server

I am writing a client/server program in boost TCP in which I want to send a HEARTBEAT message to the client every 2 seconds for which I am trying to create a new thread by which I can send it easily but unable to solve it. I am creating thread using boost::thread t(hearbeatSender,sock); this. but giving lots of errors. I also use bind to bind function name with the socket but not resolved the error.

void process(boost::asio::ip::tcp::socket & sock);
std::string read_data(boost::asio::ip::tcp::socket & sock);
void write_data(boost::asio::ip::tcp::socket & sock,std::string);
void hearbeatSender(boost::asio::ip::tcp::socket & sock);
int main()
{

    unsigned short port_num = 3333;
    boost::asio::ip::tcp::endpoint ep(boost::asio::ip::address_v4::any(), port_num);
    boost::asio::io_service io;
    try
    {
        boost::asio::ip::tcp::acceptor acceptor(io, ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        boost::asio::ip::tcp::socket sock(io);
        acceptor.accept(sock);
        boost::thread t(hearbeatSender,sock); 
        process(sock);
        t.join();

    }
    catch (boost::system::system_error &e)
    {
        std::cout << "Error occured! Error code = " << e.code()
        << ". Message: " << e.what();

        return e.code().value();
    }
  return 0;

}
void process(boost::asio::ip::tcp::socket & sock)
{
    while(1){
    std::string data = read_data(sock);
    std::cout<<"Client's request is: "<<data<<std::endl;
    write_data(sock,data);
    }
}
std::string read_data(boost::asio::ip::tcp::socket & sock)
{
    boost::asio::streambuf buf;
    boost::asio::read_until(sock, buf, "\n");
    std::string data = boost::asio::buffer_cast<const char*>(buf.data());
    return data;
}
void write_data(boost::asio::ip::tcp::socket & sock,std::string data)
{
    boost::system::error_code error;
    std::string msg;
    int ch = data[0]-'0';
    switch(ch)
    {
        case 1: msg = "Case 1\n"; break;
        case 2: msg = "Case 2\n"; break;
        case 3: msg = "Case 3\n"; break;
        case 4: msg = "Case 4\n"; break;
        default: msg  = "Case default\n"; break;
    }
    boost::asio::write( sock, boost::asio::buffer(msg+ "\n"), error );
     if( !error ) {
        std::cout << "Server sent hello message!" << std::endl;
     }
     else {
        std::cout << "send failed: " << error.message() << std::endl;
     }
}
void hearbeatSender(boost::asio::ip::tcp::socket & sock)
{
    boost::system::error_code error;
    std::string msg = "HEARTBEAT";
    while(1)
    {
        sleep(2);
        std::cout<<msg<<std::endl;
        boost::asio::write( sock, boost::asio::buffer(msg+ "\n"), error );
        if( !error ) {
        std::cout << "Server sent HEARTBEAT message!" << std::endl;
        }
        else {
            std::cout << "send failed: " << error.message() << std::endl;
        }
    }
}

This is a server-side code for responding to the message of the client and sending heartbeat to the client. This is a synchronous TCP server.

Upvotes: 2

Views: 4202

Answers (3)

sachinsaini
sachinsaini

Reputation: 73

Instead of this:

try
    {
        boost::asio::ip::tcp::acceptor acceptor(io, ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        auto sock = acceptor.accept();
        std::thread t([&sock]() {
            hearbeatSender(sock);
        });
        process(sock);
        t.join();

    }

Use it:

try{
        boost::asio::ip::tcp::acceptor acceptor(io, ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        boost::asio::ip::tcp::socket sock(io);
        acceptor.accept(sock);

        std::thread t([&sock]() {
            hearbeatSender(sock);
        });
        process(sock);
        t.join();
}

and also include header files:

#include <thread>
#include <chrono>

(Optional) you can also use this_thread::sleep_for instead of sleep() std::this_thread::sleep_for(std::chrono::seconds(10));

The problem of passing a socket to the thread is solved.

Now, for conversing a HEARTBEAT between a client and a server. Complete code can be checked from here:

Client code HEARTBEAT transfer in every 5 seconds

Server code for giving response to the client

Upvotes: 1

sehe
sehe

Reputation: 392979

It's more than a little weird to use a heartbeat... "sender" thread with async IO.

What's more, there is no synchronization on the socket object, so that's a data race which is Undefined Behavior.

Finally, this is unsafe:

    std::string data = boost::asio::buffer_cast<const char*>(buf.data());

It assumes that the data() will be NUL-terminated (which isn't true).

Typical, Single Threaded ASIO

You would not spawn threads for timers, but use e.g. boost::asio::deadline_timer or boost::asio::highresolution_timer. It can wait asynchronously, so you can do other tasks on the IO service until it expires.

Similarly you can do the request/response reading/writing asynchronously. The only "complicating" factor is that asynchronous calls don't complete before returning, so you have to make sure the buffers live long enough (they should not be a local variable).

Now, you already have a logical "unit" of lifetime that practically JUMPS out of the code at you:

enter image description here

That just screams to be rewritten as 

struct LifeTimeUnit {
    boost::asio::ip::tcp::socket sock;

    void process();
    std::string read_data();
    void write_data(std::string);
    void hearbeatSender(sock);
};

Of course LifeTimeUnit is a funny name, so let's think of a better one: Session seems meaningful!

Now that we have a unit of lifetime, it can handsomely contain other things like buffers and the timer:

struct Session {
    Session(tcp::socket&& s) : sock(std::move(s)) {}

    void start() {
        hb_wait();
        req_loop();
    }

    void cancel() {
        hbtimer.cancel();
        sock.cancel(); // or shutdown() e.g.
    }

  private:
    bool checked(error_code ec, std::string const& msg = "error") {
        if (ec) {
            std::clog << msg << ": " << ec.message() << "\n";
            cancel();
        }
        return !ec.failed();;
    }

    void req_loop(error_code ec = {}) {
        if (!checked(ec, "req_loop")) {
            async_read_until(sock, buf, "\n",
                    [this](error_code ec, size_t xfr) { on_request(ec, xfr); });
        }
    }

    void on_request(error_code ec, size_t n) {
        if (checked(ec, "on_request")) {
            request.resize(n);
            buf.sgetn(request.data(), n);

            response = "Case " + std::to_string(request.at(0) - '0') + "\n";
            async_write(sock, buffer(response), 
                    [this](error_code ec, size_t) { req_loop(ec); });
        }
    }

    void hb_wait(error_code ec = {}) {
        if (checked(ec, "hb_wait")) {
            hbtimer.expires_from_now(2s);
            hbtimer.async_wait([this](error_code ec) { hb_send(ec); });
        }
    }

    void hb_send(error_code ec) {
        if (checked(ec, "hb_send")) {
            async_write(sock, buffer(hbmsg), [this](error_code ec, size_t) { hb_wait(ec); });
        }
    }

    tcp::socket sock;
    boost::asio::high_resolution_timer hbtimer { sock.get_executor() };
    const std::string hbmsg = "HEARTBEAT\n";
    boost::asio::streambuf buf;
    std::string request, response;
};

The only public things are start() (actually we don't have a need for cancel() for now, but you know).

The main program can be much un-altered:

tcp::acceptor acceptor(io, tcp::v4());
acceptor.bind({{}, 3333});
acceptor.listen();

tcp::socket sock(io);
acceptor.accept(sock);

Session sess(std::move(sock));
sess.start(); // does both request loop and the heartbeat

io.run();

No more threads, perfect asynchrony! Using bash and netcat to test:

while sleep 4; do printf "%d request\n" {1..10}; done | netcat localhost 3333

Prints:

host 3333
HEARTBEAT
Case 1
Case 2
Case 3
Case 4
Case 5
Case 6
Case 7
Case 8
Case 9
Case 1
HEARTBEAT
HEARTBEAT
HEARTBEAT
Case 1
Case 2
Case 3
Case 4
Case 5
Case 6
Case 7
Case 8
Case 9
Case 1
^C

After stopping the client, the server exits with

on_request: End of file
hb_send: Operation canceled

Single-Thread / Multi-Session

A big advantage is that now you can accept multiple clients on a single server thread. In fact, thousands of them concurrently without a problem.

int main() {
    boost::asio::thread_pool io(1);
    try {
        tcp::acceptor acceptor(io, tcp::v4());
        acceptor.bind({{}, 3333});
        acceptor.listen();

        std::list<Session> sessions;

        while (true) {
            tcp::socket sock(io);
            acceptor.accept(sock);

            auto& sess = sessions.emplace_back(std::move(sock));
            sess.start(); // does both request loop and the heartbeat

            sessions.remove_if([](Session& s) { return !s.is_active(); });
        }

        io.join();
    } catch (boost::system::system_error& e) {
        std::cout << "Error occured! Error code = " << e.code() << ". Message: " << e.code().message() << "\n";
        return e.code().value();
    }
}

Note how we subtly changed our execution-context to a singleton thread pool. This means we still run all sessions on a single thread, but that's a different thread than running main(), meaning we can continue to accept connections.

To avoid ever-increasing sessions list, we weed out the inactive ones using a trivially implemented is_active() property.

Note that we can ALMOST force a shutdown by doing

for (auto& sess: sessions)
    sess.cancel();

That's ALMOST, because it requires posting the cancel operations on the pool thread:

for (auto& sess: sessions)
    post(io, [&sess] { sess.cancel(); });

This is to avoid racing with any tasks on the IO pool

Since only the main thread ever touches sessions there is no need for locking.

Live On Coliru

Testing with 

for a in 3 2 1; do (sleep $a; echo "$a request" | nc 127.0.0.1 3333)& done; time wait

Prints:

Case 1
Case 2
Case 3
HEARTBEAT
HEARTBEAT
...

Multi-Threading For The Win?

Now we could add multi-threading. The changes are mild:

  • we want to associate the socket with a strand (see Why do I need strand per connection when using boost::asio?)
  • note we already use sock's executor to run the timer

  • We have to take extra precautions to make all of the public interface in Session thread-safe:

    • post actions from start() and cancel() on the strand
    • make the active flag atomic_bool

  • next up, we simply increase the number of threads in the pool from 1 to, say 10

Note, in practice it rarely makes sense to use more threads than logical cores. Also, in this simple example everything is IO bound, so a single thread probably already serves as well. This is just for demonstration

Live On Coliru

boost::asio::thread_pool io(10);
try {
    tcp::acceptor acceptor(io, tcp::v4());
    acceptor.set_option(tcp::acceptor::reuse_address(true));
    acceptor.bind({{}, 3333});
    acceptor.listen();

    std::list<Session> sessions;

    while (true) {

        tcp::socket sock(make_strand(io)); // NOTE STRAND!
// ...
// ...

    io.join();

And the changes in Session:

   void start() {
        active = true;
        post(sock.get_executor(), [this]{
            hb_wait();
            req_loop();
        });
    }

    void cancel() {
        post(sock.get_executor(), [this]{
            hbtimer.cancel();
            sock.cancel(); // or shutdown() e.g.
            active = false;
        });
    }

// ....

    std::atomic_bool active {false};
}

Upvotes: 1

selbie
selbie

Reputation: 104514

Instead of this:

    boost::asio::ip::tcp::socket sock(io);
    acceptor.accept(sock);
    boost::thread t(hearbeatSender,sock);

this:

    auto sock = acceptor.accept();
    std::thread t([&sock]() {
        hearbeatSender(sock);
    });

And instead of sleep, just used std::this_thread::sleep for compiling universally.

Here's the complete program that compiles and runs

#include <boost/asio.hpp>
#include <iostream>


void process(boost::asio::ip::tcp::socket& sock);
std::string read_data(boost::asio::ip::tcp::socket& sock);
void write_data(boost::asio::ip::tcp::socket& sock, std::string);
void hearbeatSender(boost::asio::ip::tcp::socket& sock);
int main()
{

    unsigned short port_num = 3333;
    boost::asio::ip::tcp::endpoint ep(boost::asio::ip::address_v4::any(), port_num);
    boost::asio::io_service io;
    try
    {
        boost::asio::ip::tcp::acceptor acceptor(io, ep.protocol());
        acceptor.bind(ep);
        acceptor.listen();
        auto sock = acceptor.accept();
        std::thread t([&sock]() {
            hearbeatSender(sock);
        });
        process(sock);
        t.join();

    }
    catch (boost::system::system_error& e)
    {
        std::cout << "Error occured! Error code = " << e.code()
            << ". Message: " << e.what();

        return e.code().value();
    }
    return 0;

}
void process(boost::asio::ip::tcp::socket& sock)
{
    while (1) {
        std::string data = read_data(sock);
        std::cout << "Client's request is: " << data << std::endl;
        write_data(sock, data);
    }
}
std::string read_data(boost::asio::ip::tcp::socket& sock)
{
    boost::asio::streambuf buf;
    boost::asio::read_until(sock, buf, "\n");
    std::string data = boost::asio::buffer_cast<const char*>(buf.data());
    return data;
}
void write_data(boost::asio::ip::tcp::socket& sock, std::string data)
{
    boost::system::error_code error;
    std::string msg;
    int ch = data[0] - '0';
    switch (ch)
    {
    case 1: msg = "Case 1\n"; break;
    case 2: msg = "Case 2\n"; break;
    case 3: msg = "Case 3\n"; break;
    case 4: msg = "Case 4\n"; break;
    default: msg = "Case default\n"; break;
    }
    boost::asio::write(sock, boost::asio::buffer(msg + "\n"), error);
    if (!error) {
        std::cout << "Server sent hello message!" << std::endl;
    }
    else {
        std::cout << "send failed: " << error.message() << std::endl;
    }
}
void hearbeatSender(boost::asio::ip::tcp::socket& sock)
{
    boost::system::error_code error;
    std::string msg = "HEARTBEAT";
    while (1)
    {
        std::this_thread::sleep_for(std::chrono::seconds(2));
        std::cout << msg << std::endl;
        boost::asio::write(sock, boost::asio::buffer(msg + "\n"), error);
        if (!error) {
            std::cout << "Server sent HEARTBEAT message!" << std::endl;
        }
        else {
            std::cout << "send failed: " << error.message() << std::endl;
        }
    }
}

Upvotes: 2

Related Questions