<commit_before>package com.robertpreeves.agreed.paxos;

import com.google.gson.Gson;
import com.robertpreeves.agreed.NoConsensusException;
import com.robertpreeves.agreed.paxos.messages.Accept;
import com.robertpreeves.agreed.paxos.messages.Accepted;
import com.robertpreeves.agreed.paxos.messages.PaxosResponseException;
import com.robertpreeves.agreed.paxos.messages.Prepare;
import com.robertpreeves.agreed.paxos.messages.Promise;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.HttpClients;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class PaxosAcceptorsProxy<T> implements PaxosAcceptor<T>, AutoCloseable {
    private static final Logger LOGGER = LogManager.getLogger("[PROPOSER]");
    private static final int TIMEOUT = 180000; //high for demo purposes
    private static final Gson GSON = new Gson();
    private final LocalPaxosAcceptor<T> localAcceptor;
    private final Set<String> otherNodes;
    private final int majorityCount;
    private final ExecutorService executor = Executors.newCachedThreadPool();
    private final boolean slow;

    public PaxosAcceptorsProxy(
            LocalPaxosAcceptor<T> localAcceptor,
            Set<String> otherNodes,
            boolean slow) {
        this.localAcceptor = localAcceptor;
        this.otherNodes = otherNodes;
        this.slow = slow;

        //the total number of nodes must be odd so otherNodes will be even.
        //the majority is half of the otherNodes + 1.
        majorityCount = (otherNodes.size() / 2) + 1;
    }

    /**
     * Gets the most recently promised sequence number
     * @return
     */
    public long getSequenceNumber() {
        return localAcceptor.getSequenceNumber();
    }

    @Override
    public Promise<T> prepare(Prepare prepare) {
        //Send prepare messages to remote acceptors
        List<Future<Promise>> promises = makeRequests(Uris.PREPARE, prepare, Promise.class);

        //Send prepare message to local acceptor
        Future<Promise> localPromise = executor.submit(() -> localAcceptor.prepare(prepare));
        promises.add(localPromise);

        Promise reducedPromise = reducePromises(promises);
        LOGGER.info("Prepare: {}, Reduced Promise: {}", prepare, reducedPromise);
        return reducedPromise;
    }

    private Promise<T> reducePromises(List<Future<Promise>> promises) {
        //Get promises
        int promiseCount = 0;
        Accept<T> previouslyAccepted = null;
        for (Future<Promise> promiseFuture :
                promises) {
            Promise<T> promise;
            try {
                promise = promiseFuture.get(TIMEOUT, TimeUnit.SECONDS);
            } catch (Exception e) {
                promise = new Promise<>(false, null);
            }

            if (promise.promised) {
                ++promiseCount;
            }

            //Check if any acceptors have accepted a previous value.
            //Choose the value with the highest sequence number.
            if (promise.acceptedValue != null) {
                if (previouslyAccepted == null) {
                    previouslyAccepted = promise.acceptedValue;
                } else {
                    int sequenceCompare = Long.compareUnsigned(
                            promise.acceptedValue.sequenceNumber,
                            previouslyAccepted.sequenceNumber);

                    if (sequenceCompare > 0) {
                        previouslyAccepted = promise.acceptedValue;
                    }
                }
            }
        }

        //Verify a quorum has promised
        Promise<T> promiseAggregate;
        if (promiseCount >= majorityCount) {
            promiseAggregate = new Promise<>(true, previouslyAccepted);
        } else {
            promiseAggregate = new Promise<>(false, null);
        }

        return promiseAggregate;
    }

    @Override
    public Accepted accept(Accept<T> accept) {
        //Send accept messages to remote acceptors
        List<Future<Accepted>> accepteds = makeRequests(Uris.ACCEPT, accept, Accepted.class);

        //Send accept message to local acceptor
        Future<Accepted> localAccepted = executor.submit(() -> localAcceptor.accept(accept));
        accepteds.add(localAccepted);

        Accepted reducedAccepteds = reduceAccepteds(accept.sequenceNumber, accepteds);
        LOGGER.info("Accept: {}, Reduced Accepteds: {}", accept, reducedAccepteds);
        return reducedAccepteds;
    }

    private Accepted reduceAccepteds(long sequenceNumber, List<Future<Accepted>> accepteds) {
        int acceptedCount = 0;

        long highestSequenceNumber = -1;
        for (Future<Accepted> acceptedFuture :
                accepteds) {
            Accepted accepted = null;
            try {
                accepted = acceptedFuture.get(TIMEOUT, TimeUnit.SECONDS);
            } catch (Exception e) {
            }

            if (accepted != null) {
                if (Long.compareUnsigned(accepted.sequenceNumber, sequenceNumber) == 0) {
                    ++acceptedCount;<commit_msg>bug with default accept value. cant be negative since it is unsigned long<commit_after>package com.robertpreeves.agreed.paxos;

import com.google.gson.Gson;
import com.robertpreeves.agreed.NoConsensusException;
import com.robertpreeves.agreed.paxos.messages.Accept;
import com.robertpreeves.agreed.paxos.messages.Accepted;
import com.robertpreeves.agreed.paxos.messages.PaxosResponseException;
import com.robertpreeves.agreed.paxos.messages.Prepare;
import com.robertpreeves.agreed.paxos.messages.Promise;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.HttpClients;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;
import java.util.Scanner;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public class PaxosAcceptorsProxy<T> implements PaxosAcceptor<T>, AutoCloseable {
    private static final Logger LOGGER = LogManager.getLogger("[PROPOSER]");
    private static final int TIMEOUT = 180000; //high for demo purposes
    private static final Gson GSON = new Gson();
    private final LocalPaxosAcceptor<T> localAcceptor;
    private final Set<String> otherNodes;
    private final int majorityCount;
    private final ExecutorService executor = Executors.newCachedThreadPool();
    private final boolean slow;

    public PaxosAcceptorsProxy(
            LocalPaxosAcceptor<T> localAcceptor,
            Set<String> otherNodes,
            boolean slow) {
        this.localAcceptor = localAcceptor;
        this.otherNodes = otherNodes;
        this.slow = slow;

        //the total number of nodes must be odd so otherNodes will be even.
        //the majority is half of the otherNodes + 1.
        majorityCount = (otherNodes.size() / 2) + 1;
    }

    /**
     * Gets the most recently promised sequence number
     * @return
     */
    public long getSequenceNumber() {
        return localAcceptor.getSequenceNumber();
    }

    @Override
    public Promise<T> prepare(Prepare prepare) {
        //Send prepare messages to remote acceptors
        List<Future<Promise>> promises = makeRequests(Uris.PREPARE, prepare, Promise.class);

        //Send prepare message to local acceptor
        Future<Promise> localPromise = executor.submit(() -> localAcceptor.prepare(prepare));
        promises.add(localPromise);

        Promise reducedPromise = reducePromises(promises);
        LOGGER.info("Prepare: {}, Reduced Promise: {}", prepare, reducedPromise);
        return reducedPromise;
    }

    private Promise<T> reducePromises(List<Future<Promise>> promises) {
        //Get promises
        int promiseCount = 0;
        Accept<T> previouslyAccepted = null;
        for (Future<Promise> promiseFuture :
                promises) {
            Promise<T> promise;
            try {
                promise = promiseFuture.get(TIMEOUT, TimeUnit.SECONDS);
            } catch (Exception e) {
                promise = new Promise<>(false, null);
            }

            if (promise.promised) {
                ++promiseCount;
            }

            //Check if any acceptors have accepted a previous value.
            //Choose the value with the highest sequence number.
            if (promise.acceptedValue != null) {
                if (previouslyAccepted == null) {
                    previouslyAccepted = promise.acceptedValue;
                } else {
                    int sequenceCompare = Long.compareUnsigned(
                            promise.acceptedValue.sequenceNumber,
                            previouslyAccepted.sequenceNumber);

                    if (sequenceCompare > 0) {
                        previouslyAccepted = promise.acceptedValue;
                    }
                }
            }
        }

        //Verify a quorum has promised
        Promise<T> promiseAggregate;
        if (promiseCount >= majorityCount) {
            promiseAggregate = new Promise<>(true, previouslyAccepted);
        } else {
            promiseAggregate = new Promise<>(false, null);
        }

        return promiseAggregate;
    }

    @Override
    public Accepted accept(Accept<T> accept) {
        //Send accept messages to remote acceptors
        List<Future<Accepted>> accepteds = makeRequests(Uris.ACCEPT, accept, Accepted.class);

        //Send accept message to local acceptor
        Future<Accepted> localAccepted = executor.submit(() -> localAcceptor.accept(accept));
        accepteds.add(localAccepted);

        Accepted reducedAccepteds = reduceAccepteds(accept.sequenceNumber, accepteds);
        LOGGER.info("Accept: {}, Reduced Accepteds: {}", accept, reducedAccepteds);
        return reducedAccepteds;
    }

    private Accepted reduceAccepteds(long sequenceNumber, List<Future<Accepted>> accepteds) {
        int acceptedCount = 0;

        long highestSequenceNumber = 0;
        for (Future<Accepted> acceptedFuture :
                accepteds) {
            Accepted accepted = null;
            try {
                accepted = acceptedFuture.get(TIMEOUT, TimeUnit.SECONDS);
            } catch (Exception e) {
            }

            if (accepted != null) {
                if (Long.compareUnsigned(accepted.sequenceNumber, sequenceNumber) == 0) {
                    ++acceptedCount;
{'fraction_non_alphanumeric': 0.09036197121674662, 'fraction_numerical': 0.0027911033580462277, 'mean_word_length': 2.5487465181058497, 'pattern_counts': {'":': 0, '<': 57, '<?xml version=': 0, '>': 65, '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': '536886', 'n_tokens_mistral': 3090, 'n_tokens_neox': 3047, 'n_words': 727}