c++ 随机boost：：interprocess_exception：：library_error on boost：：interprocess：：message_queue

xkftehaa  于 2024-01-09  发布在  其他
关注(0)|答案(1)|浏览(250)

我已经使用消息队列为IPC创建了一个模板类。
我在无限while循环中运行我的程序（称为主循环）。
我通过以太网从各种子系统（传感器）收集数据，并使用消息队列将接收到的数据传递给适当的进程（它们是多个不同的进程，可以充当数据接收器，每个进程都有自己的消息队列）。
我刚刚运行了程序，没有执行任何活动。这是唯一运行的程序，每次运行前我都会重新启动操作系统。
程序只是在while循环中运行，其中所有标志都设置为false;因此程序只是运行一个空循环。
我随机得到boost::interprocess_exception::library_error。因为他们没有活动，我希望他们应该没有错误。
我注解掉了以太网相关的代码，但仍然得到同样的错误。
我在声明中得到错误：

  1. if (primaryNode == true)
  2. {
  3.     this->mSecondaryToPrimaryMessageQueue->receive(
  4.         &receiveData,
  5.         sizeof(receiveData),
  6.         receiveLength,
  7.         priority
  8.     );
  9. }
  10. else
  11. {
  12.     this->mPrimaryToSecondaryMessageQueue->receive(
  13.         &receiveData,
  14.         sizeof(receiveData),
  15.         receiveLength,
  16.         priority
  17.     );
  18. }

字符串
我尝试将primaryNode设置为true或false。我得到了相同的错误。
代码：

