* A leader instance starts up in * {@link org.eclipse.jgit.internal.ketch.KetchLeader.State#CANDIDATE} and tries * to begin a new term by sending an * {@link org.eclipse.jgit.internal.ketch.ElectionRound} to all replicas. Its * term starts if a majority of replicas have accepted this leader instance for * the term. *
* Once elected by a majority the instance enters * {@link org.eclipse.jgit.internal.ketch.KetchLeader.State#LEADER} and runs * proposals offered to {@link #queueProposal(Proposal)}. This continues until * the leader is timed out for inactivity, or is deposed by a competing leader * gaining its own majority. *
* Once timed out or deposed this {@code KetchLeader} instance should be * discarded, and a new instance takes over. *
* Each leader instance coordinates a group of * {@link org.eclipse.jgit.internal.ketch.KetchReplica}s. Replica instances are * owned by the leader instance and must be discarded when the leader is * discarded. *
* In Ketch all push requests are issued through the leader. The steps are as * follows (see {@link org.eclipse.jgit.internal.ketch.KetchPreReceive} for an * example): *
* The leader gains consensus by first pushing the needed objects and a * {@link org.eclipse.jgit.internal.storage.reftree.RefTree} representing the * desired target repository state to the {@code refs/txn/accepted} branch on * each of the replicas. Once a majority has succeeded, the leader commits the * state by either pushing the {@code refs/txn/accepted} value to * {@code refs/txn/committed} (for Ketch-aware replicas) or by pushing updates * to {@code refs/heads/master}, etc. for stock Git replicas. *
* Internally, the actual transport to replicas is performed on background * threads via the {@link org.eclipse.jgit.internal.ketch.KetchSystem}'s * executor service. For performance, the * {@link org.eclipse.jgit.internal.ketch.KetchLeader}, * {@link org.eclipse.jgit.internal.ketch.KetchReplica} and * {@link org.eclipse.jgit.internal.ketch.Proposal} objects share some state, * and may invoke each other's methods on different threads. This access is * protected by the leader's {@link #lock} object. Care must be taken to prevent * concurrent access by correctly obtaining the leader's lock. */ public abstract class KetchLeader { private static final Logger log = LoggerFactory.getLogger(KetchLeader.class); /** Current state of the leader instance. */ public enum State { /** Newly created instance trying to elect itself leader. */ CANDIDATE, /** Leader instance elected by a majority. */ LEADER, /** Instance has been deposed by another with a more recent term. */ DEPOSED, /** Leader has been gracefully shutdown, e.g. due to inactivity. */ SHUTDOWN; } private final KetchSystem system; /** Leader's knowledge of replicas for this repository. */ private KetchReplica[] voters; private KetchReplica[] followers; private LocalReplica self; /** * Lock protecting all data within this leader instance. *
* This lock extends into the {@link KetchReplica} instances used by the * leader. They share the same lock instance to simplify concurrency. */ final Lock lock; private State state = CANDIDATE; /** Term of this leader, once elected. */ private long term; /** * Pending proposals accepted into the queue in FIFO order. *
* These proposals were preflighted and do not contain any conflicts with
* each other and their expectations matched the leader's local view of the
* agreed upon {@code refs/txn/accepted} tree.
*/
private final List
* Must be deep-copied with {@link RefTree#copy()} if
* {@link #roundHoldsReferenceToRefTree} is {@code true}.
*/
private RefTree refTree;
/**
* If {@code true} {@link #refTree} must be duplicated before queuing the
* next proposal. The {@link #refTree} was passed into the constructor of a
* {@link ProposalRound}, and that external reference to the {@link RefTree}
* object is held by the proposal until it materializes the tree object in
* the object store. This field is set {@code true} when the proposal begins
* execution and set {@code false} once tree objects are persisted in the
* local repository's object store or {@link #refTree} is replaced with a
* copy to isolate it from any running rounds.
*
* If proposals arrive less frequently than the {@code RefTree} is written
* out to the repository the {@link #roundHoldsReferenceToRefTree} behavior
* avoids duplicating {@link #refTree}, reducing both time and memory used.
* However if proposals arrive more frequently {@link #refTree} must be
* duplicated to prevent newly queued proposals from corrupting the
* {@link #runningRound}.
*/
volatile boolean roundHoldsReferenceToRefTree;
/** End of the leader's log. */
private LogIndex headIndex;
/** Leader knows this (and all prior) states are committed. */
private LogIndex committedIndex;
/**
* Is the leader idle with no work pending? If {@code true} there is no work
* for the leader (normal state). This field is {@code false} when the
* leader thread is scheduled for execution, or while {@link #runningRound}
* defines a round in progress.
*/
private boolean idle;
/** Current round the leader is preparing and waiting for a vote on. */
private Round runningRound;
/**
* Construct a leader for a Ketch instance.
*
* @param system
* Ketch system configuration the leader must adhere to.
*/
protected KetchLeader(KetchSystem system) {
this.system = system;
this.lock = new ReentrantLock(true /* fair */);
this.queued = new ArrayList<>(4);
this.idle = true;
}
/** @return system configuration. */
KetchSystem getSystem() {
return system;
}
/**
* Configure the replicas used by this Ketch instance.
*
* Replicas should be configured once at creation before any proposals are
* executed. Once elections happen, reconfiguration is a complicated
* concept that is not currently supported.
*
* @param replicas
* members participating with the same repository.
*/
public void setReplicas(Collection
* The caller will close the repository.
*
* @return opened repository for use by the leader thread.
* @throws java.io.IOException
* cannot reopen the repository for the leader.
*/
protected abstract Repository openRepository() throws IOException;
/**
* Queue a reference update proposal for consensus.
*
* This method does not wait for consensus to be reached. The proposal is
* checked to look for risks of conflicts, and then submitted into the queue
* for distribution as soon as possible.
*
* Callers must use {@link org.eclipse.jgit.internal.ketch.Proposal#await()}
* to see if the proposal is done.
*
* @param proposal
* the proposed reference updates to queue for consideration.
* Once execution is complete the individual reference result
* fields will be populated with the outcome.
* @throws java.lang.InterruptedException
* current thread was interrupted. The proposal may have been
* aborted if it was not yet queued for execution.
* @throws java.io.IOException
* unrecoverable error preventing proposals from being attempted
* by this leader.
*/
public void queueProposal(Proposal proposal)
throws InterruptedException, IOException {
try {
lock.lockInterruptibly();
} catch (InterruptedException e) {
proposal.abort();
throw e;
}
try {
if (refTree == null) {
initialize();
for (Proposal p : queued) {
refTree.apply(p.getCommands());
}
} else if (roundHoldsReferenceToRefTree) {
refTree = refTree.copy();
roundHoldsReferenceToRefTree = false;
}
if (!refTree.apply(proposal.getCommands())) {
// A conflict exists so abort the proposal.
proposal.abort();
return;
}
queued.add(proposal);
proposal.notifyState(QUEUED);
if (idle) {
scheduleLeader();
}
} finally {
lock.unlock();
}
}
private void initialize() throws IOException {
try (Repository git = openRepository(); RevWalk rw = new RevWalk(git)) {
self.initialize(git);
ObjectId accepted = self.getTxnAccepted();
if (!ObjectId.zeroId().equals(accepted)) {
RevCommit c = rw.parseCommit(accepted);
headIndex = LogIndex.unknown(accepted);
refTree = RefTree.read(rw.getObjectReader(), c.getTree());
} else {
headIndex = LogIndex.unknown(ObjectId.zeroId());
refTree = RefTree.newEmptyTree();
}
}
}
private void scheduleLeader() {
idle = false;
system.getExecutor().execute(this::runLeader);
}
private void runLeader() {
Round round;
lock.lock();
try {
switch (state) {
case CANDIDATE:
round = new ElectionRound(this, headIndex);
break;
case LEADER:
round = newProposalRound();
break;
case DEPOSED:
case SHUTDOWN:
default:
log.warn("Leader cannot run {}", state); //$NON-NLS-1$
// TODO(sop): Redirect proposals.
return;
}
} finally {
lock.unlock();
}
try {
round.start();
} catch (IOException e) {
// TODO(sop) Depose leader if it cannot use its repository.
log.error(KetchText.get().leaderFailedToStore, e);
lock.lock();
try {
nextRound();
} finally {
lock.unlock();
}
}
}
private ProposalRound newProposalRound() {
List
* Must be called while {@link #lock} is held by the replica.
*
* @param replica
* replica posting a completion event.
*/
void onReplicaUpdate(KetchReplica replica) {
if (log.isDebugEnabled()) {
log.debug("Replica {} finished:\n{}", //$NON-NLS-1$
replica.describeForLog(), snapshot());
}
if (replica.getParticipation() == FOLLOWER_ONLY) {
// Followers cannot vote, so votes haven't changed.
return;
} else if (runningRound == null) {
// No round running, no need to tally votes.
return;
}
assert headIndex.equals(runningRound.acceptedNewIndex);
int matching = 0;
for (KetchReplica r : voters) {
if (r.hasAccepted(headIndex)) {
matching++;
}
}
int quorum = voters.length / 2 + 1;
boolean success = matching >= quorum;
if (!success) {
return;
}
switch (state) {
case CANDIDATE:
term = ((ElectionRound) runningRound).getTerm();
state = LEADER;
if (log.isDebugEnabled()) {
log.debug("Won election, running term " + term); //$NON-NLS-1$
}
//$FALL-THROUGH$
case LEADER:
committedIndex = headIndex;
if (log.isDebugEnabled()) {
log.debug("Committed {} in term {}", //$NON-NLS-1$
committedIndex.describeForLog(),
Long.valueOf(term));
}
nextRound();
commitAsync(replica);
notifySuccess(runningRound);
if (log.isDebugEnabled()) {
log.debug("Leader state:\n{}", snapshot()); //$NON-NLS-1$
}
break;
default:
log.debug("Leader ignoring replica while in {}", state); //$NON-NLS-1$
break;
}
}
private void notifySuccess(Round round) {
// Drop the leader lock while notifying Proposal listeners.
lock.unlock();
try {
round.success();
} finally {
lock.lock();
}
}
private void commitAsync(KetchReplica caller) {
for (KetchReplica r : voters) {
if (r == caller) {
continue;
}
if (r.shouldPushUnbatchedCommit(committedIndex, isIdle())) {
r.pushCommitAsync(committedIndex);
}
}
for (KetchReplica r : followers) {
if (r == caller) {
continue;
}
if (r.shouldPushUnbatchedCommit(committedIndex, isIdle())) {
r.pushCommitAsync(committedIndex);
}
}
}
/** Schedule the next round; invoked while {@link #lock} is held. */
void nextRound() {
runningRound = null;
if (queued.isEmpty()) {
idle = true;
} else {
// Caller holds lock. Reschedule leader on a new thread so
// the call stack can unwind and lock is not held unexpectedly
// during prepare for the next round.
scheduleLeader();
}
}
/**
* Snapshot this leader
*
* @return snapshot of this leader
*/
public LeaderSnapshot snapshot() {
lock.lock();
try {
LeaderSnapshot s = new LeaderSnapshot();
s.state = state;
s.term = term;
s.headIndex = headIndex;
s.committedIndex = committedIndex;
s.idle = isIdle();
for (KetchReplica r : voters) {
s.replicas.add(r.snapshot());
}
for (KetchReplica r : followers) {
s.replicas.add(r.snapshot());
}
return s;
} finally {
lock.unlock();
}
}
/**
* Gracefully shutdown this leader and cancel outstanding operations.
*/
public void shutdown() {
lock.lock();
try {
if (state != SHUTDOWN) {
state = SHUTDOWN;
for (KetchReplica r : voters) {
r.shutdown();
}
for (KetchReplica r : followers) {
r.shutdown();
}
}
} finally {
lock.unlock();
}
}
/** {@inheritDoc} */
@Override
public String toString() {
return snapshot().toString();
}
}