rdev/gossip-revisited
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Apply formatting

Browse Source
This commit is contained in:
Edward Capriolo
2015-01-17 18:21:57 -05:00
parent 4506fe4a79
commit 17f1ad6f4f
14 changed files with 811 additions and 725 deletions
Download Patch File Download Diff File Expand all files Collapse all files

85
src/main/java/com/google/code/gossip/manager/ActiveGossipThread.java
View File

@ -8,51 +8,52 @@ import com.google.code.gossip.GossipService;
import com.google.code.gossip.LocalGossipMember;
/**
* [The active thread: periodically send gossip request.]
* The class handles gossiping the membership list.
* This information is important to maintaining a common
* state among all the nodes, and is important for detecting
* failures.
* [The active thread: periodically send gossip request.] The class handles gossiping the membership
* list. This information is important to maintaining a common state among all the nodes, and is
* important for detecting failures.
*/
abstract public class ActiveGossipThread implements Runnable {
private GossipManager _gossipManager;
private final AtomicBoolean _keepRunning;
public ActiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
_keepRunning = new AtomicBoolean(true);
}
private GossipManager _gossipManager;
@Override
public void run() {
while(_keepRunning.get()) {
try {
TimeUnit.MILLISECONDS.sleep(_gossipManager.getSettings().getGossipInterval());
sendMembershipList(_gossipManager.getMyself(), _gossipManager.getMemberList());
} catch (InterruptedException e) {
GossipService.LOGGER.error(e);
_keepRunning.set(false);
}
}
shutdown();
}
public void shutdown(){
_keepRunning.set(false);
}
/**
* Performs the sending of the membership list, after we have
* incremented our own heartbeat.
*/
abstract protected void sendMembershipList(LocalGossipMember me, ArrayList<LocalGossipMember> memberList);
private final AtomicBoolean _keepRunning;
/**
* Abstract method which should be implemented by a subclass.
* This method should return a member of the list to gossip with.
* @param memberList The list of members which are stored in the local list of members.
* @return The chosen LocalGossipMember to gossip with.
*/
abstract protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList);
public ActiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
_keepRunning = new AtomicBoolean(true);
}
@Override
public void run() {
while (_keepRunning.get()) {
try {
TimeUnit.MILLISECONDS.sleep(_gossipManager.getSettings().getGossipInterval());
sendMembershipList(_gossipManager.getMyself(), _gossipManager.getMemberList());
} catch (InterruptedException e) {
GossipService.LOGGER.error(e);
_keepRunning.set(false);
}
}
shutdown();
}
public void shutdown() {
_keepRunning.set(false);
}
/**
* Performs the sending of the membership list, after we have incremented our own heartbeat.
*/
abstract protected void sendMembershipList(LocalGossipMember me,
ArrayList<LocalGossipMember> memberList);
/**
* Abstract method which should be implemented by a subclass. This method should return a member
* of the list to gossip with.
*
* @param memberList
* The list of members which are stored in the local list of members.
* @return The chosen LocalGossipMember to gossip with.
*/
abstract protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList);
}

233
src/main/java/com/google/code/gossip/manager/GossipManager.java
View File