ipc.hpp

  1. #pragma once
  3. #include <thread>
  4. #include <string>
  5. #include <atomic>
  6. #include <memory>
  7. #include <variant>
  8. #include <optional>
  9. #include <iostream>
  10. #include <functional>
  11. #include <boost/lockfree/spsc_queue.hpp>
  12. #include <boost/interprocess/ipc/message_queue.hpp>
  14. /// @brief 
  15. /// @tparam T1 Specifies the data-type that has to be sent
  16. /// @tparam T2 Specifies the data-type that has will be received
  17. /// @tparam primaryNode Denotes if the RTP is the primaryNode owner/creater of the message queue
  18. template<typename T1, typename T2, bool primaryNode>
  19. class Ipc
  20. {
  21. private:
  22.     static const std::uint8_t MAX_MESSAGE_DEPTH = 5; //Specifies the number of messages will the message queue hold
  23.     using callback_t = std::function<void(void)>;
  24.     callback_t mCallback;
  25.     std::unique_ptr<boost::interprocess::message_queue> mPrimaryToSecondaryMessageQueue;
  26.     std::unique_ptr<boost::interprocess::message_queue> mSecondaryToPrimaryMessageQueue;
  28.     std::string         mPrimaryToSecondaryMessageQueueName;
  29.     std::string         mSecondaryToPrimaryMessageQueueName;
  31.     std::thread         mReceiveThread;
  32.     std::atomic_bool    mExitReceiveThread{ false };
  33.     boost::lockfree::spsc_queue<T2, boost::lockfree::capacity<MAX_MESSAGE_DEPTH>> mReceiveDataQueue;
  35.     void listen(void);
  36. public:
  37.     Ipc() {}
  38.     bool open(const std::string& queueName);
  39.     bool close(void);
  40.     bool send(const T1& data, std::uint32_t priority = 10);
  41.     std::optional<T2> receive(void);
  42.     bool register_callback(callback_t callback_implementation);
  43.     bool isDataAvailableInReceiveDataQueue(void) const;
  44. };
  46. template<typename T1, typename T2, bool primaryNode>
  47. inline void Ipc<T1, T2, primaryNode>::listen(void)
  48. {
  49.     T2                          receiveData;//Buffer to store received data
  50.     std::uint64_t               receiveLength;
  51.     std::uint32_t               priority;
  52.     while(this->mExitReceiveThread.load() == false)
  53.     {
  54.         try
  55.         {
  56.             std::memset(&receiveData, 0, sizeof(receiveData)); //Initialize buffer to 0
  57.             receiveLength = 0; //Initialize read length to 0
  58.             priority = 0; //Initialize priority to 0
  59.             if (primaryNode == true)
  60.             {
  61.                 this->mSecondaryToPrimaryMessageQueue->receive(
  62.                     &receiveData,
  63.                     sizeof(receiveData),
  64.                     receiveLength,
  65.                     priority
  66.                 );
  67.             }
  68.             else
  69.             {
  70.                 this->mPrimaryToSecondaryMessageQueue->receive(
  71.                     &receiveData,
  72.                     sizeof(receiveData),
  73.                     receiveLength,
  74.                     priority
  75.                 );
  76.             }
  77.             this->mReceiveDataQueue.push(receiveData);
  78.             this->mCallback();
  79.         }
  80.         catch (const std::exception& ex)
  81.         {
  82.             std::cout << "Inside Listen Exception\n";
  83.             std::cout << ex.what() << std::endl;
  84.         }
  85.     }
  86. }
  88. template<typename T1, typename T2, bool primaryNode>
  89. inline bool Ipc<T1, T2, primaryNode>::open(const std::string& queueName)
  90. {
  91.     try
  92.     {
  93.         if(primaryNode == true)
  94.         {
  95.             this->mPrimaryToSecondaryMessageQueueName = queueName + std::string("_out");
  96.             this->mSecondaryToPrimaryMessageQueueName = queueName + std::string("_in");
  97.         }
  98.         else
  99.         {
  100.             this->mPrimaryToSecondaryMessageQueueName = queueName + std::string("_out");
  101.             this->mSecondaryToPrimaryMessageQueueName = queueName + std::string("_in");
  102.         }
  103.         //Open-Create message queue to send data from primaryNode node to secondary node
  104.         this->mPrimaryToSecondaryMessageQueue = std::make_unique<boost::interprocess::message_queue>(
  105.             boost::interprocess::open_or_create,
  106.             this->mPrimaryToSecondaryMessageQueueName.c_str(),
  107.             MAX_MESSAGE_DEPTH,
  108.             sizeof(T1)
  109.         );
  111.         //Open-Create message queue to send data from secondary node to primaryNode node  
  112.         this->mSecondaryToPrimaryMessageQueue = std::make_unique<boost::interprocess::message_queue>(
  113.             boost::interprocess::open_or_create,
  114.             this->mSecondaryToPrimaryMessageQueueName.c_str(),
  115.             MAX_MESSAGE_DEPTH,
  116.             sizeof(T2)
  117.         );
  119.         //Start Listner Thread
  120.         this->mReceiveThread = std::thread(&Ipc::listen, this);
  121.         return true;
  122.     }
  123.     catch (const std::exception& ex)
  124.     {
  125.         std::cout << ex.what() << std::endl;
  126.         return false;
  127.     }
  128. }
  130. template<typename T1, typename T2, bool primaryNode>
  131. inline bool Ipc<T1, T2, primaryNode>::close(void)
  132. {
  133.     try
  134.     {
  135.         this->mExitReceiveThread.store(true); //Marked to close thread
  136.         boost::interprocess::message_queue::remove(this->mPrimaryToSecondaryMessageQueueName.c_str());//Delete Primary to Secondary Message Queue
  137.         boost::interprocess::message_queue::remove(this->mSecondaryToPrimaryMessageQueueName.c_str());//Delete Secondary to Primary Message Queue
  138.     }
  139.     catch (const std::exception& ex)
  140.     {
  141.         std::cout << ex.what() << std::endl;
  142.         return false;
  143.     }
  144. }
  146. template<typename T1, typename T2, bool primaryNode>
  147. inline bool Ipc<T1, T2, primaryNode>::send(const T1& data, std::uint32_t priority)
  148. {
  149.     try
  150.     {
  151.         if (primaryNode == true) //Send message on Primary to Secondary Queue
  152.         {
  153.             this->mPrimaryToSecondaryMessageQueue->send(&data, sizeof(data), priority);
  154.         }
  155.         else //Send message on Secondary to Primary Queue
  156.         {
  157.             this->mSecondaryToPrimaryMessageQueue->send(&data, sizeof(data), priority);
  158.         }
  159.         return true;
  160.     }
  161.     catch (const std::exception& ex)
  162.     {
  163.         std::cout << ex.what() << std::endl;
  164.         return false;
  165.     }
  166. }
  168. template<typename T1, typename T2, bool primaryNode>
  169. inline std::optional<T2> Ipc<T1, T2, primaryNode>::receive(void)
  170. {
  171.     std::optional<T2> data{ std::nullopt };
  172.     if (this->mReceiveDataQueue.read_available() > 0) //If data is avaiable, pop first element
  173.     {
  174.         data = this->mReceiveDataQueue.front();
  175.         this->mReceiveDataQueue.pop();
  176.     }
  177.     else
  178.     {
  179.         //data = std::nullopt; //Not needed 
  180.     }
  181.     return data;
  182. }
  184. template<typename T1, typename T2, bool primaryNode>
  185. inline bool Ipc<T1, T2, primaryNode>::register_callback(callback_t callbackImplementation)
  186. {
  187.     try
  188.     {
  189.         this->mCallback = callbackImplementation;
  190.         return true;
  191.     }
  192.     catch (const std::exception& ex)
  193.     {
  194.         std::cerr << ex.what() << '\n';
  195.     }
  196.     return false;
  197. }
  199. template<typename T1, typename T2, bool primaryNode>
  200. inline bool Ipc<T1, T2, primaryNode>::isDataAvailableInReceiveDataQueue(void) const
  201. {
  202.     if (this->mReceiveDataQueue.read_available() > 0) //If data is avaiable
  203.     {
  204.         return true;
  205.     }
  206.     else
  207.     {
  208.         return false;
  209.     }
  210. }

