Description of the code itself:

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 it's 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?

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);
        });
    }   
}
```