Method loop

src/cluster/replication.cc:178–265

}

void FeedSlaveThread::loop() {
  // is_first_repl_batch was used to fix that replication may be stuck in a dead loop
  // when some seqs might be lost in the middle of the WAL log, so forced to replicate
  // first batch here to work around this issue instead of waiting for enough batch size.
  bool is_first_repl_batch = true;
  uint32_t yield_microseconds = 2 * 1000;
  std::string batches_bulk;
  size_t updates_in_batches = 0;
  while (!IsStopped()) {
    auto curr_seq = next_repl_seq_.load();

    // Check replication lag - disconnect slow consumers before WAL is exhausted
    // Skip check if max_replication_lag_ is 0 (feature disabled)
    if (max_replication_lag_ > 0) {
      auto latest_seq = srv_->storage->LatestSeqNumber();
      if (latest_seq > curr_seq) {
        auto lag = static_cast<int64_t>(latest_seq - curr_seq);
        if (lag > max_replication_lag_) {
          ERROR("Replication lag {} exceeds max allowed {} for slave {}:{}, disconnecting to prevent WAL exhaustion",
                lag, max_replication_lag_, conn_->GetAnnounceIP(), conn_->GetListeningPort());
          Stop();
          return;
        }
      }
    }

    if (!iter_ || !iter_->Valid()) {
      if (iter_) INFO("WAL was rotated, would reopen again");
      if (!srv_->storage->WALHasNewData(curr_seq) || !srv_->storage->GetWALIter(curr_seq, &iter_).IsOK()) {
        iter_ = nullptr;
        usleep(yield_microseconds);
        checkLivenessIfNeed();
        continue;
      }
    }
    // iter_ would be always valid here
    auto batch = iter_->GetBatch();
    if (batch.sequence != curr_seq) {
      ERROR(
          "Fatal error encountered, WAL iterator is discrete, some seq might be lost, sequence {} expected, but got {}",
          curr_seq, batch.sequence);
      Stop();
      return;
    }
    updates_in_batches += batch.writeBatchPtr->Count();
    batches_bulk += redis::BulkString(batch.writeBatchPtr->Data());
    // 1. We must send the first replication batch, as said above.
    // 2. To avoid frequently calling 'write' system call to send replication stream,
    //    we pack multiple batches into one big bulk if possible, and only send once.
    //    But we should send the bulk of batches if its size exceed kMaxDelayBytes,
    //    16Kb by default. Moreover, we also send if updates count in all bathes is
    //    more that kMaxDelayUpdates, to void too many delayed updates.
    // 3. To avoid master don't send replication stream to slave since of packing
    //    batches strategy, we still send batches if current batch sequence is less
    //    kMaxDelayUpdates than latest sequence.
    if (is_first_repl_batch || batches_bulk.size() >= max_delay_bytes_ || updates_in_batches >= max_delay_updates_ ||
        srv_->storage->LatestSeqNumber() - batch.sequence <= max_delay_updates_) {
      // get the last sequence number of the batch, because WAIT uses