c++ Fast IPC - boost message queue seems slow?

Question

I have a problem i dont seem to be able to solve myself. I have Process1 that calculate data in a while-loop. This process has to be executed as fast as possible. I need the data calculated in Process1 for later analysis and writing to a file is to slow.

I have never worked with IPC but thought it was a good way to store the data from Process1 in memory and access it from another Process2 (separate program) that is not time critical and write date to file.

I have created my little test program (to learn about IPC) so:

1. Process1 will run even if Process2 is not accessible - it will then skip IPC and just execute
2. When running Process2 it will wait for Process1 - fetch data if Process1 starts-up, and then later write to disk.
3. Process2 will only fetch x-amount of data (maxRunTime) in below 10 samples.

The current programs i have created are terrible slow, when sending messages over IPC it is 6 times slower. Currently i only pass three floats at each "TimeStep" but this could be 100. And RunTime could be 10.000.

To do: I would be pleased if someone could guide me in the right direction. Below code is working, it might be luck as it is not pretty.

I need to find a solution that is as fast as possible, but doesnt have to be realtime. As im not a pro-programmer i also need to compromise the complexity as i need to understand what im doing.

Hope someone can help.

Code:

1. Using Boost.1.59 and MSVC 11.0_x86
2. Two separate programs - ConsoleApps

Process1:

#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/date_time.hpp>
#include <iostream>
#include <vector>
#include <windows.h>
#include <string>
#include <ctime>
#include <iostream>
#include <fstream>
#include <map>
#include <stdio.h>
#include <conio.h>
#include <tchar.h>
#include <time.h>


#pragma comment(lib, "user32.lib")

using namespace std;
using namespace boost::interprocess;
using namespace boost::posix_time;
using boost::posix_time::microsec_clock; 


bool InitCreateMsgQ()
{
    bool initOK = false;
    //Create a msgQ for parsing data
    try
    {
        message_queue::remove("msgQData");
        //Create a message_queue.
        message_queue mqData
        (open_or_create     //create q 
        ,"msgQData"         //name
        ,1000000                //max message number
        ,sizeof(float)      //max message size
        );
        initOK = true;
    }
    catch(interprocess_exception &ex)
    {
        return false;
    }
//Create State
    try
    {
        message_queue::remove("msgState");
        //Create a message_queue.
        message_queue mqState
        (open_or_create     //create q 
        ,"msgState"     //name
        ,1                  //max message number
        ,sizeof(int)        //max message size
        );
        initOK = true;
    }
    catch(interprocess_exception &ex)
    {
        return false;
    }
    return initOK;
}
bool SetState(int state)
{
    bool timeout = true;
    try
    {
        //Open a message queue.
        message_queue mqState
        (open_only       //only oepn q
        ,"msgState"  //name
        );

        timeout = !mqState.timed_send(&state, sizeof(int), 0, 
                                        ptime(boost::posix_time::microsec_clock::universal_time()) + milliseconds(100));
    }
    catch(interprocess_exception &ex)
    {
        message_queue::remove("msgState");
        timeout = true;
    }
    return timeout;
}
bool SetData(float data)
{
    bool timeout = true;
    try
    {
        //Open a message queue.
        message_queue mqData
        (open_only       //only oepn q
        ,"msgQData"  //name
        );

        timeout = !mqData.timed_send(&data, sizeof(float), 0, 
                                        ptime(boost::posix_time::microsec_clock::universal_time()) + milliseconds(1));
        //mqData.send(&data, sizeof(float), 0);
    }
    catch(interprocess_exception &ex)
    {
        message_queue::remove("msgQData");
        timeout = true;
    }
    return timeout;
}

