mirror of
https://github.com/ebean-orm/ebean.git
synced 2024-04-21 10:51:47 +00:00
No effective change - code format only
This commit is contained in:
Robin Bygrave
@@ -5,39 +5,37 @@ import java.util.List;
|
||||
|
||||
/**
|
||||
* Holds a list of ACK and RESEND messages that should be sent out.
|
||||
*
|
||||
* @author rbygrave
|
||||
*/
|
||||
public class AckResendMessages {
|
||||
|
||||
final ArrayList<Message> messages = new ArrayList<Message>();
|
||||
|
||||
public String toString() {
|
||||
return messages.toString();
|
||||
}
|
||||
|
||||
public int size() {
|
||||
return messages.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a ACK message to send.
|
||||
*/
|
||||
public void add(MessageAck ack){
|
||||
messages.add(ack);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a RESEND message to send.
|
||||
*/
|
||||
public void add(MessageResend resend){
|
||||
messages.add(resend);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return all the messages to be sent out.
|
||||
*/
|
||||
public List<Message> getMessages() {
|
||||
return messages;
|
||||
}
|
||||
final ArrayList<Message> messages = new ArrayList<Message>();
|
||||
|
||||
public String toString() {
|
||||
return messages.toString();
|
||||
}
|
||||
|
||||
public int size() {
|
||||
return messages.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a ACK message to send.
|
||||
*/
|
||||
public void add(MessageAck ack) {
|
||||
messages.add(ack);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a RESEND message to send.
|
||||
*/
|
||||
public void add(MessageResend resend) {
|
||||
messages.add(resend);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return all the messages to be sent out.
|
||||
*/
|
||||
public List<Message> getMessages() {
|
||||
return messages;
|
||||
}
|
||||
}
|
||||
|
||||
+32
-34
@@ -11,43 +11,41 @@ import java.util.List;
|
||||
* </p>
|
||||
* Thread Safety note: Object only used by McastClusterBroadcast Manager thread.
|
||||
* So Single Threaded access.
|
||||
*
|
||||
* @author rbygrave
|
||||
*/
|
||||
public class IncomingPacketsLastAck {
|
||||
|
||||
private final HashMap<String,MessageAck> lastAckMap = new HashMap<String, MessageAck>();
|
||||
private final HashMap<String, MessageAck> lastAckMap = new HashMap<String, MessageAck>();
|
||||
|
||||
public String toString() {
|
||||
return lastAckMap.values().toString();
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a member of the cluster who has left.
|
||||
*/
|
||||
public void remove(String memberHostPort) {
|
||||
lastAckMap.remove(memberHostPort);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the last Ack point for a given member of the cluster.
|
||||
*/
|
||||
public MessageAck getLastAck(String memberHostPort) {
|
||||
return lastAckMap.get(memberHostPort);
|
||||
}
|
||||
|
||||
/**
|
||||
* For the ACK messages in AckResendMessages update the
|
||||
* last Ack packetId.
|
||||
*/
|
||||
public void updateLastAck(AckResendMessages ackResendMessages) {
|
||||
List<Message> messages = ackResendMessages.getMessages();
|
||||
for (int i = 0; i < messages.size(); i++) {
|
||||
Message msg = messages.get(i);
|
||||
if (msg instanceof MessageAck){
|
||||
MessageAck lastAck = (MessageAck)msg;
|
||||
lastAckMap.put(lastAck.getToHostPort(), lastAck);
|
||||
}
|
||||
}
|
||||
public String toString() {
|
||||
return lastAckMap.values().toString();
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove a member of the cluster who has left.
|
||||
*/
|
||||
public void remove(String memberHostPort) {
|
||||
lastAckMap.remove(memberHostPort);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the last Ack point for a given member of the cluster.
|
||||
*/
|
||||
public MessageAck getLastAck(String memberHostPort) {
|
||||
return lastAckMap.get(memberHostPort);
|
||||
}
|
||||
|
||||
/**
|
||||
* For the ACK messages in AckResendMessages update the
|
||||
* last Ack packetId.
|
||||
*/
|
||||
public void updateLastAck(AckResendMessages ackResendMessages) {
|
||||
List<Message> messages = ackResendMessages.getMessages();
|
||||
for (int i = 0; i < messages.size(); i++) {
|
||||
Message msg = messages.get(i);
|
||||
if (msg instanceof MessageAck) {
|
||||
MessageAck lastAck = (MessageAck) msg;
|
||||
lastAckMap.put(lastAck.getToHostPort(), lastAck);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+217
-221
@@ -20,256 +20,252 @@ import java.util.concurrent.ConcurrentHashMap;
|
||||
* have received and RESEND messages to fill the missing packets we have
|
||||
* detected.
|
||||
* </p>
|
||||
*
|
||||
* @author rbygrave
|
||||
*
|
||||
*/
|
||||
public class IncomingPacketsProcessed {
|
||||
|
||||
private final ConcurrentHashMap<String, GotAllPoint> mapByMember = new ConcurrentHashMap<String, GotAllPoint>();
|
||||
private final ConcurrentHashMap<String, GotAllPoint> mapByMember = new ConcurrentHashMap<String, GotAllPoint>();
|
||||
|
||||
private final int maxResendIncoming;
|
||||
|
||||
public IncomingPacketsProcessed(int maxResendIncoming) {
|
||||
this.maxResendIncoming = maxResendIncoming;
|
||||
}
|
||||
|
||||
public void removeMember(String memberKey) {
|
||||
mapByMember.remove(memberKey);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return true if we should process this packet. Return false if we have
|
||||
* already processed the packet.
|
||||
*/
|
||||
public boolean isProcessPacket(String memberKey, long packetId) {
|
||||
|
||||
GotAllPoint memberPackets = getMemberPackets(memberKey);
|
||||
return memberPackets.processPacket(packetId);
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the list of ACK and RESEND messages that we should send out
|
||||
* to the other members of the cluster.
|
||||
*/
|
||||
public AckResendMessages getAckResendMessages(IncomingPacketsLastAck lastAck) {
|
||||
|
||||
// Called by the McastClusterBroadcast manager thread
|
||||
|
||||
AckResendMessages response = new AckResendMessages();
|
||||
|
||||
for (GotAllPoint member : mapByMember.values()) {
|
||||
|
||||
MessageAck lastAckMessage = lastAck.getLastAck(member.getMemberKey());
|
||||
|
||||
member.addAckResendMessages(response, lastAckMessage);
|
||||
}
|
||||
|
||||
return response;
|
||||
}
|
||||
|
||||
private GotAllPoint getMemberPackets(String memberKey) {
|
||||
|
||||
// This method is only called single threaded
|
||||
// by the listener thread so I'm happy that this
|
||||
// put into mapByMember is ok.
|
||||
GotAllPoint memberGotAllPoint = mapByMember.get(memberKey);
|
||||
if (memberGotAllPoint == null) {
|
||||
memberGotAllPoint = new GotAllPoint(memberKey, maxResendIncoming);
|
||||
mapByMember.put(memberKey, memberGotAllPoint);
|
||||
}
|
||||
return memberGotAllPoint;
|
||||
}
|
||||
|
||||
/**
|
||||
* Keeps track of packets received from a particular member of the cluster.
|
||||
* <p>
|
||||
* It notes the packetIds of the packets received and uses those to maintain
|
||||
* the 'gotAllPoint'. The 'gotAllPoint' is the packetId which we know we
|
||||
* received all the previous packets.
|
||||
* </p>
|
||||
*/
|
||||
public static class GotAllPoint {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(GotAllPoint.class);
|
||||
|
||||
private final String memberKey;
|
||||
private final int maxResendIncoming;
|
||||
|
||||
public IncomingPacketsProcessed(int maxResendIncoming) {
|
||||
this.maxResendIncoming = maxResendIncoming;
|
||||
}
|
||||
|
||||
public void removeMember(String memberKey) {
|
||||
mapByMember.remove(memberKey);
|
||||
|
||||
private long gotAllPoint;
|
||||
|
||||
private long gotMaxPoint;
|
||||
|
||||
/**
|
||||
* Packets received out of order.
|
||||
*/
|
||||
private final ArrayList<Long> outOfOrderList = new ArrayList<Long>();
|
||||
|
||||
private final HashMap<Long, Integer> resendCountMap = new HashMap<Long, Integer>();
|
||||
|
||||
public GotAllPoint(String memberKey, int maxResendIncoming) {
|
||||
this.memberKey = memberKey;
|
||||
this.maxResendIncoming = maxResendIncoming;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return true if we should process this packet. Return false if we have
|
||||
* already processed the packet.
|
||||
* Add ACK and RESEND messages if required.
|
||||
*/
|
||||
public boolean isProcessPacket(String memberKey, long packetId) {
|
||||
public void addAckResendMessages(AckResendMessages response, MessageAck lastAckMessage) {
|
||||
|
||||
GotAllPoint memberPackets = getMemberPackets(memberKey);
|
||||
return memberPackets.processPacket(packetId);
|
||||
synchronized (this) {
|
||||
if (lastAckMessage != null && lastAckMessage.getGotAllPacketId() >= gotAllPoint) {
|
||||
// nothing has changed
|
||||
} else {
|
||||
// ACK that we have got every packet up to gotAllPoint
|
||||
response.add(new MessageAck(memberKey, gotAllPoint));
|
||||
}
|
||||
|
||||
if (getMissingPacketCount() > 0) {
|
||||
// Ask for these Packets to be RESENT
|
||||
List<Long> missingPackets = getMissingPackets();
|
||||
response.add(new MessageResend(memberKey, missingPackets));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the list of ACK and RESEND messages that we should send out
|
||||
* to the other members of the cluster.
|
||||
*/
|
||||
public AckResendMessages getAckResendMessages(IncomingPacketsLastAck lastAck) {
|
||||
|
||||
// Called by the McastClusterBroadcast manager thread
|
||||
|
||||
AckResendMessages response = new AckResendMessages();
|
||||
|
||||
for (GotAllPoint member : mapByMember.values()) {
|
||||
|
||||
MessageAck lastAckMessage = lastAck.getLastAck(member.getMemberKey());
|
||||
|
||||
member.addAckResendMessages(response, lastAckMessage);
|
||||
}
|
||||
|
||||
return response;
|
||||
public String getMemberKey() {
|
||||
return memberKey;
|
||||
}
|
||||
|
||||
private GotAllPoint getMemberPackets(String memberKey) {
|
||||
|
||||
// This method is only called single threaded
|
||||
// by the listener thread so I'm happy that this
|
||||
// put into mapByMember is ok.
|
||||
GotAllPoint memberGotAllPoint = mapByMember.get(memberKey);
|
||||
if (memberGotAllPoint == null) {
|
||||
memberGotAllPoint = new GotAllPoint(memberKey, maxResendIncoming);
|
||||
mapByMember.put(memberKey, memberGotAllPoint);
|
||||
}
|
||||
return memberGotAllPoint;
|
||||
public long getGotAllPoint() {
|
||||
synchronized (this) {
|
||||
return gotAllPoint;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Keeps track of packets received from a particular member of the cluster.
|
||||
* <p>
|
||||
* It notes the packetIds of the packets received and uses those to maintain
|
||||
* the 'gotAllPoint'. The 'gotAllPoint' is the packetId which we know we
|
||||
* received all the previous packets.
|
||||
* </p>
|
||||
*/
|
||||
public static class GotAllPoint {
|
||||
public long getGotMaxPoint() {
|
||||
synchronized (this) {
|
||||
return gotMaxPoint;
|
||||
}
|
||||
}
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(GotAllPoint.class);
|
||||
|
||||
private final String memberKey;
|
||||
private final int maxResendIncoming;
|
||||
|
||||
private long gotAllPoint;
|
||||
|
||||
private long gotMaxPoint;
|
||||
|
||||
/**
|
||||
* Packets received out of order.
|
||||
*/
|
||||
private final ArrayList<Long> outOfOrderList = new ArrayList<Long>();
|
||||
|
||||
private final HashMap<Long,Integer> resendCountMap = new HashMap<Long,Integer>();
|
||||
|
||||
public GotAllPoint(String memberKey, int maxResendIncoming) {
|
||||
this.memberKey = memberKey;
|
||||
this.maxResendIncoming = maxResendIncoming;
|
||||
private int getMissingPacketCount() {
|
||||
if (gotMaxPoint <= gotAllPoint) {
|
||||
if (!resendCountMap.isEmpty()) {
|
||||
resendCountMap.clear();
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
return (int) (gotMaxPoint - gotAllPoint) - outOfOrderList.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Add ACK and RESEND messages if required.
|
||||
*/
|
||||
public void addAckResendMessages(AckResendMessages response, MessageAck lastAckMessage) {
|
||||
public List<Long> getMissingPackets() {
|
||||
|
||||
synchronized (this) {
|
||||
if (lastAckMessage != null && lastAckMessage.getGotAllPacketId() >= gotAllPoint) {
|
||||
// nothing has changed
|
||||
} else {
|
||||
// ACK that we have got every packet up to gotAllPoint
|
||||
response.add(new MessageAck(memberKey, gotAllPoint));
|
||||
}
|
||||
synchronized (this) {
|
||||
ArrayList<Long> missingList = new ArrayList<Long>();
|
||||
|
||||
if (getMissingPacketCount() > 0) {
|
||||
// Ask for these Packets to be RESENT
|
||||
List<Long> missingPackets = getMissingPackets();
|
||||
response.add(new MessageResend(memberKey, missingPackets));
|
||||
}
|
||||
}
|
||||
}
|
||||
// this is not particularly efficient but expecting
|
||||
// the outOfOrderList to be relatively small
|
||||
|
||||
public String getMemberKey() {
|
||||
return memberKey;
|
||||
}
|
||||
boolean lostPacket = false;
|
||||
|
||||
public long getGotAllPoint() {
|
||||
synchronized (this) {
|
||||
return gotAllPoint;
|
||||
}
|
||||
}
|
||||
|
||||
public long getGotMaxPoint() {
|
||||
synchronized (this) {
|
||||
return gotMaxPoint;
|
||||
}
|
||||
}
|
||||
|
||||
private int getMissingPacketCount() {
|
||||
if (gotMaxPoint <= gotAllPoint) {
|
||||
if (!resendCountMap.isEmpty()) {
|
||||
resendCountMap.clear();
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
return (int) (gotMaxPoint - gotAllPoint) - outOfOrderList.size();
|
||||
}
|
||||
|
||||
public List<Long> getMissingPackets() {
|
||||
|
||||
synchronized (this) {
|
||||
ArrayList<Long> missingList = new ArrayList<Long>();
|
||||
|
||||
// this is not particularly efficient but expecting
|
||||
// the outOfOrderList to be relatively small
|
||||
|
||||
boolean lostPacket = false;
|
||||
|
||||
for (long i = gotAllPoint + 1; i < gotMaxPoint; i++) {
|
||||
Long packetId = i;
|
||||
if (!outOfOrderList.contains(packetId)) {
|
||||
if (incrementResendCount(packetId)) {
|
||||
// request this packet be resent
|
||||
missingList.add(packetId);
|
||||
} else {
|
||||
lostPacket = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (lostPacket){
|
||||
checkOutOfOrderList();
|
||||
}
|
||||
|
||||
return missingList;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Return true if this packet has not yet exceeded the maxResendCount.
|
||||
*/
|
||||
private boolean incrementResendCount(Long packetId){
|
||||
Integer resendCount = resendCountMap.get(packetId);
|
||||
if (resendCount != null){
|
||||
int i = resendCount + 1;
|
||||
if (i > maxResendIncoming){
|
||||
// we are going to give up trying to get this packet now
|
||||
logger.warn("Exceeded maxResendIncoming["+maxResendIncoming+"] for packet["+packetId+"]. Giving up on requesting it.");
|
||||
resendCountMap.remove(packetId);
|
||||
outOfOrderList.add(packetId);
|
||||
return false;
|
||||
}
|
||||
resendCount = i;
|
||||
resendCountMap.put(packetId, resendCount);
|
||||
for (long i = gotAllPoint + 1; i < gotMaxPoint; i++) {
|
||||
Long packetId = i;
|
||||
if (!outOfOrderList.contains(packetId)) {
|
||||
if (incrementResendCount(packetId)) {
|
||||
// request this packet be resent
|
||||
missingList.add(packetId);
|
||||
} else {
|
||||
resendCountMap.put(packetId, ONE);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static final Integer ONE = 1;
|
||||
|
||||
public boolean processPacket(long packetId) {
|
||||
synchronized (this) {
|
||||
|
||||
if (gotAllPoint == 0) {
|
||||
gotAllPoint = packetId;
|
||||
return true;
|
||||
}
|
||||
if (packetId <= gotAllPoint) {
|
||||
// already processed this packet
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!resendCountMap.isEmpty()){
|
||||
resendCountMap.remove(Long.valueOf(packetId));
|
||||
}
|
||||
|
||||
if (packetId == gotAllPoint + 1) {
|
||||
gotAllPoint = packetId;
|
||||
} else {
|
||||
if (packetId > gotMaxPoint) {
|
||||
gotMaxPoint = packetId;
|
||||
}
|
||||
outOfOrderList.add(packetId);
|
||||
}
|
||||
checkOutOfOrderList();
|
||||
return true;
|
||||
lostPacket = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void checkOutOfOrderList() {
|
||||
|
||||
if (outOfOrderList.size() == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
boolean continueCheck;
|
||||
do {
|
||||
continueCheck = false;
|
||||
long nextPoint = gotAllPoint + 1;
|
||||
|
||||
Iterator<Long> it = outOfOrderList.iterator();
|
||||
while (it.hasNext()) {
|
||||
Long id = it.next();
|
||||
if (id == nextPoint) {
|
||||
// we found the next one in the outOfOrderList
|
||||
it.remove();
|
||||
gotAllPoint = nextPoint;
|
||||
continueCheck = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
} while (continueCheck);
|
||||
|
||||
if (lostPacket) {
|
||||
checkOutOfOrderList();
|
||||
}
|
||||
|
||||
return missingList;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Return true if this packet has not yet exceeded the maxResendCount.
|
||||
*/
|
||||
private boolean incrementResendCount(Long packetId) {
|
||||
Integer resendCount = resendCountMap.get(packetId);
|
||||
if (resendCount != null) {
|
||||
int i = resendCount + 1;
|
||||
if (i > maxResendIncoming) {
|
||||
// we are going to give up trying to get this packet now
|
||||
logger.warn("Exceeded maxResendIncoming[" + maxResendIncoming + "] for packet[" + packetId + "]. Giving up on requesting it.");
|
||||
resendCountMap.remove(packetId);
|
||||
outOfOrderList.add(packetId);
|
||||
return false;
|
||||
}
|
||||
resendCount = i;
|
||||
resendCountMap.put(packetId, resendCount);
|
||||
} else {
|
||||
resendCountMap.put(packetId, ONE);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private static final Integer ONE = 1;
|
||||
|
||||
public boolean processPacket(long packetId) {
|
||||
synchronized (this) {
|
||||
|
||||
if (gotAllPoint == 0) {
|
||||
gotAllPoint = packetId;
|
||||
return true;
|
||||
}
|
||||
if (packetId <= gotAllPoint) {
|
||||
// already processed this packet
|
||||
return false;
|
||||
}
|
||||
|
||||
if (!resendCountMap.isEmpty()) {
|
||||
resendCountMap.remove(Long.valueOf(packetId));
|
||||
}
|
||||
|
||||
if (packetId == gotAllPoint + 1) {
|
||||
gotAllPoint = packetId;
|
||||
} else {
|
||||
if (packetId > gotMaxPoint) {
|
||||
gotMaxPoint = packetId;
|
||||
}
|
||||
outOfOrderList.add(packetId);
|
||||
}
|
||||
checkOutOfOrderList();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
private void checkOutOfOrderList() {
|
||||
|
||||
if (outOfOrderList.size() == 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
boolean continueCheck;
|
||||
do {
|
||||
continueCheck = false;
|
||||
long nextPoint = gotAllPoint + 1;
|
||||
|
||||
Iterator<Long> it = outOfOrderList.iterator();
|
||||
while (it.hasNext()) {
|
||||
Long id = it.next();
|
||||
if (id == nextPoint) {
|
||||
// we found the next one in the outOfOrderList
|
||||
it.remove();
|
||||
gotAllPoint = nextPoint;
|
||||
continueCheck = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
} while (continueCheck);
|
||||
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -215,7 +215,7 @@ public class McastClusterManager implements ClusterBroadcast, Runnable {
|
||||
|
||||
if (port == 0 || addr == null) {
|
||||
String msg = "One of these Multicast settings has not been set. " + "ebean.cluster.mcast.listen.port="
|
||||
+ port + ", ebean.cluster.mcast.listen.address=" + addr;
|
||||
+ port + ", ebean.cluster.mcast.listen.address=" + addr;
|
||||
|
||||
throw new IllegalArgumentException(msg);
|
||||
}
|
||||
@@ -229,7 +229,7 @@ public class McastClusterManager implements ClusterBroadcast, Runnable {
|
||||
this.packageControl = new McastPacketControl(this, localSenderHostPort, maxResendIncoming);
|
||||
|
||||
this.listener = new McastListener(this, packageControl, port, addr, bufferSize, timeout, localSenderHostPort,
|
||||
disableLoopback, ttl, mcastBindAddress);
|
||||
disableLoopback, ttl, mcastBindAddress);
|
||||
}
|
||||
|
||||
|
||||
@@ -285,9 +285,9 @@ public class McastClusterManager implements ClusterBroadcast, Runnable {
|
||||
String lastAcks = incomingPacketsLastAck.toString();
|
||||
|
||||
return new McastStatus(currentGroupSize, outgoingPacketsCache.size(), currentPacketId, minAcked, lastAcks,
|
||||
totalTxnEventsSent, totalTxnEventsReceived, totalPacketsSent, totalPacketsResent,
|
||||
totalPacketsReceived,
|
||||
totalBytesSent, totalBytesResent, totalBytesReceived);
|
||||
totalTxnEventsSent, totalTxnEventsReceived, totalPacketsSent, totalPacketsResent,
|
||||
totalPacketsReceived,
|
||||
totalBytesSent, totalBytesResent, totalBytesReceived);
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,22 +1,3 @@
|
||||
/**
|
||||
* Copyright (C) 2009 Authors
|
||||
*
|
||||
* This file is part of Ebean.
|
||||
*
|
||||
* Ebean is free software; you can redistribute it and/or modify it
|
||||
* under the terms of the GNU Lesser General Public License as published by
|
||||
* the Free Software Foundation; either version 2.1 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* Ebean is distributed in the hope that it will be useful, but
|
||||
* WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Lesser General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Lesser General Public License
|
||||
* along with Ebean; if not, write to the Free Software Foundation, Inc.,
|
||||
* 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
|
||||
*/
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import com.avaje.ebeaninternal.api.SpiEbeanServer;
|
||||
@@ -36,223 +17,221 @@ import java.net.MulticastSocket;
|
||||
|
||||
/**
|
||||
* Listens for Incoming packets.
|
||||
*
|
||||
* @author rbygrave
|
||||
*/
|
||||
public class McastListener implements Runnable {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(McastListener.class);
|
||||
private static final Logger logger = LoggerFactory.getLogger(McastListener.class);
|
||||
|
||||
private final McastClusterManager owner;
|
||||
|
||||
private final McastPacketControl packetControl;
|
||||
|
||||
private final MulticastSocket sock;
|
||||
private final McastClusterManager owner;
|
||||
|
||||
private final Thread listenerThread;
|
||||
private final McastPacketControl packetControl;
|
||||
|
||||
private final String localSenderHostPort;
|
||||
private final MulticastSocket sock;
|
||||
|
||||
private final InetAddress group;
|
||||
private final Thread listenerThread;
|
||||
|
||||
private DatagramPacket pack;
|
||||
private final String localSenderHostPort;
|
||||
|
||||
private final byte[] receiveBuffer;
|
||||
private final InetAddress group;
|
||||
|
||||
private volatile boolean shutdown;
|
||||
private volatile boolean shutdownComplete;
|
||||
|
||||
private long totalPacketsReceived;
|
||||
private long totalBytesReceived;
|
||||
private long totalTxnEventsReceived;
|
||||
|
||||
public McastListener(McastClusterManager owner, McastPacketControl packetControl, int port, String address,
|
||||
int bufferSize, int timeout, String localSenderHostPort,
|
||||
boolean disableLoopback, int ttl, InetAddress mcastBindAddress) {
|
||||
private DatagramPacket pack;
|
||||
|
||||
this.owner = owner;
|
||||
this.packetControl = packetControl;
|
||||
this.localSenderHostPort = localSenderHostPort;
|
||||
this.receiveBuffer = new byte[bufferSize];
|
||||
this.listenerThread = new Thread(this, "EbeanClusterMcastListener");
|
||||
private final byte[] receiveBuffer;
|
||||
|
||||
String msg = "Cluster Multicast Listening address["+address+"] port["+port+"] disableLoopback["+disableLoopback+"]";
|
||||
if (ttl >= 0){
|
||||
msg +=" ttl["+ttl+"]";
|
||||
}
|
||||
if (mcastBindAddress != null){
|
||||
msg += " mcastBindAddress["+mcastBindAddress+"]";
|
||||
}
|
||||
logger.info(msg);
|
||||
private volatile boolean shutdown;
|
||||
private volatile boolean shutdownComplete;
|
||||
|
||||
try {
|
||||
this.group = InetAddress.getByName(address);
|
||||
this.sock = new MulticastSocket(port);
|
||||
this.sock.setSoTimeout(timeout);
|
||||
private long totalPacketsReceived;
|
||||
private long totalBytesReceived;
|
||||
private long totalTxnEventsReceived;
|
||||
|
||||
if (disableLoopback){
|
||||
sock.setLoopbackMode(true);
|
||||
public McastListener(McastClusterManager owner, McastPacketControl packetControl, int port, String address,
|
||||
int bufferSize, int timeout, String localSenderHostPort,
|
||||
boolean disableLoopback, int ttl, InetAddress mcastBindAddress) {
|
||||
|
||||
this.owner = owner;
|
||||
this.packetControl = packetControl;
|
||||
this.localSenderHostPort = localSenderHostPort;
|
||||
this.receiveBuffer = new byte[bufferSize];
|
||||
this.listenerThread = new Thread(this, "EbeanClusterMcastListener");
|
||||
|
||||
String msg = "Cluster Multicast Listening address[" + address + "] port[" + port + "] disableLoopback[" + disableLoopback + "]";
|
||||
if (ttl >= 0) {
|
||||
msg += " ttl[" + ttl + "]";
|
||||
}
|
||||
if (mcastBindAddress != null) {
|
||||
msg += " mcastBindAddress[" + mcastBindAddress + "]";
|
||||
}
|
||||
logger.info(msg);
|
||||
|
||||
try {
|
||||
this.group = InetAddress.getByName(address);
|
||||
this.sock = new MulticastSocket(port);
|
||||
this.sock.setSoTimeout(timeout);
|
||||
|
||||
if (disableLoopback) {
|
||||
sock.setLoopbackMode(true);
|
||||
}
|
||||
|
||||
if (mcastBindAddress != null) {
|
||||
// bind to a specific interface
|
||||
sock.setInterface(mcastBindAddress);
|
||||
}
|
||||
|
||||
if (ttl >= 0) {
|
||||
sock.setTimeToLive(ttl);
|
||||
}
|
||||
sock.setReuseAddress(true);
|
||||
pack = new DatagramPacket(receiveBuffer, receiveBuffer.length);
|
||||
sock.joinGroup(group);
|
||||
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
public void startListening() {
|
||||
this.listenerThread.setDaemon(true);
|
||||
this.listenerThread.start();
|
||||
|
||||
logger.info("Cluster Multicast Listener up and joined Group");
|
||||
}
|
||||
|
||||
/**
|
||||
* Shutdown this listener.
|
||||
*/
|
||||
public void shutdown() {
|
||||
|
||||
shutdown = true;
|
||||
synchronized (listenerThread) {
|
||||
try {
|
||||
// wait max 20 seconds
|
||||
listenerThread.wait(20000);
|
||||
} catch (InterruptedException e) {
|
||||
logger.info("InterruptedException:" + e);
|
||||
}
|
||||
}
|
||||
|
||||
if (!shutdownComplete) {
|
||||
String msg = "WARNING: Shutdown of McastListener did not complete?";
|
||||
System.err.println(msg);
|
||||
logger.warn(msg);
|
||||
}
|
||||
|
||||
try {
|
||||
sock.leaveGroup(group);
|
||||
} catch (IOException e) {
|
||||
// send to syserr in case logging already shutdown
|
||||
e.printStackTrace();
|
||||
String msg = "Error leaving Multicast group";
|
||||
logger.info(msg, e);
|
||||
}
|
||||
try {
|
||||
sock.close();
|
||||
} catch (Exception e) {
|
||||
// send to syserr in case logging already shutdown
|
||||
e.printStackTrace();
|
||||
String msg = "Error closing Multicast socket";
|
||||
logger.info(msg, e);
|
||||
}
|
||||
}
|
||||
|
||||
public void run() {
|
||||
while (!shutdown) {
|
||||
try {
|
||||
pack.setLength(receiveBuffer.length);
|
||||
sock.receive(pack);
|
||||
|
||||
InetSocketAddress senderAddr = (InetSocketAddress) pack.getSocketAddress();
|
||||
|
||||
String senderHostPort = senderAddr.getAddress().getHostAddress() + ":" + senderAddr.getPort();
|
||||
|
||||
if (senderHostPort.equals(localSenderHostPort)) {
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.info("Ignoring message as sent by localSender: " + localSenderHostPort);
|
||||
}
|
||||
} else {
|
||||
|
||||
byte[] data = pack.getData();
|
||||
|
||||
|
||||
ByteArrayInputStream bi = new ByteArrayInputStream(data);
|
||||
DataInputStream dataInput = new DataInputStream(bi);
|
||||
|
||||
++totalPacketsReceived;
|
||||
totalBytesReceived += pack.getLength();
|
||||
|
||||
Packet header = Packet.readHeader(dataInput);
|
||||
|
||||
long packetId = header.getPacketId();
|
||||
boolean ackMsg = packetId == 0;
|
||||
|
||||
boolean processThisPacket = ackMsg || packetControl.isProcessPacket(senderHostPort, header.getPacketId());
|
||||
|
||||
if (!processThisPacket) {
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.info("Already processed packet: " + header.getPacketId() + " type:" + header.getPacketType() + " len:" + data.length);
|
||||
}
|
||||
|
||||
if (mcastBindAddress != null) {
|
||||
// bind to a specific interface
|
||||
sock.setInterface(mcastBindAddress);
|
||||
} else {
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.info("Incoming packet:" + header.getPacketId() + " type:" + header.getPacketType() + " len:" + data.length);
|
||||
}
|
||||
|
||||
if (ttl >= 0) {
|
||||
sock.setTimeToLive(ttl);
|
||||
}
|
||||
sock.setReuseAddress(true);
|
||||
pack = new DatagramPacket(receiveBuffer, receiveBuffer.length);
|
||||
sock.joinGroup(group);
|
||||
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
processPacket(senderHostPort, header, dataInput);
|
||||
}
|
||||
}
|
||||
|
||||
} catch (java.net.SocketTimeoutException e) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("timeout", e);
|
||||
}
|
||||
packetControl.onListenerTimeout();
|
||||
|
||||
} catch (IOException e) {
|
||||
logger.info("error ?", e);
|
||||
}
|
||||
}
|
||||
|
||||
public void startListening() {
|
||||
this.listenerThread.setDaemon(true);
|
||||
this.listenerThread.start();
|
||||
|
||||
logger.info("Cluster Multicast Listener up and joined Group");
|
||||
shutdownComplete = true;
|
||||
|
||||
synchronized (listenerThread) {
|
||||
listenerThread.notifyAll();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shutdown this listener.
|
||||
*/
|
||||
public void shutdown() {
|
||||
|
||||
shutdown = true;
|
||||
synchronized (listenerThread) {
|
||||
try {
|
||||
// wait max 20 seconds
|
||||
listenerThread.wait(20000);
|
||||
} catch (InterruptedException e) {
|
||||
logger.info("InterruptedException:"+e);
|
||||
}
|
||||
}
|
||||
|
||||
if (!shutdownComplete){
|
||||
String msg = "WARNING: Shutdown of McastListener did not complete?";
|
||||
System.err.println(msg);
|
||||
logger.warn(msg);
|
||||
}
|
||||
|
||||
try {
|
||||
sock.leaveGroup(group);
|
||||
} catch (IOException e) {
|
||||
// send to syserr in case logging already shutdown
|
||||
e.printStackTrace();
|
||||
String msg = "Error leaving Multicast group";
|
||||
logger.info(msg, e);
|
||||
}
|
||||
try {
|
||||
sock.close();
|
||||
} catch (Exception e) {
|
||||
// send to syserr in case logging already shutdown
|
||||
e.printStackTrace();
|
||||
String msg = "Error closing Multicast socket";
|
||||
logger.info(msg, e);
|
||||
}
|
||||
protected void processPacket(String senderHostPort, Packet header, DataInput dataInput) {
|
||||
try {
|
||||
switch (header.getPacketType()) {
|
||||
case Packet.TYPE_MESSAGES:
|
||||
packetControl.processMessagesPacket(senderHostPort, header, dataInput,
|
||||
totalPacketsReceived, totalBytesReceived, totalTxnEventsReceived);
|
||||
break;
|
||||
|
||||
case Packet.TYPE_TRANSEVENT:
|
||||
++totalTxnEventsReceived;
|
||||
processTransactionEventPacket(header, dataInput);
|
||||
break;
|
||||
|
||||
default:
|
||||
String msg = "Unknown Packet type:" + header.getPacketType();
|
||||
logger.error(msg);
|
||||
break;
|
||||
}
|
||||
} catch (IOException e) {
|
||||
// need to ask to get this packet resent...
|
||||
String msg = "Error reading Packet " + header.getPacketId() + " type:" + header.getPacketType();
|
||||
logger.error(msg, e);
|
||||
}
|
||||
|
||||
public void run() {
|
||||
while (!shutdown) {
|
||||
try {
|
||||
pack.setLength(receiveBuffer.length);
|
||||
sock.receive(pack);
|
||||
}
|
||||
|
||||
InetSocketAddress senderAddr = (InetSocketAddress)pack.getSocketAddress();
|
||||
private void processTransactionEventPacket(Packet header, DataInput dataInput) throws IOException {
|
||||
|
||||
String senderHostPort = senderAddr.getAddress().getHostAddress()+":"+senderAddr.getPort();
|
||||
|
||||
if (senderHostPort.equals(localSenderHostPort)){
|
||||
if (logger.isTraceEnabled()){
|
||||
logger.info("Ignoring message as sent by localSender: "+localSenderHostPort);
|
||||
}
|
||||
} else {
|
||||
|
||||
byte[] data = pack.getData();
|
||||
|
||||
|
||||
ByteArrayInputStream bi = new ByteArrayInputStream(data);
|
||||
DataInputStream dataInput = new DataInputStream(bi);
|
||||
|
||||
++totalPacketsReceived;
|
||||
totalBytesReceived += pack.getLength();
|
||||
|
||||
Packet header = Packet.readHeader(dataInput);
|
||||
|
||||
long packetId = header.getPacketId();
|
||||
boolean ackMsg = packetId == 0;
|
||||
|
||||
boolean processThisPacket = ackMsg || packetControl.isProcessPacket(senderHostPort, header.getPacketId());
|
||||
|
||||
if (!processThisPacket){
|
||||
if (logger.isTraceEnabled()){
|
||||
logger.info("Already processed packet: "+header.getPacketId()+" type:"+header.getPacketType()+" len:"+data.length);
|
||||
}
|
||||
} else {
|
||||
if (logger.isTraceEnabled()){
|
||||
logger.info("Incoming packet:"+header.getPacketId()+" type:"+header.getPacketType()+" len:"+data.length);
|
||||
}
|
||||
processPacket(senderHostPort, header, dataInput);
|
||||
}
|
||||
}
|
||||
|
||||
} catch (java.net.SocketTimeoutException e) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("timeout", e);
|
||||
}
|
||||
packetControl.onListenerTimeout();
|
||||
|
||||
} catch (IOException e) {
|
||||
logger.info("error ?", e);
|
||||
}
|
||||
}
|
||||
|
||||
shutdownComplete = true;
|
||||
|
||||
synchronized (listenerThread) {
|
||||
listenerThread.notifyAll();
|
||||
}
|
||||
}
|
||||
SpiEbeanServer server = owner.getEbeanServer(header.getServerName());
|
||||
|
||||
protected void processPacket(String senderHostPort, Packet header, DataInput dataInput) {
|
||||
try {
|
||||
switch (header.getPacketType()) {
|
||||
case Packet.TYPE_MESSAGES:
|
||||
packetControl.processMessagesPacket(senderHostPort, header, dataInput,
|
||||
totalPacketsReceived, totalBytesReceived, totalTxnEventsReceived);
|
||||
break;
|
||||
|
||||
case Packet.TYPE_TRANSEVENT:
|
||||
++totalTxnEventsReceived;
|
||||
processTransactionEventPacket(header, dataInput);
|
||||
break;
|
||||
|
||||
default:
|
||||
String msg = "Unknown Packet type:" + header.getPacketType();
|
||||
logger.error(msg);
|
||||
break;
|
||||
}
|
||||
} catch (IOException e) {
|
||||
// need to ask to get this packet resent...
|
||||
String msg = "Error reading Packet " + header.getPacketId() + " type:" + header.getPacketType();
|
||||
logger.error(msg, e);
|
||||
}
|
||||
}
|
||||
|
||||
private void processTransactionEventPacket(Packet header, DataInput dataInput) throws IOException {
|
||||
PacketTransactionEvent tranEventPacket = PacketTransactionEvent.forRead(header, server);
|
||||
tranEventPacket.read(dataInput);
|
||||
|
||||
SpiEbeanServer server = owner.getEbeanServer(header.getServerName());
|
||||
|
||||
PacketTransactionEvent tranEventPacket = PacketTransactionEvent.forRead(header, server);
|
||||
tranEventPacket.read(dataInput);
|
||||
|
||||
server.remoteTransactionEvent(tranEventPacket.getEvent());
|
||||
}
|
||||
server.remoteTransactionEvent(tranEventPacket.getEvent());
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
+94
-94
@@ -1,125 +1,125 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import java.io.DataInput;
|
||||
import java.io.IOException;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.Packet;
|
||||
import com.avaje.ebeaninternal.server.cluster.PacketMessages;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.DataInput;
|
||||
import java.io.IOException;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Helps co-ordinate Packet information between the McastListener and the
|
||||
* McastClusterManager.
|
||||
*
|
||||
*
|
||||
* @author rbygrave
|
||||
*/
|
||||
public class McastPacketControl {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(McastPacketControl.class);
|
||||
private static final Logger logger = LoggerFactory.getLogger(McastPacketControl.class);
|
||||
|
||||
private final String localSenderHostPort;
|
||||
private final String localSenderHostPort;
|
||||
|
||||
private final McastClusterManager owner;
|
||||
private final McastClusterManager owner;
|
||||
|
||||
private final HashSet<String> groupMembers = new HashSet<String>();
|
||||
private final HashSet<String> groupMembers = new HashSet<String>();
|
||||
|
||||
private final OutgoingPacketsAcked outgoingPacketsAcked = new OutgoingPacketsAcked();
|
||||
private final OutgoingPacketsAcked outgoingPacketsAcked = new OutgoingPacketsAcked();
|
||||
|
||||
private final IncomingPacketsProcessed incomingPacketsProcessed;
|
||||
private final IncomingPacketsProcessed incomingPacketsProcessed;
|
||||
|
||||
public McastPacketControl(McastClusterManager owner, String localSenderHostPort, int maxResendIncoming) {
|
||||
this.owner = owner;
|
||||
this.localSenderHostPort = localSenderHostPort;
|
||||
this.incomingPacketsProcessed = new IncomingPacketsProcessed(maxResendIncoming);
|
||||
public McastPacketControl(McastClusterManager owner, String localSenderHostPort, int maxResendIncoming) {
|
||||
this.owner = owner;
|
||||
this.localSenderHostPort = localSenderHostPort;
|
||||
this.incomingPacketsProcessed = new IncomingPacketsProcessed(maxResendIncoming);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle special case where cluster doesn't have any members and we don't
|
||||
* get any responses. Need to tell the sender side that the group size is 0.
|
||||
*/
|
||||
protected void onListenerTimeout() {
|
||||
if (groupMembers.size() == 0) {
|
||||
owner.fromListenerTimeoutNoMembers();
|
||||
}
|
||||
}
|
||||
|
||||
protected void processMessagesPacket(String senderHostPort, Packet header, DataInput dataInput,
|
||||
long totalPacketsReceived, long totalBytesReceived, long totalTransEventsReceived) throws IOException {
|
||||
|
||||
PacketMessages packetMessages = PacketMessages.forRead(header);
|
||||
packetMessages.read(dataInput);
|
||||
List<Message> messages = packetMessages.getMessages();
|
||||
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("INCOMING Messages " + messages);
|
||||
}
|
||||
// messages are for all nodes in the cluster so
|
||||
// we need to filter looking for messages pertaining
|
||||
// to this (senderHostPort)
|
||||
|
||||
MessageControl control = null;
|
||||
MessageAck ack = null;
|
||||
MessageResend resend = null;
|
||||
|
||||
// filter for relevant messages to this node
|
||||
for (int i = 0; i < messages.size(); i++) {
|
||||
Message message = messages.get(i);
|
||||
if (message.isControlMessage()) {
|
||||
// any 'control' message is interesting
|
||||
control = (MessageControl) message;
|
||||
|
||||
} else if (localSenderHostPort.equals(message.getToHostPort())) {
|
||||
if (message instanceof MessageAck) {
|
||||
ack = (MessageAck) message;
|
||||
} else if (message instanceof MessageResend) {
|
||||
resend = (MessageResend) message;
|
||||
} else {
|
||||
logger.error("Expecting a MessageAck or MessageResend but got a "
|
||||
+ message.getClass().getName());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle special case where cluster doesn't have any members and we don't
|
||||
* get any responses. Need to tell the sender side that the group size is 0.
|
||||
*/
|
||||
protected void onListenerTimeout() {
|
||||
if (groupMembers.size() == 0) {
|
||||
owner.fromListenerTimeoutNoMembers();
|
||||
}
|
||||
if (control != null) {
|
||||
if (control.getControlType() == MessageControl.TYPE_LEAVE) {
|
||||
groupMembers.remove(senderHostPort);
|
||||
logger.info("Cluster member leaving [" + senderHostPort + "] " + groupMembers.size()
|
||||
+ " other members left");
|
||||
outgoingPacketsAcked.removeMember(senderHostPort);
|
||||
incomingPacketsProcessed.removeMember(senderHostPort);
|
||||
} else {
|
||||
groupMembers.add(senderHostPort);
|
||||
}
|
||||
}
|
||||
|
||||
protected void processMessagesPacket(String senderHostPort, Packet header, DataInput dataInput,
|
||||
long totalPacketsReceived, long totalBytesReceived, long totalTransEventsReceived) throws IOException {
|
||||
|
||||
PacketMessages packetMessages = PacketMessages.forRead(header);
|
||||
packetMessages.read(dataInput);
|
||||
List<Message> messages = packetMessages.getMessages();
|
||||
|
||||
if (logger.isTraceEnabled()) {
|
||||
logger.trace("INCOMING Messages " + messages);
|
||||
}
|
||||
// messages are for all nodes in the cluster so
|
||||
// we need to filter looking for messages pertaining
|
||||
// to this (senderHostPort)
|
||||
|
||||
MessageControl control = null;
|
||||
MessageAck ack = null;
|
||||
MessageResend resend = null;
|
||||
|
||||
// filter for relevant messages to this node
|
||||
for (int i = 0; i < messages.size(); i++) {
|
||||
Message message = messages.get(i);
|
||||
if (message.isControlMessage()) {
|
||||
// any 'control' message is interesting
|
||||
control = (MessageControl) message;
|
||||
|
||||
} else if (localSenderHostPort.equals(message.getToHostPort())) {
|
||||
if (message instanceof MessageAck) {
|
||||
ack = (MessageAck) message;
|
||||
} else if (message instanceof MessageResend) {
|
||||
resend = (MessageResend) message;
|
||||
} else {
|
||||
logger.error("Expecting a MessageAck or MessageResend but got a "
|
||||
+ message.getClass().getName());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (control != null) {
|
||||
if (control.getControlType() == MessageControl.TYPE_LEAVE) {
|
||||
groupMembers.remove(senderHostPort);
|
||||
logger.info("Cluster member leaving [" + senderHostPort + "] " + groupMembers.size()
|
||||
+ " other members left");
|
||||
outgoingPacketsAcked.removeMember(senderHostPort);
|
||||
incomingPacketsProcessed.removeMember(senderHostPort);
|
||||
} else {
|
||||
groupMembers.add(senderHostPort);
|
||||
}
|
||||
}
|
||||
|
||||
long newMin = 0;
|
||||
if (ack != null) {
|
||||
newMin = outgoingPacketsAcked.receivedAck(senderHostPort, ack);
|
||||
}
|
||||
|
||||
if (newMin > 0 || control != null || resend != null) {
|
||||
int groupSize = groupMembers.size();
|
||||
// synchronised on the managerThread
|
||||
owner.fromListener(newMin, control, resend, groupSize,
|
||||
totalPacketsReceived, totalBytesReceived, totalTransEventsReceived);
|
||||
}
|
||||
long newMin = 0;
|
||||
if (ack != null) {
|
||||
newMin = outgoingPacketsAcked.receivedAck(senderHostPort, ack);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return true if we should process this packet. Return false if we have
|
||||
* already processed the packet.
|
||||
*/
|
||||
public boolean isProcessPacket(String memberKey, long packetId) {
|
||||
|
||||
return incomingPacketsProcessed.isProcessPacket(memberKey, packetId);
|
||||
if (newMin > 0 || control != null || resend != null) {
|
||||
int groupSize = groupMembers.size();
|
||||
// synchronised on the managerThread
|
||||
owner.fromListener(newMin, control, resend, groupSize,
|
||||
totalPacketsReceived, totalBytesReceived, totalTransEventsReceived);
|
||||
}
|
||||
}
|
||||
|
||||
public AckResendMessages getAckResendMessages(IncomingPacketsLastAck lastAck) {
|
||||
/**
|
||||
* Return true if we should process this packet. Return false if we have
|
||||
* already processed the packet.
|
||||
*/
|
||||
public boolean isProcessPacket(String memberKey, long packetId) {
|
||||
|
||||
return incomingPacketsProcessed.getAckResendMessages(lastAck);
|
||||
}
|
||||
return incomingPacketsProcessed.isProcessPacket(memberKey, packetId);
|
||||
}
|
||||
|
||||
public AckResendMessages getAckResendMessages(IncomingPacketsLastAck lastAck) {
|
||||
|
||||
return incomingPacketsProcessed.getAckResendMessages(lastAck);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,5 +1,9 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.Packet;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.DatagramPacket;
|
||||
import java.net.DatagramSocket;
|
||||
@@ -7,110 +11,104 @@ import java.net.InetAddress;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.List;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.Packet;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* Handles the sending of Packets via DatagramPacket.
|
||||
*
|
||||
* @author rbygrave
|
||||
*/
|
||||
public class McastSender {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(McastSender.class);
|
||||
private static final Logger logger = LoggerFactory.getLogger(McastSender.class);
|
||||
|
||||
private final int port;
|
||||
private final int port;
|
||||
|
||||
private final InetAddress inetAddress;
|
||||
private final InetAddress inetAddress;
|
||||
|
||||
private final DatagramSocket sock;
|
||||
private final DatagramSocket sock;
|
||||
|
||||
private final InetSocketAddress sendAddr;
|
||||
private final InetSocketAddress sendAddr;
|
||||
|
||||
private final String senderHostPort;
|
||||
private final String senderHostPort;
|
||||
|
||||
|
||||
public McastSender(int port, String address, int sendPort, String sendAddress) {
|
||||
|
||||
try {
|
||||
this.port = port;
|
||||
this.inetAddress = InetAddress.getByName(address);
|
||||
public McastSender(int port, String address, int sendPort, String sendAddress) {
|
||||
|
||||
InetAddress sendInetAddress;
|
||||
if (sendAddress != null) {
|
||||
sendInetAddress = InetAddress.getByName(sendAddress);
|
||||
} else {
|
||||
sendInetAddress = InetAddress.getLocalHost();
|
||||
}
|
||||
try {
|
||||
this.port = port;
|
||||
this.inetAddress = InetAddress.getByName(address);
|
||||
|
||||
if (sendPort > 0) {
|
||||
this.sock = new DatagramSocket(sendPort, sendInetAddress);
|
||||
} else {
|
||||
this.sock = new DatagramSocket(new InetSocketAddress(sendInetAddress, 0));
|
||||
}
|
||||
InetAddress sendInetAddress;
|
||||
if (sendAddress != null) {
|
||||
sendInetAddress = InetAddress.getByName(sendAddress);
|
||||
} else {
|
||||
sendInetAddress = InetAddress.getLocalHost();
|
||||
}
|
||||
|
||||
String msg = "Cluster Multicast Sender on["+sendInetAddress.getHostAddress()+":"+sock.getLocalPort()+"]";
|
||||
logger.info(msg);
|
||||
if (sendPort > 0) {
|
||||
this.sock = new DatagramSocket(sendPort, sendInetAddress);
|
||||
} else {
|
||||
this.sock = new DatagramSocket(new InetSocketAddress(sendInetAddress, 0));
|
||||
}
|
||||
|
||||
this.sendAddr = new InetSocketAddress(sendInetAddress, sock.getLocalPort());
|
||||
this.senderHostPort = sendInetAddress.getHostAddress()+":"+sock.getLocalPort();
|
||||
|
||||
} catch (Exception e) {
|
||||
String msg = "McastSender port:" + port + " sendPort:" + sendPort + " " + address;
|
||||
throw new RuntimeException(msg, e);
|
||||
}
|
||||
String msg = "Cluster Multicast Sender on[" + sendInetAddress.getHostAddress() + ":" + sock.getLocalPort() + "]";
|
||||
logger.info(msg);
|
||||
|
||||
this.sendAddr = new InetSocketAddress(sendInetAddress, sock.getLocalPort());
|
||||
this.senderHostPort = sendInetAddress.getHostAddress() + ":" + sock.getLocalPort();
|
||||
|
||||
} catch (Exception e) {
|
||||
String msg = "McastSender port:" + port + " sendPort:" + sendPort + " " + address;
|
||||
throw new RuntimeException(msg, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the send Address so that if we have loopback messages we can
|
||||
* detect if they where sent by this local sender and hence should be
|
||||
* ignored.
|
||||
*/
|
||||
public InetSocketAddress getAddress() {
|
||||
return sendAddr;
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the Host and Port of the sender. This is used to uniquely identify
|
||||
* this instance in the cluster.
|
||||
*/
|
||||
public String getSenderHostPort() {
|
||||
return senderHostPort;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send the packet.
|
||||
*/
|
||||
public int sendPacket(Packet packet) throws IOException {
|
||||
|
||||
byte[] pktBytes = packet.getBytes();
|
||||
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug("OUTGOING packet: " + packet.getPacketId() + " size:" + pktBytes.length);
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the send Address so that if we have loopback messages we can
|
||||
* detect if they where sent by this local sender and hence should be
|
||||
* ignored.
|
||||
*/
|
||||
public InetSocketAddress getAddress() {
|
||||
return sendAddr;
|
||||
if (pktBytes.length > 65507) {
|
||||
logger.warn("OUTGOING packet: " + packet.getPacketId() + " size:" + pktBytes.length
|
||||
+ " likely to be truncated using UDP with a MAXIMUM length of 65507");
|
||||
}
|
||||
|
||||
/**
|
||||
* Return the Host and Port of the sender. This is used to uniquely identify
|
||||
* this instance in the cluster.
|
||||
*/
|
||||
public String getSenderHostPort() {
|
||||
return senderHostPort;
|
||||
DatagramPacket pack = new DatagramPacket(pktBytes, pktBytes.length, inetAddress, port);
|
||||
sock.send(pack);
|
||||
|
||||
return pktBytes.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send the list of Packets.
|
||||
*/
|
||||
public int sendPackets(List<Packet> packets) throws IOException {
|
||||
|
||||
int totalBytes = 0;
|
||||
for (int i = 0; i < packets.size(); i++) {
|
||||
totalBytes += sendPacket(packets.get(i));
|
||||
}
|
||||
return totalBytes;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send the packet.
|
||||
*/
|
||||
public int sendPacket(Packet packet) throws IOException {
|
||||
|
||||
byte[] pktBytes = packet.getBytes();
|
||||
|
||||
if (logger.isDebugEnabled()){
|
||||
logger.debug("OUTGOING packet: " + packet.getPacketId() + " size:" + pktBytes.length);
|
||||
}
|
||||
|
||||
if (pktBytes.length > 65507){
|
||||
logger.warn("OUTGOING packet: " + packet.getPacketId() + " size:" + pktBytes.length
|
||||
+" likely to be truncated using UDP with a MAXIMUM length of 65507");
|
||||
}
|
||||
|
||||
DatagramPacket pack = new DatagramPacket(pktBytes, pktBytes.length, inetAddress, port);
|
||||
sock.send(pack);
|
||||
|
||||
return pktBytes.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Send the list of Packets.
|
||||
*/
|
||||
public int sendPackets(List<Packet> packets) throws IOException {
|
||||
|
||||
int totalBytes = 0;
|
||||
for (int i = 0; i < packets.size(); i++) {
|
||||
totalBytes += sendPacket(packets.get(i));
|
||||
}
|
||||
return totalBytes;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -5,133 +5,130 @@ package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
* <p>
|
||||
* Ideally you want to see relatively low Re-send statistics.
|
||||
* </p>
|
||||
*
|
||||
* @author rbygrave
|
||||
*
|
||||
*/
|
||||
public class McastStatus {
|
||||
|
||||
private final long totalTxnEventsSent;
|
||||
private final long totalTxnEventsReceived;
|
||||
|
||||
private final long totalPacketsSent;
|
||||
private final long totalPacketsResent;
|
||||
private final long totalPacketsReceived;
|
||||
|
||||
private final long totalBytesSent;
|
||||
private final long totalBytesResent;
|
||||
private final long totalBytesReceived;
|
||||
private final long totalTxnEventsSent;
|
||||
private final long totalTxnEventsReceived;
|
||||
|
||||
private final int currentGroupSize;
|
||||
private final int outgoingPacketsCacheSize;
|
||||
|
||||
private final long currentPacketId;
|
||||
private final long minAckedPacketId;
|
||||
private final String lastOutgoingAcks;
|
||||
|
||||
public String getSummary() {
|
||||
private final long totalPacketsSent;
|
||||
private final long totalPacketsResent;
|
||||
private final long totalPacketsReceived;
|
||||
|
||||
//noinspection StringBufferReplaceableByString
|
||||
StringBuilder sb = new StringBuilder(80);
|
||||
sb.append("txnOut:").append(totalTxnEventsSent).append("; ");
|
||||
sb.append("txnIn:").append(totalTxnEventsReceived).append("; ");
|
||||
sb.append("outPackets:").append(totalPacketsSent).append("; ");
|
||||
sb.append("outBytes:").append(totalBytesSent).append("; ");
|
||||
sb.append("inPackets:").append(totalPacketsReceived).append("; ");
|
||||
sb.append("inBytes:").append(totalBytesReceived).append("; ");
|
||||
sb.append("resentPackets:").append(totalPacketsResent).append("; ");
|
||||
sb.append("resentBytes:").append(totalBytesResent).append("; ");
|
||||
sb.append("groupSize:").append(currentGroupSize).append("; ");
|
||||
sb.append("cache:").append(outgoingPacketsCacheSize).append("; ");
|
||||
sb.append("currentPacket:").append(currentPacketId).append("; ");
|
||||
sb.append("minAckedPacket:").append(minAckedPacketId).append("; ");
|
||||
sb.append("lastAck:").append(lastOutgoingAcks).append("; ");
|
||||
private final long totalBytesSent;
|
||||
private final long totalBytesResent;
|
||||
private final long totalBytesReceived;
|
||||
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
public McastStatus(int currentGroupSize,
|
||||
int outgoingPacketsCacheSize,
|
||||
long currentPacketId,
|
||||
long minAckedPacketId,
|
||||
String lastOutgoingAcks,
|
||||
long totalTransEventsSent,
|
||||
long totalTransEventsReceived,
|
||||
long totalPacketsSent,
|
||||
long totalPacketsResent,
|
||||
long totalPacketsReceived,
|
||||
long totalBytesSent,
|
||||
long totalBytesResent,
|
||||
long totalBytesReceived) {
|
||||
|
||||
this.currentGroupSize = currentGroupSize;
|
||||
this.outgoingPacketsCacheSize = outgoingPacketsCacheSize;
|
||||
this.currentPacketId = currentPacketId;
|
||||
this.minAckedPacketId = minAckedPacketId;
|
||||
this.lastOutgoingAcks = lastOutgoingAcks;
|
||||
this.totalTxnEventsSent = totalTransEventsSent;
|
||||
this.totalTxnEventsReceived = totalTransEventsReceived;
|
||||
this.totalPacketsSent = totalPacketsSent;
|
||||
this.totalPacketsResent = totalPacketsResent;
|
||||
this.totalPacketsReceived = totalPacketsReceived;
|
||||
private final int currentGroupSize;
|
||||
private final int outgoingPacketsCacheSize;
|
||||
|
||||
this.totalBytesSent = totalBytesSent;
|
||||
this.totalBytesResent = totalBytesResent;
|
||||
this.totalBytesReceived = totalBytesReceived;
|
||||
private final long currentPacketId;
|
||||
private final long minAckedPacketId;
|
||||
private final String lastOutgoingAcks;
|
||||
|
||||
}
|
||||
public String getSummary() {
|
||||
|
||||
|
||||
public long getTotalTxnEventsReceived() {
|
||||
return totalTxnEventsReceived;
|
||||
}
|
||||
//noinspection StringBufferReplaceableByString
|
||||
StringBuilder sb = new StringBuilder(80);
|
||||
sb.append("txnOut:").append(totalTxnEventsSent).append("; ");
|
||||
sb.append("txnIn:").append(totalTxnEventsReceived).append("; ");
|
||||
sb.append("outPackets:").append(totalPacketsSent).append("; ");
|
||||
sb.append("outBytes:").append(totalBytesSent).append("; ");
|
||||
sb.append("inPackets:").append(totalPacketsReceived).append("; ");
|
||||
sb.append("inBytes:").append(totalBytesReceived).append("; ");
|
||||
sb.append("resentPackets:").append(totalPacketsResent).append("; ");
|
||||
sb.append("resentBytes:").append(totalBytesResent).append("; ");
|
||||
sb.append("groupSize:").append(currentGroupSize).append("; ");
|
||||
sb.append("cache:").append(outgoingPacketsCacheSize).append("; ");
|
||||
sb.append("currentPacket:").append(currentPacketId).append("; ");
|
||||
sb.append("minAckedPacket:").append(minAckedPacketId).append("; ");
|
||||
sb.append("lastAck:").append(lastOutgoingAcks).append("; ");
|
||||
|
||||
public long getTotalPacketsReceived() {
|
||||
return totalPacketsReceived;
|
||||
}
|
||||
return sb.toString();
|
||||
}
|
||||
|
||||
public long getTotalBytesSent() {
|
||||
return totalBytesSent;
|
||||
}
|
||||
public McastStatus(int currentGroupSize,
|
||||
int outgoingPacketsCacheSize,
|
||||
long currentPacketId,
|
||||
long minAckedPacketId,
|
||||
String lastOutgoingAcks,
|
||||
long totalTransEventsSent,
|
||||
long totalTransEventsReceived,
|
||||
long totalPacketsSent,
|
||||
long totalPacketsResent,
|
||||
long totalPacketsReceived,
|
||||
long totalBytesSent,
|
||||
long totalBytesResent,
|
||||
long totalBytesReceived) {
|
||||
|
||||
public long getTotalBytesResent() {
|
||||
return totalBytesResent;
|
||||
}
|
||||
this.currentGroupSize = currentGroupSize;
|
||||
this.outgoingPacketsCacheSize = outgoingPacketsCacheSize;
|
||||
this.currentPacketId = currentPacketId;
|
||||
this.minAckedPacketId = minAckedPacketId;
|
||||
this.lastOutgoingAcks = lastOutgoingAcks;
|
||||
this.totalTxnEventsSent = totalTransEventsSent;
|
||||
this.totalTxnEventsReceived = totalTransEventsReceived;
|
||||
this.totalPacketsSent = totalPacketsSent;
|
||||
this.totalPacketsResent = totalPacketsResent;
|
||||
this.totalPacketsReceived = totalPacketsReceived;
|
||||
|
||||
public long getTotalBytesReceived() {
|
||||
return totalBytesReceived;
|
||||
}
|
||||
this.totalBytesSent = totalBytesSent;
|
||||
this.totalBytesResent = totalBytesResent;
|
||||
this.totalBytesReceived = totalBytesReceived;
|
||||
|
||||
public String getLastOutgoingAcks() {
|
||||
return lastOutgoingAcks;
|
||||
}
|
||||
}
|
||||
|
||||
public int getOutgoingPacketsCacheSize() {
|
||||
return outgoingPacketsCacheSize;
|
||||
}
|
||||
|
||||
public long getCurrentPacketId() {
|
||||
return currentPacketId;
|
||||
}
|
||||
public long getTotalTxnEventsReceived() {
|
||||
return totalTxnEventsReceived;
|
||||
}
|
||||
|
||||
public long getMinAckedPacketId() {
|
||||
return minAckedPacketId;
|
||||
}
|
||||
public long getTotalPacketsReceived() {
|
||||
return totalPacketsReceived;
|
||||
}
|
||||
|
||||
public long getTotalTxnEventsSent() {
|
||||
return totalTxnEventsSent;
|
||||
}
|
||||
public long getTotalBytesSent() {
|
||||
return totalBytesSent;
|
||||
}
|
||||
|
||||
public long getTotalPacketsSent() {
|
||||
return totalPacketsSent;
|
||||
}
|
||||
public long getTotalBytesResent() {
|
||||
return totalBytesResent;
|
||||
}
|
||||
|
||||
public long getTotalPacketsResent() {
|
||||
return totalPacketsResent;
|
||||
}
|
||||
public long getTotalBytesReceived() {
|
||||
return totalBytesReceived;
|
||||
}
|
||||
|
||||
public String getLastOutgoingAcks() {
|
||||
return lastOutgoingAcks;
|
||||
}
|
||||
|
||||
public int getOutgoingPacketsCacheSize() {
|
||||
return outgoingPacketsCacheSize;
|
||||
}
|
||||
|
||||
public long getCurrentPacketId() {
|
||||
return currentPacketId;
|
||||
}
|
||||
|
||||
public long getMinAckedPacketId() {
|
||||
return minAckedPacketId;
|
||||
}
|
||||
|
||||
public long getTotalTxnEventsSent() {
|
||||
return totalTxnEventsSent;
|
||||
}
|
||||
|
||||
public long getTotalPacketsSent() {
|
||||
return totalPacketsSent;
|
||||
}
|
||||
|
||||
public long getTotalPacketsResent() {
|
||||
return totalPacketsResent;
|
||||
}
|
||||
|
||||
public long getCurrentGroupSize() {
|
||||
return currentGroupSize;
|
||||
}
|
||||
|
||||
public long getCurrentGroupSize() {
|
||||
return currentGroupSize;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -1,14 +1,14 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
public interface Message {
|
||||
|
||||
void writeBinaryMessage(BinaryMessageList msgList) throws IOException;
|
||||
void writeBinaryMessage(BinaryMessageList msgList) throws IOException;
|
||||
|
||||
boolean isControlMessage();
|
||||
|
||||
String getToHostPort();
|
||||
boolean isControlMessage();
|
||||
|
||||
String getToHostPort();
|
||||
}
|
||||
|
||||
@@ -1,57 +1,57 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
import java.io.DataInput;
|
||||
import java.io.DataOutputStream;
|
||||
import java.io.IOException;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
public class MessageAck implements Message {
|
||||
|
||||
private final String toHostPort;
|
||||
|
||||
private final long gotAllPacketId;
|
||||
private final String toHostPort;
|
||||
|
||||
public MessageAck(String toHostPort, long gotAllPacketId) {
|
||||
this.toHostPort = toHostPort;
|
||||
this.gotAllPacketId = gotAllPacketId;
|
||||
}
|
||||
private final long gotAllPacketId;
|
||||
|
||||
public String toString() {
|
||||
return "Ack "+toHostPort+" "+gotAllPacketId;
|
||||
}
|
||||
|
||||
public boolean isControlMessage() {
|
||||
return false;
|
||||
}
|
||||
public MessageAck(String toHostPort, long gotAllPacketId) {
|
||||
this.toHostPort = toHostPort;
|
||||
this.gotAllPacketId = gotAllPacketId;
|
||||
}
|
||||
|
||||
public String getToHostPort() {
|
||||
return toHostPort;
|
||||
}
|
||||
public String toString() {
|
||||
return "Ack " + toHostPort + " " + gotAllPacketId;
|
||||
}
|
||||
|
||||
public long getGotAllPacketId() {
|
||||
return gotAllPacketId;
|
||||
}
|
||||
|
||||
|
||||
public static MessageAck readBinaryMessage(DataInput dataInput) throws IOException {
|
||||
public boolean isControlMessage() {
|
||||
return false;
|
||||
}
|
||||
|
||||
String hostPort = dataInput.readUTF();
|
||||
long gotAllPacketId = dataInput.readLong();
|
||||
return new MessageAck(hostPort, gotAllPacketId);
|
||||
}
|
||||
public String getToHostPort() {
|
||||
return toHostPort;
|
||||
}
|
||||
|
||||
public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
|
||||
|
||||
BinaryMessage m = new BinaryMessage(toHostPort.length() * 2 + 20);
|
||||
|
||||
DataOutputStream os = m.getOs();
|
||||
os.writeInt(BinaryMessage.TYPE_MSGACK);
|
||||
os.writeUTF(toHostPort);
|
||||
os.writeLong(gotAllPacketId);
|
||||
os.flush();
|
||||
|
||||
msgList.add(m);
|
||||
}
|
||||
public long getGotAllPacketId() {
|
||||
return gotAllPacketId;
|
||||
}
|
||||
|
||||
|
||||
public static MessageAck readBinaryMessage(DataInput dataInput) throws IOException {
|
||||
|
||||
String hostPort = dataInput.readUTF();
|
||||
long gotAllPacketId = dataInput.readLong();
|
||||
return new MessageAck(hostPort, gotAllPacketId);
|
||||
}
|
||||
|
||||
public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
|
||||
|
||||
BinaryMessage m = new BinaryMessage(toHostPort.length() * 2 + 20);
|
||||
|
||||
DataOutputStream os = m.getOs();
|
||||
os.writeInt(BinaryMessage.TYPE_MSGACK);
|
||||
os.writeUTF(toHostPort);
|
||||
os.writeLong(gotAllPacketId);
|
||||
os.flush();
|
||||
|
||||
msgList.add(m);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,73 +1,77 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
import java.io.DataInput;
|
||||
import java.io.DataOutputStream;
|
||||
import java.io.IOException;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
public class MessageControl implements Message {
|
||||
|
||||
public static final short TYPE_JOIN = 1;
|
||||
public static final short TYPE_LEAVE = 2;
|
||||
public static final short TYPE_PING = 3;
|
||||
public static final short TYPE_JOINRESPONSE = 7;
|
||||
public static final short TYPE_PINGRESPONSE = 8;
|
||||
|
||||
private final short controlType;
|
||||
private final String fromHostPort;
|
||||
|
||||
public static MessageControl readBinaryMessage(DataInput dataInput) throws IOException {
|
||||
short controlType = dataInput.readShort();
|
||||
String hostPort = dataInput.readUTF();
|
||||
return new MessageControl(controlType, hostPort);
|
||||
}
|
||||
|
||||
public MessageControl(short controlType, String helloFromHostPort) {
|
||||
this.controlType = controlType;
|
||||
this.fromHostPort = helloFromHostPort;
|
||||
}
|
||||
public static final short TYPE_JOIN = 1;
|
||||
public static final short TYPE_LEAVE = 2;
|
||||
public static final short TYPE_PING = 3;
|
||||
public static final short TYPE_JOINRESPONSE = 7;
|
||||
public static final short TYPE_PINGRESPONSE = 8;
|
||||
|
||||
private final short controlType;
|
||||
private final String fromHostPort;
|
||||
|
||||
public static MessageControl readBinaryMessage(DataInput dataInput) throws IOException {
|
||||
short controlType = dataInput.readShort();
|
||||
String hostPort = dataInput.readUTF();
|
||||
return new MessageControl(controlType, hostPort);
|
||||
}
|
||||
|
||||
public MessageControl(short controlType, String helloFromHostPort) {
|
||||
this.controlType = controlType;
|
||||
this.fromHostPort = helloFromHostPort;
|
||||
}
|
||||
|
||||
|
||||
public String toString() {
|
||||
switch (controlType) {
|
||||
case TYPE_JOIN: return "Join "+fromHostPort;
|
||||
case TYPE_LEAVE: return "Leave "+fromHostPort;
|
||||
case TYPE_PING: return "Ping "+fromHostPort;
|
||||
case TYPE_PINGRESPONSE: return "PingResponse "+fromHostPort;
|
||||
public String toString() {
|
||||
switch (controlType) {
|
||||
case TYPE_JOIN:
|
||||
return "Join " + fromHostPort;
|
||||
case TYPE_LEAVE:
|
||||
return "Leave " + fromHostPort;
|
||||
case TYPE_PING:
|
||||
return "Ping " + fromHostPort;
|
||||
case TYPE_PINGRESPONSE:
|
||||
return "PingResponse " + fromHostPort;
|
||||
|
||||
default:
|
||||
throw new RuntimeException("Invalid controlType "+controlType);
|
||||
}
|
||||
}
|
||||
|
||||
public boolean isControlMessage() {
|
||||
return true;
|
||||
default:
|
||||
throw new RuntimeException("Invalid controlType " + controlType);
|
||||
}
|
||||
}
|
||||
|
||||
public short getControlType() {
|
||||
return controlType;
|
||||
}
|
||||
public boolean isControlMessage() {
|
||||
return true;
|
||||
}
|
||||
|
||||
public String getToHostPort() {
|
||||
return "*";
|
||||
}
|
||||
|
||||
public String getFromHostPort() {
|
||||
return fromHostPort;
|
||||
}
|
||||
public short getControlType() {
|
||||
return controlType;
|
||||
}
|
||||
|
||||
public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
|
||||
|
||||
BinaryMessage m = new BinaryMessage(fromHostPort.length() * 2 + 10);
|
||||
|
||||
DataOutputStream os = m.getOs();
|
||||
os.writeInt(BinaryMessage.TYPE_MSGCONTROL);
|
||||
os.writeShort(controlType);
|
||||
os.writeUTF(fromHostPort);
|
||||
os.flush();
|
||||
|
||||
msgList.add(m);
|
||||
}
|
||||
public String getToHostPort() {
|
||||
return "*";
|
||||
}
|
||||
|
||||
public String getFromHostPort() {
|
||||
return fromHostPort;
|
||||
}
|
||||
|
||||
public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
|
||||
|
||||
BinaryMessage m = new BinaryMessage(fromHostPort.length() * 2 + 10);
|
||||
|
||||
DataOutputStream os = m.getOs();
|
||||
os.writeInt(BinaryMessage.TYPE_MSGCONTROL);
|
||||
os.writeShort(controlType);
|
||||
os.writeUTF(fromHostPort);
|
||||
os.flush();
|
||||
|
||||
msgList.add(m);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,77 +1,77 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
import java.io.DataInput;
|
||||
import java.io.DataOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessage;
|
||||
import com.avaje.ebeaninternal.server.cluster.BinaryMessageList;
|
||||
|
||||
public class MessageResend implements Message {
|
||||
|
||||
private final String toHostPort;
|
||||
|
||||
private final List<Long> resendPacketIds;
|
||||
private final String toHostPort;
|
||||
|
||||
public MessageResend(String toHostPort, List<Long> resendPacketIds) {
|
||||
this.toHostPort = toHostPort;
|
||||
this.resendPacketIds = resendPacketIds;
|
||||
}
|
||||
|
||||
public MessageResend(String toHostPort) {
|
||||
this(toHostPort, new ArrayList<Long>(4));
|
||||
private final List<Long> resendPacketIds;
|
||||
|
||||
public MessageResend(String toHostPort, List<Long> resendPacketIds) {
|
||||
this.toHostPort = toHostPort;
|
||||
this.resendPacketIds = resendPacketIds;
|
||||
}
|
||||
|
||||
public MessageResend(String toHostPort) {
|
||||
this(toHostPort, new ArrayList<Long>(4));
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return "Resend " + toHostPort + " " + resendPacketIds;
|
||||
}
|
||||
|
||||
public boolean isControlMessage() {
|
||||
return false;
|
||||
}
|
||||
|
||||
public String getToHostPort() {
|
||||
return toHostPort;
|
||||
}
|
||||
|
||||
public void add(long packetId) {
|
||||
resendPacketIds.add(packetId);
|
||||
}
|
||||
|
||||
public List<Long> getResendPacketIds() {
|
||||
return resendPacketIds;
|
||||
}
|
||||
|
||||
public static MessageResend readBinaryMessage(DataInput dataInput) throws IOException {
|
||||
|
||||
String hostPort = dataInput.readUTF();
|
||||
|
||||
MessageResend msg = new MessageResend(hostPort);
|
||||
|
||||
int numberOfPacketIds = dataInput.readInt();
|
||||
for (int i = 0; i < numberOfPacketIds; i++) {
|
||||
long packetId = dataInput.readLong();
|
||||
msg.add(packetId);
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return "Resend "+toHostPort+" "+resendPacketIds;
|
||||
}
|
||||
|
||||
public boolean isControlMessage() {
|
||||
return false;
|
||||
}
|
||||
return msg;
|
||||
}
|
||||
|
||||
public String getToHostPort() {
|
||||
return toHostPort;
|
||||
}
|
||||
public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
|
||||
|
||||
public void add(long packetId){
|
||||
resendPacketIds.add(packetId);
|
||||
}
|
||||
|
||||
public List<Long> getResendPacketIds() {
|
||||
return resendPacketIds;
|
||||
}
|
||||
BinaryMessage m = new BinaryMessage(toHostPort.length() * 2 + 20);
|
||||
|
||||
public static MessageResend readBinaryMessage(DataInput dataInput) throws IOException {
|
||||
|
||||
String hostPort = dataInput.readUTF();
|
||||
|
||||
MessageResend msg = new MessageResend(hostPort);
|
||||
|
||||
int numberOfPacketIds = dataInput.readInt();
|
||||
for (int i = 0; i < numberOfPacketIds; i++) {
|
||||
long packetId = dataInput.readLong();
|
||||
msg.add(packetId);
|
||||
}
|
||||
|
||||
return msg;
|
||||
}
|
||||
|
||||
public void writeBinaryMessage(BinaryMessageList msgList) throws IOException {
|
||||
|
||||
BinaryMessage m = new BinaryMessage(toHostPort.length() * 2 + 20);
|
||||
|
||||
DataOutputStream os = m.getOs();
|
||||
os.writeInt(BinaryMessage.TYPE_MSGRESEND);
|
||||
os.writeUTF(toHostPort);
|
||||
os.writeInt(resendPacketIds.size());
|
||||
for (int i = 0; i < resendPacketIds.size(); i++) {
|
||||
Long packetId = resendPacketIds.get(i);
|
||||
os.writeLong(packetId.longValue());
|
||||
}
|
||||
os.flush();
|
||||
msgList.add(m);
|
||||
DataOutputStream os = m.getOs();
|
||||
os.writeInt(BinaryMessage.TYPE_MSGRESEND);
|
||||
os.writeUTF(toHostPort);
|
||||
os.writeInt(resendPacketIds.size());
|
||||
for (int i = 0; i < resendPacketIds.size(); i++) {
|
||||
Long packetId = resendPacketIds.get(i);
|
||||
os.writeLong(packetId.longValue());
|
||||
}
|
||||
os.flush();
|
||||
msgList.add(m);
|
||||
}
|
||||
}
|
||||
|
||||
+76
-76
@@ -5,90 +5,90 @@ import java.util.Map;
|
||||
|
||||
public class OutgoingPacketsAcked {
|
||||
|
||||
private long minimumGotAllPacketId;
|
||||
private long minimumGotAllPacketId;
|
||||
|
||||
private final Map<String, GroupMemberAck> recievedByMap = new HashMap<String, GroupMemberAck>();
|
||||
private final Map<String, GroupMemberAck> recievedByMap = new HashMap<String, GroupMemberAck>();
|
||||
|
||||
public int getGroupSize() {
|
||||
synchronized (this) {
|
||||
return recievedByMap.size();
|
||||
}
|
||||
public int getGroupSize() {
|
||||
synchronized (this) {
|
||||
return recievedByMap.size();
|
||||
}
|
||||
}
|
||||
|
||||
public long getMinimumGotAllPacketId() {
|
||||
synchronized (this) {
|
||||
return minimumGotAllPacketId;
|
||||
}
|
||||
}
|
||||
|
||||
public void removeMember(String groupMember) {
|
||||
synchronized (this) {
|
||||
recievedByMap.remove(groupMember);
|
||||
resetGotAllMin();
|
||||
}
|
||||
}
|
||||
|
||||
private boolean resetGotAllMin() {
|
||||
|
||||
long tempMin = Long.MAX_VALUE;
|
||||
|
||||
for (GroupMemberAck groupMemAck : recievedByMap.values()) {
|
||||
long memberMin = groupMemAck.getGotAllPacketId();
|
||||
if (memberMin < tempMin) {
|
||||
tempMin = memberMin;
|
||||
}
|
||||
}
|
||||
|
||||
public long getMinimumGotAllPacketId() {
|
||||
synchronized (this) {
|
||||
return minimumGotAllPacketId;
|
||||
}
|
||||
if (tempMin != minimumGotAllPacketId) {
|
||||
minimumGotAllPacketId = tempMin;
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
|
||||
public void removeMember(String groupMember){
|
||||
synchronized (this) {
|
||||
recievedByMap.remove(groupMember);
|
||||
resetGotAllMin();
|
||||
}
|
||||
}
|
||||
|
||||
public long receivedAck(String groupMember, MessageAck ack) {
|
||||
|
||||
synchronized (this) {
|
||||
|
||||
boolean checkMin;
|
||||
|
||||
GroupMemberAck groupMemberAck = recievedByMap.get(groupMember);
|
||||
if (groupMemberAck == null) {
|
||||
groupMemberAck = new GroupMemberAck();
|
||||
groupMemberAck.setIfBigger(ack.getGotAllPacketId());
|
||||
recievedByMap.put(groupMember, groupMemberAck);
|
||||
checkMin = true;
|
||||
} else {
|
||||
checkMin = groupMemberAck.getGotAllPacketId() == minimumGotAllPacketId;
|
||||
groupMemberAck.setIfBigger(ack.getGotAllPacketId());
|
||||
}
|
||||
|
||||
boolean minChanged = false;
|
||||
|
||||
if (checkMin || minimumGotAllPacketId == 0) {
|
||||
minChanged = resetGotAllMin();
|
||||
}
|
||||
|
||||
return minChanged ? minimumGotAllPacketId : 0;
|
||||
}
|
||||
}
|
||||
|
||||
private static class GroupMemberAck {
|
||||
|
||||
private long gotAllPacketId;
|
||||
|
||||
private GroupMemberAck() {
|
||||
}
|
||||
|
||||
private boolean resetGotAllMin() {
|
||||
|
||||
long tempMin = Long.MAX_VALUE;
|
||||
|
||||
for (GroupMemberAck groupMemAck : recievedByMap.values()) {
|
||||
long memberMin = groupMemAck.getGotAllPacketId();
|
||||
if (memberMin < tempMin) {
|
||||
tempMin = memberMin;
|
||||
}
|
||||
}
|
||||
|
||||
if (tempMin != minimumGotAllPacketId) {
|
||||
minimumGotAllPacketId = tempMin;
|
||||
return true;
|
||||
} else {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
public long receivedAck(String groupMember, MessageAck ack) {
|
||||
|
||||
synchronized (this) {
|
||||
|
||||
boolean checkMin;
|
||||
|
||||
GroupMemberAck groupMemberAck = recievedByMap.get(groupMember);
|
||||
if (groupMemberAck == null) {
|
||||
groupMemberAck = new GroupMemberAck();
|
||||
groupMemberAck.setIfBigger(ack.getGotAllPacketId());
|
||||
recievedByMap.put(groupMember, groupMemberAck);
|
||||
checkMin = true;
|
||||
} else {
|
||||
checkMin = groupMemberAck.getGotAllPacketId() == minimumGotAllPacketId;
|
||||
groupMemberAck.setIfBigger(ack.getGotAllPacketId());
|
||||
}
|
||||
|
||||
boolean minChanged = false;
|
||||
|
||||
if (checkMin || minimumGotAllPacketId == 0){
|
||||
minChanged = resetGotAllMin();
|
||||
}
|
||||
|
||||
return minChanged ? minimumGotAllPacketId : 0;
|
||||
}
|
||||
private long getGotAllPacketId() {
|
||||
return gotAllPacketId;
|
||||
}
|
||||
|
||||
private static class GroupMemberAck {
|
||||
|
||||
private long gotAllPacketId;
|
||||
|
||||
private GroupMemberAck() {
|
||||
}
|
||||
|
||||
private long getGotAllPacketId() {
|
||||
return gotAllPacketId;
|
||||
}
|
||||
|
||||
private void setIfBigger(long newGotAll) {
|
||||
if (newGotAll > gotAllPacketId) {
|
||||
gotAllPacketId = newGotAll;
|
||||
}
|
||||
}
|
||||
private void setIfBigger(long newGotAll) {
|
||||
if (newGotAll > gotAllPacketId) {
|
||||
gotAllPacketId = newGotAll;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+36
-39
@@ -1,66 +1,63 @@
|
||||
package com.avaje.ebeaninternal.server.cluster.mcast;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.Packet;
|
||||
|
||||
import java.util.Iterator;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.TreeMap;
|
||||
|
||||
import com.avaje.ebeaninternal.server.cluster.Packet;
|
||||
|
||||
/**
|
||||
* Cache of the outgoing packets.
|
||||
* <p>
|
||||
* These are held until we receive ACKs from the other members of the cluster to
|
||||
* say they have received the packets.
|
||||
* </p>
|
||||
*
|
||||
* @author rbygrave
|
||||
*
|
||||
*/
|
||||
public class OutgoingPacketsCache {
|
||||
|
||||
private final Map<Long, Packet> packetMap = new TreeMap<Long, Packet>();
|
||||
private final Map<Long, Packet> packetMap = new TreeMap<Long, Packet>();
|
||||
|
||||
public int size() {
|
||||
return packetMap.size();
|
||||
}
|
||||
public int size() {
|
||||
return packetMap.size();
|
||||
}
|
||||
|
||||
public Packet getPacket(Long packetId) {
|
||||
return packetMap.get(packetId);
|
||||
}
|
||||
public Packet getPacket(Long packetId) {
|
||||
return packetMap.get(packetId);
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return packetMap.keySet().toString();
|
||||
}
|
||||
public String toString() {
|
||||
return packetMap.keySet().toString();
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove the packet when we give up trying to send it out.
|
||||
*/
|
||||
public void remove(Packet packet) {
|
||||
packetMap.remove(packet.getPacketId());
|
||||
}
|
||||
/**
|
||||
* Remove the packet when we give up trying to send it out.
|
||||
*/
|
||||
public void remove(Packet packet) {
|
||||
packetMap.remove(packet.getPacketId());
|
||||
}
|
||||
|
||||
public void registerPackets(List<Packet> packets) {
|
||||
for (int i = 0; i < packets.size(); i++) {
|
||||
Packet p = packets.get(i);
|
||||
packetMap.put(p.getPacketId(), p);
|
||||
}
|
||||
public void registerPackets(List<Packet> packets) {
|
||||
for (int i = 0; i < packets.size(); i++) {
|
||||
Packet p = packets.get(i);
|
||||
packetMap.put(p.getPacketId(), p);
|
||||
}
|
||||
}
|
||||
|
||||
public int trimAll() {
|
||||
int size = packetMap.size();
|
||||
packetMap.clear();
|
||||
return size;
|
||||
}
|
||||
public int trimAll() {
|
||||
int size = packetMap.size();
|
||||
packetMap.clear();
|
||||
return size;
|
||||
}
|
||||
|
||||
public void trimAcknowledgedMessages(long minAcked) {
|
||||
Iterator<Long> it = packetMap.keySet().iterator();
|
||||
while (it.hasNext()) {
|
||||
Long pktId = it.next();
|
||||
if (minAcked >= pktId) {
|
||||
it.remove();
|
||||
}
|
||||
}
|
||||
public void trimAcknowledgedMessages(long minAcked) {
|
||||
Iterator<Long> it = packetMap.keySet().iterator();
|
||||
while (it.hasNext()) {
|
||||
Long pktId = it.next();
|
||||
if (minAcked >= pktId) {
|
||||
it.remove();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user