Method SyncData

src/replica/replicate_node.cc:117–151

}

void ReplicateNode::SyncData() {
    uint32_t coffee_time = 0;
    while (is_running_.load(std::memory_order_relaxed)) {
        if (coffee_time > 0) {
            bthread_usleep(coffee_time * 1000);
            coffee_time = 0;
        }
        {
            std::unique_lock<bthread::Mutex> lock(*mu_);
            // no new data append and wait
            while (last_sync_offset_ >=
                   leader_log_offset_->load(std::memory_order_relaxed)) {
                cv_->wait_for(lock, FLAGS_binlog_sync_wait_time * 1000);
                if (!is_running_.load(std::memory_order_relaxed)) {
                    PDLOG(INFO,
                          "replicate log to endpoint %s for table #tid %u #pid "
                          "%u exist",
                          endpoint_.c_str(), tid_, pid_);
                    return;
                }
            }
        }
        int ret;
        if (rep_node_.load(std::memory_order_relaxed)) {
            ret = SyncData(follower_offset_->load(std::memory_order_relaxed));
        } else {
            ret = SyncData(leader_log_offset_->load(std::memory_order_relaxed));
        }
        if (ret == 1) {
            coffee_time = FLAGS_binlog_coffee_time;
        }
    }
    PDLOG(INFO, "replicate log to endpoint %s for table #tid %u #pid %u exist",
          endpoint_.c_str(), tid_, pid_);
}

int ReplicateNode::GetLogIndex() { return log_reader_.GetLogIndex(); }