型

main.cpp

  1. #include <ipc.hpp>
  2. #include <iostream>
  4. //P1 stands for Process 1
  5. //P2 stands for Process 2
  6. struct P1ToP2
  7. {
  8.     float a;
  9.     int b;
  10. };
  12. struct P2ToP1
  13. {
  14.     int a;
  15.     int b;
  16. };
  18. Ipc<P1ToP2, P2ToP1, false> ipc1; //Global IPC object
  20. void message_queue_data_received(void)
  21. {
  22.     if (ipc1.isDataAvailableInReceiveDataQueue() == true)
  23.     {
  24.         auto s = ipc1.receive();
  25.         if (s.has_value() == true)
  26.         {
  27.             std::cout << "a : " << s->a << "\tb : " << s->b << std::endl;
  28.         }
  30.     }
  31. }
  33. int main(int argc, char *argv[])
  34. {
  35.     bool dataReceivedOnEthernet = false;
  36.     
  37.     ipc1.register_callback(message_queue_data_received);
  38.     this->ipc1.open("ipc1");
  39.     
  40.     while(true)
  41.     {
  42.         if(dataReceivedOnEthernet == true) //Flag set by other thread
  43.         {
  44.             P1ToP2 p;
  45.             p.a = 10.23; //Some Data received over ethernet
  46.             p.b = 10; //Some Data received over ethernet
  47.             ipc1.send(p); //Send data over IPC
  48.         }
  49.         //Other Code
  50.     }
  51. }

型
误差

  1. boost::interprocess_exception::library_error

型

c++

1条答案

按热度按时间
zzwlnbp8

zzwlnbp81#

为什么进程使用不同的消息类型，同时默认它们的大小相同（以及琐碎和标准布局等）。你混淆了各种类型和队列吗？看起来是这样。
如果你能给事物起个好名字会很有帮助。
我将按消息类型分隔队列。按角色命名它们：

  1. // Requests are sent by client to server.
  2. // Responses are sent by server to client.
  3. struct Request { int a, b; };
  4. struct Response { double a; int b; };
  5. using ClientIpc = Ipc<Request, Response, true>;
  6. using ServerIpc = Ipc<Request, Response, false>;

