<commit_before>/*
 * Copyright 2015-present Open Networking Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */
package io.atomix.protocols.raft.roles;

import io.atomix.protocols.raft.RaftException;
import io.atomix.protocols.raft.RaftServer;
import io.atomix.protocols.raft.cluster.RaftMember;
import io.atomix.protocols.raft.cluster.impl.RaftMemberContext;
import io.atomix.protocols.raft.protocol.AppendRequest;
import io.atomix.protocols.raft.protocol.AppendResponse;
import io.atomix.protocols.raft.protocol.ConfigureRequest;
import io.atomix.protocols.raft.protocol.ConfigureResponse;
import io.atomix.protocols.raft.protocol.InstallRequest;
import io.atomix.protocols.raft.protocol.InstallResponse;
import io.atomix.protocols.raft.protocol.RaftRequest;
import io.atomix.protocols.raft.storage.snapshot.Snapshot;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * The leader appender is responsible for sending {@link AppendRequest}s on behalf of a leader to followers.
 * Append requests are sent by the leader only to other active members of the cluster.
 */
final class LeaderAppender extends AbstractAppender {
  private static final long MAX_HEARTBEAT_WAIT = 60000;
  private static final int MIN_BACKOFF_FAILURE_COUNT = 5;
  private static final int MIN_STEP_DOWN_FAILURE_COUNT = 3;

  private final long leaderTime;
  private final long leaderIndex;
  private final long electionTimeout;
  private final long heartbeatInterval;
  private long heartbeatTime;
  private final Map<Long, CompletableFuture<Long>> appendFutures = new HashMap<>();
  private final List<TimestampedFuture<Long>> heartbeatFutures = new ArrayList<>();

  LeaderAppender(LeaderRole leader) {
    super(leader.raft);
    this.leaderTime = System.currentTimeMillis();
    this.leaderIndex = raft.getLogWriter().getNextIndex();
    this.heartbeatTime = leaderTime;
    this.electionTimeout = raft.getElectionTimeout().toMillis();
    this.heartbeatInterval = raft.getHeartbeatInterval().toMillis();
  }

  /**
   * Returns the current commit time.
   *
   * @return The current commit time.
   */
  public long getTime() {
    return heartbeatTime;
  }

  /**
   * Returns the leader index.
   *
   * @return The leader index.
   */
  public long getIndex() {
    return leaderIndex;
  }

  /**
   * Returns the current quorum index.
   *
   * @return The current quorum index.
   */
  private int getQuorumIndex() {
    return raft.getCluster().getQuorum() - 2;
  }

  /**
   * Triggers a heartbeat to a majority of the cluster.
   * <p>
   * For followers to which no AppendRequest is currently being sent, a new empty AppendRequest will be
   * created and sent. For followers to which an AppendRequest is already being sent, the appendEntries()
   * call will piggyback on the *next* AppendRequest. Thus, multiple calls to this method will only ever
   * result in a single AppendRequest to each follower at any given time, and the returned future will be
   * shared by all concurrent calls.
   *
   * @return A completable future to be completed the next time a heartbeat is received by a majority of the cluster.
   */
  public CompletableFuture<Long> appendEntries() {
    raft.checkThread();

    // If there are no other active members in the cluster, simply complete the append operation.
    if (raft.getCluster().getRemoteMemberStates().isEmpty()) {
      return CompletableFuture.completedFuture(null);
    }

    // Create a heartbeat future and add it to the heartbeat futures list.
    TimestampedFuture<Long> future = new TimestampedFuture<>();
    heartbeatFutures.add(future);

    // Iterate through members and append entries. Futures will be completed on responses from followers.
    for (RaftMemberContext member : raft.getCluster().getRemoteMemberStates()) {
      appendEntries(member);
    }
    return future;
  }

  /**
   * Registers a commit handler for the given commit index.
   *
   * @param index The index for which to register the handler.
   * @return A completable future to be completed once the given log index has been committed.
   */
  public CompletableFuture<Long> appendEntries(long index) {
    raft.checkThread();

    if (index == 0) {
      return appendEntries();
    }

    if (index <= raft.getCommitIndex()) {
      return CompletableFuture.completedFuture(index);
    }

    // If there are no other stateful servers in the cluster, immediately commit the index.
    if (raft.getCluster().getActiveMemberStates().isEmpty() && raft.getCluster().getPassiveMemberStates().isEmpty()) {
      long previousCommitIndex = raft.getCommitIndex();
      raft.setCommitIndex(index);
      completeCommits(previousCommitIndex, index);
      return CompletableFuture.completedFuture(index);
    }
    // If there are no other active members in the cluster, update the commit index and complete the commit.
    // The updated commit index will be sent to passive/reserve members on heartbeats.
    else if (raft.getCluster().getActiveMemberStates().isEmpty()) {
      long previousCommitIndex = raft.getCommitIndex();
      raft.setCommitIndex(index);
      completeCommits(previousCommitIndex, index);
      return CompletableFuture.completedFuture(index);
    }

    // Only send entry-specific AppendRequests to active members of the cluster.
    return appendFutures.computeIfAbsent(index, i -> {
      for (RaftMemberContext member : raft.getCluster().getActiveMemberStates()) {
        appendEntries(member);
      }
      return new CompletableFuture<>();
    });
  }