int main ()
{
    time_t start,end;

    int runTime = 0; //just for testing
    int dummyState = 2;
    float x;
    int state = 0;
    if (InitCreateMsgQ()){state = 1;} //If all msQ ok set state 1
    if (SetState(state)){state = 0;}// If timeout to set state go to state 0
    //Do twice to get error if observer is not started
    if (SetState(dummyState)){state = 0;}// Set Dummy state for obersver
                                         // If timeout to set state go to state 0

    time (&start);
    //Runtime!
    while(runTime<1000)
    {
        switch (state) 
        {
            case 0:
                state = 0;//force next state 0 - should not be needed
                //Do nothing and break loop if monitor tool is not ready                
                break;
            case 1:
                state = 1;
                cout << "Try SEND DATA" << endl;
                for (int i = 0; i < 3; i++)
                {
                    x = rand() % 100;
                    if (SetData(x)){state = 0;}
                }               
                break;
            default:
                break;
        }
        runTime++;
        cout << "runTime: " << runTime <<" state: " << state << endl;
    }

    message_queue::remove("msgQData");
    message_queue::remove("msgState");
    cout << "done - state: " << state << endl;

    time (&end);
    double dif = difftime (end,start);
    printf ("Elasped time is %.2lf seconds.", dif );

    getchar();
}

Process2:

#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/date_time.hpp>
#include <iostream>
#include <vector>
#include <windows.h>
#include <string>
#include <ctime>
#include <iostream>
#include <fstream>
#include <map>
#include <stdio.h>
#include <conio.h>
#include <tchar.h>
#include <time.h>


#pragma comment(lib, "user32.lib")

using namespace std;
using namespace boost::interprocess;
using namespace boost::posix_time;
using boost::posix_time::microsec_clock; 

ofstream debugOut;      // Output file for debug    (DEBUG)

int getState()
{
    int state = 0;
    bool timeout = true;
    try
    {
        //Open a message queue.
        message_queue mqState
        (open_only       //only oepn q
        ,"msgState"  //name
        );

        unsigned int priority;
        message_queue::size_type recvd_size;

        timeout = !mqState.try_receive(&state, sizeof(state), recvd_size, priority);    
    }
    catch(interprocess_exception &ex)
    {
        timeout = true;
    }

    if(timeout){state = 0;}

    return state;
}
float getData()
{
    float Data = -123456;
    bool timeout = true;
    try
    {
        //Open a message queue.
        message_queue mqData
        (open_only       //only oepn q
        ,"msgQData"  //name
        );

        unsigned int priority;
        message_queue::size_type recvd_size;

        //Receive the data
        //mqData.try_receive(&Data, sizeof(Data), recvd_size, priority);
        timeout = !mqData.timed_receive(&Data, sizeof(Data), recvd_size, priority,
                                        ptime(boost::posix_time::microsec_clock::universal_time()) + milliseconds(10));
    }
    catch(interprocess_exception &ex)
    {
        timeout = true;
    }

    if(timeout){Data = -123456;}

    return Data;
}

int main ()
{
    int state = 0;
    int maxRunTime = 10;
    float Data;
    float DataArray[100000];

    debugOut.open("IPCWriteTest.txt", std::ios::trunc);
    debugOut.close();

    while(true)
    {
        switch (state) 
        {
            case 0: 
                //Do nothing - data not ready state
                if(getState() == 1)
                {
                    state = 1;
                    cout << "State: 1" <<endl;
                } //If all msQ ok set state 1
                else{state = 0;}
                break;
            case 1:
                for (int runTime = 0; runTime < maxRunTime; runTime++)
                {
                    cout << "runTime: " << runTime << " Data: ";
                    for (int i = 0; i < 3; i++)
                    {
                        Data = getData();
                        cout << Data << "   ";
                        DataArray[runTime]=Data;
                    }   
                    cout << endl;
                }

                debugOut.open("IPCWriteTest.txt", std::ios::app);
                for (int runTime = 0; runTime < maxRunTime; runTime++)
                {
                    debugOut << "runTime: " << runTime << " Data: ";
                    for (int i = 0; i < 3; i++)
                    {
                        debugOut << DataArray[runTime] << " ";

                    }   
                    debugOut << endl;
                }
                debugOut.close();
                state = 0;
                break;
            default:
                break;
        }
    }

    std::cout << "done" << endl;
    getchar();
}

Answer 1

You are opening the queue for each operation.

You should try opening once and passing a reference to all related code (typically you would store it as a member in a class).

Also, having separate queues is recipe for slowness. It seems to me you're "abusing" mqState as an interprocess::condition_variable or semaphore:

• http://www.boost.org/doc/libs/1_59_0/doc/html/interprocess/synchronization_mechanisms.html#interprocess.synchronization_mechanisms.conditions
• http://www.boost.org/doc/libs/1_59_0/doc/html/interprocess/synchronization_mechanisms.html#interprocess.synchronization_mechanisms.semaphores

Translating exceptions to tedious error codes like that is not very productive anyways. You're doing, manually, what exception handling should be doing.

Also, the fact that you trace debug messages to standard output will vastly slow down the program, especially on Windows

Notes on the observer

The same things go, and also for the debugOutput file which should probably not be reopened continuously.

It's strange to be "hardlooping" in triples. If it's a queue, just pop 1 message at a time. If a message "logically" consists of three floats, send messages containing three floats. Right now I even think this is a bug:

            for (int i = 0; i < 3; i++) {
                data = getData();
                std::cout << data << "   ";
                DataArray[runTime] = data;
            }

It assigns three different values to the same index (runTime)...

Simplified code

Code for the producer after I "reviewed it" (cleaned it up):

Live¹ On Coliru

#include <boost/date_time.hpp>
#include <boost/interprocess/ipc/message_queue.hpp>
#include <fstream>
#include <algorithm>
#include <iterator>
#include <iostream>
#include <map>
#include <string>
#include <vector>

namespace bip = boost::interprocess;
namespace pt  = boost::posix_time;

struct QueueLogic {

    bool forced_remove = bip::message_queue::remove("msgQData");
    bip::message_queue mqData{ bip::open_or_create, "msgQData", 1000000, sizeof(float) };

    bool SetData(float data) {
        return !mqData.timed_send(&data, sizeof(float), 0, pt::ptime(pt::microsec_clock::universal_time()) + pt::milliseconds(1));
    }
};

#include <boost/chrono.hpp>
#include <boost/chrono/chrono_io.hpp>
using Clock = boost::chrono::high_resolution_clock;

int main() {
    std::vector<float> pre_calculated;
    std::generate_n(back_inserter(pre_calculated), 10000*100, [] { return rand()%100; });

    auto start = Clock::now();

    try {
        QueueLogic instance;

        for (auto v : pre_calculated)
            instance.SetData(v);

    } catch(std::exception const& e) {
        std::cout << "Exception thrown: " << e.what() << "\n";
        bip::message_queue::remove("msgQData");
        throw;
    }

    auto end = Clock::now();
    std::cout << boost::chrono::duration_cast<boost::chrono::milliseconds>(end-start) << "\n";
}

Code for the consumer:

Live¹ On Coliru

#include <iostream>
#include <fstream>
#include <vector>

#include <boost/interprocess/ipc/message_queue.hpp>
#include <boost/date_time.hpp>

using namespace std;
namespace bip = boost::interprocess;
namespace pt  = boost::posix_time;

#include <boost/chrono.hpp>
#include <boost/chrono/chrono_io.hpp>
using Clock = boost::chrono::high_resolution_clock;

struct ObserverLogic {

    bip::message_queue mqData{bip::open_only, "msgQData"};

    float getData() {
        float data;
        bip::message_queue::size_type recvd_size;
        unsigned int priority;
        if (!mqData.timed_receive(&data, sizeof(data), recvd_size, priority,
                                  pt::ptime(pt::microsec_clock::universal_time()) + pt::milliseconds(10))) 
        {
            throw std::runtime_error("timeout in timed_receive");
        }

        return data;
    }
};

int main() {
    std::vector<float> DataArray;
    DataArray.reserve(100000);

    ObserverLogic instance;

    try {
        while (DataArray.size() <= 100000) {
            DataArray.push_back(instance.getData());
        }
    } catch (std::exception const &e) {
        std::cout << "Exception caught: " << e.what() << "\n";
    }

    std::cout << "Received " << DataArray.size() << " messages\n";
    std::copy(DataArray.begin(), DataArray.end(), std::ostream_iterator<float>(std::cout, "; "));

    std::cout << "\n\ndone" << std::endl;
}

---

Notes

Live¹ - shared memory is not allowed on Coliru