Method process

src/parameter/shared_parameter.h:98–162

template <typename K>
void SharedParameter<K>::process(const MessagePtr& msg) {
  bool req = msg->task.request();
  int chl = msg->task.key_channel();
  auto call = get(msg);
  bool push = call.cmd() == CallSharedPara::PUSH;
  bool pull = call.cmd() == CallSharedPara::PULL;
  MessagePtr reply;
  if (pull && req) {
    reply = MessagePtr(new Message(*msg));
    reply->task.set_request(false);
    std::swap(reply->sender, reply->recver);
  }

  // this->sys_.hb().startTimer(HeartbeatInfo::TimerType::BUSY);
  // process
  if (call.replica()) {
    if (pull && !req && Range<K>(msg->task.key_range()) == myKeyRange()) {
      recoverFrom(msg);
    } else if ((push && req) || (pull && !req)) {
      setReplica(msg);
    } else if (pull && req) {
      getReplica(reply);
    }
  } else if (call.has_tail_filter()) {
    if (key_filter_ignore_chl_) chl = 0;
    auto cmd = call.tail_filter();
    // push the key count
    if (cmd.insert_count() && req && !msg->key.empty()) {
      CHECK(!msg->value.empty());
      auto& filter = key_filter_[chl];
      if (filter.empty()) {
        double w = (double)FLAGS_num_workers;
        filter.resize(w*cmd.countmin_n()/log(w+1), cmd.countmin_k());
      }
      filter.insertKeys(SArray<K>(msg->key), SArray<uint8>(msg->value[0]));
    }
    // pull the filtered keys
    if (cmd.has_query_key() && pull) {
      if (req) {
        reply->clearData();
        reply->setKey(key_filter_[chl].queryKeys(
            SArray<K>(msg->key), cmd.query_key()));
        if (cmd.has_query_value()) {
          getValue(reply);
        }
      } else {
        setValue(msg);
      }
    }
  } else {
    if ((push && req) || (pull && !req)) {
      setValue(msg);
    } else if (pull && req) {
      getValue(reply);
    }
  }
  // this->sys_.hb().stopTimer(HeartbeatInfo::TimerType::BUSY);