  @Override
  protected void appendEntries(RaftMemberContext member) {
    // Prevent recursive, asynchronous appends from being executed if the appender has been closed.
    if (!open) {
      return;
    }

    // If prior requests to the member have failed, build an empty append request to send to the member
    // to prevent having to read from disk to configure, install, or append to an unavailable member.
    if (member.getFailureCount() >= MIN_BACKOFF_FAILURE_COUNT) {
      // To prevent the leader from unnecessarily attempting to connect to a down follower on every heartbeat,
      // use exponential backoff to back off up to 60 second heartbeat intervals.
      if (System.currentTimeMillis() - member.getFailureTime() > Math.min(heartbeatInterval * Math.pow(2, member.getFailureCount()), MAX_HEARTBEAT_WAIT)) {
        sendAppendRequest(member, buildAppendEmptyRequest(member));
      }
    }
    // If the member term is less than the current term or the member's configuration index is less
    // than the local configuration index, send a configuration update to the member.
    // Ensure that only one configuration attempt per member is attempted at any given time by storing the
    // member state in a set of configuring members.
    // Once the configuration is complete sendAppendRequest will be called recursively.
    else if (member.getConfigTerm() < raft.getTerm()
        || member.getConfigIndex() < raft.getCluster().getConfiguration().index()
        && member.canConfigure()) {
      sendConfigureRequest(member, buildConfigureRequest(member));
    }
    // If there's a snapshot at the member's nextIndex, replicate the snapshot.
    else if (member.getMember().getType() == RaftMember.Type.ACTIVE
        || member.getMember().getType() == RaftMember.Type.PROMOTABLE
        || member.getMember().getType() == RaftMember.Type.PASSIVE) {
      long currentIndex = member.getLogReader().getCurrentIndex();
      Collection<Snapshot> snapshots = raft.getSnapshotStore().getSnapshotsByIndex(currentIndex);
      if (snapshots != null && member.getSnapshotIndex() <= currentIndex) {
        if (!member.canInstall()) {<commit_msg>Allow empty heartbeats to be sent to followers during configuration.<commit_after>/*
 * Copyright 2015-present Open Networking Foundation
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License
 */
package io.atomix.protocols.raft.roles;

import io.atomix.protocols.raft.RaftException;
import io.atomix.protocols.raft.RaftServer;
import io.atomix.protocols.raft.cluster.RaftMember;
import io.atomix.protocols.raft.cluster.impl.RaftMemberContext;
import io.atomix.protocols.raft.protocol.AppendRequest;
import io.atomix.protocols.raft.protocol.AppendResponse;
import io.atomix.protocols.raft.protocol.ConfigureRequest;
import io.atomix.protocols.raft.protocol.ConfigureResponse;
import io.atomix.protocols.raft.protocol.InstallRequest;
import io.atomix.protocols.raft.protocol.InstallResponse;
import io.atomix.protocols.raft.protocol.RaftRequest;
import io.atomix.protocols.raft.storage.snapshot.Snapshot;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * The leader appender is responsible for sending {@link AppendRequest}s on behalf of a leader to followers.
 * Append requests are sent by the leader only to other active members of the cluster.
 */
final class LeaderAppender extends AbstractAppender {
  private static final long MAX_HEARTBEAT_WAIT = 60000;
  private static final int MIN_BACKOFF_FAILURE_COUNT = 5;
  private static final int MIN_STEP_DOWN_FAILURE_COUNT = 3;

  private final long leaderTime;
  private final long leaderIndex;
  private final long electionTimeout;
  private final long heartbeatInterval;
  private long heartbeatTime;
  private final Map<Long, CompletableFuture<Long>> appendFutures = new HashMap<>();
  private final List<TimestampedFuture<Long>> heartbeatFutures = new ArrayList<>();

  LeaderAppender(LeaderRole leader) {
    super(leader.raft);
    this.leaderTime = System.currentTimeMillis();
    this.leaderIndex = raft.getLogWriter().getNextIndex();
    this.heartbeatTime = leaderTime;
    this.electionTimeout = raft.getElectionTimeout().toMillis();
    this.heartbeatInterval = raft.getHeartbeatInterval().toMillis();
  }

  /**
   * Returns the current commit time.
   *
   * @return The current commit time.
   */
  public long getTime() {
    return heartbeatTime;
  }

  /**
   * Returns the leader index.
   *
   * @return The leader index.
   */
  public long getIndex() {
    return leaderIndex;
  }

  /**
   * Returns the current quorum index.
   *
   * @return The current quorum index.
   */
  private int getQuorumIndex() {
    return raft.getCluster().getQuorum() - 2;
  }

  /**
   * Triggers a heartbeat to a majority of the cluster.
   * <p>
   * For followers to which no AppendRequest is currently being sent, a new empty AppendRequest will be
   * created and sent. For followers to which an AppendRequest is already being sent, the appendEntries()
   * call will piggyback on the *next* AppendRequest. Thus, multiple calls to this method will only ever
   * result in a single AppendRequest to each follower at any given time, and the returned future will be
   * shared by all concurrent calls.
   *
   * @return A completable future to be completed the next time a heartbeat is received by a majority of the cluster.
   */
  public CompletableFuture<Long> appendEntries() {
    raft.checkThread();

    // If there are no other active members in the cluster, simply complete the append operation.
    if (raft.getCluster().getRemoteMemberStates().isEmpty()) {
      return CompletableFuture.completedFuture(null);
    }

    // Create a heartbeat future and add it to the heartbeat futures list.
    TimestampedFuture<Long> future = new TimestampedFuture<>();
    heartbeatFutures.add(future);

    // Iterate through members and append entries. Futures will be completed on responses from followers.
    for (RaftMemberContext member : raft.getCluster().getRemoteMemberStates()) {
      appendEntries(member);
    }
    return future;
  }

  /**
   * Registers a commit handler for the given commit index.
   *
   * @param index The index for which to register the handler.
   * @return A completable future to be completed once the given log index has been committed.
   */
  public CompletableFuture<Long> appendEntries(long index) {
    raft.checkThread();

    if (index == 0) {
      return appendEntries();
    }

    if (index <= raft.getCommitIndex()) {
      return CompletableFuture.completedFuture(index);
    }

    // If there are no other stateful servers in the cluster, immediately commit the index.
    if (raft.getCluster().getActiveMemberStates().isEmpty() && raft.getCluster().getPassiveMemberStates().isEmpty()) {
      long previousCommitIndex = raft.getCommitIndex();
      raft.setCommitIndex(index);
      completeCommits(previousCommitIndex, index);
      return CompletableFuture.completedFuture(index);
    }
    // If there are no other active members in the cluster, update the commit index and complete the commit.
    // The updated commit index will be sent to passive/reserve members on heartbeats.
    else if (raft.getCluster().getActiveMemberStates().isEmpty()) {
      long previousCommitIndex = raft.getCommitIndex();
      raft.setCommitIndex(index);
      completeCommits(previousCommitIndex, index);
      return CompletableFuture.completedFuture(index);
    }

    // Only send entry-specific AppendRequests to active members of the cluster.
    return appendFutures.computeIfAbsent(index, i -> {
      for (RaftMemberContext member : raft.getCluster().getActiveMemberStates()) {
        appendEntries(member);
      }
      return new CompletableFuture<>();
    });
  }

  @Override
  protected void appendEntries(RaftMemberContext member) {
    // Prevent recursive, asynchronous appends from being executed if the appender has been closed.
    if (!open) {
      return;
    }

    // If prior requests to the member have failed, build an empty append request to send to the member
    // to prevent having to read from disk to configure, install, or append to an unavailable member.
    if (member.getFailureCount() >= MIN_BACKOFF_FAILURE_COUNT) {
      // To prevent the leader from unnecessarily attempting to connect to a down follower on every heartbeat,
      // use exponential backoff to back off up to 60 second heartbeat intervals.
      if (System.currentTimeMillis() - member.getFailureTime() > Math.min(heartbeatInterval * Math.pow(2, member.getFailureCount()), MAX_HEARTBEAT_WAIT)) {
        sendAppendRequest(member, buildAppendEmptyRequest(member));
      }
    }
    // If the member term is less than the current term or the member's configuration index is less
    // than the local configuration index, send a configuration update to the member.
    // Ensure that only one configuration attempt per member is attempted at any given time by storing the
    // member state in a set of configuring members.
    // Once the configuration is complete sendAppendRequest will be called recursively.
    else if (member.getConfigTerm() < raft.getTerm()
        || member.getConfigIndex() < raft.getCluster().getConfiguration().index()) {
      if (member.canConfigure()) {
        sendConfigureRequest(member, buildConfigureRequest(member));
      } else if (member.canHeartbeat()) {
        sendAppendRequest(member, buildAppendEmptyRequest(member));
      }
    }
    // If there's a snapshot at the member's nextIndex, replicate the snapshot.
    else if (member.getMember().getType() == RaftMember.Type.ACTIVE
        || member.getMember().getType() == RaftMember.Type.PROMOTABLE
        || member.getMember().getType() == RaftMember.Type.PASSIVE) {
      long currentIndex = member.getLogReader().getCurrentIndex();
      Collection<Snapshot> snapshots = raft.getSnapshotStore().getSnapshotsByIndex(currentIndex);
      if (snapshots != null && member.getSnapshotIndex() <= currentIndex) {
        if (!member.canInstall()) {
{'fraction_non_alphanumeric': 0.08894159501927068, 'fraction_numerical': 0.0023717758671805513, 'mean_word_length': 3.787397104740278, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6892403', 'n_tokens_mistral': 4554, 'n_tokens_neox': 4448, 'n_words': 1543}