@ -16,142 +16,147 @@ import com.google.code.gossip.GossipSettings;
import com.google.code.gossip.LocalGossipMember;
public abstract class GossipManager extends Thread implements NotificationListener {
/** The maximal number of bytes the packet with the GOSSIP may be. (Default is 100 kb) */
public static final int MAX_PACKET_SIZE = 102400;
/** The list of members which are in the gossip group (not including myself). */
private ArrayList<LocalGossipMember> _memberList;
/** The maximal number of bytes the packet with the GOSSIP may be. (Default is 100 kb) */
public static final int MAX_PACKET_SIZE = 102400;
/** The list of members which are known to be dead. */
private ArrayList<LocalGossipMember> _deadList;
/** The member I am representing. */
private LocalGossipMember _me;
/** The settings for gossiping. */
private GossipSettings _settings;
/** A boolean whether the gossip service should keep running. */
private AtomicBoolean _gossipServiceRunning;
/** The list of members which are in the gossip group (not including myself). */
private ArrayList<LocalGossipMember> _memberList;
/** The list of members which are known to be dead. */
private ArrayList<LocalGossipMember> _deadList;
/** The member I am representing. */
private LocalGossipMember _me;
/** The settings for gossiping. */
private GossipSettings _settings;
/** A boolean whether the gossip service should keep running. */
private AtomicBoolean _gossipServiceRunning;
/** A ExecutorService used for executing the active and passive gossip threads. */
private ExecutorService _gossipThreadExecutor;
private Class<? extends PassiveGossipThread> _passiveGossipThreadClass;
private PassiveGossipThread passiveGossipThread;
private Class<? extends ActiveGossipThread> _activeGossipThreadClass;
private ActiveGossipThread activeGossipThread;
/** A ExecutorService used for executing the active and passive gossip threads. */
private ExecutorService _gossipThreadExecutor;
private Class<? extends PassiveGossipThread> _passiveGossipThreadClass;
private PassiveGossipThread passiveGossipThread;
private Class<? extends ActiveGossipThread> _activeGossipThreadClass;
private ActiveGossipThread activeGossipThread;
public GossipManager(Class<? extends PassiveGossipThread> passiveGossipThreadClass,
Class<? extends ActiveGossipThread> activeGossipThreadClass, String address, int port,
String id, GossipSettings settings, ArrayList<GossipMember> gossipMembers) {
_passiveGossipThreadClass = passiveGossipThreadClass;
_activeGossipThreadClass = activeGossipThreadClass;
_settings = settings;
_me = new LocalGossipMember(address, port, id, 0, this, settings.getCleanupInterval());
_memberList = new ArrayList<LocalGossipMember>();
_deadList = new ArrayList<LocalGossipMember>();
for (GossipMember startupMember : gossipMembers) {
if (!startupMember.equals(_me)) {
_passiveGossipThreadClass = passiveGossipThreadClass;
_activeGossipThreadClass = activeGossipThreadClass;
_settings = settings;
_me = new LocalGossipMember(address, port, id, 0, this, settings.getCleanupInterval());
_memberList = new ArrayList<LocalGossipMember>();
_deadList = new ArrayList<LocalGossipMember>();
for (GossipMember startupMember : gossipMembers) {
if (!startupMember.equals(_me)) {
LocalGossipMember member = new LocalGossipMember(startupMember.getHost(),
startupMember.getPort(), startupMember.getId(), 0, this,
settings.getCleanupInterval());
_memberList.add(member);
GossipService.LOGGER.debug(member);
}
}
_gossipServiceRunning = new AtomicBoolean(true);
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
public void run() {
GossipService.LOGGER.info("Service has been shutdown...");
}
}));
}
/**
* All timers associated with a member will trigger this method when it goes
* off. The timer will go off if we have not heard from this member in
* <code> _settings.T_CLEANUP </code> time.
*/
@Override
public void handleNotification(Notification notification, Object handback) {
LocalGossipMember deadMember = (LocalGossipMember) notification.getUserData();
GossipService.LOGGER.info("Dead member detected: " + deadMember);
synchronized (this._memberList) {
this._memberList.remove(deadMember);
}
synchronized (this._deadList) {
this._deadList.add(deadMember);
}
}
_memberList.add(member);
GossipService.LOGGER.debug(member);
}
}
public GossipSettings getSettings() {
return _settings;
}
_gossipServiceRunning = new AtomicBoolean(true);
Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
public void run() {
GossipService.LOGGER.info("Service has been shutdown...");
}
}));
}
/**
* Get a clone of the memberlist.
* @return
*/
public ArrayList<LocalGossipMember> getMemberList() {
return _memberList;
}
public LocalGossipMember getMyself() {
return _me;
}
/**
* All timers associated with a member will trigger this method when it goes off. The timer will
* go off if we have not heard from this member in <code> _settings.T_CLEANUP </code> time.
*/
@Override
public void handleNotification(Notification notification, Object handback) {
LocalGossipMember deadMember = (LocalGossipMember) notification.getUserData();
GossipService.LOGGER.info("Dead member detected: " + deadMember);
synchronized (this._memberList) {
this._memberList.remove(deadMember);
}
synchronized (this._deadList) {
this._deadList.add(deadMember);
}
}
public ArrayList<LocalGossipMember> getDeadList() {
return _deadList;
}
/**
* Starts the client. Specifically, start the various cycles for this protocol.
* Start the gossip thread and start the receiver thread.
* @throws InterruptedException
*/
public void run() {
for (LocalGossipMember member : _memberList) {
if (member != _me) {
member.startTimeoutTimer();
}
}
public GossipSettings getSettings() {
return _settings;
}
/**
* Get a clone of the memberlist.
*
* @return
*/
public ArrayList<LocalGossipMember> getMemberList() {
return _memberList;
}
public LocalGossipMember getMyself() {
return _me;
}
public ArrayList<LocalGossipMember> getDeadList() {
return _deadList;
}
/**
* Starts the client. Specifically, start the various cycles for this protocol. Start the gossip
* thread and start the receiver thread.
*
* @throws InterruptedException
*/
public void run() {
for (LocalGossipMember member : _memberList) {
if (member != _me) {
member.startTimeoutTimer();
}
}
_gossipThreadExecutor = Executors.newCachedThreadPool();
try {
passiveGossipThread = _passiveGossipThreadClass.getConstructor(GossipManager.class).newInstance(this);
passiveGossipThread = _passiveGossipThreadClass.getConstructor(GossipManager.class)
.newInstance(this);
_gossipThreadExecutor.execute(passiveGossipThread);
activeGossipThread = _activeGossipThreadClass.getConstructor(GossipManager.class).newInstance(this);
activeGossipThread = _activeGossipThreadClass.getConstructor(GossipManager.class)
.newInstance(this);
_gossipThreadExecutor.execute(activeGossipThread);
} catch (InstantiationException | IllegalAccessException | IllegalArgumentException
| InvocationTargetException | NoSuchMethodException | SecurityException e1) {
throw new RuntimeException(e1);
}
GossipService.LOGGER.info("The GossipService is started.");
while(_gossipServiceRunning.get()) {
try {
//TODO
TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException e) {
GossipService.LOGGER.info("The GossipClient was interrupted.");
}
}
}
/**
* Shutdown the gossip service.
*/
public void shutdown() {
_gossipThreadExecutor.shutdown();
passiveGossipThread.shutdown();
activeGossipThread.shutdown();
try {
GossipService.LOGGER.info("The GossipService is started.");
while (_gossipServiceRunning.get()) {
try {
// TODO
TimeUnit.MILLISECONDS.sleep(1);
} catch (InterruptedException e) {
GossipService.LOGGER.info("The GossipClient was interrupted.");
}
}
}
/**
* Shutdown the gossip service.
*/
public void shutdown() {
_gossipThreadExecutor.shutdown();
passiveGossipThread.shutdown();
activeGossipThread.shutdown();
try {
boolean result = _gossipThreadExecutor.awaitTermination(1000, TimeUnit.MILLISECONDS);
System.err.println("Terminate retuned " + result);
} catch (InterruptedException e) {
e.printStackTrace();
}
_gossipServiceRunning.set(false);
}
_gossipServiceRunning.set(false);
}
}

