You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@zookeeper.apache.org by an...@apache.org on 2018/10/10 10:30:33 UTC
[34/51] [partial] zookeeper git commit: ZOOKEEPER-3032: MAVEN
MIGRATION - branch 3.4 move java server, client
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/main/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java b/src/java/main/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java
deleted file mode 100644
index d32a725..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/AuthFastLeaderElection.java
+++ /dev/null
@@ -1,980 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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 org.apache.zookeeper.server.quorum;
-
-import java.io.IOException;
-import java.net.DatagramPacket;
-import java.net.DatagramSocket;
-import java.net.InetSocketAddress;
-import java.net.SocketException;
-import java.nio.ByteBuffer;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.Semaphore;
-
-import java.util.concurrent.TimeUnit;
-import java.util.Random;
-
-import org.apache.zookeeper.common.Time;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.apache.zookeeper.jmx.MBeanRegistry;
-import org.apache.zookeeper.server.ZooKeeperThread;
-import org.apache.zookeeper.server.quorum.Election;
-import org.apache.zookeeper.server.quorum.Vote;
-import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
-import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
-
-/**
- * @deprecated This class has been deprecated as of release 3.4.0.
- */
-@Deprecated
-public class AuthFastLeaderElection implements Election {
- private static final Logger LOG = LoggerFactory.getLogger(AuthFastLeaderElection.class);
-
- /* Sequence numbers for messages */
- static int sequencer = 0;
- static int maxTag = 0;
-
- /*
- * Determine how much time a process has to wait once it believes that it
- * has reached the end of leader election.
- */
- static int finalizeWait = 100;
-
- /*
- * Challenge counter to avoid replay attacks
- */
-
- static int challengeCounter = 0;
-
- /*
- * Flag to determine whether to authenticate or not
- */
-
- private boolean authEnabled = false;
-
- static public class Notification {
- /*
- * Proposed leader
- */
- long leader;
-
- /*
- * zxid of the proposed leader
- */
- long zxid;
-
- /*
- * Epoch
- */
- long epoch;
-
- /*
- * current state of sender
- */
- QuorumPeer.ServerState state;
-
- /*
- * Address of the sender
- */
- InetSocketAddress addr;
- }
-
- /*
- * Messages to send, both Notifications and Acks
- */
- static public class ToSend {
- static enum mType {
- crequest, challenge, notification, ack
- }
-
- ToSend(mType type, long tag, long leader, long zxid, long epoch,
- ServerState state, InetSocketAddress addr) {
-
- switch (type) {
- case crequest:
- this.type = 0;
- this.tag = tag;
- this.leader = leader;
- this.zxid = zxid;
- this.epoch = epoch;
- this.state = state;
- this.addr = addr;
-
- break;
- case challenge:
- this.type = 1;
- this.tag = tag;
- this.leader = leader;
- this.zxid = zxid;
- this.epoch = epoch;
- this.state = state;
- this.addr = addr;
-
- break;
- case notification:
- this.type = 2;
- this.leader = leader;
- this.zxid = zxid;
- this.epoch = epoch;
- this.state = QuorumPeer.ServerState.LOOKING;
- this.tag = tag;
- this.addr = addr;
-
- break;
- case ack:
- this.type = 3;
- this.tag = tag;
- this.leader = leader;
- this.zxid = zxid;
- this.epoch = epoch;
- this.state = state;
- this.addr = addr;
-
- break;
- default:
- break;
- }
- }
-
- /*
- * Message type: 0 notification, 1 acknowledgement
- */
- int type;
-
- /*
- * Proposed leader in the case of notification
- */
- long leader;
-
- /*
- * id contains the tag for acks, and zxid for notifications
- */
- long zxid;
-
- /*
- * Epoch
- */
- long epoch;
-
- /*
- * Current state;
- */
- QuorumPeer.ServerState state;
-
- /*
- * Message tag
- */
- long tag;
-
- InetSocketAddress addr;
- }
-
- LinkedBlockingQueue<ToSend> sendqueue;
-
- LinkedBlockingQueue<Notification> recvqueue;
-
- private class Messenger {
-
- final DatagramSocket mySocket;
- long lastProposedLeader;
- long lastProposedZxid;
- long lastEpoch;
- final Set<Long> ackset;
- final ConcurrentHashMap<Long, Long> challengeMap;
- final ConcurrentHashMap<Long, Semaphore> challengeMutex;
- final ConcurrentHashMap<Long, Semaphore> ackMutex;
- final ConcurrentHashMap<InetSocketAddress, ConcurrentHashMap<Long, Long>> addrChallengeMap;
-
- class WorkerReceiver implements Runnable {
-
- DatagramSocket mySocket;
- Messenger myMsg;
-
- WorkerReceiver(DatagramSocket s, Messenger msg) {
- mySocket = s;
- myMsg = msg;
- }
-
- boolean saveChallenge(long tag, long challenge) {
- Semaphore s = challengeMutex.get(tag);
- if (s != null) {
- synchronized (Messenger.this) {
- challengeMap.put(tag, challenge);
- challengeMutex.remove(tag);
- }
-
-
- s.release();
- } else {
- LOG.error("No challenge mutex object");
- }
-
-
- return true;
- }
-
- public void run() {
- byte responseBytes[] = new byte[48];
- ByteBuffer responseBuffer = ByteBuffer.wrap(responseBytes);
- DatagramPacket responsePacket = new DatagramPacket(
- responseBytes, responseBytes.length);
- while (true) {
- // Sleeps on receive
- try {
- responseBuffer.clear();
- mySocket.receive(responsePacket);
- } catch (IOException e) {
- LOG.warn("Ignoring exception receiving", e);
- }
- // Receive new message
- if (responsePacket.getLength() != responseBytes.length) {
- LOG.warn("Got a short response: "
- + responsePacket.getLength() + " "
- + responsePacket.toString());
- continue;
- }
- responseBuffer.clear();
- int type = responseBuffer.getInt();
- if ((type > 3) || (type < 0)) {
- LOG.warn("Got bad Msg type: " + type);
- continue;
- }
- long tag = responseBuffer.getLong();
-
- QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
- switch (responseBuffer.getInt()) {
- case 0:
- ackstate = QuorumPeer.ServerState.LOOKING;
- break;
- case 1:
- ackstate = QuorumPeer.ServerState.LEADING;
- break;
- case 2:
- ackstate = QuorumPeer.ServerState.FOLLOWING;
- break;
- }
-
- Vote current = self.getCurrentVote();
-
- switch (type) {
- case 0:
- // Receive challenge request
- ToSend c = new ToSend(ToSend.mType.challenge, tag,
- current.getId(), current.getZxid(),
- logicalclock, self.getPeerState(),
- (InetSocketAddress) responsePacket
- .getSocketAddress());
- sendqueue.offer(c);
- break;
- case 1:
- // Receive challenge and store somewhere else
- long challenge = responseBuffer.getLong();
- saveChallenge(tag, challenge);
-
- break;
- case 2:
- Notification n = new Notification();
- n.leader = responseBuffer.getLong();
- n.zxid = responseBuffer.getLong();
- n.epoch = responseBuffer.getLong();
- n.state = ackstate;
- n.addr = (InetSocketAddress) responsePacket
- .getSocketAddress();
-
- if ((myMsg.lastEpoch <= n.epoch)
- && ((n.zxid > myMsg.lastProposedZxid)
- || ((n.zxid == myMsg.lastProposedZxid)
- && (n.leader > myMsg.lastProposedLeader)))) {
- myMsg.lastProposedZxid = n.zxid;
- myMsg.lastProposedLeader = n.leader;
- myMsg.lastEpoch = n.epoch;
- }
-
- long recChallenge;
- InetSocketAddress addr = (InetSocketAddress) responsePacket
- .getSocketAddress();
- if (authEnabled) {
- ConcurrentHashMap<Long, Long> tmpMap = addrChallengeMap.get(addr);
- if(tmpMap != null){
- if (tmpMap.get(tag) != null) {
- recChallenge = responseBuffer.getLong();
-
- if (tmpMap.get(tag) == recChallenge) {
- recvqueue.offer(n);
-
- ToSend a = new ToSend(ToSend.mType.ack,
- tag, current.getId(),
- current.getZxid(),
- logicalclock, self.getPeerState(),
- addr);
-
- sendqueue.offer(a);
- } else {
- LOG.warn("Incorrect challenge: "
- + recChallenge + ", "
- + addrChallengeMap.toString());
- }
- } else {
- LOG.warn("No challenge for host: " + addr
- + " " + tag);
- }
- }
- } else {
- recvqueue.offer(n);
-
- ToSend a = new ToSend(ToSend.mType.ack, tag,
- current.getId(), current.getZxid(),
- logicalclock, self.getPeerState(),
- (InetSocketAddress) responsePacket
- .getSocketAddress());
-
- sendqueue.offer(a);
- }
- break;
-
- // Upon reception of an ack message, remove it from the
- // queue
- case 3:
- Semaphore s = ackMutex.get(tag);
-
- if(s != null)
- s.release();
- else LOG.error("Empty ack semaphore");
-
- ackset.add(tag);
-
- if (authEnabled) {
- ConcurrentHashMap<Long, Long> tmpMap = addrChallengeMap.get(responsePacket
- .getSocketAddress());
- if(tmpMap != null) {
- tmpMap.remove(tag);
- } else {
- LOG.warn("No such address in the ensemble configuration " + responsePacket
- .getSocketAddress());
- }
- }
-
- if (ackstate != QuorumPeer.ServerState.LOOKING) {
- Notification outofsync = new Notification();
- outofsync.leader = responseBuffer.getLong();
- outofsync.zxid = responseBuffer.getLong();
- outofsync.epoch = responseBuffer.getLong();
- outofsync.state = ackstate;
- outofsync.addr = (InetSocketAddress) responsePacket
- .getSocketAddress();
-
- recvqueue.offer(outofsync);
- }
-
- break;
- // Default case
- default:
- LOG.warn("Received message of incorrect type " + type);
- break;
- }
- }
- }
- }
-
- class WorkerSender implements Runnable {
-
- Random rand;
- int maxAttempts;
- int ackWait = finalizeWait;
-
- /*
- * Receives a socket and max number of attempts as input
- */
-
- WorkerSender(int attempts) {
- maxAttempts = attempts;
- rand = new Random(java.lang.Thread.currentThread().getId()
- + Time.currentElapsedTime());
- }
-
- long genChallenge() {
- byte buf[] = new byte[8];
-
- buf[0] = (byte) ((challengeCounter & 0xff000000) >>> 24);
- buf[1] = (byte) ((challengeCounter & 0x00ff0000) >>> 16);
- buf[2] = (byte) ((challengeCounter & 0x0000ff00) >>> 8);
- buf[3] = (byte) ((challengeCounter & 0x000000ff));
-
- challengeCounter++;
- int secret = rand.nextInt(java.lang.Integer.MAX_VALUE);
-
- buf[4] = (byte) ((secret & 0xff000000) >>> 24);
- buf[5] = (byte) ((secret & 0x00ff0000) >>> 16);
- buf[6] = (byte) ((secret & 0x0000ff00) >>> 8);
- buf[7] = (byte) ((secret & 0x000000ff));
-
- return (((long)(buf[0] & 0xFF)) << 56)
- + (((long)(buf[1] & 0xFF)) << 48)
- + (((long)(buf[2] & 0xFF)) << 40)
- + (((long)(buf[3] & 0xFF)) << 32)
- + (((long)(buf[4] & 0xFF)) << 24)
- + (((long)(buf[5] & 0xFF)) << 16)
- + (((long)(buf[6] & 0xFF)) << 8)
- + ((long)(buf[7] & 0xFF));
- }
-
- public void run() {
- while (true) {
- try {
- ToSend m = sendqueue.take();
- process(m);
- } catch (InterruptedException e) {
- break;
- }
-
- }
- }
-
- private void process(ToSend m) {
- int attempts = 0;
- byte zeroes[];
- byte requestBytes[] = new byte[48];
- DatagramPacket requestPacket = new DatagramPacket(requestBytes,
- requestBytes.length);
- ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
-
- switch (m.type) {
- case 0:
- /*
- * Building challenge request packet to send
- */
- requestBuffer.clear();
- requestBuffer.putInt(ToSend.mType.crequest.ordinal());
- requestBuffer.putLong(m.tag);
- requestBuffer.putInt(m.state.ordinal());
- zeroes = new byte[32];
- requestBuffer.put(zeroes);
-
- requestPacket.setLength(48);
- try {
- requestPacket.setSocketAddress(m.addr);
- } catch (IllegalArgumentException e) {
- // Sun doesn't include the address that causes this
- // exception to be thrown, so we wrap the exception
- // in order to capture this critical detail.
- throw new IllegalArgumentException(
- "Unable to set socket address on packet, msg:"
- + e.getMessage() + " with addr:" + m.addr,
- e);
- }
-
- try {
- if (challengeMap.get(m.tag) == null) {
- mySocket.send(requestPacket);
- }
- } catch (IOException e) {
- LOG.warn("Exception while sending challenge: ", e);
- }
-
- break;
- case 1:
- /*
- * Building challenge packet to send
- */
-
- long newChallenge;
- ConcurrentHashMap<Long, Long> tmpMap = addrChallengeMap.get(m.addr);
- if(tmpMap != null){
- Long tmpLong = tmpMap.get(m.tag);
- if (tmpLong != null) {
- newChallenge = tmpLong;
- } else {
- newChallenge = genChallenge();
- }
-
- tmpMap.put(m.tag, newChallenge);
-
- requestBuffer.clear();
- requestBuffer.putInt(ToSend.mType.challenge.ordinal());
- requestBuffer.putLong(m.tag);
- requestBuffer.putInt(m.state.ordinal());
- requestBuffer.putLong(newChallenge);
- zeroes = new byte[24];
- requestBuffer.put(zeroes);
-
- requestPacket.setLength(48);
- try {
- requestPacket.setSocketAddress(m.addr);
- } catch (IllegalArgumentException e) {
- // Sun doesn't include the address that causes this
- // exception to be thrown, so we wrap the exception
- // in order to capture this critical detail.
- throw new IllegalArgumentException(
- "Unable to set socket address on packet, msg:"
- + e.getMessage() + " with addr:" + m.addr,
- e);
- }
-
-
- try {
- mySocket.send(requestPacket);
- } catch (IOException e) {
- LOG.warn("Exception while sending challenge: ", e);
- }
- } else {
- LOG.error("Address is not in the configuration: " + m.addr);
- }
-
- break;
- case 2:
-
- /*
- * Building notification packet to send
- */
-
- requestBuffer.clear();
- requestBuffer.putInt(m.type);
- requestBuffer.putLong(m.tag);
- requestBuffer.putInt(m.state.ordinal());
- requestBuffer.putLong(m.leader);
- requestBuffer.putLong(m.zxid);
- requestBuffer.putLong(m.epoch);
- zeroes = new byte[8];
- requestBuffer.put(zeroes);
-
- requestPacket.setLength(48);
- try {
- requestPacket.setSocketAddress(m.addr);
- } catch (IllegalArgumentException e) {
- // Sun doesn't include the address that causes this
- // exception to be thrown, so we wrap the exception
- // in order to capture this critical detail.
- throw new IllegalArgumentException(
- "Unable to set socket address on packet, msg:"
- + e.getMessage() + " with addr:" + m.addr,
- e);
- }
-
-
- boolean myChallenge = false;
- boolean myAck = false;
-
- while (attempts < maxAttempts) {
- try {
- /*
- * Try to obtain a challenge only if does not have
- * one yet
- */
-
- if (!myChallenge && authEnabled) {
- ToSend crequest = new ToSend(
- ToSend.mType.crequest, m.tag, m.leader,
- m.zxid, m.epoch,
- QuorumPeer.ServerState.LOOKING, m.addr);
- sendqueue.offer(crequest);
-
- try {
- double timeout = ackWait
- * java.lang.Math.pow(2, attempts);
-
- Semaphore s = new Semaphore(0);
- synchronized(Messenger.this) {
- challengeMutex.put(m.tag, s);
- s.tryAcquire((long) timeout, TimeUnit.MILLISECONDS);
- myChallenge = challengeMap
- .containsKey(m.tag);
- }
- } catch (InterruptedException e) {
- LOG.warn("Challenge request exception: ", e);
- }
- }
-
- /*
- * If don't have challenge yet, skip sending
- * notification
- */
-
- if (authEnabled && !myChallenge) {
- attempts++;
- continue;
- }
-
- if (authEnabled) {
- requestBuffer.position(40);
- Long tmpLong = challengeMap.get(m.tag);
- if(tmpLong != null){
- requestBuffer.putLong(tmpLong);
- } else {
- LOG.warn("No challenge with tag: " + m.tag);
- }
- }
- mySocket.send(requestPacket);
- try {
- Semaphore s = new Semaphore(0);
- double timeout = ackWait
- * java.lang.Math.pow(10, attempts);
- ackMutex.put(m.tag, s);
- s.tryAcquire((int) timeout, TimeUnit.MILLISECONDS);
- } catch (InterruptedException e) {
- LOG.warn("Ack exception: ", e);
- }
-
- if(ackset.remove(m.tag)){
- myAck = true;
- }
-
- } catch (IOException e) {
- LOG.warn("Sending exception: ", e);
- /*
- * Do nothing, just try again
- */
- }
- if (myAck) {
- /*
- * Received ack successfully, so return
- */
- challengeMap.remove(m.tag);
-
- return;
- } else
- attempts++;
- }
- /*
- * Return message to queue for another attempt later if
- * epoch hasn't changed.
- */
- if (m.epoch == logicalclock) {
- challengeMap.remove(m.tag);
- sendqueue.offer(m);
- }
- break;
- case 3:
-
- requestBuffer.clear();
- requestBuffer.putInt(m.type);
- requestBuffer.putLong(m.tag);
- requestBuffer.putInt(m.state.ordinal());
- requestBuffer.putLong(m.leader);
- requestBuffer.putLong(m.zxid);
- requestBuffer.putLong(m.epoch);
-
- requestPacket.setLength(48);
- try {
- requestPacket.setSocketAddress(m.addr);
- } catch (IllegalArgumentException e) {
- // Sun doesn't include the address that causes this
- // exception to be thrown, so we wrap the exception
- // in order to capture this critical detail.
- throw new IllegalArgumentException(
- "Unable to set socket address on packet, msg:"
- + e.getMessage() + " with addr:" + m.addr,
- e);
- }
-
-
- try {
- mySocket.send(requestPacket);
- } catch (IOException e) {
- LOG.warn("Exception while sending ack: ", e);
- }
- break;
- }
- }
- }
-
- public boolean queueEmpty() {
- return (sendqueue.isEmpty() || ackset.isEmpty() || recvqueue
- .isEmpty());
- }
-
- Messenger(int threads, DatagramSocket s) {
- mySocket = s;
- ackset = Collections.<Long>newSetFromMap(new ConcurrentHashMap<Long, Boolean>());
- challengeMap = new ConcurrentHashMap<Long, Long>();
- challengeMutex = new ConcurrentHashMap<Long, Semaphore>();
- ackMutex = new ConcurrentHashMap<Long, Semaphore>();
- addrChallengeMap = new ConcurrentHashMap<InetSocketAddress, ConcurrentHashMap<Long, Long>>();
- lastProposedLeader = 0;
- lastProposedZxid = 0;
- lastEpoch = 0;
-
- for (int i = 0; i < threads; ++i) {
- Thread t = new ZooKeeperThread(new WorkerSender(3),
- "WorkerSender Thread: " + (i + 1));
- t.setDaemon(true);
- t.start();
- }
-
- for (QuorumServer server : self.getVotingView().values()) {
- InetSocketAddress saddr = new InetSocketAddress(server.addr
- .getAddress(), port);
- addrChallengeMap.put(saddr, new ConcurrentHashMap<Long, Long>());
- }
-
- Thread t = new ZooKeeperThread(new WorkerReceiver(s, this),
- "WorkerReceiver-" + s.getRemoteSocketAddress());
- t.start();
- }
-
- }
-
- QuorumPeer self;
- int port;
- volatile long logicalclock; /* Election instance */
- DatagramSocket mySocket;
- long proposedLeader;
- long proposedZxid;
-
- public AuthFastLeaderElection(QuorumPeer self,
- boolean auth) {
- this.authEnabled = auth;
- starter(self);
- }
-
- public AuthFastLeaderElection(QuorumPeer self) {
- starter(self);
- }
-
- private void starter(QuorumPeer self) {
- this.self = self;
- port = self.getVotingView().get(self.getId()).electionAddr.getPort();
- proposedLeader = -1;
- proposedZxid = -1;
-
- try {
- mySocket = new DatagramSocket(port);
- // mySocket.setSoTimeout(20000);
- } catch (SocketException e1) {
- e1.printStackTrace();
- throw new RuntimeException();
- }
- sendqueue = new LinkedBlockingQueue<ToSend>(2 * self.getVotingView().size());
- recvqueue = new LinkedBlockingQueue<Notification>(2 * self.getVotingView()
- .size());
- new Messenger(self.getVotingView().size() * 2, mySocket);
- }
-
- private void leaveInstance() {
- logicalclock++;
- }
-
- private void sendNotifications() {
- for (QuorumServer server : self.getView().values()) {
-
- ToSend notmsg = new ToSend(ToSend.mType.notification,
- AuthFastLeaderElection.sequencer++, proposedLeader,
- proposedZxid, logicalclock, QuorumPeer.ServerState.LOOKING,
- self.getView().get(server.id).electionAddr);
-
- sendqueue.offer(notmsg);
- }
- }
-
- private boolean totalOrderPredicate(long id, long zxid) {
- if ((zxid > proposedZxid)
- || ((zxid == proposedZxid) && (id > proposedLeader)))
- return true;
- else
- return false;
-
- }
-
- private boolean termPredicate(HashMap<InetSocketAddress, Vote> votes,
- long l, long zxid) {
-
-
- Collection<Vote> votesCast = votes.values();
- int count = 0;
- /*
- * First make the views consistent. Sometimes peers will have different
- * zxids for a server depending on timing.
- */
- for (Vote v : votesCast) {
- if ((v.getId() == l) && (v.getZxid() == zxid))
- count++;
- }
-
- if (count > (self.getVotingView().size() / 2))
- return true;
- else
- return false;
-
- }
-
- /**
- * There is nothing to shutdown in this implementation of
- * leader election, so we simply have an empty method.
- */
- public void shutdown(){}
-
- /**
- * Invoked in QuorumPeer to find or elect a new leader.
- *
- * @throws InterruptedException
- */
- public Vote lookForLeader() throws InterruptedException {
- try {
- self.jmxLeaderElectionBean = new LeaderElectionBean();
- MBeanRegistry.getInstance().register(
- self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
- } catch (Exception e) {
- LOG.warn("Failed to register with JMX", e);
- self.jmxLeaderElectionBean = null;
- }
-
- try {
- HashMap<InetSocketAddress, Vote> recvset =
- new HashMap<InetSocketAddress, Vote>();
-
- HashMap<InetSocketAddress, Vote> outofelection =
- new HashMap<InetSocketAddress, Vote>();
-
- logicalclock++;
-
- proposedLeader = self.getId();
- proposedZxid = self.getLastLoggedZxid();
-
- LOG.info("Election tally");
- sendNotifications();
-
- /*
- * Loop in which we exchange notifications until we find a leader
- */
-
- while (self.getPeerState() == ServerState.LOOKING) {
- /*
- * Remove next notification from queue, times out after 2 times
- * the termination time
- */
- Notification n = recvqueue.poll(2 * finalizeWait,
- TimeUnit.MILLISECONDS);
-
- /*
- * Sends more notifications if haven't received enough.
- * Otherwise processes new notification.
- */
- if (n == null) {
- if (((!outofelection.isEmpty()) || (recvset.size() > 1)))
- sendNotifications();
- } else
- switch (n.state) {
- case LOOKING:
- if (n.epoch > logicalclock) {
- logicalclock = n.epoch;
- recvset.clear();
- if (totalOrderPredicate(n.leader, n.zxid)) {
- proposedLeader = n.leader;
- proposedZxid = n.zxid;
- }
- sendNotifications();
- } else if (n.epoch < logicalclock) {
- break;
- } else if (totalOrderPredicate(n.leader, n.zxid)) {
- proposedLeader = n.leader;
- proposedZxid = n.zxid;
-
- sendNotifications();
- }
-
- recvset.put(n.addr, new Vote(n.leader, n.zxid));
-
- // If have received from all nodes, then terminate
- if (self.getVotingView().size() == recvset.size()) {
- self.setPeerState((proposedLeader == self.getId()) ?
- ServerState.LEADING: ServerState.FOLLOWING);
- // if (self.state == ServerState.FOLLOWING) {
- // Thread.sleep(100);
- // }
- leaveInstance();
- return new Vote(proposedLeader, proposedZxid);
-
- } else if (termPredicate(recvset, proposedLeader,
- proposedZxid)) {
- // Otherwise, wait for a fixed amount of time
- LOG.info("Passed predicate");
- Thread.sleep(finalizeWait);
-
- // Notification probe = recvqueue.peek();
-
- // Verify if there is any change in the proposed leader
- while ((!recvqueue.isEmpty())
- && !totalOrderPredicate(
- recvqueue.peek().leader, recvqueue
- .peek().zxid)) {
- recvqueue.poll();
- }
- if (recvqueue.isEmpty()) {
- // LOG.warn("Proposed leader: " +
- // proposedLeader);
- self.setPeerState(
- (proposedLeader == self.getId()) ?
- ServerState.LEADING :
- ServerState.FOLLOWING);
-
- leaveInstance();
- return new Vote(proposedLeader, proposedZxid);
- }
- }
- break;
- case LEADING:
- outofelection.put(n.addr, new Vote(n.leader, n.zxid));
-
- if (termPredicate(outofelection, n.leader, n.zxid)) {
-
- self.setPeerState((n.leader == self.getId()) ?
- ServerState.LEADING: ServerState.FOLLOWING);
-
- leaveInstance();
- return new Vote(n.leader, n.zxid);
- }
- break;
- case FOLLOWING:
- outofelection.put(n.addr, new Vote(n.leader, n.zxid));
-
- if (termPredicate(outofelection, n.leader, n.zxid)) {
-
- self.setPeerState((n.leader == self.getId()) ?
- ServerState.LEADING: ServerState.FOLLOWING);
-
- leaveInstance();
- return new Vote(n.leader, n.zxid);
- }
- break;
- default:
- break;
- }
- }
-
- return null;
- } finally {
- try {
- if(self.jmxLeaderElectionBean != null){
- MBeanRegistry.getInstance().unregister(
- self.jmxLeaderElectionBean);
- }
- } catch (Exception e) {
- LOG.warn("Failed to unregister with JMX", e);
- }
- self.jmxLeaderElectionBean = null;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/main/org/apache/zookeeper/server/quorum/CommitProcessor.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/CommitProcessor.java b/src/java/main/org/apache/zookeeper/server/quorum/CommitProcessor.java
deleted file mode 100644
index 3419f08..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/CommitProcessor.java
+++ /dev/null
@@ -1,195 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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 org.apache.zookeeper.server.quorum;
-
-import java.util.ArrayList;
-import java.util.LinkedList;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.apache.zookeeper.ZooDefs.OpCode;
-import org.apache.zookeeper.server.Request;
-import org.apache.zookeeper.server.RequestProcessor;
-import org.apache.zookeeper.server.ZooKeeperCriticalThread;
-import org.apache.zookeeper.server.ZooKeeperServerListener;
-
-/**
- * This RequestProcessor matches the incoming committed requests with the
- * locally submitted requests. The trick is that locally submitted requests that
- * change the state of the system will come back as incoming committed requests,
- * so we need to match them up.
- */
-public class CommitProcessor extends ZooKeeperCriticalThread implements RequestProcessor {
- private static final Logger LOG = LoggerFactory.getLogger(CommitProcessor.class);
-
- /**
- * Requests that we are holding until the commit comes in.
- */
- LinkedList<Request> queuedRequests = new LinkedList<Request>();
-
- /**
- * Requests that have been committed.
- */
- LinkedList<Request> committedRequests = new LinkedList<Request>();
-
- RequestProcessor nextProcessor;
- ArrayList<Request> toProcess = new ArrayList<Request>();
-
- /**
- * This flag indicates whether we need to wait for a response to come back from the
- * leader or we just let the sync operation flow through like a read. The flag will
- * be false if the CommitProcessor is in a Leader pipeline.
- */
- boolean matchSyncs;
-
- public CommitProcessor(RequestProcessor nextProcessor, String id,
- boolean matchSyncs, ZooKeeperServerListener listener) {
- super("CommitProcessor:" + id, listener);
- this.nextProcessor = nextProcessor;
- this.matchSyncs = matchSyncs;
- }
-
- volatile boolean finished = false;
-
- @Override
- public void run() {
- try {
- Request nextPending = null;
- while (!finished) {
- int len = toProcess.size();
- for (int i = 0; i < len; i++) {
- nextProcessor.processRequest(toProcess.get(i));
- }
- toProcess.clear();
- synchronized (this) {
- if ((queuedRequests.size() == 0 || nextPending != null)
- && committedRequests.size() == 0) {
- wait();
- continue;
- }
- // First check and see if the commit came in for the pending
- // request
- if ((queuedRequests.size() == 0 || nextPending != null)
- && committedRequests.size() > 0) {
- Request r = committedRequests.remove();
- /*
- * We match with nextPending so that we can move to the
- * next request when it is committed. We also want to
- * use nextPending because it has the cnxn member set
- * properly.
- */
- if (nextPending != null
- && nextPending.sessionId == r.sessionId
- && nextPending.cxid == r.cxid) {
- // we want to send our version of the request.
- // the pointer to the connection in the request
- nextPending.hdr = r.hdr;
- nextPending.txn = r.txn;
- nextPending.zxid = r.zxid;
- toProcess.add(nextPending);
- nextPending = null;
- } else {
- // this request came from someone else so just
- // send the commit packet
- toProcess.add(r);
- }
- }
- }
-
- // We haven't matched the pending requests, so go back to
- // waiting
- if (nextPending != null) {
- continue;
- }
-
- synchronized (this) {
- // Process the next requests in the queuedRequests
- while (nextPending == null && queuedRequests.size() > 0) {
- Request request = queuedRequests.remove();
- switch (request.type) {
- case OpCode.create:
- case OpCode.delete:
- case OpCode.setData:
- case OpCode.multi:
- case OpCode.setACL:
- case OpCode.createSession:
- case OpCode.closeSession:
- nextPending = request;
- break;
- case OpCode.sync:
- if (matchSyncs) {
- nextPending = request;
- } else {
- toProcess.add(request);
- }
- break;
- default:
- toProcess.add(request);
- }
- }
- }
- }
- } catch (InterruptedException e) {
- LOG.warn("Interrupted exception while waiting", e);
- } catch (Throwable e) {
- LOG.error("Unexpected exception causing CommitProcessor to exit", e);
- }
- LOG.info("CommitProcessor exited loop!");
- }
-
- synchronized public void commit(Request request) {
- if (!finished) {
- if (request == null) {
- LOG.warn("Committed a null!",
- new Exception("committing a null! "));
- return;
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("Committing request:: " + request);
- }
- committedRequests.add(request);
- notifyAll();
- }
- }
-
- synchronized public void processRequest(Request request) {
- // request.addRQRec(">commit");
- if (LOG.isDebugEnabled()) {
- LOG.debug("Processing request:: " + request);
- }
-
- if (!finished) {
- queuedRequests.add(request);
- notifyAll();
- }
- }
-
- public void shutdown() {
- LOG.info("Shutting down");
- synchronized (this) {
- finished = true;
- queuedRequests.clear();
- notifyAll();
- }
- if (nextProcessor != null) {
- nextProcessor.shutdown();
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/main/org/apache/zookeeper/server/quorum/Election.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/Election.java b/src/java/main/org/apache/zookeeper/server/quorum/Election.java
deleted file mode 100644
index 8990638..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/Election.java
+++ /dev/null
@@ -1,27 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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 org.apache.zookeeper.server.quorum;
-
-
-import org.apache.zookeeper.server.quorum.Vote;
-
-public interface Election {
- public Vote lookForLeader() throws InterruptedException;
- public void shutdown();
-}
http://git-wip-us.apache.org/repos/asf/zookeeper/blob/c0aa3b3f/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
----------------------------------------------------------------------
diff --git a/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java b/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
deleted file mode 100644
index 91b6af8..0000000
--- a/src/java/main/org/apache/zookeeper/server/quorum/FastLeaderElection.java
+++ /dev/null
@@ -1,1015 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * 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 org.apache.zookeeper.server.quorum;
-
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
-
-import org.apache.zookeeper.common.Time;
-import org.apache.zookeeper.jmx.MBeanRegistry;
-import org.apache.zookeeper.server.ZooKeeperThread;
-import org.apache.zookeeper.server.quorum.QuorumCnxManager.Message;
-import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
-import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
-import org.apache.zookeeper.server.quorum.QuorumPeer.ServerState;
-import org.apache.zookeeper.server.util.ZxidUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-
-/**
- * Implementation of leader election using TCP. It uses an object of the class
- * QuorumCnxManager to manage connections. Otherwise, the algorithm is push-based
- * as with the other UDP implementations.
- *
- * There are a few parameters that can be tuned to change its behavior. First,
- * finalizeWait determines the amount of time to wait until deciding upon a leader.
- * This is part of the leader election algorithm.
- */
-
-
-public class FastLeaderElection implements Election {
- private static final Logger LOG = LoggerFactory.getLogger(FastLeaderElection.class);
-
- /**
- * Determine how much time a process has to wait
- * once it believes that it has reached the end of
- * leader election.
- */
- final static int finalizeWait = 200;
-
-
- /**
- * Upper bound on the amount of time between two consecutive
- * notification checks. This impacts the amount of time to get
- * the system up again after long partitions. Currently 60 seconds.
- */
-
- final static int maxNotificationInterval = 60000;
-
- /**
- * Connection manager. Fast leader election uses TCP for
- * communication between peers, and QuorumCnxManager manages
- * such connections.
- */
-
- QuorumCnxManager manager;
-
-
- /**
- * Notifications are messages that let other peers know that
- * a given peer has changed its vote, either because it has
- * joined leader election or because it learned of another
- * peer with higher zxid or same zxid and higher server id
- */
-
- static public class Notification {
- /*
- * Format version, introduced in 3.4.6
- */
-
- public final static int CURRENTVERSION = 0x1;
- int version;
-
- /*
- * Proposed leader
- */
- long leader;
-
- /*
- * zxid of the proposed leader
- */
- long zxid;
-
- /*
- * Epoch
- */
- long electionEpoch;
-
- /*
- * current state of sender
- */
- QuorumPeer.ServerState state;
-
- /*
- * Address of sender
- */
- long sid;
-
- /*
- * epoch of the proposed leader
- */
- long peerEpoch;
-
- @Override
- public String toString() {
- return Long.toHexString(version) + " (message format version), "
- + leader + " (n.leader), 0x"
- + Long.toHexString(zxid) + " (n.zxid), 0x"
- + Long.toHexString(electionEpoch) + " (n.round), " + state
- + " (n.state), " + sid + " (n.sid), 0x"
- + Long.toHexString(peerEpoch) + " (n.peerEpoch) ";
- }
- }
-
- static ByteBuffer buildMsg(int state,
- long leader,
- long zxid,
- long electionEpoch,
- long epoch) {
- byte requestBytes[] = new byte[40];
- ByteBuffer requestBuffer = ByteBuffer.wrap(requestBytes);
-
- /*
- * Building notification packet to send
- */
-
- requestBuffer.clear();
- requestBuffer.putInt(state);
- requestBuffer.putLong(leader);
- requestBuffer.putLong(zxid);
- requestBuffer.putLong(electionEpoch);
- requestBuffer.putLong(epoch);
- requestBuffer.putInt(Notification.CURRENTVERSION);
-
- return requestBuffer;
- }
-
- /**
- * Messages that a peer wants to send to other peers.
- * These messages can be both Notifications and Acks
- * of reception of notification.
- */
- static public class ToSend {
- static enum mType {crequest, challenge, notification, ack}
-
- ToSend(mType type,
- long leader,
- long zxid,
- long electionEpoch,
- ServerState state,
- long sid,
- long peerEpoch) {
-
- this.leader = leader;
- this.zxid = zxid;
- this.electionEpoch = electionEpoch;
- this.state = state;
- this.sid = sid;
- this.peerEpoch = peerEpoch;
- }
-
- /*
- * Proposed leader in the case of notification
- */
- long leader;
-
- /*
- * id contains the tag for acks, and zxid for notifications
- */
- long zxid;
-
- /*
- * Epoch
- */
- long electionEpoch;
-
- /*
- * Current state;
- */
- QuorumPeer.ServerState state;
-
- /*
- * Address of recipient
- */
- long sid;
-
- /*
- * Leader epoch
- */
- long peerEpoch;
- }
-
- LinkedBlockingQueue<ToSend> sendqueue;
- LinkedBlockingQueue<Notification> recvqueue;
-
- /**
- * Multi-threaded implementation of message handler. Messenger
- * implements two sub-classes: WorkReceiver and WorkSender. The
- * functionality of each is obvious from the name. Each of these
- * spawns a new thread.
- */
-
- protected class Messenger {
-
- /**
- * Receives messages from instance of QuorumCnxManager on
- * method run(), and processes such messages.
- */
-
- class WorkerReceiver extends ZooKeeperThread {
- volatile boolean stop;
- QuorumCnxManager manager;
-
- WorkerReceiver(QuorumCnxManager manager) {
- super("WorkerReceiver");
- this.stop = false;
- this.manager = manager;
- }
-
- public void run() {
-
- Message response;
- while (!stop) {
- // Sleeps on receive
- try{
- response = manager.pollRecvQueue(3000, TimeUnit.MILLISECONDS);
- if(response == null) continue;
-
- /*
- * If it is from an observer, respond right away.
- * Note that the following predicate assumes that
- * if a server is not a follower, then it must be
- * an observer. If we ever have any other type of
- * learner in the future, we'll have to change the
- * way we check for observers.
- */
- if(!validVoter(response.sid)){
- Vote current = self.getCurrentVote();
- ToSend notmsg = new ToSend(ToSend.mType.notification,
- current.getId(),
- current.getZxid(),
- logicalclock.get(),
- self.getPeerState(),
- response.sid,
- current.getPeerEpoch());
-
- sendqueue.offer(notmsg);
- } else {
- // Receive new message
- if (LOG.isDebugEnabled()) {
- LOG.debug("Receive new notification message. My id = "
- + self.getId());
- }
-
- /*
- * We check for 28 bytes for backward compatibility
- */
- if (response.buffer.capacity() < 28) {
- LOG.error("Got a short response: "
- + response.buffer.capacity());
- continue;
- }
- boolean backCompatibility = (response.buffer.capacity() == 28);
- response.buffer.clear();
-
- // Instantiate Notification and set its attributes
- Notification n = new Notification();
-
- // State of peer that sent this message
- QuorumPeer.ServerState ackstate = QuorumPeer.ServerState.LOOKING;
- switch (response.buffer.getInt()) {
- case 0:
- ackstate = QuorumPeer.ServerState.LOOKING;
- break;
- case 1:
- ackstate = QuorumPeer.ServerState.FOLLOWING;
- break;
- case 2:
- ackstate = QuorumPeer.ServerState.LEADING;
- break;
- case 3:
- ackstate = QuorumPeer.ServerState.OBSERVING;
- break;
- default:
- continue;
- }
-
- n.leader = response.buffer.getLong();
- n.zxid = response.buffer.getLong();
- n.electionEpoch = response.buffer.getLong();
- n.state = ackstate;
- n.sid = response.sid;
- if(!backCompatibility){
- n.peerEpoch = response.buffer.getLong();
- } else {
- if(LOG.isInfoEnabled()){
- LOG.info("Backward compatibility mode, server id=" + n.sid);
- }
- n.peerEpoch = ZxidUtils.getEpochFromZxid(n.zxid);
- }
-
- /*
- * Version added in 3.4.6
- */
-
- n.version = (response.buffer.remaining() >= 4) ?
- response.buffer.getInt() : 0x0;
-
- /*
- * Print notification info
- */
- if(LOG.isInfoEnabled()){
- printNotification(n);
- }
-
- /*
- * If this server is looking, then send proposed leader
- */
-
- if(self.getPeerState() == QuorumPeer.ServerState.LOOKING){
- recvqueue.offer(n);
-
- /*
- * Send a notification back if the peer that sent this
- * message is also looking and its logical clock is
- * lagging behind.
- */
- if((ackstate == QuorumPeer.ServerState.LOOKING)
- && (n.electionEpoch < logicalclock.get())){
- Vote v = getVote();
- ToSend notmsg = new ToSend(ToSend.mType.notification,
- v.getId(),
- v.getZxid(),
- logicalclock.get(),
- self.getPeerState(),
- response.sid,
- v.getPeerEpoch());
- sendqueue.offer(notmsg);
- }
- } else {
- /*
- * If this server is not looking, but the one that sent the ack
- * is looking, then send back what it believes to be the leader.
- */
- Vote current = self.getCurrentVote();
- if(ackstate == QuorumPeer.ServerState.LOOKING){
- if(LOG.isDebugEnabled()){
- LOG.debug("Sending new notification. My id = " +
- self.getId() + " recipient=" +
- response.sid + " zxid=0x" +
- Long.toHexString(current.getZxid()) +
- " leader=" + current.getId());
- }
-
- ToSend notmsg;
- if(n.version > 0x0) {
- notmsg = new ToSend(
- ToSend.mType.notification,
- current.getId(),
- current.getZxid(),
- current.getElectionEpoch(),
- self.getPeerState(),
- response.sid,
- current.getPeerEpoch());
-
- } else {
- Vote bcVote = self.getBCVote();
- notmsg = new ToSend(
- ToSend.mType.notification,
- bcVote.getId(),
- bcVote.getZxid(),
- bcVote.getElectionEpoch(),
- self.getPeerState(),
- response.sid,
- bcVote.getPeerEpoch());
- }
- sendqueue.offer(notmsg);
- }
- }
- }
- } catch (InterruptedException e) {
- System.out.println("Interrupted Exception while waiting for new message" +
- e.toString());
- }
- }
- LOG.info("WorkerReceiver is down");
- }
- }
-
-
- /**
- * This worker simply dequeues a message to send and
- * and queues it on the manager's queue.
- */
-
- class WorkerSender extends ZooKeeperThread {
- volatile boolean stop;
- QuorumCnxManager manager;
-
- WorkerSender(QuorumCnxManager manager){
- super("WorkerSender");
- this.stop = false;
- this.manager = manager;
- }
-
- public void run() {
- while (!stop) {
- try {
- ToSend m = sendqueue.poll(3000, TimeUnit.MILLISECONDS);
- if(m == null) continue;
-
- process(m);
- } catch (InterruptedException e) {
- break;
- }
- }
- LOG.info("WorkerSender is down");
- }
-
- /**
- * Called by run() once there is a new message to send.
- *
- * @param m message to send
- */
- void process(ToSend m) {
- ByteBuffer requestBuffer = buildMsg(m.state.ordinal(),
- m.leader,
- m.zxid,
- m.electionEpoch,
- m.peerEpoch);
- manager.toSend(m.sid, requestBuffer);
- }
- }
-
-
- WorkerSender ws;
- WorkerReceiver wr;
-
- /**
- * Constructor of class Messenger.
- *
- * @param manager Connection manager
- */
- Messenger(QuorumCnxManager manager) {
-
- this.ws = new WorkerSender(manager);
-
- Thread t = new Thread(this.ws,
- "WorkerSender[myid=" + self.getId() + "]");
- t.setDaemon(true);
- t.start();
-
- this.wr = new WorkerReceiver(manager);
-
- t = new Thread(this.wr,
- "WorkerReceiver[myid=" + self.getId() + "]");
- t.setDaemon(true);
- t.start();
- }
-
- /**
- * Stops instances of WorkerSender and WorkerReceiver
- */
- void halt(){
- this.ws.stop = true;
- this.wr.stop = true;
- }
-
- }
-
- QuorumPeer self;
- Messenger messenger;
- AtomicLong logicalclock = new AtomicLong(); /* Election instance */
- long proposedLeader;
- long proposedZxid;
- long proposedEpoch;
-
-
- /**
- * Returns the current vlue of the logical clock counter
- */
- public long getLogicalClock(){
- return logicalclock.get();
- }
-
- /**
- * Constructor of FastLeaderElection. It takes two parameters, one
- * is the QuorumPeer object that instantiated this object, and the other
- * is the connection manager. Such an object should be created only once
- * by each peer during an instance of the ZooKeeper service.
- *
- * @param self QuorumPeer that created this object
- * @param manager Connection manager
- */
- public FastLeaderElection(QuorumPeer self, QuorumCnxManager manager){
- this.stop = false;
- this.manager = manager;
- starter(self, manager);
- }
-
- /**
- * This method is invoked by the constructor. Because it is a
- * part of the starting procedure of the object that must be on
- * any constructor of this class, it is probably best to keep as
- * a separate method. As we have a single constructor currently,
- * it is not strictly necessary to have it separate.
- *
- * @param self QuorumPeer that created this object
- * @param manager Connection manager
- */
- private void starter(QuorumPeer self, QuorumCnxManager manager) {
- this.self = self;
- proposedLeader = -1;
- proposedZxid = -1;
-
- sendqueue = new LinkedBlockingQueue<ToSend>();
- recvqueue = new LinkedBlockingQueue<Notification>();
- this.messenger = new Messenger(manager);
- }
-
- private void leaveInstance(Vote v) {
- if(LOG.isDebugEnabled()){
- LOG.debug("About to leave FLE instance: leader="
- + v.getId() + ", zxid=0x" +
- Long.toHexString(v.getZxid()) + ", my id=" + self.getId()
- + ", my state=" + self.getPeerState());
- }
- recvqueue.clear();
- }
-
- public QuorumCnxManager getCnxManager(){
- return manager;
- }
-
- volatile boolean stop;
- public void shutdown(){
- stop = true;
- LOG.debug("Shutting down connection manager");
- manager.halt();
- LOG.debug("Shutting down messenger");
- messenger.halt();
- LOG.debug("FLE is down");
- }
-
-
- /**
- * Send notifications to all peers upon a change in our vote
- */
- private void sendNotifications() {
- for (QuorumServer server : self.getVotingView().values()) {
- long sid = server.id;
-
- ToSend notmsg = new ToSend(ToSend.mType.notification,
- proposedLeader,
- proposedZxid,
- logicalclock.get(),
- QuorumPeer.ServerState.LOOKING,
- sid,
- proposedEpoch);
- if(LOG.isDebugEnabled()){
- LOG.debug("Sending Notification: " + proposedLeader + " (n.leader), 0x" +
- Long.toHexString(proposedZxid) + " (n.zxid), 0x" + Long.toHexString(logicalclock.get()) +
- " (n.round), " + sid + " (recipient), " + self.getId() +
- " (myid), 0x" + Long.toHexString(proposedEpoch) + " (n.peerEpoch)");
- }
- sendqueue.offer(notmsg);
- }
- }
-
-
- private void printNotification(Notification n){
- LOG.info("Notification: " + n.toString()
- + self.getPeerState() + " (my state)");
- }
-
- /**
- * Check if a pair (server id, zxid) succeeds our
- * current vote.
- *
- * @param id Server identifier
- * @param zxid Last zxid observed by the issuer of this vote
- */
- protected boolean totalOrderPredicate(long newId, long newZxid, long newEpoch, long curId, long curZxid, long curEpoch) {
- LOG.debug("id: " + newId + ", proposed id: " + curId + ", zxid: 0x" +
- Long.toHexString(newZxid) + ", proposed zxid: 0x" + Long.toHexString(curZxid));
- if(self.getQuorumVerifier().getWeight(newId) == 0){
- return false;
- }
-
- /*
- * We return true if one of the following three cases hold:
- * 1- New epoch is higher
- * 2- New epoch is the same as current epoch, but new zxid is higher
- * 3- New epoch is the same as current epoch, new zxid is the same
- * as current zxid, but server id is higher.
- */
-
- return ((newEpoch > curEpoch) ||
- ((newEpoch == curEpoch) &&
- ((newZxid > curZxid) || ((newZxid == curZxid) && (newId > curId)))));
- }
-
- /**
- * Termination predicate. Given a set of votes, determines if
- * have sufficient to declare the end of the election round.
- *
- * @param votes Set of votes
- * @param l Identifier of the vote received last
- * @param zxid zxid of the the vote received last
- */
- protected boolean termPredicate(
- HashMap<Long, Vote> votes,
- Vote vote) {
-
- HashSet<Long> set = new HashSet<Long>();
-
- /*
- * First make the views consistent. Sometimes peers will have
- * different zxids for a server depending on timing.
- */
- for (Map.Entry<Long,Vote> entry : votes.entrySet()) {
- if (vote.equals(entry.getValue())){
- set.add(entry.getKey());
- }
- }
-
- return self.getQuorumVerifier().containsQuorum(set);
- }
-
- /**
- * In the case there is a leader elected, and a quorum supporting
- * this leader, we have to check if the leader has voted and acked
- * that it is leading. We need this check to avoid that peers keep
- * electing over and over a peer that has crashed and it is no
- * longer leading.
- *
- * @param votes set of votes
- * @param leader leader id
- * @param electionEpoch epoch id
- */
- protected boolean checkLeader(
- HashMap<Long, Vote> votes,
- long leader,
- long electionEpoch){
-
- boolean predicate = true;
-
- /*
- * If everyone else thinks I'm the leader, I must be the leader.
- * The other two checks are just for the case in which I'm not the
- * leader. If I'm not the leader and I haven't received a message
- * from leader stating that it is leading, then predicate is false.
- */
-
- if(leader != self.getId()){
- if(votes.get(leader) == null) predicate = false;
- else if(votes.get(leader).getState() != ServerState.LEADING) predicate = false;
- } else if(logicalclock.get() != electionEpoch) {
- predicate = false;
- }
-
- return predicate;
- }
-
- /**
- * This predicate checks that a leader has been elected. It doesn't
- * make a lot of sense without context (check lookForLeader) and it
- * has been separated for testing purposes.
- *
- * @param recv map of received votes
- * @param ooe map containing out of election votes (LEADING or FOLLOWING)
- * @param n Notification
- * @return
- */
- protected boolean ooePredicate(HashMap<Long,Vote> recv,
- HashMap<Long,Vote> ooe,
- Notification n) {
-
- return (termPredicate(recv, new Vote(n.version,
- n.leader,
- n.zxid,
- n.electionEpoch,
- n.peerEpoch,
- n.state))
- && checkLeader(ooe, n.leader, n.electionEpoch));
-
- }
-
- synchronized void updateProposal(long leader, long zxid, long epoch){
- if(LOG.isDebugEnabled()){
- LOG.debug("Updating proposal: " + leader + " (newleader), 0x"
- + Long.toHexString(zxid) + " (newzxid), " + proposedLeader
- + " (oldleader), 0x" + Long.toHexString(proposedZxid) + " (oldzxid)");
- }
- proposedLeader = leader;
- proposedZxid = zxid;
- proposedEpoch = epoch;
- }
-
- synchronized Vote getVote(){
- return new Vote(proposedLeader, proposedZxid, proposedEpoch);
- }
-
- /**
- * A learning state can be either FOLLOWING or OBSERVING.
- * This method simply decides which one depending on the
- * role of the server.
- *
- * @return ServerState
- */
- private ServerState learningState(){
- if(self.getLearnerType() == LearnerType.PARTICIPANT){
- LOG.debug("I'm a participant: " + self.getId());
- return ServerState.FOLLOWING;
- }
- else{
- LOG.debug("I'm an observer: " + self.getId());
- return ServerState.OBSERVING;
- }
- }
-
- /**
- * Returns the initial vote value of server identifier.
- *
- * @return long
- */
- private long getInitId(){
- if(self.getLearnerType() == LearnerType.PARTICIPANT)
- return self.getId();
- else return Long.MIN_VALUE;
- }
-
- /**
- * Returns initial last logged zxid.
- *
- * @return long
- */
- private long getInitLastLoggedZxid(){
- if(self.getLearnerType() == LearnerType.PARTICIPANT)
- return self.getLastLoggedZxid();
- else return Long.MIN_VALUE;
- }
-
- /**
- * Returns the initial vote value of the peer epoch.
- *
- * @return long
- */
- private long getPeerEpoch(){
- if(self.getLearnerType() == LearnerType.PARTICIPANT)
- try {
- return self.getCurrentEpoch();
- } catch(IOException e) {
- RuntimeException re = new RuntimeException(e.getMessage());
- re.setStackTrace(e.getStackTrace());
- throw re;
- }
- else return Long.MIN_VALUE;
- }
-
- /**
- * Starts a new round of leader election. Whenever our QuorumPeer
- * changes its state to LOOKING, this method is invoked, and it
- * sends notifications to all other peers.
- */
- public Vote lookForLeader() throws InterruptedException {
- try {
- self.jmxLeaderElectionBean = new LeaderElectionBean();
- MBeanRegistry.getInstance().register(
- self.jmxLeaderElectionBean, self.jmxLocalPeerBean);
- } catch (Exception e) {
- LOG.warn("Failed to register with JMX", e);
- self.jmxLeaderElectionBean = null;
- }
- if (self.start_fle == 0) {
- self.start_fle = Time.currentElapsedTime();
- }
- try {
- HashMap<Long, Vote> recvset = new HashMap<Long, Vote>();
-
- HashMap<Long, Vote> outofelection = new HashMap<Long, Vote>();
-
- int notTimeout = finalizeWait;
-
- synchronized(this){
- logicalclock.incrementAndGet();
- updateProposal(getInitId(), getInitLastLoggedZxid(), getPeerEpoch());
- }
-
- LOG.info("New election. My id = " + self.getId() +
- ", proposed zxid=0x" + Long.toHexString(proposedZxid));
- sendNotifications();
-
- /*
- * Loop in which we exchange notifications until we find a leader
- */
-
- while ((self.getPeerState() == ServerState.LOOKING) &&
- (!stop)){
- /*
- * Remove next notification from queue, times out after 2 times
- * the termination time
- */
- Notification n = recvqueue.poll(notTimeout,
- TimeUnit.MILLISECONDS);
-
- /*
- * Sends more notifications if haven't received enough.
- * Otherwise processes new notification.
- */
- if(n == null){
- if(manager.haveDelivered()){
- sendNotifications();
- } else {
- manager.connectAll();
- }
-
- /*
- * Exponential backoff
- */
- int tmpTimeOut = notTimeout*2;
- notTimeout = (tmpTimeOut < maxNotificationInterval?
- tmpTimeOut : maxNotificationInterval);
- LOG.info("Notification time out: " + notTimeout);
- }
- else if(validVoter(n.sid) && validVoter(n.leader)) {
- /*
- * Only proceed if the vote comes from a replica in the
- * voting view for a replica in the voting view.
- */
- switch (n.state) {
- case LOOKING:
- // If notification > current, replace and send messages out
- if (n.electionEpoch > logicalclock.get()) {
- logicalclock.set(n.electionEpoch);
- recvset.clear();
- if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
- getInitId(), getInitLastLoggedZxid(), getPeerEpoch())) {
- updateProposal(n.leader, n.zxid, n.peerEpoch);
- } else {
- updateProposal(getInitId(),
- getInitLastLoggedZxid(),
- getPeerEpoch());
- }
- sendNotifications();
- } else if (n.electionEpoch < logicalclock.get()) {
- if(LOG.isDebugEnabled()){
- LOG.debug("Notification election epoch is smaller than logicalclock. n.electionEpoch = 0x"
- + Long.toHexString(n.electionEpoch)
- + ", logicalclock=0x" + Long.toHexString(logicalclock.get()));
- }
- break;
- } else if (totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
- proposedLeader, proposedZxid, proposedEpoch)) {
- updateProposal(n.leader, n.zxid, n.peerEpoch);
- sendNotifications();
- }
-
- if(LOG.isDebugEnabled()){
- LOG.debug("Adding vote: from=" + n.sid +
- ", proposed leader=" + n.leader +
- ", proposed zxid=0x" + Long.toHexString(n.zxid) +
- ", proposed election epoch=0x" + Long.toHexString(n.electionEpoch));
- }
-
- recvset.put(n.sid, new Vote(n.leader, n.zxid, n.electionEpoch, n.peerEpoch));
-
- if (termPredicate(recvset,
- new Vote(proposedLeader, proposedZxid,
- logicalclock.get(), proposedEpoch))) {
-
- // Verify if there is any change in the proposed leader
- while((n = recvqueue.poll(finalizeWait,
- TimeUnit.MILLISECONDS)) != null){
- if(totalOrderPredicate(n.leader, n.zxid, n.peerEpoch,
- proposedLeader, proposedZxid, proposedEpoch)){
- recvqueue.put(n);
- break;
- }
- }
-
- /*
- * This predicate is true once we don't read any new
- * relevant message from the reception queue
- */
- if (n == null) {
- self.setPeerState((proposedLeader == self.getId()) ?
- ServerState.LEADING: learningState());
-
- Vote endVote = new Vote(proposedLeader,
- proposedZxid,
- logicalclock.get(),
- proposedEpoch);
- leaveInstance(endVote);
- return endVote;
- }
- }
- break;
- case OBSERVING:
- LOG.debug("Notification from observer: " + n.sid);
- break;
- case FOLLOWING:
- case LEADING:
- /*
- * Consider all notifications from the same epoch
- * together.
- */
- if(n.electionEpoch == logicalclock.get()){
- recvset.put(n.sid, new Vote(n.leader,
- n.zxid,
- n.electionEpoch,
- n.peerEpoch));
-
- if(ooePredicate(recvset, outofelection, n)) {
- self.setPeerState((n.leader == self.getId()) ?
- ServerState.LEADING: learningState());
-
- Vote endVote = new Vote(n.leader,
- n.zxid,
- n.electionEpoch,
- n.peerEpoch);
- leaveInstance(endVote);
- return endVote;
- }
- }
-
- /*
- * Before joining an established ensemble, verify
- * a majority is following the same leader.
- */
- outofelection.put(n.sid, new Vote(n.version,
- n.leader,
- n.zxid,
- n.electionEpoch,
- n.peerEpoch,
- n.state));
-
- if(ooePredicate(outofelection, outofelection, n)) {
- synchronized(this){
- logicalclock.set(n.electionEpoch);
- self.setPeerState((n.leader == self.getId()) ?
- ServerState.LEADING: learningState());
- }
- Vote endVote = new Vote(n.leader,
- n.zxid,
- n.electionEpoch,
- n.peerEpoch);
- leaveInstance(endVote);
- return endVote;
- }
- break;
- default:
- LOG.warn("Notification state unrecognized: {} (n.state), {} (n.sid)",
- n.state, n.sid);
- break;
- }
- } else {
- if (!validVoter(n.leader)) {
- LOG.warn("Ignoring notification for non-cluster member sid {} from sid {}", n.leader, n.sid);
- }
- if (!validVoter(n.sid)) {
- LOG.warn("Ignoring notification for sid {} from non-quorum member sid {}", n.leader, n.sid);
- }
- }
- }
- return null;
- } finally {
- try {
- if(self.jmxLeaderElectionBean != null){
- MBeanRegistry.getInstance().unregister(
- self.jmxLeaderElectionBean);
- }
- } catch (Exception e) {
- LOG.warn("Failed to unregister with JMX", e);
- }
- self.jmxLeaderElectionBean = null;
- LOG.debug("Number of connection processing threads: {}",
- manager.getConnectionThreadCount());
- }
- }
-
- /**
- * Check if a given sid is represented in either the current or
- * the next voting view
- *
- * @param sid Server identifier
- * @return boolean
- */
- private boolean validVoter(long sid) {
- return self.getVotingView().containsKey(sid);
- }
-}