package com.orbekk.paxos; import java.util.List; import java.util.concurrent.Callable; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; import java.util.concurrent.FutureTask; import java.util.concurrent.atomic.AtomicInteger; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.google.protobuf.RpcCallback; import com.orbekk.protobuf.Rpc; import com.orbekk.same.ConnectionManager; import com.orbekk.same.Services; import com.orbekk.same.Services.ClientState; import com.orbekk.same.Services.PaxosRequest; import com.orbekk.same.Services.PaxosResponse; public class MasterProposer extends Thread { private final Logger logger = LoggerFactory.getLogger(getClass()); private final ClientState client; private final List paxosLocations; private final ConnectionManager connections; public MasterProposer(ClientState client, List paxosLocations, ConnectionManager connections) { this.client = client; this.paxosLocations = paxosLocations; this.connections = connections; } private class ResponseHandler implements RpcCallback { final int proposalNumber; final int numRequests; AtomicInteger bestPromise = new AtomicInteger(); AtomicInteger numPromises = new AtomicInteger(0); AtomicInteger numResponses = new AtomicInteger(0); AtomicInteger result = new AtomicInteger(); CountDownLatch done = new CountDownLatch(1); public ResponseHandler(int proposalNumber, int numRequests) { this.proposalNumber = proposalNumber; this.numRequests = numRequests; bestPromise.set(-proposalNumber); } @Override public void run(PaxosResponse response) { numResponses.incrementAndGet(); if (response != null) { int result = response.getResult(); if (result == proposalNumber) { numPromises.incrementAndGet(); } while (true) { int oldVal = bestPromise.get(); int update = Math.min(oldVal, result); if (bestPromise.compareAndSet(oldVal, update)) { break; } } } checkDone(); } private void checkDone() { if (done.getCount() > 0) { if (numPromises.get() >= numRequests / 2) { result.set(proposalNumber); done.countDown(); } else if (numResponses.get() >= numRequests) { result.set(bestPromise.get()); done.countDown(); } } } public int getResult() throws InterruptedException { done.await(); return result.get(); } } private int internalPropose(int proposalNumber) throws InterruptedException { ResponseHandler handler = new ResponseHandler(proposalNumber, paxosLocations.size()); for (String location : paxosLocations) { Rpc rpc = new Rpc(); Services.Paxos paxos = connections.getPaxos0(location); if (paxos == null) { handler.run(null); continue; } PaxosRequest request = PaxosRequest.newBuilder() .setClient(client) .setProposalNumber(proposalNumber) .build(); paxos.propose(rpc, request, handler); } return handler.getResult(); } private int internalAcceptRequest(int proposalNumber) throws InterruptedException { ResponseHandler handler = new ResponseHandler(proposalNumber, paxosLocations.size()); for (String location : paxosLocations) { Rpc rpc = new Rpc(); Services.Paxos paxos = connections.getPaxos0(location); PaxosRequest request = PaxosRequest.newBuilder() .setClient(client) .setProposalNumber(proposalNumber) .build(); paxos.acceptRequest(rpc, request, handler); } return handler.getResult(); } boolean propose(int proposalNumber) throws InterruptedException { int result = internalPropose(proposalNumber); if (result == proposalNumber) { result = internalAcceptRequest(proposalNumber); } if (result == proposalNumber) { return true; } else { return false; } } boolean proposeRetry(int proposalNumber) throws InterruptedException { return proposeRetry(proposalNumber, null) != null; } Integer proposeRetry(int proposalNumber, Runnable retryAction) throws InterruptedException { assert proposalNumber > 0; int nextProposal = proposalNumber; int result = nextProposal - 1; while (!Thread.currentThread().isInterrupted() && result != nextProposal) { result = internalPropose(nextProposal); if (result == nextProposal) { result = internalAcceptRequest(nextProposal); } logger.info("Proposed value {}, result {}", nextProposal, result); if (result < 0) { nextProposal = -result + 1; if (retryAction != null) { retryAction.run(); } } } if (Thread.interrupted()) { throw new InterruptedException(); } return result; } public Future startProposalTask(final int proposalNumber, final Runnable retryAction) { Callable proposalCallable = new Callable() { @Override public Integer call() throws Exception { return proposeRetry(proposalNumber, retryAction); } }; FutureTask task = new FutureTask(proposalCallable); new Thread(task).start(); return task; } }