<commit_before>package com.rvprg.raft.protocol.impl;

import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import com.google.inject.Inject;
import com.rvprg.raft.configuration.Configuration;
import com.rvprg.raft.protocol.Log;
import com.rvprg.raft.protocol.Raft;
import com.rvprg.raft.protocol.RaftObserver;
import com.rvprg.raft.protocol.Role;
import com.rvprg.raft.protocol.messages.ProtocolMessages.AppendEntries;
import com.rvprg.raft.protocol.messages.ProtocolMessages.AppendEntriesResponse;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage.MessageType;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RequestVote;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RequestVoteResponse;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RequestVoteResponse.Builder;
import com.rvprg.raft.transport.MemberConnector;
import com.rvprg.raft.transport.MessageReceiver;

import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.ScheduledFuture;

public class RaftImpl implements Raft {
    private final Object thisLock = new Object();

    private Role role = Role.Follower;

    private final Configuration configuration;

    private final MemberConnector memberConnector;
    private final MessageReceiver messageReceiver;

    private final long heartbeatTimeout;

    private final int electionMinTimeout = 150;
    private final int electionMaxTimeout = 300;

    private final EventLoopGroup eventLoop = new NioEventLoopGroup();

    private final AtomicReference<ScheduledFuture<?>> electionTrigger = new AtomicReference<ScheduledFuture<?>>();
    private final AtomicReference<ScheduledFuture<?>> electionTimeoutTrigger = new AtomicReference<ScheduledFuture<?>>();

    private final AtomicReference<String> votedFor = new AtomicReference<>(null);

    private final RaftObserver observer;

    private final AtomicInteger currentTerm = new AtomicInteger(0);
    private final AtomicInteger votesReceived = new AtomicInteger(0);

    private final String selfId;

    private final Random random = new Random();

    private final Log log;

    @Inject
    public RaftImpl(Configuration configuration, MemberConnector memberConnector, MessageReceiver messageReceiver, Log log, RaftObserver observer) {
        this.heartbeatTimeout = configuration.getHeartbeatTimeout();
        this.memberConnector = memberConnector;
        this.messageReceiver = messageReceiver;
        this.configuration = configuration;
        this.observer = observer;
        this.selfId = messageReceiver.getId();
        this.log = log;
    }

    @Override
    public void start() throws InterruptedException {
        // FIXME: Check if started.
        messageReceiver.start(this);
        scheduleHeartbeatTask();
    }

    @Override
    public void shutdown() {
        cancelHeartbeatTask();
        cancelElectionTimeoutTask();
        this.messageReceiver.shutdown();
        this.eventLoop.shutdownGracefully().awaitUninterruptibly();
    }

    @Override
    public void consumeRequestVote(Channel senderChannel, RequestVote requestVote) {
        Builder response = RequestVoteResponse.newBuilder().setTerm(getCurrentTerm());
        String currVotedFor = votedFor.get();
        boolean grantVote = false;

        if (requestVote.getTerm() < getCurrentTerm()) {
            grantVote = false;
        } else if (currVotedFor == null) {
            grantVote = true;
        } else if (currVotedFor.equals(requestVote.getCandidateId())) {
            grantVote = checkCandidatesLogIsUpToDate(requestVote);
        }

        if (grantVote) {
            votedFor.set(requestVote.getCandidateId());
        }

        RaftMessage responseMessage = RaftMessage.newBuilder()
                .setType(MessageType.RequestVoteResponse)
                .setRequestVoteResponse(response.setVoteGranted(grantVote).build())
                .build();
        senderChannel.writeAndFlush(responseMessage);
    }

    private boolean checkCandidatesLogIsUpToDate(RequestVote requestVote) {
        if (log.getLast().getTerm() == requestVote.getLastLogTerm()) {
            return requestVote.getLastLogIndex() >= log.length();
        }
        return requestVote.getLastLogTerm() >= log.getLast().getTerm();
    }

    @Override
    public void consumeRequestVoteResponse(Channel senderChannel, RequestVoteResponse requestVoteResponse) {
        observer.voteReceived();

        if (requestVoteResponse.getTerm() == getCurrentTerm()) {
            votesReceived.incrementAndGet();
        }

        if (votesReceived.get() >= getMajority()) {
            cancelElectionTimeoutTask();
            changeRole(Role.Leader);
            observer.electionWon();
        }
    }

    @Override
    public void consumeAppendEntries(Channel senderChannel, AppendEntries appendEntries) {
        if (appendEntries.getTerm() >= getCurrentTerm() && getRole() != Role.Follower) {
            cancelElectionTimeoutTask();
            changeRole(Role.Follower);
        }

        if (appendEntries.getLogEntriesCount() == 0) {
            processHeartbeat(appendEntries);
        }
    }

    @Override
    public void consumeAppendEntriesResponse(Channel senderChannel, AppendEntriesResponse appendEntriesResponse) {
    }

    private ScheduledFuture<?> scheduleNextElectionTask() {
        observer.nextElectionScheduled();
        return this.eventLoop.schedule(() -> RaftImpl.this.heartbeatTimedout(), heartbeatTimeout, TimeUnit.MILLISECONDS);
    }