78
src/main/java/com/google/code/gossip/manager/PassiveGossipThread.java
View File

@ -18,36 +18,36 @@ import com.google.code.gossip.GossipService;
import com.google.code.gossip.RemoteGossipMember;
/**
* [The passive thread: reply to incoming gossip request.]
* This class handles the passive cycle, where this client
* has received an incoming message. For now, this message
* is always the membership list, but if you choose to gossip
* additional information, you will need some logic to determine
* the incoming message.
* [The passive thread: reply to incoming gossip request.] This class handles the passive cycle,
* where this client has received an incoming message. For now, this message is always the
* membership list, but if you choose to gossip additional information, you will need some logic to
* determine the incoming message.
*/
abstract public class PassiveGossipThread implements Runnable {
/** The socket used for the passive thread of the gossip service. */
private DatagramSocket _server;
private GossipManager _gossipManager;
private AtomicBoolean _keepRunning;
/** The socket used for the passive thread of the gossip service. */
private DatagramSocket _server;
public PassiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
try {
SocketAddress socketAddress = new InetSocketAddress(_gossipManager.getMyself().getHost(), _gossipManager.getMyself().getPort());
_server = new DatagramSocket(socketAddress);
GossipService.LOGGER.info("Gossip service successfully initialized on port " + _gossipManager.getMyself().getPort());
GossipService.LOGGER.debug("I am " + _gossipManager.getMyself());
} catch (SocketException ex) {
GossipService.LOGGER.error(ex);
_server = null;
throw new RuntimeException(ex);
}
_keepRunning = new AtomicBoolean(true);
}
private GossipManager _gossipManager;
private AtomicBoolean _keepRunning;
public PassiveGossipThread(GossipManager gossipManager) {
_gossipManager = gossipManager;
try {
SocketAddress socketAddress = new InetSocketAddress(_gossipManager.getMyself().getHost(),
_gossipManager.getMyself().getPort());
_server = new DatagramSocket(socketAddress);
GossipService.LOGGER.info("Gossip service successfully initialized on port "
+ _gossipManager.getMyself().getPort());
GossipService.LOGGER.debug("I am " + _gossipManager.getMyself());
} catch (SocketException ex) {
GossipService.LOGGER.error(ex);
_server = null;
throw new RuntimeException(ex);
}
_keepRunning = new AtomicBoolean(true);
}
@Override
public void run() {
@ -127,16 +127,22 @@ abstract public class PassiveGossipThread implements Runnable {
}
shutdown();
}
public void shutdown(){
public void shutdown() {
_server.close();
}
/**
* Abstract method for merging the local and remote list.
* @param gossipManager The GossipManager for retrieving the local members and dead members list.
* @param senderMember The member who is sending this list, this could be used to send a response if the remote list contains out-dated information.
* @param remoteList The list of members known at the remote side.
*/
abstract protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember, ArrayList<GossipMember> remoteList);
/**
* Abstract method for merging the local and remote list.
*
* @param gossipManager
* The GossipManager for retrieving the local members and dead members list.
* @param senderMember
* The member who is sending this list, this could be used to send a response if the
* remote list contains out-dated information.
* @param remoteList
* The list of members known at the remote side.
*/
abstract protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember,
ArrayList<GossipMember> remoteList);
}

