从状态内的自定义函数（不是操作）提升 MSM 调用 process_event？

Question

我正在尝试使用std::thread 在使用 Boost MSM 库编码的状态机中实现并行行为。我正在使用std::thread 从状态A 的on_entry 方法启动一个线程，我想知道如何调用process_event 方法以便从该线程中触发一个事件。这是一个最小的工作示例：

#include <iostream>
#include <thread>
#include <unistd.h>

#include <boost/msm/back/state_machine.hpp>
#include <boost/msm/front/state_machine_def.hpp>
#include <boost/msm/front/functor_row.hpp>

namespace msm = boost::msm;
namespace msmf = boost::msm::front;
namespace mpl = boost::mpl;

// Events
struct timeout {};

struct outer_:msmf::state_machine_def<outer_> {
  struct inner_:msmf::state_machine_def<inner_> {
    template <class Event, class Fsm>
    void on_entry(Event const&, Fsm&) {
      std::cout << "[state machine entry] inner" << std::endl;
    }
    struct A:msmf::state<> {
      template <class Event, class Fsm>
      void on_entry(Event const&, Fsm& f) {
        std::cout << "[state entry] A" << std::endl;
        stop_threads_ = false;
        thread_ = new std::thread(&A::func,this);
      }
      template <class Event, class Fsm>
      void on_exit(Event const&, Fsm&) {
        stop_threads_ = true;
        thread_->join(); // wait for threads to finish
        delete thread_;
        std::cout << "[state exit] A" << std::endl;
      }
      void func() {
        while (!stop_threads_) {
          usleep(1000000);
          std::cout << "Hello" << std::endl;
          // QUESTION: how to call process_event(timeout()) here?
        }
      }
     public:
      std::thread* thread_;
      bool stop_threads_;
    };
    struct Action {
      template <class Event, class Fsm, class SourceState, class TargetState>
      void operator()(Event const&, Fsm&, SourceState&, TargetState&) {
        std::cout << "Trying again..." << std::endl;
      }
    };
    typedef A initial_state;
    struct transition_table:mpl::vector<
        msmf::Row <A,timeout,A,Action>
    > {};
  };
  typedef msm::back::state_machine<inner_> inner;
  typedef inner initial_state;
};
typedef msm::back::state_machine<outer_> outer;

void waiting_thread() {
  while(true) {
    usleep(2000000);
  }
}

int main() {
  outer sm;
  sm.start();
  std::thread wait(waiting_thread);
  wait.join();
}

我的问题是评论// QUESTION...，我想要一种方法来执行process_event 方法。感谢您的帮助！

Answer 1

很遗憾，你不能这样做。

这是尝试处理func() 中的事件的代码。 这与Boost MSM parallel behavior with delayed self-transitions?的策略相似

成功完成。但我遇到了一个例外。

#include <iostream>
#include <thread>
#include <mutex>
#include <unistd.h>

#include <boost/msm/back/state_machine.hpp>
#include <boost/msm/front/state_machine_def.hpp>
#include <boost/msm/front/functor_row.hpp>

namespace msm = boost::msm;
namespace msmf = boost::msm::front;
namespace mpl = boost::mpl;

// Events
struct timeout {};

struct outer_:msmf::state_machine_def<outer_> {
  std::mutex mtx;
  typedef msm::back::state_machine<outer_> outer;
  std::weak_ptr<outer> wp;

  static std::shared_ptr<outer> create() {
    auto p = std::make_shared<outer>();
    p->wp = p; // set wp after creation.
    return p;
  }

  template <typename Ev>
  void process(Ev&& ev) {
    // process_event via backend weak_ptr
    std::lock_guard<std::mutex> g(wp.lock()->mtx);
    wp.lock()->process_event(std::forward<Ev>(ev));
  }
  struct inner_:msmf::state_machine_def<inner_> {
    std::weak_ptr<outer> wp;