字符串
然后，通过定义Channel类型：

  1. using Prio = uint32_t;
  2. template <typename T> struct Channel {
  3.     Channel(std::string name);
  4.     ~Channel();
  6.     std::tuple<T, Prio> receive(std::stop_token token);
  7.     bool send(T const& msg, Prio prio, std::stop_token token);
  9.   private:
  10.     using Queue = boost::interprocess::message_queue;
  11.     std::string name_;
  12.     Queue       queue_ = open();
  14.     Queue open() {
  15.         if constexpr (IsClient) {
  16.             return {boost::interprocess::open_only, name_.c_str()};
  17.         } else {
  18.             return {boost::interprocess::open_or_create, name_.c_str(), MAX_DEPTH, sizeof(T)};
  19.         }
  20.     }
  21. };

型
现在我们可以简单地说：

  1. Channel<Request>  reqQ;
  2. Channel<Response> resQ;

型
与建筑一样，

  1. Ipc(std::string const& queueName) try
  2.     : reqQ(queueName + "_req")
  3.     , resQ(queueName + "_res")
  4.     , mThread(bind(&Ipc::listen, this, std::placeholders::_1)) {
  5. } catch (std::exception const& ex) {
  6.     std::cerr << "Ipc: " << ex.what() << std::endl;
  7.     throw;
  8. }

型
侦听器将接收到的消息排队。类型取决于客户端/服务器模式：

  1. using Incoming = std::conditional_t<IsClient, Response, Request>;
  2. boost::lockfree::spsc_queue<Incoming, boost::lockfree::capacity<MAX_DEPTH>> mInbox;

型
请注意我是如何选择更安全的jthread和停止令牌来协调线程退出的：

  1. std::jthread      mThread;
  2. std::stop_token   mToken = mThread.get_stop_token();
  4. void listen(std::stop_token token) {
  5.     while (!token.stop_requested()) {
  6.         try {
  7.             if constexpr (IsClient)
  8.                 mInbox.push(get<0>(resQ.receive(token)));
  9.             else
  10.                 mInbox.push(get<0>(reqQ.receive(token)));
  12.             if (mCallback)
  13.                 mCallback();
  14.         } catch (std::exception const& ex) {
  15.             std::cerr << "Listen: " << ex.what() << std::endl;
  16.         }
  17.     }
  18. }

型
外部操作看起来更简单，如：

  1. void wait() { mThread.join(); }
  3. void close() {
  4.     mThread.request_stop();
  5.     if (std::this_thread::get_id() != mThread.get_id())
  6.         wait();
  7. }
  9. bool send(Request  const& msg, Prio prio = 10) { return reqQ.send(msg, prio, mToken); }
  10. bool send(Response const& msg, Prio prio = 10) { return resQ.send(msg, prio, mToken); }
  12. std::optional<Incoming> consume() {
  13.     if (Incoming val; mInbox.consume_one([&val](Incoming res) { val = std::move(res); }))
  14.         return val;
  15.     return {};
  16. }
  18. void register_callback(callback_t cb) { mCallback = cb; }
  19. bool haveMessage() const { return mInbox.read_available(); }

型

客户端/服务器示例

让我们定义上面的服务器通过发送a和b/2的平方根来响应Requests：

  1. void server() {
  2.     ServerIpc ipc(IPC_NAME);
  4.     auto handler = [&ipc] {
  5.         assert(ipc.haveMessage());
  6.         if (std::optional<Request> req = ipc.consume()) {
  7.             auto [a, b] = *req;
  8.             std::cout << "server received request a:" << a << " b:" << b << std::endl;
  10.             if (a == -42 && b == -42) {
  11.                 std::cout << " -> server handling close request" << std::endl;
  12.                 ipc.close();
  13.             } else {
  14.                 // send response
  15.                 ipc.send(Response{sqrt(a), b / 2});
  16.             }
  17.         }
  18.     };
  20.     ipc.register_callback(handler);
  21.     ipc.wait();
  22. }