164
src/main/java/com/google/code/gossip/manager/impl/OnlyProcessReceivedPassiveGossipThread.java
View File

@ -11,83 +11,101 @@ import com.google.code.gossip.manager.PassiveGossipThread;
public class OnlyProcessReceivedPassiveGossipThread extends PassiveGossipThread {
public OnlyProcessReceivedPassiveGossipThread(GossipManager gossipManager) {
super(gossipManager);
}
public OnlyProcessReceivedPassiveGossipThread(GossipManager gossipManager) {
super(gossipManager);
}
/**
* Merge remote list (received from peer), and our local member list.
* Simply, we must update the heartbeats that the remote list has with
* our list. Also, some additional logic is needed to make sure we have
* not timed out a member and then immediately received a list with that
* member.
* @param remoteList
*/
protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember, ArrayList<GossipMember> remoteList) {
/**
* Merge remote list (received from peer), and our local member list. Simply, we must update the
* heartbeats that the remote list has with our list. Also, some additional logic is needed to
* make sure we have not timed out a member and then immediately received a list with that member.
*
* @param remoteList
*/
protected void mergeLists(GossipManager gossipManager, RemoteGossipMember senderMember,
ArrayList<GossipMember> remoteList) {
synchronized (gossipManager.getDeadList()) {
synchronized (gossipManager.getDeadList()) {
synchronized (gossipManager.getMemberList()) {
synchronized (gossipManager.getMemberList()) {
for (GossipMember remoteMember : remoteList) {
// Skip myself. We don't want ourselves in the local member list.
if (!remoteMember.equals(gossipManager.getMyself())) {
if (gossipManager.getMemberList().contains(remoteMember)) {
GossipService.LOGGER.debug("The local list already contains the remote member (" + remoteMember + ").");
// The local memberlist contains the remote member.
LocalGossipMember localMember = gossipManager.getMemberList().get(gossipManager.getMemberList().indexOf(remoteMember));
for (GossipMember remoteMember : remoteList) {
// Skip myself. We don't want ourselves in the local member list.
if (!remoteMember.equals(gossipManager.getMyself())) {
if (gossipManager.getMemberList().contains(remoteMember)) {
GossipService.LOGGER.debug("The local list already contains the remote member ("
+ remoteMember + ").");
// The local memberlist contains the remote member.
LocalGossipMember localMember = gossipManager.getMemberList().get(
gossipManager.getMemberList().indexOf(remoteMember));
// Let's synchronize it's heartbeat.
if (remoteMember.getHeartbeat() > localMember.getHeartbeat()) {
// update local list with latest heartbeat
localMember.setHeartbeat(remoteMember.getHeartbeat());
// and reset the timeout of that member
localMember.resetTimeoutTimer();
}
// TODO: Otherwise, should we inform the other when the heartbeat is already higher?
} else {
// The local list does not contain the remote member.
GossipService.LOGGER.debug("The local list does not contain the remote member (" + remoteMember + ").");
// Let's synchronize it's heartbeat.
if (remoteMember.getHeartbeat() > localMember.getHeartbeat()) {
// update local list with latest heartbeat
localMember.setHeartbeat(remoteMember.getHeartbeat());
// and reset the timeout of that member
localMember.resetTimeoutTimer();
}
// TODO: Otherwise, should we inform the other when the heartbeat is already higher?
} else {
// The local list does not contain the remote member.
GossipService.LOGGER.debug("The local list does not contain the remote member ("
+ remoteMember + ").");
// The remote member is either brand new, or a previously declared dead member.
// If its dead, check the heartbeat because it may have come back from the dead.
if (gossipManager.getDeadList().contains(remoteMember)) {
// The remote member is known here as a dead member.
GossipService.LOGGER.debug("The remote member is known here as a dead member.");
LocalGossipMember localDeadMember = gossipManager.getDeadList().get(
gossipManager.getDeadList().indexOf(remoteMember));
// If a member is restarted the heartbeat will restart from 1, so we should check
// that here.
// So a member can become from the dead when it is either larger than a previous
// heartbeat (due to network failure)
// or when the heartbeat is 1 (after a restart of the service).
// TODO: What if the first message of a gossip service is sent to a dead node? The
// second member will receive a heartbeat of two.
// TODO: The above does happen. Maybe a special message for a revived member?
// TODO: Or maybe when a member is declared dead for more than
// _settings.getCleanupInterval() ms, reset the heartbeat to 0.
// It will then accept a revived member.
// The above is now handle by checking whether the heartbeat differs
// _settings.getCleanupInterval(), it must be restarted.
if (remoteMember.getHeartbeat() == 1
|| ((localDeadMember.getHeartbeat() - remoteMember.getHeartbeat()) * -1) > (gossipManager
.getSettings().getCleanupInterval() / 1000)
|| remoteMember.getHeartbeat() > localDeadMember.getHeartbeat()) {
GossipService.LOGGER
.debug("The remote member is back from the dead. We will remove it from the dead list and add it as a new member.");
// The remote member is back from the dead.
// Remove it from the dead list.
gossipManager.getDeadList().remove(localDeadMember);
// Add it as a new member and add it to the member list.
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(),
remoteMember.getPort(), remoteMember.getId(),
remoteMember.getHeartbeat(), gossipManager, gossipManager.getSettings()
.getCleanupInterval());
gossipManager.getMemberList().add(newLocalMember);
newLocalMember.startTimeoutTimer();
GossipService.LOGGER.info("Removed remote member " + remoteMember.getAddress()
+ " from dead list and added to local member list.");
}
} else {
// Brand spanking new member - welcome.
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(),
remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(),
gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.getMemberList().add(newLocalMember);
newLocalMember.startTimeoutTimer();
GossipService.LOGGER.info("Added new remote member " + remoteMember.getAddress()
+ " to local member list.");
}
}
}
}
}
}
}
// The remote member is either brand new, or a previously declared dead member.
// If its dead, check the heartbeat because it may have come back from the dead.
if (gossipManager.getDeadList().contains(remoteMember)) {
// The remote member is known here as a dead member.
GossipService.LOGGER.debug("The remote member is known here as a dead member.");
LocalGossipMember localDeadMember = gossipManager.getDeadList().get(gossipManager.getDeadList().indexOf(remoteMember));
// If a member is restarted the heartbeat will restart from 1, so we should check that here.
// So a member can become from the dead when it is either larger than a previous heartbeat (due to network failure)
// or when the heartbeat is 1 (after a restart of the service).
// TODO: What if the first message of a gossip service is sent to a dead node? The second member will receive a heartbeat of two.
// TODO: The above does happen. Maybe a special message for a revived member?
// TODO: Or maybe when a member is declared dead for more than _settings.getCleanupInterval() ms, reset the heartbeat to 0.
// It will then accept a revived member.
// The above is now handle by checking whether the heartbeat differs _settings.getCleanupInterval(), it must be restarted.
if (remoteMember.getHeartbeat() == 1
|| ((localDeadMember.getHeartbeat() - remoteMember.getHeartbeat()) * -1) > (gossipManager.getSettings().getCleanupInterval() / 1000)
|| remoteMember.getHeartbeat() > localDeadMember.getHeartbeat()) {
GossipService.LOGGER.debug("The remote member is back from the dead. We will remove it from the dead list and add it as a new member.");
// The remote member is back from the dead.
// Remove it from the dead list.
gossipManager.getDeadList().remove(localDeadMember);
// Add it as a new member and add it to the member list.
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(), remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(), gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.getMemberList().add(newLocalMember);
newLocalMember.startTimeoutTimer();
GossipService.LOGGER.info("Removed remote member " + remoteMember.getAddress() + " from dead list and added to local member list.");
}
} else {
// Brand spanking new member - welcome.
LocalGossipMember newLocalMember = new LocalGossipMember(remoteMember.getHost(), remoteMember.getPort(), remoteMember.getId(), remoteMember.getHeartbeat(), gossipManager, gossipManager.getSettings().getCleanupInterval());
gossipManager.getMemberList().add(newLocalMember);
newLocalMember.startTimeoutTimer();
GossipService.LOGGER.info("Added new remote member " + remoteMember.getAddress() + " to local member list.");
}
}
}
}
}
}
}
}

