Tool for asynchronous IPC using pipes

Question

Description of the code:

The code provides asynchronous IPC functionality in C++ using the Boost libraries and pipes.

Each process asynchronously "listens" on the read end of the pipe and the messages are added to a queue, which means you won't have to call read() manually: you just have to send messages and they're automatically received on the other end.

I also plan to add a feature which enables you to send the name of a function and its arguments to the other process to call it remotely.

My question:

This is pretty much the first non Tic-tac-toe project I've done, and I've never really written production code for a big project, so I always have doubt about whether I'm doing things the proper way or I'm just making stuff up.

I'd really appreciate it if you could look at the code and point out what you would've done differently if you were to create this project yourself.

Any bugs or ways to make the code faster are also very welcome.

Unrelated question, could this be of use to anyone? Should I put it up on GitHub?

The code:

example_server.cpp

// Code for "Connector" is provided below, I thought it would be better
// to show how it's used first.

#include "Connector.hpp"

#include <iostream>
#include <string>
#include <memory>

#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>


using namespace std;

int main()
{
    // Create two pipes and convert the file descriptors to char.
    int pipe_fd1[2];
    int pipe_fd2[2];
    pipe(pipe_fd1);
    pipe(pipe_fd2);
    char reader_for_client[4];
    char writer_for_client[4];
    sprintf(&reader_for_client[0], "%d", pipe_fd1[0]);
    sprintf(&writer_for_client[0], "%d", pipe_fd2[1]);

    // Create the child process or "Client" and pass 
    // one read handle and one write handle to it.
    pid_t cpid = fork();
    if(cpid == 0)
    {
        execl("./example_client", &reader_for_client[0], &writer_for_client[0], (char*) NULL);
    }

    // Create a "Connector" object and pass the file descriptors to it.
    // Also specify the number of threads it should use.
    auto node = make_unique<Connector>(pipe_fd2[0], pipe_fd1[1], 1);

    // Provide a callback function that gets called
    // whenever a new message is received.
    node->AssignHandler([](const string& data)
    {
        cout << "[Server] " << data << endl;
    });

    // Read messages from the terminal and send them to the "Client".
    string message;
    do
    {
        cin >> message;
        node->SendData(message);
    }
    // Send the message "die" to make both processes exit.
    // The child process ("Client") sends back "die" to the
    // parent process ("Server") to tell it to exit as well.
    while(message != "die");
    
    // Wait for the child process to exit. This is probably
    // not necessary since the parent process won't exit
    // until it receives "die" from the child process.
    waitpid(cpid, NULL, 0);
}

example_client.cpp

#include "Connector.hpp"

#include <memory>
#include <iostream>
#include <string>

using namespace std;

int main(int argc, char* argv[])
{
    // Receive the pipe file descriptors from the "server".
    int reader_fd = atoi(argv[0]);
    int writer_fd = atoi(argv[1]);

    // Create a "Connector" object and pass the file descriptors to it.
    // Also specify the number of threads it should use.
    auto node = make_unique<Connector>(reader_fd, writer_fd, 1);

    // Provide a callback function that gets called
    // whenever a new message is received.
    node->AssignHandler([](const string& data)
    {
        cout << "[Client] " << data << endl;
    });

    // Send a messsage to the "Server".
    node->SendData("hello, this is the client.");
}

Connector.hpp

#pragma once
#include <queue>
#include <memory>
#include <string>
#include <functional>
#include <boost/asio/readable_pipe.hpp>
#include <boost/asio/writable_pipe.hpp>
#include <boost/asio/io_service.hpp>
#include <boost/system/error_code.hpp>
#include <boost/asio/thread_pool.hpp>

typedef std::shared_ptr<boost::asio::io_service> io_ptr;

class Connector
{
    public:
        Connector(int reader_fd, int writer_fd, int numThreads);
        ~Connector();

        void CleanUp();

        // Used to send simple string.
        void SendData(const std::string& data);

        // NOT IMPLEMENTED YET.
        // Used in order to send the name of a function that exists
        // in the other process to run it remotely.
        // Planning to implement it using a map since reflection
        // doesn't exist in c++.
        void SendCommand(const std::string& command);

        // "AssignHandler" takes in a function which is called
        // everytime a message is received.
        void AssignHandler(const std::function<void(const std::string&)>& func);

