c++ Fast IPC - boost 消息队列似乎很慢？

Question

我有一个我自己似乎无法解决的问题。我有 Process1 在 while 循环中计算数据。这个过程必须尽快执行。我需要在 Process1 中计算的数据供以后分析，写入文件速度很慢。

我从未使用过 IPC，但我认为这是一种将 Process1 中的数据存储在内存中并从另一个对时间要求不严格的 Process2(独立程序)访问它并将日期写入文件的好方法。

我已经创建了我的小测试程序(以了解 IPC):

1. 即使 Process2 不可访问，Process1 也会运行 - 然后它将跳过 IPC 并直接执行
2. 当运行 Process2 时，它将等待 Process1 - 如果 Process1 启动，则获取数据，然后写入磁盘。
3. Process2 将仅在 10 个样本以下获取 x 数量的数据 (maxRunTime)。

我创建的当前程序非常慢，当通过 IPC 发送消息时，速度慢了 6 倍。目前我只在每个“TimeStep”传递三个 float ，但这可能是 100。而 RunTime 可能是 10.000。

要做的事:如果有人能指导我正确的方向，我会很高兴。下面的代码是有效的，可能是运气不好，因为它并不漂亮。

我需要找到一个尽可能快的解决方案，但不必是实时的。由于我不是专业程序员，我还需要在复杂性上妥协，因为我需要了解我在做什么。

希望有人能帮忙。

代码:

1. 使用 Boost.1.59 和 MSVC 11.0_x86
2. 两个独立的程序 - ConsoleApps

进程 1:

#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();
}

进程 2:

#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

您正在为每个操作打开队列。

您应该尝试打开一次并传递对所有相关代码的引用(通常您会将其存储为类中的成员)。

此外，拥有单独的队列是导致速度缓慢的原因。在我看来，您将 mqState 作为 interprocess::condition_variable 或信号量“滥用”:

• 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

无论如何，将异常转换为乏味的错误代码并不是很有效率。您正在手动执行异常处理应该执行的操作。

此外，将调试消息跟踪到标准输出这一事实将大大降低程序速度，尤其是在 Windows 上

关于观察者的注释

同样的事情发生了，debugOutput 文件也不应该连续重新打开。

在三元组中“硬循环”很奇怪。如果是队列，一次只弹出 1 条消息。如果消息“逻辑上”由三个 float 组成，则发送包含三个 float 的消息。现在我什至认为这是一个错误:

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

它将三个不同的值分配给同一个索引 (runTime)...

简化代码

我“审查”(清理)后制作人的代码:

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";
}

消费者代码:

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

---

注意事项

Live ¹ - Coliru 不允许共享内存