    template <class Event, class Fsm>
    void on_entry(Event const&, Fsm& f) {
      std::cout << "[state machine entry] inner" << std::endl;
      wp = f.wp;
    }
    struct A:msmf::state<> {
      template <class Event, class Fsm>
      void on_entry(Event const&, Fsm& f) {
        std::cout << "[state entry] A" << std::endl;
        stop_threads_ = false;
        thread_ = new std::thread(&A::func,this, std::ref(f));
      }
      template <class Event, class Fsm>
      void on_exit(Event const&, Fsm&) {
        stop_threads_ = true;
        thread_->join(); // wait for threads to finish
        delete thread_;
        std::cout << "[state exit] A" << std::endl;
      }
      void func(inner_& f) {
        while (!stop_threads_) {
          usleep(1000000);
          std::cout << "Hello" << std::endl;
          // QUESTION: how to call process_event(timeout()) here?
          f.wp.lock()->process(timeout());
        }
      }
     public:
      std::thread* thread_;
      bool stop_threads_;
    };
    struct Action {
      template <class Event, class Fsm, class SourceState, class TargetState>
      void operator()(Event const&, Fsm&, SourceState&, TargetState&) {
        std::cout << "Trying again..." << std::endl;
      }
    };
    typedef A initial_state;
    struct transition_table:mpl::vector<
        msmf::Row <A,timeout,A,Action>
    > {};
    template <class FSM,class Event>
    void exception_caught (Event const&,FSM&,std::exception& e) {
        std::cout << e.what() << std::endl;
    }
  };
  typedef msm::back::state_machine<inner_> inner;
  typedef inner initial_state;
  template <class FSM,class Event>
  void exception_caught (Event const&,FSM&,std::exception& e) {
      std::cout << e.what() << std::endl;
  }
};

void waiting_thread() {
  while(true) {
    usleep(2000000);
  }
}

int main() {
  std::shared_ptr<outer_::outer> sm = outer_::create();
  sm->start();
  std::cout << "started" << std::endl;
  std::thread wait(waiting_thread);
  wait.join();
  std::cout << "joined" << std::endl;
}

我添加了异常打印功能如下。当 Boost.MSM 在 process_event 期间捕获异常时，将调用 exception_caught。

    template <class FSM,class Event>
    void exception_caught (Event const&,FSM&,std::exception& e) {
        std::cout << e.what() << std::endl;
    }

我收到的消息是Resource deadlock avoided。它最初是由 pthread 库引起的。如果您创建一个线程，然后从另一个线程加入该线程，则会引发异常。

让我们看看代码在哪里创建线程。

重点是：

      template <class Event, class Fsm>
      void on_entry(Event const&, Fsm& f) {
        std::cout << "[state entry] A" << std::endl;
        stop_threads_ = false;
        thread_ = new std::thread(&A::func,this, std::ref(f));
      }

它是从主线程调用的。因为它是从sm->start()调用的。

那我们检查一下连接点/

重点是：

      template <class Event, class Fsm>
      void on_exit(Event const&, Fsm&) {
        stop_threads_ = true;
        thread_->join(); // wait for threads to finish
        delete thread_;
        std::cout << "[state exit] A" << std::endl;
      }

它是从void func(inner_& f) 中的f.wp.lock()->process(timeout()); 调用的。 func 是由thread_ = new std::thread(&A::func,this, std::ref(f)); 创建的线程的入口点。 这意味着thread_->join(); 正在等待自己。

这就是Resource deadlock avoided被抛出的原因。

编辑

这是评论的答案 Boost MSM call process_event from a custom function (not an Action) inside a State?

这不是一个完整的解决方案，但可能是设计的暗示。

如果我分离线程并删除join()，那么程序将按预期运行。当然detach() 不是最好的解决方案。而thread_从未被删除。您需要在产品代码中关心它们。

以下代码的目的是演示状态机行为。

struct A:msmf::state<> {
  template <class Event, class Fsm>
  void on_entry(Event const&, Fsm& f) {
    std::cout << "[state entry] A" << std::endl;
    stop_threads_ = false;
    thread_ = new std::thread(&A::func,this, std::ref(f));
    thread_->detach();
  }
  template <class Event, class Fsm>
  void on_exit(Event const&, Fsm&) {
    stop_threads_ = true;
    std::cout << "[state exit] A" << std::endl;
  }
  struct func {
    func(inner_& f):f(f) {}
      void operator()() {
      while (!stop_threads_) {
        usleep(1000000);
        std::cout << "Hello" << std::endl;
        // QUESTION: how to call process_event(timeout()) here?
        f.wp.lock()->process(timeout());
      }
    }
  };
 public:
  std::thread* thread_;
  bool stop_threads_;
};