        // This queue stores incoming messages. Messages are
        // "pop"ed after they are passed to the user provided handler.
        std::queue<std::string> data_queue;
    private:
        // Stores the messages and commands that have been
        // received before the user provided a callback function.
        std::queue<std::string> pending;

        // handler function provided by the user.
        std::function<void(const std::string&)> handler;

        // The class receives an integer with the size
        // of the message or command that is about to be received,
        // and allocates an appropriately sized string.
        int32_t size_of_incoming_buffer;

        // Shows whether we're waiting for the size of the message,
        // or for the message itself.
        bool getting_size = true;

        // The incoming message or command is stored in this string.
        std::string buffer;

        // Create boost objects that provide async functionality.
        std::shared_ptr<boost::asio::io_service> io_service;
        std::shared_ptr<boost::asio::readable_pipe> reader;
        std::shared_ptr<boost::asio::writable_pipe> writer;
        std::shared_ptr<boost::asio::thread_pool> threads;
        std::shared_ptr<boost::asio::io_service::work> work;

        // Reads a certain amout of bytes from the read-end of the pipe.
        void Receive(int32_t bytes = 4);

        // Gets called when "Receive" succussfully reads the specified
        // number of bytes from the pipe.
        void ReceiveHandler(const boost::system::error_code& ec, size_t bytes);

        // Gets called when we succussfully write to the pipe.
        // Currently doesn't actually do anything.
        void SendHandler(const boost::system::error_code& ec, size_t bytes);

        // Create the pipe objects used for communication.
        // Could be overloaded to work with sockets and stuff too.
        void InitializeCommMethod(int reader_fd, int writer_fd);

        // Posts the specified tasks to the io_service object for execution.
        // Currently only performs "Receive".
        void PerformTasks();

        // Calls the user provided handler function with the appropriate arguments.
        void CallHandler();

        // Gets called when the user provides the handler function.
        // Runs the handler on all the pending messages.
        void PerformPending();
};

Connector.cpp

#include "Connector.hpp"
#include <boost/asio/post.hpp>
#include <boost/bind/bind.hpp>
#include <boost/bind/placeholders.hpp>
#include <boost/asio/write.hpp>
#include <boost/asio/read.hpp>

#include <iostream>

using namespace std;
using namespace boost::placeholders;

Connector::Connector(int reader_fd, int writer_fd, int numThreads)
{
    // Initialize boost related objects that will help us
    // perform asynchronous io operations.

    // The io_service objects provides async_io functionality.
    io_service = make_shared<boost::asio::io_service>();

    // The "work" object doesn't allow io_service to
    // exit when it runs out of tasks.
    work = make_shared<boost::asio::io_service::work>(*io_service);

    // Create pipe objects used for communication, from native pipe
    // handles.
    InitializeCommMethod(reader_fd, writer_fd);

    threads = make_shared<boost::asio::thread_pool>(numThreads);

    // Post io_service->run to the thread pool.
    boost::asio::post(*threads, [this]()
    {
        this->io_service->run();
    });

    // Perform a number of predefined tasks.
    PerformTasks();
}

Connector::~Connector()
{
    // When the destructor is called, it doesn't allow the main thread
    // to exit until all the worker threads have finished.
    threads->join();
}

void Connector::CleanUp()
{
    // Allow the worker threads to exit by
    // removing the work object.
    work.reset();
    threads->join();
}

void Connector::SendData(const string& data)
{
    // First sends the size of the message, then sends
    // the message itself.
    int32_t size = data.length();
    boost::asio::async_write(*writer, boost::asio::buffer((void*)&size, 4), boost::bind(&Connector::SendHandler, this, _1, _2));
    boost::asio::async_write(*writer, boost::asio::buffer(data), boost::bind(&Connector::SendHandler, this, _1, _2));
}

void Connector::SendHandler(const boost::system::error_code& ec, size_t bytes)
{
    // Gets called whenever we write to the pipe.
    return;
}