型
这就是全部。注意我们如何添加一个机制来告诉服务器客户端希望它退出（因为服务器拥有资源）。客户端可能看起来像这样：

  1. void client() {
  2.     ClientIpc ipc(IPC_NAME);
  4.     auto handler = [&ipc] {
  5.         assert(ipc.haveMessage());
  6.         if (std::optional<Response> res = ipc.consume()) {
  7.             auto [a, b] = *res;
  8.             std::cout << "client received response a:" << a << " b:" << b << std::endl;
  9.         }
  10.     };
  12.     ipc.register_callback(handler);
  14.     for (int i = 0; i < 100; ++i) {
  15.         if (rand() % 30 == 0)            // Flag set by other thread
  16.             ipc.send(Request{i, 2 * i}); // Send request
  18.         std::this_thread::sleep_for(10ms);
  19.     }
  21.     std::cout << "Client sending close command" << std::endl;
  22.     ipc.send(Request{-42, -42});
  24.     std::cout << "Closing" << std::endl;
  25.     ipc.close();
  26. }

型
它所做的就是发送一些请求，持续大约10秒，然后记录响应。然后它告诉服务器退出，然后关闭。只有服务器会删除队列。
一个简单的主切换客户端/服务器：

  1. int main(int argc, char** argv) {
  2.     if (std::set<std::string_view>(argv + 1, argv + argc).contains("server"))
  3.         server();
  4.     else
  5.         client();
  7.     std::cout << "Bye" << std::endl;
  8. }

型

现场演示

Live¹ On Coliru

  • 文件test.h
  1. #pragma once
  3.  #include <boost/interprocess/ipc/message_queue.hpp>
  4.  #include <boost/lockfree/spsc_queue.hpp>
  5.  #include <iostream>
  6.  #include <optional>
  7.  #include <thread>
  8.  using namespace std::chrono_literals;
  10.  template <typename Request, typename Response, bool IsClient> class Ipc {
  11.    private:
  12.      static constexpr uint8_t MAX_DEPTH = 5;
  14.      using callback_t = std::function<void()>;
  15.      callback_t mCallback;
  17.      using Prio = uint32_t;
  18.      template <typename T> struct Channel {
  19.          Channel(std::string name) : name_(std::move(name)) { //
  20.              assert(queue_.get_max_msg_size() == sizeof(T));
  21.          }
  23.          ~Channel() {
  24.              if (!IsClient) {
  25.                  std::cerr << "Server cleaning up " << name_ << std::endl;
  26.                  Queue::remove(name_.c_str());
  27.              }
  28.          }
  30.          std::tuple<T, Prio> receive(std::stop_token token) {
  31.              size_t len  = 0;
  32.              Prio   prio = 0;
  33.              T      msg{};
  35.              while (!token.stop_requested()) {
  36.                  auto deadline = std::chrono::steady_clock::now() + 50ms;
  37.                  if (queue_.timed_receive(&msg, sizeof(msg), len, prio, deadline)) {
  38.                      assert(len == sizeof(T));
  39.                      return {std::move(msg), prio};
  40.                  }
  41.              }
  43.              throw std::runtime_error("stop requested");
  44.          }
  46.          bool send(T const& msg, Prio prio, std::stop_token token) {
  47.              while (!token.stop_requested()) {
  48.                  auto deadline = std::chrono::steady_clock::now() + 50ms;
  49.                  if (queue_.timed_send(&msg, sizeof(msg), prio, deadline))
  50.                      return true;
  51.              }
  52.              return false;
  53.          }
  55.        private:
  56.          using Queue = boost::interprocess::message_queue;
  57.          std::string name_;
  58.          Queue       queue_ = open();
  60.          Queue open() {
  61.              if constexpr (IsClient) {
  62.                  return {boost::interprocess::open_only, name_.c_str()};
  63.              } else {
  64.                  return {boost::interprocess::open_or_create, name_.c_str(), MAX_DEPTH, sizeof(T)};
  65.              }
  66.          }
  67.      };
  69.      Channel<Request>  reqQ;
  70.      Channel<Response> resQ;
  71.      std::jthread      mThread;
  72.      std::stop_token   mToken = mThread.get_stop_token();
  74.      using Incoming = std::conditional_t<IsClient, Response, Request>;
  75.      boost::lockfree::spsc_queue<Incoming, boost::lockfree::capacity<MAX_DEPTH>> mInbox;
  77.      void listen(std::stop_token token) {
  78.          while (!token.stop_requested()) {
  79.              try {
  80.                  if constexpr (IsClient)
  81.                      mInbox.push(get<0>(resQ.receive(token)));
  82.                  else
  83.                      mInbox.push(get<0>(reqQ.receive(token)));
  85.                  if (mCallback)
  86.                      mCallback();
  87.              } catch (std::exception const& ex) {
  88.                  std::cerr << "Listen: " << ex.what() << std::endl;
  89.              }
  90.          }
  91.      }
  93.    public:
  94.      Ipc(std::string const& queueName) try
  95.          : reqQ(queueName + "_req")
  96.          , resQ(queueName + "_res")
  97.          , mThread(bind(&Ipc::listen, this, std::placeholders::_1)) {
  98.      } catch (std::exception const& ex) {
  99.          std::cerr << "Ipc: " << ex.what() << std::endl;
  100.          throw;
  101.      }
  103.      void wait() { mThread.join(); }
  105.      void close() {
  106.          mThread.request_stop();
  107.          if (std::this_thread::get_id() != mThread.get_id())
  108.              wait();
  109.      }
  111.      bool send(Request  const& msg, Prio prio = 10) { return reqQ.send(msg, prio, mToken); }
  112.      bool send(Response const& msg, Prio prio = 10) { return resQ.send(msg, prio, mToken); }
  114.      std::optional<Incoming> consume() {
  115.          if (Incoming val; mInbox.consume_one([&val](Incoming res) { val = std::move(res); }))
  116.              return val;
  117.          return {};
  118.      }
  120.      void register_callback(callback_t cb) { mCallback = cb; }
  121.      bool haveMessage() const { return mInbox.read_available(); }
  122.  };