143
src/main/java/com/google/code/gossip/manager/impl/SendMembersActiveGossipThread.java
View File

@ -16,79 +16,80 @@ import com.google.code.gossip.manager.GossipManager;
abstract public class SendMembersActiveGossipThread extends ActiveGossipThread {
public SendMembersActiveGossipThread(GossipManager gossipManager) {
super(gossipManager);
}
/**
* Performs the sending of the membership list, after we have
* incremented our own heartbeat.
*/
protected void sendMembershipList(LocalGossipMember me, ArrayList<LocalGossipMember> memberList) {
GossipService.LOGGER.debug("Send sendMembershipList() is called.");
public SendMembersActiveGossipThread(GossipManager gossipManager) {
super(gossipManager);
}
// Increase the heartbeat of myself by 1.
me.setHeartbeat(me.getHeartbeat() + 1);
synchronized (memberList) {
try {
LocalGossipMember member = selectPartner(memberList);
/**
* Performs the sending of the membership list, after we have incremented our own heartbeat.
*/
protected void sendMembershipList(LocalGossipMember me, ArrayList<LocalGossipMember> memberList) {
GossipService.LOGGER.debug("Send sendMembershipList() is called.");
// Increase the heartbeat of myself by 1.
me.setHeartbeat(me.getHeartbeat() + 1);
synchronized (memberList) {
try {
LocalGossipMember member = selectPartner(memberList);
if (member != null) {
InetAddress dest = InetAddress.getByName(member.getHost());
// Create a StringBuffer for the JSON message.
JSONArray jsonArray = new JSONArray();
GossipService.LOGGER.debug("Sending memberlist to " + dest + ":" + member.getPort());
GossipService.LOGGER.debug("---------------------");
// First write myself, append the JSON representation of the member to the buffer.
jsonArray.put(me.toJSONObject());
GossipService.LOGGER.debug(me);
// Then write the others.
for (int i = 0; i < memberList.size(); i++) {
LocalGossipMember other = memberList.get(i);
// Append the JSON representation of the member to the buffer.
jsonArray.put(other.toJSONObject());
GossipService.LOGGER.debug(other);
}
GossipService.LOGGER.debug("---------------------");
// Write the objects to a byte array.
byte[] json_bytes = jsonArray.toString().getBytes();
int packet_length = json_bytes.length;
if (packet_length < GossipManager.MAX_PACKET_SIZE) {
// Convert the packet length to the byte representation of the int.
byte[] length_bytes = new byte[4];
length_bytes[0] = (byte) (packet_length >> 24);
length_bytes[1] = (byte) ((packet_length << 8) >> 24);
length_bytes[2] = (byte) ((packet_length << 16) >> 24);
length_bytes[3] = (byte) ((packet_length << 24) >> 24);
GossipService.LOGGER.debug("Sending message (" + packet_length + " bytes): "
+ jsonArray.toString());
if (member != null) {
InetAddress dest = InetAddress.getByName(member.getHost());
// Create a StringBuffer for the JSON message.
JSONArray jsonArray = new JSONArray();
GossipService.LOGGER.debug("Sending memberlist to " + dest + ":" + member.getPort());
GossipService.LOGGER.debug("---------------------");
// First write myself, append the JSON representation of the member to the buffer.
jsonArray.put(me.toJSONObject());
GossipService.LOGGER.debug(me);
// Then write the others.
for (int i=0; i<memberList.size(); i++) {
LocalGossipMember other = memberList.get(i);
// Append the JSON representation of the member to the buffer.
jsonArray.put(other.toJSONObject());
GossipService.LOGGER.debug(other);
}
GossipService.LOGGER.debug("---------------------");
// Write the objects to a byte array.
byte[] json_bytes = jsonArray.toString().getBytes();
int packet_length = json_bytes.length;
if (packet_length < GossipManager.MAX_PACKET_SIZE) {
// Convert the packet length to the byte representation of the int.
byte[] length_bytes = new byte[4];
length_bytes[0] =(byte)( packet_length >> 24 );
length_bytes[1] =(byte)( (packet_length << 8) >> 24 );
length_bytes[2] =(byte)( (packet_length << 16) >> 24 );
length_bytes[3] =(byte)( (packet_length << 24) >> 24 );
GossipService.LOGGER.debug("Sending message ("+packet_length+" bytes): " + jsonArray.toString());
ByteBuffer byteBuffer = ByteBuffer.allocate(4 + json_bytes.length);
byteBuffer.put(length_bytes);
byteBuffer.put(json_bytes);
byte[] buf = byteBuffer.array();
DatagramSocket socket = new DatagramSocket();
DatagramPacket datagramPacket = new DatagramPacket(buf, buf.length, dest, member.getPort());
socket.send(datagramPacket);
socket.close();
} else {
GossipService.LOGGER.error("The length of the to be send message is too large (" + packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ").");
}
}
byteBuffer.put(length_bytes);
byteBuffer.put(json_bytes);
byte[] buf = byteBuffer.array();
} catch (IOException e1) {
e1.printStackTrace();
}
}
}
DatagramSocket socket = new DatagramSocket();
DatagramPacket datagramPacket = new DatagramPacket(buf, buf.length, dest,
member.getPort());
socket.send(datagramPacket);
socket.close();
} else {
GossipService.LOGGER.error("The length of the to be send message is too large ("
+ packet_length + " > " + GossipManager.MAX_PACKET_SIZE + ").");
}
}
} catch (IOException e1) {
e1.printStackTrace();
}
}
}
}

