Class Executor

src/system/executor.h:24–93

Maintain all remote nodes for a customer. It has its own thread to process received tasks.

// Maintain all remote nodes for a customer. It has its own thread to process
// received tasks.
class Executor {
 public:
  Executor(Customer& obj);
  ~Executor();

  // wake up the processing thread
  void notify() { Lock l(recved_msg_mu_); dag_cond_.notify_one(); }

  // mark this message as finshed in executor
  void finish(const MessagePtr& msg);

  // stop the processing thread
  void stop();

  // accessors
  RNode* rnode(const NodeID& k);
  std::vector<RNode*>& group(const NodeID& k);
  const std::vector<Range<Key>>& keyRanges(const NodeID& k);
  const Node& myNode() { return my_node_; }

  void add(const Node& node);
  // void remove(const Node& node);
  void copyNodesFrom(const Executor& other);

  // will be called by postoffice's receiving thread
  // or the thread call wk->submit
  void accept(const MessagePtr& msg);
  // last received message
  MessagePtr activeMessage() { return active_msg_; }
  string lastRecvReply();

  Customer& obj() { return obj_; }
 private:
  void run();

  Customer& obj_;
  // Temporal buffer for received messages
  std::list<MessagePtr> recved_msgs_;
  std::mutex recved_msg_mu_;
  // the message is going to be processed or is the last one be processed
  MessagePtr active_msg_;
  std::condition_variable dag_cond_;

  std::vector<NodeID> groupIDs() {
   std::vector<NodeID> ids = {
     kServerGroup, kWorkerGroup, kCompGroup,
     kReplicaGroup, kOwnerGroup, kLiveGroup};
    return ids;
  }

  struct NodeInfo {
    NodeInfo() { };
    ~NodeInfo() { delete node; }

    void addSubNode(RNode* s);
    void removeSubNode(RNode* s);

    RNode* node = NULL;