void Connector::Receive(int32_t bytes)
{
    // Allocate an appropriately sized string ("buffer"), 
    // read "bytes" bytes from the pipe and put it in
    // said string.

    // OR, read the size of the incoming message and put it
    // in "size_of_incoming_buffer".

    buffer = string(bytes, '\0');

    if(getting_size)
    {
        boost::asio::async_read(*reader, boost::asio::buffer((void*)&size_of_incoming_buffer, 4), boost::bind(&Connector::ReceiveHandler, this, _1, _2));
        return;
    }
    boost::asio::async_read(*reader, boost::asio::buffer(buffer), boost::bind(&Connector::ReceiveHandler, this, _1, _2));
}

void Connector::ReceiveHandler(const boost::system::error_code& ec, size_t bytes)
{
    // If we've already received the size of the incoming message, 
    // read "size_of_incoming_buffer" bytes from the pipe, which is the
    // size of the message.
    if(getting_size)
    {
        getting_size = false;
        Receive(size_of_incoming_buffer);
        return;
    }

    // If getting_size is false, then the last thing we received
    // was the message, which is currently in "buffer".
    // We push the message to "data_queue", and wait for the size
    // of the next message.
    getting_size = true;
    data_queue.push(buffer);

    // If the message is "die", exit.
    if(buffer == "die")
    {
        // Tell the other process to exit as well.
        SendData("die");

        CleanUp();
        return;
    }

    // Call the user provided handler function for when we receive
    // a new message.
    io_service->post(boost::bind(&Connector::CallHandler, this));

    // Wait for the size of the next message.
    // (The default value of the "Receive" function is 4 bytes, or an int32_t).
    Receive();
}

void Connector::InitializeCommMethod(int reader_fd, int writer_fd)
{
    // Create boost pipe objects from native pipe handles.
    reader = make_shared<boost::asio::readable_pipe>(*io_service, reader_fd);
    writer = make_shared<boost::asio::writable_pipe>(*io_service, writer_fd);
}

void Connector::PerformTasks()
{
    // Receive the size of the first message.
    Receive();
}

void Connector::AssignHandler(const function<void(const string&)>& func)
{
    // Assign the user provided function to "handler".
    handler = func;

    // Run "handler" on all pending message, received before the user
    // provided a handler function.
    PerformPending();
}

void Connector::CallHandler()
{
    // If handler is provided, run it on the
    // last message in the queue.
    if(handler)
    {
        handler(data_queue.front());
        data_queue.pop();
    }
    // Otherwise, add the message to the pending queue.
    else
    {
        pending.push(data_queue.front());
        data_queue.pop();
    }
}

void Connector::PerformPending()
{
    // Gets called when the user provides a handler.
    // Runs the handler on all pending messages.
    string pending_message;
    for(int i = 0; i < pending.size(); i++)
    {
        pending_message = pending.front();
        pending.pop();
        io_service->post([this, pending_message]()
        {
            this->handler(pending_message);
        });
    }   
}

Answer 1

Unnecessary use of smart pointers

You are using std::unique_ptr and std::shared_ptr in places where there is no need for them. For example, the client can just declare the Connector like so:

int main(...) {
    ...
    Connector node(reader_fd, writer_fd, 1);
    node.AssignHandler(...);
    node.SendData("Hello, this is the client.");
}

But many more things in your code are using smart pointers unnecessarily.

Unnecessary use of this->

You are using this-> twice in the code, but there is no reason to use it.

Thread safety

Your code starts multiple I/O processing loops. However, nothing in class Connector is thread-safe. If you are going to use multiple threads, I would expect to see a mutex guarding the queues at the very least. However, it seems you only enqueue a single async_read() at a time, so in that sense the code is probably safe, but then what's the point of the async I/O? But still, you could receive messages faster than they can be handled, in which case there are multiple threads running CallHandler(), and those do modify data_queue and pending.

Missing error checking

Things can go wrong, but nothing in your code does error checking. What if the server or the client is killed before "die" has been sent? What if ReceiveHandler() is called when getting_size == false but bytes is not the same as size_of_incoming?

Make sure you check the return value of any system calls like fork() and execl(), and for ASIO callbacks check the boost::system::error_code parameter.

In case of an error, make sure you handle it appropriately; if you cannot recover from it at the place it occurs, throw an exception so the error can be handled at a higher level, or if it's really fatal, print an error message to std::cerr and call std::exit(EXIT_FAILURE).

Use '\n' instead of std::endl

Prefer to use '\n' instead of std::endl; the latter is equivalent to the former, but also forces the output to be flushed, which is often unnecessary and may negatively impact performance.