型

  • 文件test.cpp
  1. #include "test.h"
  2.  #include <cmath>
  3.  #include <set>
  5.  // Requests are sent by client to server.
  6.  // Responses are sent by server to client.
  7.  struct Request { int a, b; };
  8.  struct Response { double a; int b; };
  9.  using ClientIpc = Ipc<Request, Response, true>;
  10.  using ServerIpc = Ipc<Request, Response, false>;
  12.  static std::string IPC_NAME = "so_demo_ipc";
  13.  void server() {
  14.      ServerIpc ipc(IPC_NAME);
  16.      auto handler = [&ipc] {
  17.          assert(ipc.haveMessage());
  18.          if (std::optional<Request> req = ipc.consume()) {
  19.              auto [a, b] = *req;
  20.              std::cout << "server received request a:" << a << " b:" << b << std::endl;
  22.              if (a == -42 && b == -42) {
  23.                  std::cout << " -> server handling close request" << std::endl;
  24.                  ipc.close();
  25.              } else {
  26.                  // send response
  27.                  ipc.send(Response{sqrt(a), b / 2});
  28.              }
  29.          }
  30.      };
  32.      ipc.register_callback(handler);
  33.      ipc.wait();
  34.  }
  36.  void client() {
  37.      ClientIpc ipc(IPC_NAME);
  39.      auto handler = [&ipc] {
  40.          assert(ipc.haveMessage());
  41.          if (std::optional<Response> res = ipc.consume()) {
  42.              auto [a, b] = *res;
  43.              std::cout << "client received response a:" << a << " b:" << b << std::endl;
  44.          }
  45.      };
  47.      ipc.register_callback(handler);
  49.      for (int i = 0; i < 100; ++i) {
  50.          if (rand() % 30 == 0)            // Flag set by other thread
  51.              ipc.send(Request{i, 2 * i}); // Send request
  53.          std::this_thread::sleep_for(10ms);
  54.      }
  56.      std::cout << "Client sending close command" << std::endl;
  57.      ipc.send(Request{-42, -42});
  59.      std::cout << "Closing" << std::endl;
  60.      ipc.close();
  61.  }
  63.  int main(int argc, char** argv) {
  64.      if (std::set<std::string_view>(argv + 1, argv + argc).contains("server"))
  65.          server();
  66.      else
  67.          client();
  69.      std::cout << "Bye" << std::endl;
  70.  }

型
在本地进行现场演示：
x1c 0d1x的数据
遗憾的是，在线编译器不允许共享内存访问

展开查看全部
赞(0）分享  回复(0）举报  2024-01-09
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com