Method run

src/system/executor.cc:148–267

a simple implementation of the DAG execution engine.

// a simple implementation of the DAG execution engine.
void Executor::run() {
  while (!done_) {
    bool process = false;
    if (FLAGS_verbose) {
      LI << obj_.name() << " before entering task loop; recved_msgs_. size [" <<
        recved_msgs_.size() << "]";
    }
    {
      std::unique_lock<std::mutex> lk(recved_msg_mu_);
      // pickup a message with dependency satisfied
      for (auto it = recved_msgs_.begin(); it != recved_msgs_.end(); ++it) {
        auto& msg = *it;
        auto sender = rnode(msg->sender);
        if (!sender) {
          LL << my_node_.id() << ": " << msg->sender
             << " does not exist, ignore\n" << msg->debugString();
          recved_msgs_.erase(it);
          break;
        }
        // ack message, no dependency constraint
        process = !msg->task.request();
        if (!process) {
          // check if the dependency constraints are satisfied
          bool satisfied = true;
          for (int i = 0; i < msg->task.wait_time_size(); ++i) {
            int wait_time = msg->task.wait_time(i);
            if (wait_time > Message::kInvalidTime &&
                !sender->tryWaitIncomingTask(wait_time)) {
              satisfied = false;
            }
          }
          process = satisfied;
        }
        if (process) {
          active_msg_ = msg;
          recved_msgs_.erase(it);
          if (FLAGS_verbose) {
            LI << obj_.name() << " picked up an active_msg_ from recved_msgs_. " <<
              "remaining size [" << recved_msgs_.size() << "]; msg [" <<
              active_msg_->shortDebugString() << "]";
          }
          break;
        }
      }
      if (!process) {
        if (FLAGS_verbose) {
          LI << obj_.name() << " picked nothing from recved_msgs_. size [" <<
            recved_msgs_.size() << "] waiting Executor::accept";
        }
        dag_cond_.wait(lk);
        continue;
      }
    }
    // process the picked message
    bool req = active_msg_->task.request();
    int t = active_msg_->task.time();
    auto sender = rnode(active_msg_->sender);
    CHECK(sender) << "unknow node: " << active_msg_->sender;