<commit_before>/**
 * Copyright 2015 <NAME>
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * 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 com.palantir.leader;

import java.util.AbstractMap;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.CheckForNull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Defaults;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.palantir.common.base.Throwables;
import com.palantir.paxos.BooleanPaxosResponse;
import com.palantir.paxos.PaxosAcceptor;
import com.palantir.paxos.PaxosLearner;
import com.palantir.paxos.PaxosProposer;
import com.palantir.paxos.PaxosQuorumChecker;
import com.palantir.paxos.PaxosResponse;
import com.palantir.paxos.PaxosRoundFailureException;
import com.palantir.paxos.PaxosUpdate;
import com.palantir.paxos.PaxosValue;

/**
 * Implementation of a paxos member than can be a designated proposer (leader) and designated
 * learner (informer).
 *
 * @author rullman
 */
public class PaxosLeaderElectionService implements PingableLeader, LeaderElectionService {
    private static final Logger log = LoggerFactory.getLogger(PaxosLeaderElectionService.class);

    public static final long DEFAULT_UPDATE_POLLING_WAIT_IN_MS = 1000;
    public static final long DEFAULT_RANDOM_WAIT_BEFORE_PROPOSING_LEADERSHIP_IN_MS = 2000;
    public static final long DEFAULT_LEADER_PING_RESPONSE_WAIT_IN_MS = 2000;

    private final ReentrantLock lock;

    final PaxosProposer proposer;
    final PaxosLearner knowledge;

    final List<PingableLeader> potentialLeaders;
    final ImmutableList<PaxosAcceptor> acceptors;
    final ImmutableList<PaxosLearner> learners;

    final long updatePollingRateInMs;
    final long randomWaitBeforeProposingLeadership;
    final long leaderPingResponseWaitMs;

    final Executor executor;

    final ConcurrentMap<String, PingableLeader> uuidToServiceCache = Maps.newConcurrentMap();

    public PaxosLeaderElectionService(PaxosProposer proposer,
                                      PaxosLearner knowledge,
                                      List<PingableLeader> potentialLeaders,
                                      List<PaxosAcceptor> acceptors,
                                      List<PaxosLearner> learners,
                                      Executor executor,
                                      long updatePollingWaitInMs,
                                      long randomWaitBeforeProposingLeadership,
                                      long leaderPingResponseWaitMs) {
        this.proposer = proposer;
        this.knowledge = knowledge;
        this.potentialLeaders = Lists.newArrayList(potentialLeaders);
        this.acceptors = ImmutableList.copyOf(acceptors);
        this.learners = ImmutableList.copyOf(learners);
        this.executor = executor;
        this.updatePollingRateInMs = updatePollingWaitInMs;
        this.randomWaitBeforeProposingLeadership = randomWaitBeforeProposingLeadership;
        this.leaderPingResponseWaitMs = leaderPingResponseWaitMs;
        lock = new ReentrantLock();
    }

    @VisibleForTesting
    void addPeer(PingableLeader peer) {
        for (PingableLeader leader : potentialLeaders) {
            if (leader.equals(peer)) {
                return;
            }
        }
        potentialLeaders.add(peer);
    }

    @VisibleForTesting
    void removePeer(PingableLeader peer) {
        for (PingableLeader leader : potentialLeaders) {
            if (leader.equals(peer)) {
                potentialLeaders.remove(peer);
                return;
            }
        }
    }

    @Override
    public LeadershipToken blockOnBecomingLeader() throws InterruptedException {
        for (;;) {
            PaxosValue greatestLearned = knowledge.getGreatestLearnedValue();
            LeadershipToken token = genTokenFromValue(greatestLearned);

            if (isLastConfirmedLeader(greatestLearned)) {
                StillLeadingStatus leadingStatus = isStillLeading(token);
                if (leadingStatus == StillLeadingStatus.LEADING) {
                    return token;
                } else if (leadingStatus == StillLeadingStatus.NO_QUORUM) {
                    // If we don't have quorum we should just retry our calls.
                    continue;
                }
            } else {
                // We are not the leader, so we should ping them to see if they are still up.
                if (pingLeader() || pingLeader()) {
                    Thread.sleep(updatePollingRateInMs);
                    continue;
                }
            }

            boolean learnedNewState = updateLearnedStateFromPeers(greatestLearned);
            if (learnedNewState) {
                continue;
            }

            long backoffTime = (long) (randomWaitBeforeProposingLeadership * Math.random());
            Thread.sleep(backoffTime);

            proposeLeadership(token);
        }
    }

    private LeadershipToken genTokenFromValue(@CheckForNull PaxosValue value) {
        return new PaxosLeadershipToken(value);
    }

