diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java index 7d12e6b72..447194c9c 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java @@ -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 messages = new ArrayList(); - - 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 getMessages() { - return messages; - } + final ArrayList messages = new ArrayList(); + + 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 getMessages() { + return messages; + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java index a9301b926..854f3c712 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java @@ -11,43 +11,41 @@ import java.util.List; *

* Thread Safety note: Object only used by McastClusterBroadcast Manager thread. * So Single Threaded access. - * - * @author rbygrave */ public class IncomingPacketsLastAck { - private final HashMap lastAckMap = new HashMap(); + private final HashMap lastAckMap = new HashMap(); - 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 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 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); + } } + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java index c74879e09..372bb88ef 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java @@ -20,256 +20,252 @@ import java.util.concurrent.ConcurrentHashMap; * have received and RESEND messages to fill the missing packets we have * detected. *

- * - * @author rbygrave - * */ public class IncomingPacketsProcessed { - private final ConcurrentHashMap mapByMember = new ConcurrentHashMap(); + private final ConcurrentHashMap mapByMember = new ConcurrentHashMap(); + 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. + *

+ * 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. + *

+ */ + 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 outOfOrderList = new ArrayList(); + + private final HashMap resendCountMap = new HashMap(); + + 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 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. - *

- * 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. - *

- */ - 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 outOfOrderList = new ArrayList(); - - private final HashMap resendCountMap = new HashMap(); - - 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 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 missingList = new ArrayList(); - if (getMissingPacketCount() > 0) { - // Ask for these Packets to be RESENT - List 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 getMissingPackets() { - - synchronized (this) { - ArrayList missingList = new ArrayList(); - - // 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 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 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); + } - - + } + + } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java index 85e41de83..b949834a9 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java @@ -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); } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java index a5c7f3953..2eaaad97e 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java @@ -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()); + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java index 510bfa90b..6c456b4ea 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java @@ -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 groupMembers = new HashSet(); + private final HashSet groupMembers = new HashSet(); - 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 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 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); + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java index 4ea5c86af..80994cd8d 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java @@ -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 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 packets) throws IOException { - - int totalBytes = 0; - for (int i = 0; i < packets.size(); i++) { - totalBytes += sendPacket(packets.get(i)); - } - return totalBytes; - } - } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java index 3b0429665..e4c1659cf 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java @@ -5,133 +5,130 @@ package com.avaje.ebeaninternal.server.cluster.mcast; *

* Ideally you want to see relatively low Re-send statistics. *

- * - * @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; - } - } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java index 230c75fff..7f21a7cf8 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java @@ -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(); } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java index 53cf1c93c..a261a4060 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java @@ -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); + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java index 2c2053ac3..e917c9db8 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java @@ -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); + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java index 34c2fc729..c712b6c67 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java @@ -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 resendPacketIds; + private final String toHostPort; - public MessageResend(String toHostPort, List resendPacketIds) { - this.toHostPort = toHostPort; - this.resendPacketIds = resendPacketIds; - } - - public MessageResend(String toHostPort) { - this(toHostPort, new ArrayList(4)); + private final List resendPacketIds; + + public MessageResend(String toHostPort, List resendPacketIds) { + this.toHostPort = toHostPort; + this.resendPacketIds = resendPacketIds; + } + + public MessageResend(String toHostPort) { + this(toHostPort, new ArrayList(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 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 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); + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java index be95d5802..8e7a836e9 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java @@ -5,90 +5,90 @@ import java.util.Map; public class OutgoingPacketsAcked { - private long minimumGotAllPacketId; + private long minimumGotAllPacketId; - private final Map recievedByMap = new HashMap(); + private final Map recievedByMap = new HashMap(); - 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; + } } + } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java index eff026335..6ffaac2f2 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java @@ -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. *

* These are held until we receive ACKs from the other members of the cluster to * say they have received the packets. *

- * - * @author rbygrave - * */ public class OutgoingPacketsCache { - private final Map packetMap = new TreeMap(); + private final Map packetMap = new TreeMap(); - 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 packets) { - for (int i = 0; i < packets.size(); i++) { - Packet p = packets.get(i); - packetMap.put(p.getPacketId(), p); - } + public void registerPackets(List 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 it = packetMap.keySet().iterator(); - while (it.hasNext()) { - Long pktId = it.next(); - if (minAcked >= pktId) { - it.remove(); - } - } + public void trimAcknowledgedMessages(long minAcked) { + Iterator it = packetMap.keySet().iterator(); + while (it.hasNext()) { + Long pktId = it.next(); + if (minAcked >= pktId) { + it.remove(); + } } + } }