    private void cancelHeartbeatTask() {
        cancelTask(electionTrigger.get());<commit_msg>Revert to follower if received an RPC with a more recent term.<commit_after>package com.rvprg.raft.protocol.impl;

import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

import com.google.inject.Inject;
import com.rvprg.raft.configuration.Configuration;
import com.rvprg.raft.protocol.Log;
import com.rvprg.raft.protocol.Raft;
import com.rvprg.raft.protocol.RaftObserver;
import com.rvprg.raft.protocol.Role;
import com.rvprg.raft.protocol.messages.ProtocolMessages.AppendEntries;
import com.rvprg.raft.protocol.messages.ProtocolMessages.AppendEntriesResponse;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RaftMessage.MessageType;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RequestVote;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RequestVoteResponse;
import com.rvprg.raft.protocol.messages.ProtocolMessages.RequestVoteResponse.Builder;
import com.rvprg.raft.transport.MemberConnector;
import com.rvprg.raft.transport.MessageReceiver;

import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.ScheduledFuture;

public class RaftImpl implements Raft {
    private final Object thisLock = new Object();

    private Role role = Role.Follower;

    private final Configuration configuration;

    private final MemberConnector memberConnector;
    private final MessageReceiver messageReceiver;

    private final long heartbeatTimeout;

    private final int electionMinTimeout = 150;
    private final int electionMaxTimeout = 300;

    private final EventLoopGroup eventLoop = new NioEventLoopGroup();

    private final AtomicReference<ScheduledFuture<?>> electionTrigger = new AtomicReference<ScheduledFuture<?>>();
    private final AtomicReference<ScheduledFuture<?>> electionTimeoutTrigger = new AtomicReference<ScheduledFuture<?>>();

    private final AtomicReference<String> votedFor = new AtomicReference<>(null);

    private final RaftObserver observer;

    private final AtomicInteger currentTerm = new AtomicInteger(0);
    private final AtomicInteger votesReceived = new AtomicInteger(0);

    private final String selfId;

    private final Random random = new Random();

    private final Log log;

    @Inject
    public RaftImpl(Configuration configuration, MemberConnector memberConnector, MessageReceiver messageReceiver, Log log, RaftObserver observer) {
        this.heartbeatTimeout = configuration.getHeartbeatTimeout();
        this.memberConnector = memberConnector;
        this.messageReceiver = messageReceiver;
        this.configuration = configuration;
        this.observer = observer;
        this.selfId = messageReceiver.getId();
        this.log = log;
    }

    @Override
    public void start() throws InterruptedException {
        // FIXME: Check if started.
        messageReceiver.start(this);
        scheduleHeartbeatTask();
    }

    @Override
    public void shutdown() {
        cancelHeartbeatTask();
        cancelElectionTimeoutTask();
        this.messageReceiver.shutdown();
        this.eventLoop.shutdownGracefully().awaitUninterruptibly();
    }

    @Override
    public void consumeRequestVote(Channel senderChannel, RequestVote requestVote) {
        Builder response = RequestVoteResponse.newBuilder().setTerm(getCurrentTerm());
        String currVotedFor = votedFor.get();
        boolean grantVote = false;

        if (requestVote.getTerm() < getCurrentTerm()) {
            grantVote = false;
        } else if (currVotedFor == null) {
            grantVote = true;
        } else if (currVotedFor.equals(requestVote.getCandidateId())) {
            grantVote = checkCandidatesLogIsUpToDate(requestVote);
        }

        if (grantVote) {
            votedFor.set(requestVote.getCandidateId());
        }

        RaftMessage responseMessage = RaftMessage.newBuilder()
                .setType(MessageType.RequestVoteResponse)
                .setRequestVoteResponse(response.setVoteGranted(grantVote).build())
                .build();
        senderChannel.writeAndFlush(responseMessage);
    }

    private boolean checkCandidatesLogIsUpToDate(RequestVote requestVote) {
        if (log.getLast().getTerm() == requestVote.getLastLogTerm()) {
            return requestVote.getLastLogIndex() >= log.length();
        }
        return requestVote.getLastLogTerm() >= log.getLast().getTerm();
    }

    @Override
    public void consumeRequestVoteResponse(Channel senderChannel, RequestVoteResponse requestVoteResponse) {
        observer.voteReceived();

        boolean changedToFollower = becomeFollowerIfNewerTerm(requestVoteResponse.getTerm());
        if (changedToFollower) {
            return;
        }

        if (requestVoteResponse.getTerm() == getCurrentTerm()) {
            votesReceived.incrementAndGet();
        }

        if (votesReceived.get() >= getMajority()) {
            cancelElectionTimeoutTask();
            changeRole(Role.Leader);
            observer.electionWon();
        }
    }

    private boolean updateTerm(int term) {
        synchronized (thisLock) {
            if (term > getCurrentTerm()) {
                currentTerm.set(term);
                return true;
            }
        }
        return false;
    }

    private boolean becomeFollowerIfNewerTerm(int term) {
        if (updateTerm(term)) {
            if (getRole() != Role.Follower) {
                cancelElectionTimeoutTask();
                changeRole(Role.Follower);
                return true;
            }
        }
        return false;
    }

    @Override
    public void consumeAppendEntries(Channel senderChannel, AppendEntries appendEntries) {
        becomeFollowerIfNewerTerm(appendEntries.getTerm());

        if (appendEntries.getLogEntriesCount() == 0) {
            processHeartbeat(appendEntries);
        }
    }

    @Override
    public void consumeAppendEntriesResponse(Channel senderChannel, AppendEntriesResponse appendEntriesResponse) {
    }

    private ScheduledFuture<?> scheduleNextElectionTask() {
        observer.nextElectionScheduled();
        return this.eventLoop.schedule(() -> RaftImpl.this.heartbeatTimedout(), heartbeatTimeout, TimeUnit.MILLISECONDS);
    }

    private void cancelHeartbeatTask() {
        cancelTask(electionTrigger.get());
{'fraction_non_alphanumeric': 0.09185245237130117, 'fraction_numerical': 0.001459262261856506, 'mean_word_length': 3.435814455231931, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '133972', 'n_tokens_mistral': 3325, 'n_tokens_neox': 3222, 'n_words': 639}