    private boolean pingLeader() {
        final PingableLeader leader = getSuspectedLeader();
        if (leader == null) {
            return false;
        }

        CompletionService<Boolean> pingCompletionService = new ExecutorCompletionService<Boolean>(
                executor);

        // kick off all the requests
        pingCompletionService.submit(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return leader.ping();
            }
        });

        try {
            Future<Boolean> pingFuture = pingCompletionService.poll(
                    leaderPingResponseWaitMs,
                    TimeUnit.MILLISECONDS);
            return pingFuture != null && pingFuture.get();
        } catch (InterruptedException e) {
            return false;
        } catch (ExecutionException e) {
            log.warn("cannot ping leader", e);
            return false;
        }
    }

    private PingableLeader getSuspectedLeader() {<commit_msg>warn to info<commit_after>/**
 * Copyright 2015 <NAME>
 *
 * Licensed under the BSD-3 License (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://opensource.org/licenses/BSD-3-Clause
 *
 * 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 com.palantir.leader;

import java.util.AbstractMap;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

import javax.annotation.CheckForNull;
import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Defaults;
import com.google.common.base.Function;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableCollection;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.palantir.common.base.Throwables;
import com.palantir.paxos.BooleanPaxosResponse;
import com.palantir.paxos.PaxosAcceptor;
import com.palantir.paxos.PaxosLearner;
import com.palantir.paxos.PaxosProposer;
import com.palantir.paxos.PaxosQuorumChecker;
import com.palantir.paxos.PaxosResponse;
import com.palantir.paxos.PaxosRoundFailureException;
import com.palantir.paxos.PaxosUpdate;
import com.palantir.paxos.PaxosValue;

/**
 * Implementation of a paxos member than can be a designated proposer (leader) and designated
 * learner (informer).
 *
 * @author rullman
 */
public class PaxosLeaderElectionService implements PingableLeader, LeaderElectionService {
    private static final Logger log = LoggerFactory.getLogger(PaxosLeaderElectionService.class);

    public static final long DEFAULT_UPDATE_POLLING_WAIT_IN_MS = 1000;
    public static final long DEFAULT_RANDOM_WAIT_BEFORE_PROPOSING_LEADERSHIP_IN_MS = 2000;
    public static final long DEFAULT_LEADER_PING_RESPONSE_WAIT_IN_MS = 2000;

    private final ReentrantLock lock;

    final PaxosProposer proposer;
    final PaxosLearner knowledge;

    final List<PingableLeader> potentialLeaders;
    final ImmutableList<PaxosAcceptor> acceptors;
    final ImmutableList<PaxosLearner> learners;

    final long updatePollingRateInMs;
    final long randomWaitBeforeProposingLeadership;
    final long leaderPingResponseWaitMs;

    final Executor executor;

    final ConcurrentMap<String, PingableLeader> uuidToServiceCache = Maps.newConcurrentMap();

    public PaxosLeaderElectionService(PaxosProposer proposer,
                                      PaxosLearner knowledge,
                                      List<PingableLeader> potentialLeaders,
                                      List<PaxosAcceptor> acceptors,
                                      List<PaxosLearner> learners,
                                      Executor executor,
                                      long updatePollingWaitInMs,
                                      long randomWaitBeforeProposingLeadership,
                                      long leaderPingResponseWaitMs) {
        this.proposer = proposer;
        this.knowledge = knowledge;
        this.potentialLeaders = Lists.newArrayList(potentialLeaders);
        this.acceptors = ImmutableList.copyOf(acceptors);
        this.learners = ImmutableList.copyOf(learners);
        this.executor = executor;
        this.updatePollingRateInMs = updatePollingWaitInMs;
        this.randomWaitBeforeProposingLeadership = randomWaitBeforeProposingLeadership;
        this.leaderPingResponseWaitMs = leaderPingResponseWaitMs;
        lock = new ReentrantLock();
    }

    @VisibleForTesting
    void addPeer(PingableLeader peer) {
        for (PingableLeader leader : potentialLeaders) {
            if (leader.equals(peer)) {
                return;
            }
        }
        potentialLeaders.add(peer);
    }

    @VisibleForTesting
    void removePeer(PingableLeader peer) {
        for (PingableLeader leader : potentialLeaders) {
            if (leader.equals(peer)) {
                potentialLeaders.remove(peer);
                return;
            }
        }
    }

    @Override
    public LeadershipToken blockOnBecomingLeader() throws InterruptedException {
        for (;;) {
            PaxosValue greatestLearned = knowledge.getGreatestLearnedValue();
            LeadershipToken token = genTokenFromValue(greatestLearned);

            if (isLastConfirmedLeader(greatestLearned)) {
                StillLeadingStatus leadingStatus = isStillLeading(token);
                if (leadingStatus == StillLeadingStatus.LEADING) {
                    return token;
                } else if (leadingStatus == StillLeadingStatus.NO_QUORUM) {
                    // If we don't have quorum we should just retry our calls.
                    continue;
                }
            } else {
                // We are not the leader, so we should ping them to see if they are still up.
                if (pingLeader() || pingLeader()) {
                    Thread.sleep(updatePollingRateInMs);
                    continue;
                }
            }

            boolean learnedNewState = updateLearnedStateFromPeers(greatestLearned);
            if (learnedNewState) {
                continue;
            }

            long backoffTime = (long) (randomWaitBeforeProposingLeadership * Math.random());
            Thread.sleep(backoffTime);

            proposeLeadership(token);
        }
    }

    private LeadershipToken genTokenFromValue(@CheckForNull PaxosValue value) {
        return new PaxosLeadershipToken(value);
    }

    private boolean pingLeader() {
        final PingableLeader leader = getSuspectedLeader();
        if (leader == null) {
            return false;
        }

        CompletionService<Boolean> pingCompletionService = new ExecutorCompletionService<Boolean>(
                executor);

        // kick off all the requests
        pingCompletionService.submit(new Callable<Boolean>() {
            @Override
            public Boolean call() throws Exception {
                return leader.ping();
            }
        });

        try {
            Future<Boolean> pingFuture = pingCompletionService.poll(
                    leaderPingResponseWaitMs,
                    TimeUnit.MILLISECONDS);
            return pingFuture != null && pingFuture.get();
        } catch (InterruptedException e) {
            return false;
        } catch (ExecutionException e) {
            log.info("cannot ping leader", e);
            return false;
        }
    }

    private PingableLeader getSuspectedLeader() {
{'fraction_non_alphanumeric': 0.07330482590103848, 'fraction_numerical': 0.0027149935518903144, 'mean_word_length': 2.577081815974751, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3637787', 'n_tokens_mistral': 4113, 'n_tokens_neox': 3886, 'n_words': 890}