48
src/main/java/com/google/code/gossip/manager/random/RandomActiveGossipThread.java
View File

@ -9,30 +9,30 @@ import com.google.code.gossip.manager.GossipManager;
import com.google.code.gossip.manager.impl.SendMembersActiveGossipThread;
public class RandomActiveGossipThread extends SendMembersActiveGossipThread {
/** The Random used for choosing a member to gossip with. */
private Random _random;
public RandomActiveGossipThread(GossipManager gossipManager) {
super(gossipManager);
_random = new Random();
}
/**
* [The selectToSend() function.]
* Find a random peer from the local membership list.
* In the case where this client is the only member in the list, this method will return null.
* @return Member random member if list is greater than 1, null otherwise
*/
protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList) {
LocalGossipMember member = null;
if (memberList.size() > 0) {
int randomNeighborIndex = _random.nextInt(memberList.size());
member = memberList.get(randomNeighborIndex);
} else {
GossipService.LOGGER.debug("I am alone in this world.");
}
return member;
}
/** The Random used for choosing a member to gossip with. */
private Random _random;
public RandomActiveGossipThread(GossipManager gossipManager) {
super(gossipManager);
_random = new Random();
}
/**
* [The selectToSend() function.] Find a random peer from the local membership list. In the case
* where this client is the only member in the list, this method will return null.
*
* @return Member random member if list is greater than 1, null otherwise
*/
protected LocalGossipMember selectPartner(ArrayList<LocalGossipMember> memberList) {
LocalGossipMember member = null;
if (memberList.size() > 0) {
int randomNeighborIndex = _random.nextInt(memberList.size());
member = memberList.get(randomNeighborIndex);
} else {
GossipService.LOGGER.debug("I am alone in this world.");
}
return member;
}
}

8
src/main/java/com/google/code/gossip/manager/random/RandomGossipManager.java
View File

@ -8,7 +8,9 @@ import com.google.code.gossip.manager.GossipManager;
import com.google.code.gossip.manager.impl.OnlyProcessReceivedPassiveGossipThread;
public class RandomGossipManager extends GossipManager {
public RandomGossipManager(String address, int port, String id, GossipSettings settings, ArrayList<GossipMember> gossipMembers) {
super(OnlyProcessReceivedPassiveGossipThread.class, RandomActiveGossipThread.class, address, port, id, settings, gossipMembers);
}
public RandomGossipManager(String address, int port, String id, GossipSettings settings,
ArrayList<GossipMember> gossipMembers) {
super(OnlyProcessReceivedPassiveGossipThread.class, RandomActiveGossipThread.class, address,
port, id, settings, gossipMembers);
}
}