public class NonBlockingCoordinator extends ChannelInterceptorBase
Title: Auto merging leader election algorithm
Description: Implementation of a simple coordinator algorithm that not only selects a coordinator, it also merges groups automatically when members are discovered that werent part of the
This algorithm is non blocking meaning it allows for transactions while the coordination phase is going on
This implementation is based on a home brewed algorithm that uses the AbsoluteOrder of a membership
to pass a token ring of the current membership.
This is not the same as just using AbsoluteOrder! Consider the following scenario:
Nodes, A,B,C,D,E on a network, in that priority. AbsoluteOrder will only work if all
nodes are receiving pings from all the other nodes.
meaning, that node{i} receives pings from node{all}-node{i}
but the following could happen if a multicast problem occurs.
A has members {B,C,D}
B has members {A,C}
C has members {D,E}
D has members {A,B,C,E}
E has members {A,C,D}
Because the default Tribes membership implementation, relies on the multicast packets to
arrive at all nodes correctly, there is nothing guaranteeing that it will.
To best explain how this algorithm works, lets take the above example:
For simplicity we assume that a send operation is O(1) for all nodes, although this algorithm will work
where messages overlap, as they all depend on absolute order
Scenario 1: A,B,C,D,E all come online at the same time
Eval phase, A thinks of itself as leader, B thinks of A as leader,
C thinks of itself as leader, D,E think of A as leader
Token phase:
(1) A sends out a message X{A-ldr, A-src, mbrs-A,B,C,D} to B where X is the id for the message(and the view)
(1) C sends out a message Y{C-ldr, C-src, mbrs-C,D,E} to D where Y is the id for the message(and the view)
(2) B receives X{A-ldr, A-src, mbrs-A,B,C,D}, sends X{A-ldr, A-src, mbrs-A,B,C,D} to C
(2) D receives Y{C-ldr, C-src, mbrs-C,D,E} D is aware of A,B, sends Y{A-ldr, C-src, mbrs-A,B,C,D,E} to E
(3) C receives X{A-ldr, A-src, mbrs-A,B,C,D}, sends X{A-ldr, A-src, mbrs-A,B,C,D,E} to D
(3) E receives Y{A-ldr, C-src, mbrs-A,B,C,D,E} sends Y{A-ldr, C-src, mbrs-A,B,C,D,E} to A
(4) D receives X{A-ldr, A-src, mbrs-A,B,C,D,E} sends sends X{A-ldr, A-src, mbrs-A,B,C,D,E} to A
(4) A receives Y{A-ldr, C-src, mbrs-A,B,C,D,E}, holds the message, add E to its list of members
(5) A receives X{A-ldr, A-src, mbrs-A,B,C,D,E}
At this point, the state looks like
A - {A-ldr, mbrs-A,B,C,D,E, id=X}
B - {A-ldr, mbrs-A,B,C,D, id=X}
C - {A-ldr, mbrs-A,B,C,D,E, id=X}
D - {A-ldr, mbrs-A,B,C,D,E, id=X}
E - {A-ldr, mbrs-A,B,C,D,E, id=Y}
A message doesn't stop until it reaches its original sender, unless its dropped by a higher leader.
As you can see, E still thinks the viewId=Y, which is not correct. But at this point we have
arrived at the same membership and all nodes are informed of each other.
To synchronize the rest we simply perform the following check at A when A receives X:
Original X{A-ldr, A-src, mbrs-A,B,C,D} == Arrived X{A-ldr, A-src, mbrs-A,B,C,D,E}
Since the condition is false, A, will resend the token, and A sends X{A-ldr, A-src, mbrs-A,B,C,D,E} to B
When A receives X again, the token is complete.
Optionally, A can send a message X{A-ldr, A-src, mbrs-A,B,C,D,E confirmed} to A,B,C,D,E who then
install and accept the view.
Lets assume that C1 arrives, C1 has lower priority than C, but higher priority than D.
Lets also assume that C1 sees the following view {B,D,E}
C1 waits for a token to arrive. When the token arrives, the same scenario as above will happen.
In the scenario where C1 sees {D,E} and A,B,C can not see C1, no token will ever arrive.
In this case, C1 sends a Z{C1-ldr, C1-src, mbrs-C1,D,E} to D
D receives Z{C1-ldr, C1-src, mbrs-C1,D,E} and sends Z{A-ldr, C1-src, mbrs-A,B,C,C1,D,E} to E
E receives Z{A-ldr, C1-src, mbrs-A,B,C,C1,D,E} and sends it to A
A sends Z{A-ldr, A-src, mbrs-A,B,C,C1,D,E} to B and the chain continues until A receives the token again.
At that time A optionally sends out Z{A-ldr, A-src, mbrs-A,B,C,C1,D,E, confirmed} to A,B,C,C1,D,E
To ensure that the view gets implemented at all nodes at the same time, A will send out a VIEW_CONF message, this is the 'confirmed' message that is optional above.
Ideally, the interceptor below this one would be the TcpFailureDetector to ensure correct memberships
The example above, of course can be simplified with a finite statemachine:
But I suck at writing state machines, my head gets all confused. One day I will document this algorithm though.
Maybe I'll do a state diagram :)
Modifier and Type | Class and Description |
---|---|
static class |
NonBlockingCoordinator.CoordinationEvent |
static class |
NonBlockingCoordinator.CoordinationMessage |
ChannelInterceptor.InterceptorEvent
Modifier and Type | Field and Description |
---|---|
protected static byte[] |
COORD_ALIVE
Alive message
|
protected static byte[] |
COORD_CONF
Coordination confirmation, for blocking installations
|
protected static byte[] |
COORD_HEADER
header for a coordination message
|
protected static byte[] |
COORD_REQUEST
Coordination request
|
protected java.util.concurrent.atomic.AtomicBoolean |
coordMsgReceived |
protected java.lang.Object |
electionMutex |
protected Membership |
membership
Our nonblocking membership
|
protected static StringManager |
sm |
protected boolean |
started |
protected int |
startsvc |
protected Membership |
suggestedView |
protected UniqueId |
suggestedviewId
indicates that we are running an election
and this is the one we are running
|
protected Membership |
view
Our current view
|
protected UniqueId |
viewId
Out current viewId
|
protected long |
waitForCoordMsgTimeout
Time to wait for coordination timeout
|
optionFlag
Constructor and Description |
---|
NonBlockingCoordinator() |
Modifier and Type | Method and Description |
---|---|
protected boolean |
alive(Member mbr) |
ChannelData |
createData(NonBlockingCoordinator.CoordinationMessage msg,
Member local) |
void |
fireInterceptorEvent(ChannelInterceptor.InterceptorEvent event) |
Member |
getCoordinator()
Returns coordinator if one is available
|
Member |
getLocalMember(boolean incAlive)
Return the member that represents this node.
|
Member |
getMember(Member mbr)
Intercepts the
Channel.getMember(Member) method |
Member[] |
getMembers()
Get all current cluster members
|
Member[] |
getView() |
UniqueId |
getViewId() |
protected void |
halt()
Block in/out messages while a election is going on
|
protected void |
handleMyToken(Member local,
NonBlockingCoordinator.CoordinationMessage msg,
Membership merged) |
protected void |
handleOtherToken(Member local,
NonBlockingCoordinator.CoordinationMessage msg,
Membership merged) |
protected void |
handleToken(NonBlockingCoordinator.CoordinationMessage msg,
Membership merged) |
protected void |
handleViewConf(NonBlockingCoordinator.CoordinationMessage msg,
Membership merged) |
protected boolean |
hasHigherPriority(Member[] complete,
Member[] local) |
boolean |
hasMembers()
has members
|
void |
heartbeat()
The
heartbeat() method gets invoked periodically
to allow interceptors to clean up resources, time out object and
perform actions that are unrelated to sending/receiving data. |
boolean |
isCoordinator() |
boolean |
isHighest() |
protected boolean |
isViewConf(NonBlockingCoordinator.CoordinationMessage msg) |
void |
memberAdded(Member member)
A member was added to the group
|
void |
memberAdded(Member member,
boolean elect) |
void |
memberDisappeared(Member member)
A member was removed from the group
If the member left voluntarily, the Member.getCommand will contain the Member.SHUTDOWN_PAYLOAD data |
protected Membership |
mergeOnArrive(NonBlockingCoordinator.CoordinationMessage msg) |
void |
messageReceived(ChannelMessage msg)
the
messageReceived is invoked when a message is received. |
protected void |
processCoordMessage(NonBlockingCoordinator.CoordinationMessage msg) |
protected void |
release()
Release lock for in/out messages election is completed
|
protected void |
sendElectionMsg(Member local,
Member next,
NonBlockingCoordinator.CoordinationMessage msg) |
protected void |
sendElectionMsgToNextInline(Member local,
NonBlockingCoordinator.CoordinationMessage msg) |
void |
sendMessage(Member[] destination,
ChannelMessage msg,
InterceptorPayload payload)
The
sendMessage method is called when a message is being sent to one more destinations. |
protected void |
setupMembership() |
void |
start(int svc)
Starts up the channel.
|
void |
startElection(boolean force) |
void |
stop(int svc)
Shuts down the channel.
|
protected void |
waitForRelease()
Wait for an election to end
|
getChannel, getNext, getOptionFlag, getPrevious, okToProcess, setChannel, setNext, setOptionFlag, setPrevious
protected static final StringManager sm
protected static final byte[] COORD_HEADER
protected static final byte[] COORD_REQUEST
protected static final byte[] COORD_CONF
protected static final byte[] COORD_ALIVE
protected final long waitForCoordMsgTimeout
protected volatile Membership view
protected UniqueId viewId
protected Membership membership
protected UniqueId suggestedviewId
protected volatile Membership suggestedView
protected volatile boolean started
protected final int startsvc
protected final java.lang.Object electionMutex
protected final java.util.concurrent.atomic.AtomicBoolean coordMsgReceived
public void startElection(boolean force) throws ChannelException
ChannelException
protected void sendElectionMsg(Member local, Member next, NonBlockingCoordinator.CoordinationMessage msg) throws ChannelException
ChannelException
protected void sendElectionMsgToNextInline(Member local, NonBlockingCoordinator.CoordinationMessage msg) throws ChannelException
ChannelException
public ChannelData createData(NonBlockingCoordinator.CoordinationMessage msg, Member local)
protected boolean alive(Member mbr)
protected Membership mergeOnArrive(NonBlockingCoordinator.CoordinationMessage msg)
protected void processCoordMessage(NonBlockingCoordinator.CoordinationMessage msg) throws ChannelException
ChannelException
protected void handleToken(NonBlockingCoordinator.CoordinationMessage msg, Membership merged) throws ChannelException
ChannelException
protected void handleMyToken(Member local, NonBlockingCoordinator.CoordinationMessage msg, Membership merged) throws ChannelException
ChannelException
protected void handleOtherToken(Member local, NonBlockingCoordinator.CoordinationMessage msg, Membership merged) throws ChannelException
ChannelException
protected void handleViewConf(NonBlockingCoordinator.CoordinationMessage msg, Membership merged) throws ChannelException
ChannelException
protected boolean isViewConf(NonBlockingCoordinator.CoordinationMessage msg)
public Member getCoordinator()
public Member[] getView()
public UniqueId getViewId()
protected void halt()
protected void release()
protected void waitForRelease()
public void start(int svc) throws ChannelException
ChannelInterceptorBase
start
in interface ChannelInterceptor
start
in class ChannelInterceptorBase
svc
- int value of ChannelException
- if a startup error occurs or the service is already started.Channel
public void stop(int svc) throws ChannelException
ChannelInterceptorBase
stop
in interface ChannelInterceptor
stop
in class ChannelInterceptorBase
svc
- int value of ChannelException
- if a startup error occurs or the service is already started.Channel
public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException
ChannelInterceptor
sendMessage
method is called when a message is being sent to one more destinations.
The interceptor can modify any of the parameters and then pass on the message down the stack by
invoking getNext().sendMessage(destination,msg,payload)
getNext().sendMessage(destination,msg,payload)
sendMessage
in interface ChannelInterceptor
sendMessage
in class ChannelInterceptorBase
destination
- Member[] - the destination for this messagemsg
- ChannelMessage - the message to be sentpayload
- InterceptorPayload - the payload, carrying an error handler and future useful data, can be nullChannelException
ErrorHandler
,
InterceptorPayload
public void messageReceived(ChannelMessage msg)
ChannelInterceptor
messageReceived
is invoked when a message is received.
ChannelMessage.getAddress()
is the sender, or the reply-to address
if it has been overwritten.messageReceived
in interface ChannelInterceptor
messageReceived
in class ChannelInterceptorBase
msg
- ChannelMessagepublic void memberAdded(Member member)
MembershipListener
memberAdded
in interface MembershipListener
memberAdded
in class ChannelInterceptorBase
member
- Member - the member that was addedpublic void memberAdded(Member member, boolean elect)
public void memberDisappeared(Member member)
MembershipListener
memberDisappeared
in interface MembershipListener
memberDisappeared
in class ChannelInterceptorBase
member
- MemberMember.SHUTDOWN_PAYLOAD
public boolean isHighest()
public boolean isCoordinator()
public void heartbeat()
ChannelInterceptor
heartbeat()
method gets invoked periodically
to allow interceptors to clean up resources, time out object and
perform actions that are unrelated to sending/receiving data.heartbeat
in interface ChannelInterceptor
heartbeat
in interface Heartbeat
heartbeat
in class ChannelInterceptorBase
public boolean hasMembers()
hasMembers
in interface ChannelInterceptor
hasMembers
in class ChannelInterceptorBase
Channel.hasMembers()
public Member[] getMembers()
getMembers
in interface ChannelInterceptor
getMembers
in class ChannelInterceptorBase
Channel.getMembers()
public Member getMember(Member mbr)
ChannelInterceptor
Channel.getMember(Member)
methodgetMember
in interface ChannelInterceptor
getMember
in class ChannelInterceptorBase
mbr
- MemberChannel.getMember(Member)
public Member getLocalMember(boolean incAlive)
getLocalMember
in interface ChannelInterceptor
getLocalMember
in class ChannelInterceptorBase
incAlive
- booleanChannel.getLocalMember(boolean)
protected void setupMembership()
public void fireInterceptorEvent(ChannelInterceptor.InterceptorEvent event)
fireInterceptorEvent
in interface ChannelInterceptor
fireInterceptorEvent
in class ChannelInterceptorBase
Copyright © 2000-2018 Apache Software Foundation. All Rights Reserved.