From 9a912390d41abc1ca09ac97a575a85c8ecacef0e Mon Sep 17 00:00:00 2001 From: Robin Bygrave Date: Wed, 30 Mar 2016 16:20:24 +1300 Subject: [PATCH] #628 - Refactor Clustering - extract ClusterBroadcast implementations (TCP and Multicast) --- .../com/avaje/ebean/EbeanServerFactory.java | 2 +- .../avaje/ebean/config/ContainerConfig.java | 452 +------------- .../server/cluster/ClusterBroadcast.java | 2 +- .../cluster/ClusterBroadcastFactory.java | 14 + .../server/cluster/ClusterManager.java | 60 +- .../server/cluster/DataHolder.java | 24 - .../ebeaninternal/server/cluster/Packet.java | 193 ------ .../server/cluster/PacketMessages.java | 69 --- .../cluster/PacketTransactionEvent.java | 64 -- .../server/cluster/PacketWriter.java | 160 ----- .../cluster/SerialiseTransactionHelper.java | 60 -- .../cluster/mcast/AckResendMessages.java | 41 -- .../cluster/mcast/IncomingPacketsLastAck.java | 51 -- .../mcast/IncomingPacketsProcessed.java | 271 --------- .../cluster/mcast/McastClusterManager.java | 571 ------------------ .../server/cluster/mcast/McastListener.java | 237 -------- .../cluster/mcast/McastPacketControl.java | 125 ---- .../server/cluster/mcast/McastSender.java | 114 ---- .../server/cluster/mcast/McastStatus.java | 134 ---- .../server/cluster/mcast/Message.java | 14 - .../server/cluster/mcast/MessageAck.java | 57 -- .../server/cluster/mcast/MessageControl.java | 77 --- .../server/cluster/mcast/MessageResend.java | 77 --- .../cluster/mcast/OutgoingPacketsAcked.java | 94 --- .../cluster/mcast/OutgoingPacketsCache.java | 63 -- .../cluster/socket/RequestProcessor.java | 60 -- .../server/cluster/socket/SocketClient.java | 132 ---- .../socket/SocketClusterBroadcast.java | 250 -------- .../cluster/socket/SocketClusterListener.java | 143 ----- .../cluster/socket/SocketClusterMessage.java | 78 --- .../cluster/socket/SocketClusterStatus.java | 39 -- .../cluster/socket/SocketConnection.java | 75 --- .../server/transaction/BeanPersistIds.java | 61 +- .../server/transaction/DeleteByIdMap.java | 6 +- .../transaction/PostCommitProcessing.java | 19 +- .../transaction/RemoteTransactionEvent.java | 25 +- .../transaction/TransactionManager.java | 26 +- .../cluster/mcast/TestMcastMemberPackets.java | 59 -- .../cluster/mcast/TestPacketsAcked.java | 32 - .../socket/SocketClusterBroadcastTest.java | 79 --- 40 files changed, 134 insertions(+), 3976 deletions(-) create mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java delete mode 100644 src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java delete mode 100644 src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestMcastMemberPackets.java delete mode 100644 src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestPacketsAcked.java delete mode 100644 src/test/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcastTest.java diff --git a/src/main/java/com/avaje/ebean/EbeanServerFactory.java b/src/main/java/com/avaje/ebean/EbeanServerFactory.java index 290def9fb..d6326b5ea 100644 --- a/src/main/java/com/avaje/ebean/EbeanServerFactory.java +++ b/src/main/java/com/avaje/ebean/EbeanServerFactory.java @@ -133,7 +133,7 @@ public class EbeanServerFactory { /** * Create the container instance using the configuration. */ - private static SpiContainer createContainer(ContainerConfig containerConfig) { + protected static SpiContainer createContainer(ContainerConfig containerConfig) { String implClassName = System.getProperty("ebean.container", DEFAULT_CONTAINER); diff --git a/src/main/java/com/avaje/ebean/config/ContainerConfig.java b/src/main/java/com/avaje/ebean/config/ContainerConfig.java index f25a97759..c8f243645 100644 --- a/src/main/java/com/avaje/ebean/config/ContainerConfig.java +++ b/src/main/java/com/avaje/ebean/config/ContainerConfig.java @@ -1,7 +1,5 @@ package com.avaje.ebean.config; -import java.util.ArrayList; -import java.util.List; import java.util.Properties; /** @@ -12,453 +10,51 @@ import java.util.Properties; */ public class ContainerConfig { + protected boolean clusterActive; + + protected Properties properties; + /** - * Communication mode used for clustering. + * Return true if clustering is active. */ - public enum ClusterMode { - - /** - * No clustering. - */ - NONE, - - /** - * Use Multicast networking for cluster wide communication. - */ - MULTICAST, - - /** - * Use TCP Sockets for cluster wide communication. - */ - SOCKET + public boolean isClusterActive() { + return clusterActive; } /** - * The cluster mode to use. + * Set to true for clustering to be active. */ - ClusterMode mode = ClusterMode.NONE; - - /** - * Configuration if using TCP sockets for clustering communication. - */ - SocketConfig socketConfig = new SocketConfig(); - - /** - * Configuration if using Multicast for clustering communication. - */ - MulticastConfig multicastConfig = new MulticastConfig(); - - - // ------------------------------------------------------------------------------------------- - // MulticastConfig - - /** - * The configuration for clustering using Multicast networking. - */ - public static class MulticastConfig { - - int managerSleepMillis = 80; - int lastSendTimeFreqSecs = 300;//5mins - int lastStatusTimeFreqSecs = 600;//10mins - int maxResendOutgoingAttempts = 200; - int maxResendIncomingRequests = 50; - - int listenPort; - String listenAddress; - int sendPort; - String sendAddress; - - // Note 1500 is Ethernet MTU and this must be less than UDP max packet size of 65507 - int maxSendPacketSize = 1500; - - // Whether to send packets even when there are no other members online - boolean sendWithNoMembers = true; - - // When multiple instances are on same box you need to broadcast back locally - boolean disableLoopback; - int listenTimeToLive = -1; - int listenTimeout = 1000; - int listenBufferSize = 65500; - // For multihomed environment the address the listener should bind to - String listenBindAddress; - - /** - * Return the manager sleep millis. - */ - public int getManagerSleepMillis() { - return managerSleepMillis; - } - - /** - * Set the manager sleep millis. - */ - public void setManagerSleepMillis(int managerSleepMillis) { - this.managerSleepMillis = managerSleepMillis; - } - - /** - * Return the last send time frequency. - */ - public int getLastSendTimeFreqSecs() { - return lastSendTimeFreqSecs; - } - - /** - * Set the last send time frequency. - */ - public void setLastSendTimeFreqSecs(int lastSendTimeFreqSecs) { - this.lastSendTimeFreqSecs = lastSendTimeFreqSecs; - } - - /** - * Return the last status time frequency. - */ - public int getLastStatusTimeFreqSecs() { - return lastStatusTimeFreqSecs; - } - - /** - * Set the last status time frequency. - */ - public void setLastStatusTimeFreqSecs(int lastStatusTimeFreqSecs) { - this.lastStatusTimeFreqSecs = lastStatusTimeFreqSecs; - } - - /** - * Return the maximum number of times we will try to re-send a given packet before giving up sending - */ - public int getMaxResendOutgoingAttempts() { - return maxResendOutgoingAttempts; - } - - /** - * Set the maximum retry attempts for outgoing messages. - */ - public void setMaxResendOutgoingAttempts(int maxResendOutgoingAttempts) { - this.maxResendOutgoingAttempts = maxResendOutgoingAttempts; - } - - /** - * Return the maximum number of times we will ask for a packet to be resent to us before giving up asking. - */ - public int getMaxResendIncomingRequests() { - return maxResendIncomingRequests; - } - - /** - * Set the maximum retry attempts for incoming messages. - */ - public void setMaxResendIncomingRequests(int maxResendIncomingRequests) { - this.maxResendIncomingRequests = maxResendIncomingRequests; - } - - /** - * Return the listen port. - */ - public int getListenPort() { - return listenPort; - } - - /** - * Set the listen port. - */ - public void setListenPort(int port) { - this.listenPort = port; - } - - /** - * Return the listen address. - */ - public String getListenAddress() { - return listenAddress; - } - - /** - * Set the listen address. - */ - public void setListenAddress(String listenAddress) { - this.listenAddress = listenAddress; - } - - /** - * Return the send port. - */ - public int getSendPort() { - return sendPort; - } - - /** - * Set the send port. - */ - public void setSendPort(int sendPort) { - this.sendPort = sendPort; - } - - /** - * Return the send address. - */ - public String getSendAddress() { - return sendAddress; - } - - /** - * Set the send address. - */ - public void setSendAddress(String sendAddress) { - this.sendAddress = sendAddress; - } - - /** - * Return the maximum send packet size. - */ - public int getMaxSendPacketSize() { - return maxSendPacketSize; - } - - /** - * Set the maximum send packet size. Note 1500 is Ethernet MTU and this must be less than UDP max packet size of 65507. - */ - public void setMaxSendPacketSize(int maxSendPacketSize) { - this.maxSendPacketSize = maxSendPacketSize; - } - - /** - * Return true if send messages when no other members in the cluster are up. - */ - public boolean isSendWithNoMembers() { - return sendWithNoMembers; - } - - /** - * Set true if send messages when no other members in the cluster are up. - */ - public void setSendWithNoMembers(boolean sendWithNoMembers) { - this.sendWithNoMembers = sendWithNoMembers; - } - - /** - * Return true if loopback is disabled. When multiple instances are on same box you need to broadcast back locally. - */ - public boolean isDisableLoopback() { - return disableLoopback; - } - - /** - * Set if loopback is disabled. When multiple instances are on same box you need to broadcast back locally. - */ - public void setDisableLoopback(boolean disableLoopback) { - this.disableLoopback = disableLoopback; - } - - /** - * Return the listen time to live. - */ - public int getListenTimeToLive() { - return listenTimeToLive; - } - - /** - * Set the listen time to live. - */ - public void setListenTimeToLive(int listenTimeToLive) { - this.listenTimeToLive = listenTimeToLive; - } - - /** - * Return the listen timeout. - */ - public int getListenTimeout() { - return listenTimeout; - } - - /** - * set the listen timeout. - */ - public void setListenTimeout(int listenTimeout) { - this.listenTimeout = listenTimeout; - } - - /** - * Return the listen buffer size. - */ - public int getListenBufferSize() { - return listenBufferSize; - } - - /** - * Set the listen buffer size. - */ - public void setListenBufferSize(int listenBufferSize) { - this.listenBufferSize = listenBufferSize; - } - - /** - * Return the listener bind address (optional). For multihomed environment the address the listener should bind to. - */ - public String getListenBindAddress() { - return listenBindAddress; - } - - /** - * Set the listener bind address (optional). For multihomed environment the address the listener should bind to. - */ - public void setListenBindAddress(String listenBindAddress) { - this.listenBindAddress = listenBindAddress; - } + public void setClusterActive(boolean clusterActive) { + this.clusterActive = clusterActive; } - // ------------------------------------------------------------------------------------------- - // SocketConfig - /** - * Configuration for clustering using TCP sockets. - *

- * This is good for when there are relatively small number of cluster members. + * Return the deployment properties. */ - public static class SocketConfig { - - /** - * This local server in host:port format. - */ - String localHostPort; - - /** - * All the cluster members in host:port format. - */ - List members = new ArrayList(); - - /** - * core threads for the associated thread pool. - */ - int coreThreads = 2; - - /** - * Max threads for the associated thread pool. - */ - int maxThreads = 16; - - String threadPoolName = "EbeanCluster"; - - /** - * Return the host and port for this server instance. - */ - public String getLocalHostPort() { - return localHostPort; - } - - /** - * Set the host and port for this server instance. - */ - public void setLocalHostPort(String localHostPort) { - this.localHostPort = localHostPort; - } - - /** - * Return all the host and port for all the members of the cluster. - */ - public List getMembers() { - return members; - } - - /** - * Set all the host and port for all the members of the cluster. - */ - public void setMembers(List members) { - this.members = members; - } - - /** - * Return the number of core threads to use. - */ - public int getCoreThreads() { - return coreThreads; - } - - /** - * Set the number of core threads to use. - */ - public void setCoreThreads(int coreThreads) { - this.coreThreads = coreThreads; - } - - /** - * Return the number of max threads to use. - */ - public int getMaxThreads() { - return maxThreads; - } - - /** - * Set the number of max threads to use. - */ - public void setMaxThreads(int maxThreads) { - this.maxThreads = maxThreads; - } - - /** - * Return the thread pool name. - */ - public String getThreadPoolName() { - return threadPoolName; - } - - /** - * Set the thread pool name. - */ - public void setThreadPoolName(String threadPoolName) { - this.threadPoolName = threadPoolName; - } + public Properties getProperties() { + return properties; } - // ------------------------------------------------------------------------------------------- - // Members + /** + * Set the deployment properties. + */ + public void setProperties(Properties properties) { + this.properties = properties; + } /** * Load the settings from properties. */ public void loadFromProperties(Properties properties) { - //TODO + this.properties = properties; + this.clusterActive = getProperty(properties, "ebean.cluster.active", clusterActive); } /** - * Return the cluster mode. + * Return the boolean property setting. */ - public ClusterMode getMode() { - return mode; + protected boolean getProperty(Properties properties, String key, boolean defaultValue) { + return "true".equalsIgnoreCase(properties.getProperty(key, Boolean.toString(defaultValue))); } - /** - * Set the cluster mode. - */ - public void setMode(ClusterMode mode) { - this.mode = mode; - } - - /** - * Return the socket communication configuration. - */ - public SocketConfig getSocketConfig() { - return socketConfig; - } - - /** - * Set the socket communication configuration. - */ - public void setSocketConfig(SocketConfig socketConfig) { - this.socketConfig = socketConfig; - } - - /** - * Return the multicast communication configuration. - */ - public MulticastConfig getMulticastConfig() { - return multicastConfig; - } - - /** - * Set the multicast communication configuration. - */ - public void setMulticastConfig(MulticastConfig multicastConfig) { - this.multicastConfig = multicastConfig; - } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java index 414c0bfaf..8bf919048 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcast.java @@ -12,7 +12,7 @@ public interface ClusterBroadcast { * Inform the other cluster members that this instance has come online and * start any listeners etc. */ - void startup(ClusterManager clusterManager); + void startup(); /** * Inform the other cluster members that this instance is leaving and diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java new file mode 100644 index 000000000..af303d919 --- /dev/null +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterBroadcastFactory.java @@ -0,0 +1,14 @@ +package com.avaje.ebeaninternal.server.cluster; + +import java.util.Properties; + +/** + * Factory to create the cluster broadcast service. + */ +public interface ClusterBroadcastFactory { + + /** + * Create the cluster transport with the manager and deployment properties. + */ + ClusterBroadcast create(ClusterManager manager, Properties properties); +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java index 67318d2d3..30af29e53 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java +++ b/src/main/java/com/avaje/ebeaninternal/server/cluster/ClusterManager.java @@ -2,12 +2,12 @@ package com.avaje.ebeaninternal.server.cluster; import com.avaje.ebean.EbeanServer; import com.avaje.ebean.config.ContainerConfig; -import com.avaje.ebeaninternal.server.cluster.mcast.McastClusterManager; -import com.avaje.ebeaninternal.server.cluster.socket.SocketClusterBroadcast; import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.Iterator; +import java.util.ServiceLoader; import java.util.concurrent.ConcurrentHashMap; /** @@ -15,6 +15,8 @@ import java.util.concurrent.ConcurrentHashMap; */ public class ClusterManager { + private static final Logger clusterLogger = LoggerFactory.getLogger("org.avaje.ebean.Cluster"); + private static final Logger logger = LoggerFactory.getLogger(ClusterManager.class); private final ConcurrentHashMap serverMap = new ConcurrentHashMap(); @@ -25,30 +27,33 @@ public class ClusterManager { private boolean started; - public ClusterManager(ContainerConfig containerConfig) { - - ContainerConfig.ClusterMode mode = containerConfig.getMode(); - try { - switch (mode) { - case SOCKET: { - this.broadcast = new SocketClusterBroadcast(containerConfig); - break; - } - case MULTICAST: { - this.broadcast = new McastClusterManager(containerConfig); - break; - } - default: { - this.broadcast = null; - } - } - - } catch (Exception e) { - logger.error("Error initialising ClusterManager type [" + mode + "]", e); - throw new RuntimeException(e); + public ClusterManager(ContainerConfig config) { + if (!config.isClusterActive()) { + broadcast = null; + } else { + ClusterBroadcastFactory factory = createFactory(); + broadcast = factory.create(this, config.getProperties()); } } + /** + * Return the ClusterTransportFactory via ServiceLoader. + */ + private ClusterBroadcastFactory createFactory() { + + ServiceLoader load = ServiceLoader.load(ClusterBroadcastFactory.class); + ClusterBroadcastFactory factory = null; + Iterator iterator = load.iterator(); + if (iterator.hasNext()) { + factory = iterator.next(); + } + if (factory == null) { + throw new IllegalStateException("No ClusterTransportFactory found in classpath. " + + " Probably need to add the avaje-ebeanorm-cluster dependency"); + } + return factory; + } + public void registerServer(EbeanServer server) { synchronized (monitor) { serverMap.put(server.getName(), server); @@ -67,7 +72,7 @@ public class ClusterManager { private void startup() { started = true; if (broadcast != null) { - broadcast.startup(this); + broadcast.startup(); } } @@ -81,9 +86,12 @@ public class ClusterManager { /** * Send the message headers and payload to every server in the cluster. */ - public void broadcast(RemoteTransactionEvent remoteTransEvent) { + public void broadcast(RemoteTransactionEvent event) { if (broadcast != null) { - broadcast.broadcast(remoteTransEvent); + if (clusterLogger.isDebugEnabled()) { + clusterLogger.debug("sending: {}", event); + } + broadcast.broadcast(event); } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java deleted file mode 100644 index 46f4e148c..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/DataHolder.java +++ /dev/null @@ -1,24 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster; - -import java.io.Serializable; - -/** - * Simple holder of binary data. - * Used to use Packet based serialisation of RemoteTransactionEvent - * with simple Java Serialisation of the DataHolder. - */ -public class DataHolder implements Serializable { - - private static final long serialVersionUID = 9090748723571322192L; - - private final byte[] data; - - public DataHolder(byte[] data) { - this.data = data; - } - - public byte[] getData() { - return data; - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java deleted file mode 100644 index 1966a34f6..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/Packet.java +++ /dev/null @@ -1,193 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster; - -import java.io.ByteArrayOutputStream; -import java.io.DataInput; -import java.io.DataOutputStream; -import java.io.IOException; - -/** - * Represents the contents sent as a single DatagramPacket. - *

- * The contents is typically multiple messages (ACK,PING etc) or all or part of - * a RemoteTransactionEvent. - *

- *

- * Due to the hard limit on the size of UDP packets a RemoteTransactionEvent - * with lots of information could be broken up into multiple packets. - *

- * - * @author rbygrave - */ -public class Packet { - - /** - * A Packet that holds protocol messages like ACK, PING etc. - */ - public static final short TYPE_MESSAGES = 1; - - /** - * A Packet that holds TransactionEvent information such as Bean - * and or Table IUD information. - */ - public static final short TYPE_TRANSEVENT = 2; - - /** - * The type of Packet. - */ - protected final short packetType; - - /** - * The PacketId. - */ - protected final long packetId; - - /** - * The timestamp the Packet was created. - */ - protected final long timestamp; - - /** - * The EbeanServer name this relates to if relevant. - */ - protected final String serverName; - - protected ByteArrayOutputStream buffer; - protected DataOutputStream dataOut; - protected byte[] bytes; - - /** - * The number of messages in this Packet. - */ - private int messageCount; - - /** - * The number of times this Packet was resent. - */ - private int resendCount; - - /** - * Create a Packet for writing messages to. - */ - public static Packet forWrite(short packetType, long packetId, long timestamp, String serverName) throws IOException { - return new Packet(true, packetType, packetId, timestamp, serverName); - } - - /** - * Create a Packet just reading the Header information. - */ - public static Packet readHeader(DataInput dataInput) throws IOException { - - short packetType = dataInput.readShort(); - long packetId = dataInput.readLong(); - long timestamp = dataInput.readLong(); - String serverName = dataInput.readUTF(); - - return new Packet(false, packetType, packetId, timestamp, serverName); - } - - protected Packet(boolean write, short packetType, long packetId, long timestamp, String serverName) throws IOException { - this.packetType = packetType; - this.packetId = packetId; - this.timestamp = timestamp; - this.serverName = serverName; - if (write) { - this.buffer = new ByteArrayOutputStream(); - this.dataOut = new DataOutputStream(buffer); - writeHeader(); - } else { - this.buffer = null; - this.dataOut = null; - } - } - - private void writeHeader() throws IOException { - dataOut.writeShort(packetType); - dataOut.writeLong(packetId); - dataOut.writeLong(timestamp); - dataOut.writeUTF(serverName); - } - - public int incrementResendCount() { - return resendCount++; - } - - public short getPacketType() { - return packetType; - } - - public long getPacketId() { - return packetId; - } - - public long getTimestamp() { - return timestamp; - } - - public String getServerName() { - return serverName; - } - - public void writeEof() throws IOException { - dataOut.writeBoolean(false); - } - - public void read(DataInput dataInput) throws IOException { - boolean more = dataInput.readBoolean(); - while (more) { - int msgType = dataInput.readInt(); - readMessage(dataInput, msgType); - // see if there is more information - more = dataInput.readBoolean(); - } - } - - /** - * Overridden by more specific Packet implementations to read the messages. - */ - protected void readMessage(DataInput dataInput, int msgType) throws IOException { - - } - - /** - * Write a binary message to this packet returning true if there was - * enough room to do so. Return false if the message was too large for - * the remaining space left - in this case another Packet should be - * created to put that message into. - */ - public boolean writeBinaryMessage(BinaryMessage msg, int maxPacketSize) throws IOException { - - byte[] bytes = msg.getByteArray(); - - if (messageCount > 0 && (bytes.length + buffer.size() > maxPacketSize)) { - // we are actually going to ignore the maxPacketSize iff we have one - // large message. - - // false = no more messages - dataOut.writeBoolean(false); - return false; - } - ++messageCount; - // true = another message follows - dataOut.writeBoolean(true); - dataOut.write(bytes); - return true; - } - - public int getSize() { - return getBytes().length; - } - - /** - * Return the Packet as raw bytes. - */ - public byte[] getBytes() { - if (bytes == null) { - bytes = buffer.toByteArray(); - buffer = null; - dataOut = null; - } - return bytes; - } - - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java deleted file mode 100644 index cc43039f7..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketMessages.java +++ /dev/null @@ -1,69 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster; - -import com.avaje.ebeaninternal.server.cluster.mcast.Message; -import com.avaje.ebeaninternal.server.cluster.mcast.MessageAck; -import com.avaje.ebeaninternal.server.cluster.mcast.MessageControl; -import com.avaje.ebeaninternal.server.cluster.mcast.MessageResend; - -import java.io.DataInput; -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; - -/** - * A Packet that contains Ack, Resend and Control messages. - * - * @author rbygrave - */ -public class PacketMessages extends Packet { - - private final ArrayList messages; - - public static PacketMessages forWrite(long packetId, long timestamp, String serverName) throws IOException { - return new PacketMessages(true, packetId, timestamp, serverName); - } - - public static PacketMessages forRead(Packet header) throws IOException { - return new PacketMessages(header); - } - - private PacketMessages(boolean write, long packetId, long timestamp, String serverName) throws IOException { - super(write, TYPE_MESSAGES, packetId, timestamp, serverName); - this.messages = null; - } - - private PacketMessages(Packet header) throws IOException { - super(false, TYPE_MESSAGES, header.packetId, header.timestamp, header.serverName); - this.messages = new ArrayList(); - } - - /** - * Return the messages contained in this Packet. - */ - public List getMessages() { - return messages; - } - - /** - * Read the messages (Ack, Resend or Control) contained in this packet. - */ - protected void readMessage(DataInput dataInput, int msgType) throws IOException { - - switch (msgType) { - case BinaryMessage.TYPE_MSGCONTROL: - messages.add(MessageControl.readBinaryMessage(dataInput)); - break; - - case BinaryMessage.TYPE_MSGACK: - messages.add(MessageAck.readBinaryMessage(dataInput)); - break; - - case BinaryMessage.TYPE_MSGRESEND: - messages.add(MessageResend.readBinaryMessage(dataInput)); - break; - - default: - throw new RuntimeException("Invalid Transaction msgType " + msgType); - } - } -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java deleted file mode 100644 index a4dac5d2b..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketTransactionEvent.java +++ /dev/null @@ -1,64 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster; - -import com.avaje.ebeaninternal.api.SpiEbeanServer; -import com.avaje.ebeaninternal.api.TransactionEventTable.TableIUD; -import com.avaje.ebeaninternal.server.transaction.BeanPersistIds; -import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; - -import java.io.DataInput; -import java.io.IOException; - -/** - * A Packet holding TransactionEvent data. - *

- * Due to the hard limit for UDP packet sizes a RemoteTransactionEvent - * is actually broken up into smaller messages. - *

- */ -public class PacketTransactionEvent extends Packet { - - private final SpiEbeanServer server; - - private final RemoteTransactionEvent event; - - public static PacketTransactionEvent forWrite(long packetId, long timestamp, String serverName) throws IOException { - return new PacketTransactionEvent(true, packetId, timestamp, serverName); - } - - private PacketTransactionEvent(boolean write, long packetId, long timestamp, String serverName) throws IOException { - super(write, TYPE_TRANSEVENT, packetId, timestamp, serverName); - this.server = null; - this.event = null; - } - - private PacketTransactionEvent(Packet header, SpiEbeanServer server) throws IOException { - super(false, TYPE_TRANSEVENT, header.packetId, header.timestamp, header.serverName); - this.server = server; - this.event = new RemoteTransactionEvent(server); - } - - public static PacketTransactionEvent forRead(Packet header, SpiEbeanServer server) throws IOException { - return new PacketTransactionEvent(header, server); - } - - public RemoteTransactionEvent getEvent() { - return event; - } - - protected void readMessage(DataInput dataInput, int msgType) throws IOException { - - switch (msgType) { - case BinaryMessage.TYPE_BEANIUD: - event.addBeanPersistIds(BeanPersistIds.readBinaryMessage(server, dataInput)); - break; - - case BinaryMessage.TYPE_TABLEIUD: - event.addTableIUD(TableIUD.readBinaryMessage(dataInput)); - break; - - default: - throw new RuntimeException("Invalid Transaction msgType " + msgType); - } - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java deleted file mode 100644 index bc3ee0775..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/PacketWriter.java +++ /dev/null @@ -1,160 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster; - -import com.avaje.ebeaninternal.server.cluster.mcast.Message; -import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; - -import java.io.IOException; -import java.util.ArrayList; -import java.util.List; - -/** - * Creates Packets for either RemoteTransactionEvents or Messages (Ping, ACK, - * Join, Leave etc). - * - * @author rbygrave - */ -public class PacketWriter { - - private final PacketIdGenerator idGenerator; - private final PacketBuilder messagesPacketBuilder; - private final PacketBuilder transEventPacketBuilder; - - /** - * Create a PacketWriter with an expected max packet size. - *

- * In theory we would prefer to create packets up to the MTU size which for - * Ethernet will likely be 1500. Note that the maxPacketSize is ignored for - * large single messages. - *

- */ - public PacketWriter(int maxPacketSize) { - this.idGenerator = new PacketIdGenerator(); - this.messagesPacketBuilder = new PacketBuilder(maxPacketSize, idGenerator, new MessagesPacketFactory()); - this.transEventPacketBuilder = new PacketBuilder(maxPacketSize, idGenerator, new TransPacketFactory()); - } - - /** - * Return the currentPacketId. - */ - public long currentPacketId() { - return idGenerator.currentPacketId(); - } - - /** - * Create Packets for a given list of messages. - *

- * Typically this creates a single Packet but there is a hard limit for UDP - * packet sizes. - *

- */ - public List write(boolean requiresAck, List messages) throws IOException { - - BinaryMessageList binaryMsgList = new BinaryMessageList(); - for (int i = 0; i < messages.size(); i++) { - Message message = messages.get(i); - message.writeBinaryMessage(binaryMsgList); - } - return messagesPacketBuilder.write(requiresAck, binaryMsgList, ""); - } - - /** - * Create Packets for a given RemoteTransactionEvent. - *

- * Typically this creates a single Packet but there is a hard limit for UDP - * packet sizes. - *

- */ - public List write(RemoteTransactionEvent transEvent) throws IOException { - - BinaryMessageList messageList = new BinaryMessageList(); - - // split into reasonably small independent messages - transEvent.writeBinaryMessage(messageList); - - return transEventPacketBuilder.write(true, messageList, transEvent.getServerName()); - } - - /** - * Reuse the same packetIdCounter for building Packets for both Message and - * RemoteTransactionEvent - */ - private static class PacketIdGenerator { - - long packetIdCounter; - - public long nextPacketId() { - return ++packetIdCounter; - } - - public long currentPacketId() { - return packetIdCounter; - } - - } - - interface PacketFactory { - - Packet createPacket(long packetId, long timestamp, String serverName) throws IOException; - } - - private static class TransPacketFactory implements PacketFactory { - - public Packet createPacket(long packetId, long timestamp, String serverName) throws IOException { - return PacketTransactionEvent.forWrite(packetId, timestamp, serverName); - } - } - - private static class MessagesPacketFactory implements PacketFactory { - - public Packet createPacket(long packetId, long timestamp, String serverName) throws IOException { - return PacketMessages.forWrite(packetId, timestamp, serverName); - } - } - - /** - * Helper class for building Packets from messages or - * RemoteTransactionEvents. - */ - private static class PacketBuilder { - - private final PacketIdGenerator idGenerator; - private final PacketFactory packetFactory; - private final int maxPacketSize; - - private PacketBuilder(int maxPacketSize, PacketIdGenerator idGenerator, PacketFactory packetFactory) { - this.maxPacketSize = maxPacketSize; - this.idGenerator = idGenerator; - this.packetFactory = packetFactory; - } - - private List write(boolean requiresAck, BinaryMessageList messageList, String serverName) - throws IOException { - - List list = messageList.getList(); - - ArrayList packets = new ArrayList(1); - - long timestamp = System.currentTimeMillis(); - - long packetId = requiresAck ? idGenerator.nextPacketId() : 0; - Packet p = packetFactory.createPacket(packetId, timestamp, serverName); - - packets.add(p); - - for (int i = 0; i < list.size(); i++) { - BinaryMessage binMsg = list.get(i); - if (!p.writeBinaryMessage(binMsg, maxPacketSize)) { - // didn't fit into the package so put into another packet - packetId = requiresAck ? idGenerator.nextPacketId() : 0; - p = packetFactory.createPacket(packetId, timestamp, serverName); - packets.add(p); - p.writeBinaryMessage(binMsg, maxPacketSize); - } - } - p.writeEof(); - - return packets; - - } - } -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java deleted file mode 100644 index daab0f50d..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/SerialiseTransactionHelper.java +++ /dev/null @@ -1,60 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster; - -import com.avaje.ebeaninternal.api.SpiEbeanServer; -import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.ByteArrayInputStream; -import java.io.DataInputStream; -import java.io.IOException; -import java.util.List; - -/** - * Mechanism to convert RemoteTransactionEvent to/from byte[] content. - */ -public abstract class SerialiseTransactionHelper { - - private static final Logger logger = LoggerFactory.getLogger(SerialiseTransactionHelper.class); - - private final PacketWriter packetWriter; - - public SerialiseTransactionHelper() { - packetWriter = new PacketWriter(Integer.MAX_VALUE); - } - - public abstract SpiEbeanServer getEbeanServer(String serverName); - - /** - * Convert the RemoteTransactionEvent to byte[] content. - */ - public DataHolder createDataHolder(RemoteTransactionEvent transEvent) throws IOException { - - List packetList = packetWriter.write(transEvent); - if (packetList.size() != 1) { - throw new RuntimeException("Always expecting 1 Packet but got " + packetList.size()); - } - byte[] data = packetList.get(0).getBytes(); - return new DataHolder(data); - } - - /** - * Convert the byte[] content to RemoteTransactionEvent. - */ - public RemoteTransactionEvent read(DataHolder dataHolder) throws IOException { - - ByteArrayInputStream bi = new ByteArrayInputStream(dataHolder.getData()); - DataInputStream dataInput = new DataInputStream(bi); - - Packet header = Packet.readHeader(dataInput); - - SpiEbeanServer server = getEbeanServer(header.getServerName()); - if (server == null) { - logger.error("server [{}] not found/registered?", header.getServerName()); - } - - PacketTransactionEvent tranEventPacket = PacketTransactionEvent.forRead(header, server); - tranEventPacket.read(dataInput); - return tranEventPacket.getEvent(); - } -} 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 deleted file mode 100644 index 447194c9c..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/AckResendMessages.java +++ /dev/null @@ -1,41 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.util.ArrayList; -import java.util.List; - -/** - * Holds a list of ACK and RESEND messages that should be sent out. - */ -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; - } -} 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 deleted file mode 100644 index 854f3c712..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsLastAck.java +++ /dev/null @@ -1,51 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.util.HashMap; -import java.util.List; - -/** - * For this node this holds the ACK gotAllPoint for each member in the cluster. - *

- * As we receive messages from other members of the cluster periodically we need - * to send them ACK messages to say we got all the packets up to the gotAllPoint. - *

- * Thread Safety note: Object only used by McastClusterBroadcast Manager thread. - * So Single Threaded access. - */ -public class IncomingPacketsLastAck { - - 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); - } - } - } -} 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 deleted file mode 100644 index 372bb88ef..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/IncomingPacketsProcessed.java +++ /dev/null @@ -1,271 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.Iterator; -import java.util.List; -import java.util.concurrent.ConcurrentHashMap; - -/** - * For Incoming Packets remembers the packets we have received and processed. - *

- * This determines the gotAllPoint per cluster member and identifies missing - * packets (gap between gotAllPoint and gotMaxPoint). - *

- *

- * This information is used by the managerThread so send ACK's for messages we - * have received and RESEND messages to fill the missing packets we have - * detected. - *

- */ -public class IncomingPacketsProcessed { - - 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; - - 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; - } - - /** - * Add ACK and RESEND messages if required. - */ - public void addAckResendMessages(AckResendMessages response, MessageAck lastAckMessage) { - - 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)); - } - } - } - - public String getMemberKey() { - return memberKey; - } - - 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); - } 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 deleted file mode 100644 index b949834a9..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastClusterManager.java +++ /dev/null @@ -1,571 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import com.avaje.ebean.config.ContainerConfig; -import com.avaje.ebeaninternal.api.SpiEbeanServer; -import com.avaje.ebeaninternal.server.cluster.ClusterBroadcast; -import com.avaje.ebeaninternal.server.cluster.ClusterManager; -import com.avaje.ebeaninternal.server.cluster.Packet; -import com.avaje.ebeaninternal.server.cluster.PacketWriter; -import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.net.InetAddress; -import java.net.UnknownHostException; -import java.util.ArrayList; -import java.util.List; -import java.util.TreeSet; - -/** - * Overall Manager of the Multicast Cluster communication for this instance. - *

- * McastListener, McastSender and McastPacketControl are the main helpers to - * this object. - *

- *

- * This Manager (thread) periodically processes the ACK, Re-send and Control - * messages. The McastListener is handling all the incoming packets and informs - * this manager when interesting packets need to be processed by the Manager. - *

- *

- * Other threads call {@link #broadcast(RemoteTransactionEvent)} to send - * transaction even information. - *

- */ -public class McastClusterManager implements ClusterBroadcast, Runnable { - - private static final Logger logger = LoggerFactory.getLogger(McastClusterManager.class); - - private ClusterManager clusterManager; - - private final Thread managerThread; - - /** - * Helps co-ordinate packet information (Acks, Missing Packets etc). - */ - private final McastPacketControl packageControl; - - /** - * Listeners for incoming packets. - */ - private final McastListener listener; - - /** - * Sends packets out to the cluster. - */ - private final McastSender localSender; - - /** - * The localSenderHostPort is used to identify this instance in the cluster. - */ - private final String localSenderHostPort; - - /** - * Creates the Packets (byte[]) from Messages and RemoteTransactionEvent. - */ - private final PacketWriter packetWriter; - - /** - * List of Re-send messages that the managerThread needs to process. - */ - private final ArrayList resendMessages = new ArrayList(); - - /** - * List of Control messages (Ping,PingResponse,Join,Leave) that the managerThread needs to process. - */ - private final ArrayList controlMessages = new ArrayList(); - - /** - * Cache of outgoing messages that have not been ACK'ed by the other cluster members yet. - */ - private final OutgoingPacketsCache outgoingPacketsCache = new OutgoingPacketsCache(); - - /** - * The last ACK we sent out to other members of the cluster. - */ - private final IncomingPacketsLastAck incomingPacketsLastAck = new IncomingPacketsLastAck(); - - /** - * A limit of the number of times we will try to send out a given packet. - * Once this is exceeded we will just drop that packet. Hopefully this does - * not happen but we don't want to keep trying forever producing network - * traffic. - */ - private final int maxResendOutgoing; - - /** - * Instead of ACK'ing immediately we periodically wake up and in a single - * packet (typically) ACK all members of the cluster everything we got since - * the last sleep time. More frequent ACK's means less memory consumption as - * Packets are cleared from the outgoingPacketsCache quicker at the cost of - * sending more packets. - */ - private final long managerSleepMillis; - - /** - * When true then packets are still sent out even when the cluster has no other online members. - */ - private final boolean sendWithNoMembers; - - /** - * The current minAcked packetId processed by the managerThread. - * All packets before this have been ACK'ed by everyone in the cluster. - */ - private long minAcked; - - /** - * The min packetId that has been ACKed by all the members of the cluster according - * to the McastListener. This will increase as the Listener receives ACK's and means - * we can trim out Packets from the sent cache. - */ - private long minAckedFromListener; - - /** - * Start the groupSize at -1 so we have to wait until the Listener times out or gets - * a control messages (Ping, PingResponse, Join, Leave etc) before we know how many - * members of the group the listener knows about. - *

- * Generally speaking we only care if the groupSize == 0 meaning there are no other - * members of the cluster that are online. In this case we can potentially not send - * the packets out (depending on sendWithNoMembers) and not cache them (for re-sending - * if they where not ACK'ed). - *

- */ - private int currentGroupSize = -1; - - /** - * The last time a packet was sent from this node. - */ - private long lastSendTime; - - /** - * The max time we go without sending any packets. - */ - private final int lastSendTimeFreqMillis; - - /** - * The last time the cluster status was logged. - */ - private long lastStatusTime = System.currentTimeMillis(); - - /** - * The max time we go before logging the cluster status. - */ - private final int lastStatusTimeFreqMillis; - - - private long totalTxnEventsSent; - private long totalTxnEventsReceived; - - private long totalPacketsSent; - private long totalBytesSent; - - private long totalPacketsResent; - private long totalBytesResent; - - private long totalPacketsReceived; - private long totalBytesReceived; - - - public McastClusterManager(ContainerConfig containerConfig) { - - ContainerConfig.MulticastConfig config = containerConfig.getMulticastConfig(); - - this.managerSleepMillis = config.getManagerSleepMillis(); - this.lastSendTimeFreqMillis = 1000 * config.getLastSendTimeFreqSecs(); - this.lastStatusTimeFreqMillis = 1000 * config.getLastStatusTimeFreqSecs(); - - // the maximum number of times we will try to re-send a given packet before giving up sending - this.maxResendOutgoing = config.getMaxResendOutgoingAttempts(); - // the maximum number of times we will ask for a packet to be resent to us before giving up asking - int maxResendIncoming = config.getMaxResendIncomingRequests(); - - - int port = config.getListenPort(); - String addr = config.getListenAddress(); - - int sendPort = config.getSendPort(); - String sendAddr = config.getSendAddress(); - - // Sender options - // Note 1500 is Ethernet MTU and this must be less than UDP max packet size of 65507 - int maxSendPacketSize = config.getMaxSendPacketSize(); - // Whether to send packets even when there are no other members online - this.sendWithNoMembers = config.isSendWithNoMembers(); - - // Listener options - // When multiple instances are on same box you need to broadcast back locally - boolean disableLoopback = config.isDisableLoopback(); - int ttl = config.getListenTimeToLive(); - int timeout = config.getListenTimeout(); - int bufferSize = config.getListenBufferSize(); - // For multihomed environment the address the listener should bind to - String mcastAddr = config.getListenBindAddress(); - - InetAddress mcastBindAddress = null; - if (mcastAddr != null) { - try { - mcastBindAddress = InetAddress.getByName(mcastAddr); - } catch (UnknownHostException e) { - String msg = "Error getting Multicast InetAddress for " + mcastAddr; - throw new RuntimeException(msg, e); - } - } - - 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; - - throw new IllegalArgumentException(msg); - } - - this.managerThread = new Thread(this, "EbeanClusterMcastManager"); - - this.packetWriter = new PacketWriter(maxSendPacketSize); - this.localSender = new McastSender(port, addr, sendPort, sendAddr); - this.localSenderHostPort = localSender.getSenderHostPort(); - - this.packageControl = new McastPacketControl(this, localSenderHostPort, maxResendIncoming); - - this.listener = new McastListener(this, packageControl, port, addr, bufferSize, timeout, localSenderHostPort, - disableLoopback, ttl, mcastBindAddress); - } - - - /** - * The McastListener tells us there are no other members of the cluster that - * are currently online. - */ - protected void fromListenerTimeoutNoMembers() { - synchronized (managerThread) { - this.currentGroupSize = 0; - } - } - - /** - * McastListener calls this method to get the manager to process messages. - * - * @param newMinAcked the minAcked packetId according to the listener - * @param msgControl a control message to process - * @param msgResend a Please re-send message to process - * @param groupSize the number of other online members - */ - protected void fromListener(long newMinAcked, MessageControl msgControl, MessageResend msgResend, - int groupSize, long totalPacketsReceived, long totalBytesReceived, - long totalTxnEventsReceived) { - - synchronized (managerThread) { - if (newMinAcked > minAckedFromListener) { - minAckedFromListener = newMinAcked; - } - if (msgControl != null) { - controlMessages.add(msgControl); - } - if (msgResend != null) { - resendMessages.add(msgResend); - } - // mostly interested when groupSize hits 0 (we are the only instance online). - this.currentGroupSize = groupSize; - - // and some stats so we know how busy the listener has been - this.totalPacketsReceived = totalPacketsReceived; - this.totalBytesReceived = totalBytesReceived; - this.totalTxnEventsReceived = totalTxnEventsReceived; - } - } - - /** - * Get the overall status and activity of this cluster node. - */ - public McastStatus getStatus() { - - synchronized (managerThread) { - long currentPacketId = packetWriter.currentPacketId(); - String lastAcks = incomingPacketsLastAck.toString(); - - return new McastStatus(currentGroupSize, outgoingPacketsCache.size(), currentPacketId, minAcked, lastAcks, - totalTxnEventsSent, totalTxnEventsReceived, totalPacketsSent, totalPacketsResent, - totalPacketsReceived, - totalBytesSent, totalBytesResent, totalBytesReceived); - - } - } - - /** - * Periodically send out Ack, Re-send and Control messages. - */ - public void run() { - - while (true) { - try { - // sleep for a little bit as we ACK packets periodically - // rather than immediately. We will typically ACK many - // messages from all cluster members in a single Packet - Thread.sleep(managerSleepMillis); - - synchronized (managerThread) { - - handleControlMessages(); - - handleResendMessages(); - - if (currentGroupSize == 0) { - // no members online so trim the entire outgoing packets cache - int trimmedCount = outgoingPacketsCache.trimAll(); - if (trimmedCount > 0) { - logger.debug("Cluster has no other members. Trimmed " + trimmedCount); - } - - } else if (minAckedFromListener > minAcked) { - // ACKs have come back so trim send packets cache - outgoingPacketsCache.trimAcknowledgedMessages(minAckedFromListener); - minAcked = minAckedFromListener; - } - - // Get list of all the ACK messages required to sent since the last time. - // This is effectively one ACK message per member of the cluster. The ACK - // message covers all the packets received from the member up to - // the gotAllPoint. - // Also get any RESEND messages asking for packets that we have not - // received between the gotAllPoint and the gotMaxPoint. - AckResendMessages ackResendMessages = packageControl.getAckResendMessages(incomingPacketsLastAck); - - if (ackResendMessages.size() > 0) { - // send the ACK and RESEND messages for all members of the - // cluster typically in a single Packet - if (sendMessages(false, ackResendMessages.getMessages())) { - // update the last Ack position - incomingPacketsLastAck.updateLastAck(ackResendMessages); - } - } - - if (lastSendTime < System.currentTimeMillis() - lastSendTimeFreqMillis) { - // been quite for too long - send a Ping out - sendPing(); - } - - if (lastStatusTimeFreqMillis > 0) { - if (lastStatusTime < System.currentTimeMillis() - lastStatusTimeFreqMillis) { - McastStatus status = getStatus(); - logger.info("Cluster Status: " + status.getSummary()); - lastStatusTime = System.currentTimeMillis(); - } - } - - } - } catch (Exception e) { - logger.error("Error with Cluster Mcast Manager thread", e); - } - } - } - - /** - * We have been asked to Re-send some packets. - */ - private void handleResendMessages() { - - if (resendMessages.size() > 0) { - - TreeSet s = new TreeSet(); - for (int i = 0; i < resendMessages.size(); i++) { - MessageResend resendMsg = resendMessages.get(i); - s.addAll(resendMsg.getResendPacketIds()); - } - - totalPacketsResent += s.size(); - - for (Long resendPacketId : s) { - Packet packet = outgoingPacketsCache.getPacket(resendPacketId); - if (packet == null) { - String msg = "Cluster unable to resend packet[" + resendPacketId + "] as it is no longer in the " + - "outgoingPacketsCache"; - logger.error(msg); - } else { - int resendCount = packet.incrementResendCount(); - if (resendCount <= maxResendOutgoing) { - resendPacket(packet); - } else { - String msg = "Cluster maxResendOutgoing [" + maxResendOutgoing + "] hit for packet " + resendPacketId - + ". We will not try to send it anymore, removing it from the outgoingPacketsCache."; - logger.error(msg); - outgoingPacketsCache.remove(packet); - } - } - } - } - } - - /** - * Re-send a packet that a member didn't seem to receive. - */ - private void resendPacket(Packet packet) { - try { - ++totalPacketsResent; - totalBytesResent += localSender.sendPacket(packet); - } catch (IOException e) { - String msg = "Error trying to resend packet " + packet.getPacketId(); - logger.error(msg, e); - } - } - - /** - * Handle Control messages (Join, Leave, Ping). - */ - private void handleControlMessages() { - - boolean pingReponse = false; - boolean joinReponse = false; - - for (int i = 0; i < controlMessages.size(); i++) { - MessageControl message = controlMessages.get(i); - - short type = message.getControlType(); - switch (type) { - case MessageControl.TYPE_JOIN: - // a new member online, send back a Join Response - logger.info("Cluster member Joined [" + message.getFromHostPort() + "]"); - joinReponse = true; - break; - - case MessageControl.TYPE_JOINRESPONSE: - logger.info("Cluster member Online [" + message.getFromHostPort() + "]"); - // do nothing - break; - - case MessageControl.TYPE_PING: - pingReponse = true; - break; - - case MessageControl.TYPE_PINGRESPONSE: - // do nothing - break; - - case MessageControl.TYPE_LEAVE: - // remove member. If/When that member comes back its - // packetIds will have been reset - incomingPacketsLastAck.remove(message.getFromHostPort()); - break; - - default: - break; - } - } - controlMessages.clear(); - - if (joinReponse) { - sendJoinResponse(); - } - if (pingReponse) { - sendPingResponse(); - } - } - - /** - * Say 'Leaving' and shutdown. - */ - public void shutdown() { - sendLeave(); - listener.shutdown(); - } - - /** - * Startup listeners and 'Join'. - */ - public void startup(ClusterManager clusterManager) { - this.clusterManager = clusterManager; - listener.startListening(); - - this.managerThread.setDaemon(true); - this.managerThread.start(); - - sendJoin(); - } - - protected SpiEbeanServer getEbeanServer(String serverName) { - return (SpiEbeanServer) clusterManager.getServer(serverName); - } - - private void sendJoin() { - sendControlMessage(true, MessageControl.TYPE_JOIN); - } - - private void sendLeave() { - sendControlMessage(false, MessageControl.TYPE_LEAVE); - } - - private void sendJoinResponse() { - sendControlMessage(true, MessageControl.TYPE_JOINRESPONSE); - } - - private void sendPingResponse() { - sendControlMessage(true, MessageControl.TYPE_PINGRESPONSE); - } - - private void sendPing() { - sendControlMessage(true, MessageControl.TYPE_PING); - } - - private void sendControlMessage(boolean requiresAck, short controlType) { - sendMessage(requiresAck, new MessageControl(controlType, localSenderHostPort)); - } - - private void sendMessage(boolean requiresAck, Message msg) { - ArrayList messages = new ArrayList(1); - messages.add(msg); - sendMessages(requiresAck, messages); - } - - private boolean sendMessages(boolean requiresAck, List messages) { - - synchronized (managerThread) { - try { - - List packets = packetWriter.write(requiresAck, messages); - sendPackets(requiresAck, packets); - return true; - - } catch (IOException e) { - String msg = "Error sending Messages " + messages; - logger.error(msg, e); - return false; - } - } - } - - private boolean sendPackets(boolean requiresAck, List packets) throws IOException { - if (currentGroupSize == 0 && !sendWithNoMembers) { - // no other members online so not sending packets - return false; - - } else { - if (requiresAck) { - // cache them until they have been ACK'ed - outgoingPacketsCache.registerPackets(packets); - } - totalPacketsSent += packets.size(); - totalBytesSent += localSender.sendPackets(packets); - lastSendTime = System.currentTimeMillis(); - return true; - } - } - - /** - * Send the remoteTransEvent to all the other members of the cluster. - */ - public void broadcast(RemoteTransactionEvent remoteTransEvent) { - - synchronized (managerThread) { - try { - List packets = packetWriter.write(remoteTransEvent); - if (sendPackets(true, packets)) { - ++totalTxnEventsSent; - } - } catch (IOException e) { - String msg = "Error sending RemoteTransactionEvent " + remoteTransEvent; - logger.error(msg, e); - } - } - } - -} 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 deleted file mode 100644 index 2eaaad97e..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastListener.java +++ /dev/null @@ -1,237 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import com.avaje.ebeaninternal.api.SpiEbeanServer; -import com.avaje.ebeaninternal.server.cluster.Packet; -import com.avaje.ebeaninternal.server.cluster.PacketTransactionEvent; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.ByteArrayInputStream; -import java.io.DataInput; -import java.io.DataInputStream; -import java.io.IOException; -import java.net.DatagramPacket; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.net.MulticastSocket; - -/** - * Listens for Incoming packets. - */ -public class McastListener implements Runnable { - - private static final Logger logger = LoggerFactory.getLogger(McastListener.class); - - private final McastClusterManager owner; - - private final McastPacketControl packetControl; - - private final MulticastSocket sock; - - private final Thread listenerThread; - - private final String localSenderHostPort; - - private final InetAddress group; - - private DatagramPacket pack; - - private final byte[] receiveBuffer; - - 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) { - - 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); - } - } 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(); - } - } - - 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 { - - SpiEbeanServer server = owner.getEbeanServer(header.getServerName()); - - PacketTransactionEvent tranEventPacket = PacketTransactionEvent.forRead(header, server); - tranEventPacket.read(dataInput); - - 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 deleted file mode 100644 index 6c456b4ea..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastPacketControl.java +++ /dev/null @@ -1,125 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -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 final String localSenderHostPort; - - private final McastClusterManager owner; - - private final HashSet groupMembers = new HashSet(); - - private final OutgoingPacketsAcked outgoingPacketsAcked = new OutgoingPacketsAcked(); - - private final IncomingPacketsProcessed incomingPacketsProcessed; - - 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()); - } - } - } - - 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); - } - } - - /** - * 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); - } - - 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 deleted file mode 100644 index 80994cd8d..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastSender.java +++ /dev/null @@ -1,114 +0,0 @@ -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; -import java.net.InetAddress; -import java.net.InetSocketAddress; -import java.util.List; - -/** - * Handles the sending of Packets via DatagramPacket. - */ -public class McastSender { - - private static final Logger logger = LoggerFactory.getLogger(McastSender.class); - - private final int port; - - private final InetAddress inetAddress; - - private final DatagramSocket sock; - - private final InetSocketAddress sendAddr; - - private final String senderHostPort; - - - public McastSender(int port, String address, int sendPort, String sendAddress) { - - try { - this.port = port; - this.inetAddress = InetAddress.getByName(address); - - InetAddress sendInetAddress; - if (sendAddress != null) { - sendInetAddress = InetAddress.getByName(sendAddress); - } else { - sendInetAddress = InetAddress.getLocalHost(); - } - - if (sendPort > 0) { - this.sock = new DatagramSocket(sendPort, sendInetAddress); - } else { - this.sock = new DatagramSocket(new InetSocketAddress(sendInetAddress, 0)); - } - - 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); - } - - 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 deleted file mode 100644 index e4c1659cf..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/McastStatus.java +++ /dev/null @@ -1,134 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -/** - * Gives an overall status of this Cluster instance. - *

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

- */ -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 int currentGroupSize; - private final int outgoingPacketsCacheSize; - - private final long currentPacketId; - private final long minAckedPacketId; - private final String lastOutgoingAcks; - - public String getSummary() { - - //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("; "); - - 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; - - this.totalBytesSent = totalBytesSent; - this.totalBytesResent = totalBytesResent; - this.totalBytesReceived = totalBytesReceived; - - } - - - public long getTotalTxnEventsReceived() { - return totalTxnEventsReceived; - } - - public long getTotalPacketsReceived() { - return totalPacketsReceived; - } - - public long getTotalBytesSent() { - return totalBytesSent; - } - - public long getTotalBytesResent() { - return totalBytesResent; - } - - 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; - } - -} 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 deleted file mode 100644 index 7f21a7cf8..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/Message.java +++ /dev/null @@ -1,14 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import com.avaje.ebeaninternal.server.cluster.BinaryMessageList; - -import java.io.IOException; - -public interface Message { - - void writeBinaryMessage(BinaryMessageList msgList) throws IOException; - - 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 deleted file mode 100644 index a261a4060..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageAck.java +++ /dev/null @@ -1,57 +0,0 @@ -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; - -public class MessageAck implements Message { - - private final String toHostPort; - - private final long gotAllPacketId; - - public MessageAck(String toHostPort, long gotAllPacketId) { - this.toHostPort = toHostPort; - this.gotAllPacketId = gotAllPacketId; - } - - public String toString() { - return "Ack " + toHostPort + " " + gotAllPacketId; - } - - public boolean isControlMessage() { - return false; - } - - public String getToHostPort() { - return toHostPort; - } - - 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 deleted file mode 100644 index e917c9db8..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageControl.java +++ /dev/null @@ -1,77 +0,0 @@ -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; - -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 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; - } - - public short getControlType() { - return controlType; - } - - 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 deleted file mode 100644 index c712b6c67..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/MessageResend.java +++ /dev/null @@ -1,77 +0,0 @@ -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; - -public class MessageResend implements Message { - - private final String toHostPort; - - 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); - } - - 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); - } -} 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 deleted file mode 100644 index 8e7a836e9..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsAcked.java +++ /dev/null @@ -1,94 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.util.HashMap; -import java.util.Map; - -public class OutgoingPacketsAcked { - - private long minimumGotAllPacketId; - - private final Map recievedByMap = new HashMap(); - - 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; - } - } - - 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 static class GroupMemberAck { - - private long gotAllPacketId; - - private GroupMemberAck() { - } - - private long getGotAllPacketId() { - return gotAllPacketId; - } - - 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 deleted file mode 100644 index 6ffaac2f2..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/mcast/OutgoingPacketsCache.java +++ /dev/null @@ -1,63 +0,0 @@ -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; - -/** - * 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. - *

- */ -public class OutgoingPacketsCache { - - private final Map packetMap = new TreeMap(); - - public int size() { - return packetMap.size(); - } - - public Packet getPacket(Long packetId) { - return packetMap.get(packetId); - } - - 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()); - } - - 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 void trimAcknowledgedMessages(long minAcked) { - Iterator it = packetMap.keySet().iterator(); - while (it.hasNext()) { - Long pktId = it.next(); - if (minAcked >= pktId) { - it.remove(); - } - } - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java deleted file mode 100644 index a4c89516d..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/RequestProcessor.java +++ /dev/null @@ -1,60 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.net.Socket; - -/** - * This parses and dispatches a request to the appropriate handler. - *

- * Looks up the appropriate RequestHandler - * and then gets it to process the Client request.

- *

- * Note that this is a Runnable because it is assigned to the ThreadPool. - */ -class RequestProcessor implements Runnable { - - private static final Logger logger = LoggerFactory.getLogger(RequestProcessor.class); - - private final Socket clientSocket; - - private final SocketClusterBroadcast owner; - - private final String hostPort; - - /** - * Create including the Listener (used to lookup the Request Handler) and - * the socket itself. - */ - public RequestProcessor(SocketClusterBroadcast owner, Socket clientSocket) { - this.clientSocket = clientSocket; - this.owner = owner; - this.hostPort = owner.getHostPort(); - } - - /** - * This will parse out the command. Lookup the appropriate Handler and - * pass the information to the handler for processing. - *

Dev Note: the command parsing is processed here so that it is preformed - * by the assigned thread rather than the listeners thread.

- */ - public void run() { - try { - logger.trace("start listening for cluster messages"); - SocketConnection sc = new SocketConnection(clientSocket); - while (true) { - if (owner.process(sc)) { - // got the offline message or timeout - break; - } - } - logger.trace("disconnecting: {}", hostPort); - sc.disconnect(); - - } catch (Exception e) { - logger.error("Error listening for messages - " + owner.getHostPort(), e); - } - } - -} \ No newline at end of file diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java deleted file mode 100644 index acefd636d..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClient.java +++ /dev/null @@ -1,132 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.ObjectOutputStream; -import java.io.OutputStream; -import java.net.InetSocketAddress; -import java.net.Socket; - - -/** - * The client side of the socket clustering. - */ -class SocketClient { - - private static final Logger logger = LoggerFactory.getLogger(SocketClient.class); - - private final InetSocketAddress address; - - private final String hostPort; - - private boolean online; - - private Socket socket; - private OutputStream os; - private ObjectOutputStream oos; - - /** - * Construct with an IP address and port. - */ - public SocketClient(InetSocketAddress address) { - this.address = address; - this.hostPort = address.getHostName() + ":" + address.getPort(); - } - - public String toString() { - return address.toString(); - } - - public String getHostPort() { - return hostPort; - } - - public int getPort() { - return address.getPort(); - } - - public boolean isOnline() { - return online; - } - - public void setOnline(boolean online) throws IOException { - if (online) { - setOnline(); - } else { - disconnect(); - } - } - - - /** - * Set whether the client is thought to be online. - */ - private void setOnline() throws IOException { - connect(); - this.online = true; - } - - public void reconnect() throws IOException { - disconnect(); - connect(); - } - - private void connect() throws IOException { - if (socket != null) { - throw new IllegalStateException("Already got a socket connection?"); - } - Socket s = new Socket(); - s.setKeepAlive(true); - s.connect(address); - - this.socket = s; - this.os = socket.getOutputStream(); - } - - public void disconnect() { - this.online = false; - if (socket != null) { - - try { - socket.close(); - } catch (IOException e) { - String msg = "Error disconnecting from Cluster member " + hostPort; - logger.info(msg, e); - } - - os = null; - oos = null; - socket = null; - } - } - - public boolean register(SocketClusterMessage registerMsg) { - - try { - setOnline(); - send(registerMsg); - return true; - } catch (IOException e) { - disconnect(); - return false; - } - } - - public void send(SocketClusterMessage msg) throws IOException { - - if (online) { - writeObject(msg); - } - } - - private void writeObject(Object object) throws IOException { - if (oos == null) { - this.oos = new ObjectOutputStream(os); - } - oos.writeObject(object); - oos.flush(); - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java deleted file mode 100644 index d77878b50..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcast.java +++ /dev/null @@ -1,250 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import com.avaje.ebean.config.ContainerConfig; -import com.avaje.ebeaninternal.api.SpiEbeanServer; -import com.avaje.ebeaninternal.server.cluster.ClusterBroadcast; -import com.avaje.ebeaninternal.server.cluster.ClusterManager; -import com.avaje.ebeaninternal.server.cluster.DataHolder; -import com.avaje.ebeaninternal.server.cluster.SerialiseTransactionHelper; -import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.EOFException; -import java.io.IOException; -import java.io.InterruptedIOException; -import java.net.InetSocketAddress; -import java.util.HashMap; -import java.util.List; -import java.util.concurrent.atomic.AtomicInteger; - -/** - * Broadcast messages across the cluster using sockets. - */ -public class SocketClusterBroadcast implements ClusterBroadcast { - - private static final Logger logger = LoggerFactory.getLogger(SocketClusterBroadcast.class); - - private final SocketClient local; - - private final HashMap clientMap; - - private final SocketClusterListener listener; - - private final SocketClient[] members; - - private ClusterManager clusterManager; - - private final TxnSerialiseHelper txnSerialiseHelper = new TxnSerialiseHelper(); - - private final AtomicInteger txnOutgoing = new AtomicInteger(); - private final AtomicInteger txnIncoming = new AtomicInteger(); - - public SocketClusterBroadcast(ContainerConfig containerConfig) { - - ContainerConfig.SocketConfig socketConfig = containerConfig.getSocketConfig(); - - String localHostPort = socketConfig.getLocalHostPort(); - List members = socketConfig.getMembers(); - - logger.info("Clustering using Sockets local[" + localHostPort + "] members[" + members + "]"); - - this.local = new SocketClient(parseFullName(localHostPort)); - this.clientMap = new HashMap(); - - for (String memberHostPort : members) { - InetSocketAddress member = parseFullName(memberHostPort); - SocketClient client = new SocketClient(member); - if (!local.getHostPort().equalsIgnoreCase(client.getHostPort())) { - // don't add the local one ... - clientMap.put(client.getHostPort(), client); - } - } - - this.members = clientMap.values().toArray(new SocketClient[clientMap.size()]); - this.listener = new SocketClusterListener(this, local.getPort(), socketConfig.getCoreThreads(), socketConfig.getMaxThreads(), socketConfig.getThreadPoolName()); - } - - public String getHostPort() { - return local.getHostPort(); - } - - /** - * Return the current status of this instance. - */ - public SocketClusterStatus getStatus() { - - // count of online members - int currentGroupSize = 0; - for (int i = 0; i < members.length; i++) { - if (members[i].isOnline()) { - ++currentGroupSize; - } - } - int txnIn = txnIncoming.get(); - int txnOut = txnOutgoing.get(); - - return new SocketClusterStatus(currentGroupSize, txnIn, txnOut); - } - - public void startup(ClusterManager clusterManager) { - - this.clusterManager = clusterManager; - listener.startListening(); - register(); - } - - public void shutdown() { - deregister(); - listener.shutdown(); - } - - /** - * Register with all the other members of the Cluster. - */ - private void register() { - - SocketClusterMessage h = SocketClusterMessage.register(local.getHostPort(), true); - - for (int i = 0; i < members.length; i++) { - boolean online = members[i].register(h); - logger.info("Cluster Member [{}] online[{}]", members[i].getHostPort(), online); - } - } - - protected void setMemberOnline(String fullName, boolean online) throws IOException { - synchronized (clientMap) { - logger.info("Cluster Member [{}] online[{}]", fullName, online); - SocketClient member = clientMap.get(fullName); - member.setOnline(online); - } - } - - private void send(SocketClient client, SocketClusterMessage msg) { - - try { - // alternative would be to connect/disconnect here but prefer to use keepalive - if (logger.isTraceEnabled()) { - logger.trace("... send to member {} broadcast msg: {}", client, msg); - } - client.send(msg); - - } catch (Exception ex) { - logger.error("Error sending message", ex); - try { - client.reconnect(); - } catch (IOException e) { - logger.error("Error trying to reconnect", ex); - } - } - } - - /** - * Send the payload to all the members of the cluster. - */ - public void broadcast(RemoteTransactionEvent remoteTransEvent) { - try { - txnOutgoing.incrementAndGet(); - DataHolder dataHolder = txnSerialiseHelper.createDataHolder(remoteTransEvent); - SocketClusterMessage msg = SocketClusterMessage.transEvent(dataHolder); - broadcast(msg); - } catch (Exception e) { - logger.error("Error sending RemoteTransactionEvent " + remoteTransEvent + " to cluster members.", e); - } - } - - protected void broadcast(SocketClusterMessage msg) { - - if (logger.isTraceEnabled()) { - logger.trace("... broadcast msg: " + msg); - } - for (int i = 0; i < members.length; i++) { - send(members[i], msg); - } - } - - /** - * Leave the cluster. - */ - private void deregister() { - - SocketClusterMessage h = SocketClusterMessage.register(local.getHostPort(), false); - broadcast(h); - for (int i = 0; i < members.length; i++) { - members[i].disconnect(); - } - } - - /** - * Process an incoming Cluster message. - */ - protected boolean process(SocketConnection request) throws ClassNotFoundException { - - try { - SocketClusterMessage h = (SocketClusterMessage) request.readObject(); - if (logger.isTraceEnabled()) { - logger.trace("... received msg: {}", h); - } - - if (h.isRegisterEvent()) { - setMemberOnline(h.getRegisterHost(), h.isRegister()); - - } else { - txnIncoming.incrementAndGet(); - DataHolder dataHolder = h.getDataHolder(); - RemoteTransactionEvent transEvent = txnSerialiseHelper.read(dataHolder); - transEvent.run(); - } - - // instance shutting down - return h.isRegisterEvent() && !h.isRegister(); - - } catch (InterruptedIOException e) { - logger.info("Timeout waiting for message", e); - try { - request.disconnect(); - } catch (IOException ex) { - logger.info("Error disconnecting after timeout", ex); - } - return true; - - } catch (EOFException e) { - logger.info("EOF disconnecting"); - return true; - } catch (IOException e) { - logger.info("IO Error waiting/reading message", e); - return true; - } - } - - /** - * Parse a host:port into a InetSocketAddress. - */ - private InetSocketAddress parseFullName(String hostAndPort) { - - try { - hostAndPort = hostAndPort.trim(); - int colonPos = hostAndPort.indexOf(":"); - if (colonPos == -1) { - String msg = "No colon \":\" in " + hostAndPort; - throw new IllegalArgumentException(msg); - } - String host = hostAndPort.substring(0, colonPos); - String sPort = hostAndPort.substring(colonPos + 1, hostAndPort.length()); - int port = Integer.parseInt(sPort); - - return new InetSocketAddress(host, port); - - } catch (Exception ex) { - throw new RuntimeException("Error parsing [" + hostAndPort + "] for the form [host:port]", ex); - } - } - - class TxnSerialiseHelper extends SerialiseTransactionHelper { - - @Override - public SpiEbeanServer getEbeanServer(String serverName) { - return (SpiEbeanServer) clusterManager.getServer(serverName); - } - } -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java deleted file mode 100644 index 10d55b978..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterListener.java +++ /dev/null @@ -1,143 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import com.avaje.ebeaninternal.server.lib.DaemonThreadPool; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.io.IOException; -import java.io.InterruptedIOException; -import java.net.ServerSocket; -import java.net.Socket; -import java.net.SocketException; - - -/** - * Serverside multithreaded socket listener. Accepts connections and dispatches - * them to an appropriate handler. - *

- * This is designed as a single port listener, where part of the connection - * protocol determines which service the client is requesting (rather than a - * port per service). - *

- *

- * It has its own daemon background thread that handles the accept() loop on the - * ServerSocket. - *

- */ -class SocketClusterListener implements Runnable { - - private static final Logger logger = LoggerFactory.getLogger(SocketClusterListener.class); - - /** - * The server socket used to listen for requests. - */ - private final ServerSocket serverListenSocket; - - /** - * The listening thread. - */ - private final Thread listenerThread; - - /** - * The pool of threads that actually do the parsing execution of requests. - */ - private final DaemonThreadPool threadPool; - - private final SocketClusterBroadcast owner; - - /** - * shutting down flag. - */ - boolean doingShutdown; - - /** - * Whether the listening thread is busy assigning a request to a thread. - */ - boolean isActive; - - /** - * Construct with a given thread pool name. - */ - public SocketClusterListener(SocketClusterBroadcast owner, int port, int coreThreads, int maxThreads, String poolName) { - this.owner = owner; - this.threadPool = new DaemonThreadPool(coreThreads, maxThreads, 60, 30, poolName); - try { - this.serverListenSocket = new ServerSocket(port); - this.serverListenSocket.setSoTimeout(60000); - this.listenerThread = new Thread(this, "EbeanClusterListener"); - - } catch (IOException e) { - String msg = "Error starting cluster socket listener on port " + port; - throw new RuntimeException(msg, e); - } - } - - /** - * Start listening for requests. - */ - public void startListening() { - logger.trace("... startListening()"); - this.listenerThread.setDaemon(true); - this.listenerThread.start(); - } - - /** - * Shutdown this listener. - */ - public void shutdown() { - doingShutdown = true; - try { - if (isActive) { - synchronized (listenerThread) { - try { - listenerThread.wait(1000); - } catch (InterruptedException e) { - // OK to ignore as expected to Interrupt for shutdown. - } - } - } - listenerThread.interrupt(); - serverListenSocket.close(); - } catch (IOException e) { - logger.error("Error shutting down listener", e); - } - - threadPool.shutdown(); - } - - /** - * This is a runnable and so this must be public. Don't call this externally - * but rather call the startListening() method. - */ - public void run() { - // run in loop until doingShutdown is true... - while (!doingShutdown) { - try { - synchronized (listenerThread) { - Socket clientSocket = serverListenSocket.accept(); - isActive = true; - - Runnable request = new RequestProcessor(owner, clientSocket); - threadPool.execute(request); - - isActive = false; - } - } catch (SocketException e) { - if (doingShutdown) { - logger.info("doingShutdown and accept threw:" + e.getMessage()); - } else { - logger.error("Error while listening", e); - } - } catch (InterruptedIOException e) { - // this will happen when the server is very quiet. - // that is, no requests - logger.debug("Possibly expected due to accept timeout? {}", e.getMessage()); - - } catch (IOException e) { - // log it and continue in the loop... - logger.error("IOException processing cluster message", e); - } - } - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java deleted file mode 100644 index 07bfd6d11..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterMessage.java +++ /dev/null @@ -1,78 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import com.avaje.ebeaninternal.server.cluster.DataHolder; -import com.avaje.ebeaninternal.server.cluster.Packet; - -import java.io.Serializable; - -/** - * The messages broadcast around the cluster. - */ -public class SocketClusterMessage implements Serializable { - - private static final long serialVersionUID = 2993350408394934473L; - - private final String registerHost; - - private final boolean register; - - private final DataHolder dataHolder; - - public static SocketClusterMessage register(String registerHost, boolean register) { - return new SocketClusterMessage(registerHost, register); - } - - public static SocketClusterMessage transEvent(DataHolder transEvent) { - return new SocketClusterMessage(transEvent); - } - - public static SocketClusterMessage packet(Packet packet) { - DataHolder d = new DataHolder(packet.getBytes()); - return new SocketClusterMessage(d); - } - - /** - * Used to construct a Child AttributeMap. - */ - private SocketClusterMessage(String registerHost, boolean register) { - this.registerHost = registerHost; - this.register = register; - this.dataHolder = null; - } - - private SocketClusterMessage(DataHolder dataHolder) { - this.dataHolder = dataHolder; - this.registerHost = null; - this.register = false; - } - - public String toString() { - StringBuilder sb = new StringBuilder(); - if (registerHost != null) { - sb.append("register "); - sb.append(register); - sb.append(" "); - sb.append(registerHost); - } else { - sb.append("transEvent "); - } - return sb.toString(); - } - - public boolean isRegisterEvent() { - return registerHost != null; - } - - public String getRegisterHost() { - return registerHost; - } - - public boolean isRegister() { - return register; - } - - public DataHolder getDataHolder() { - return dataHolder; - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java deleted file mode 100644 index cafd5bc74..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterStatus.java +++ /dev/null @@ -1,39 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -/** - * The current state of this cluster member. - */ -public class SocketClusterStatus { - - private final int currentGroupSize; - private final int txnIncoming; - private final int txtOutgoing; - - public SocketClusterStatus(int currentGroupSize, int txnIncoming, int txnOutgoing) { - this.currentGroupSize = currentGroupSize; - this.txnIncoming = txnIncoming; - this.txtOutgoing = txnOutgoing; - } - - /** - * Return the number of members of the cluster currently online. - */ - public int getCurrentGroupSize() { - return currentGroupSize; - } - - /** - * Return the number of Remote transactions received. - */ - public int getTxnIncoming() { - return txnIncoming; - } - - /** - * Return the number of transactions sent to the cluster. - */ - public int getTxtOutgoing() { - return txtOutgoing; - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java b/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java deleted file mode 100644 index 81f8fd154..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/cluster/socket/SocketConnection.java +++ /dev/null @@ -1,75 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import java.io.IOException; -import java.io.InputStream; -import java.io.ObjectInputStream; -import java.io.OutputStream; -import java.net.Socket; - -/** - * The client side of a TCP Sockect connection. - */ -class SocketConnection { - - /** - * The underlying ObjectInputStream. - */ - ObjectInputStream ois; - - /** - * The underlying inputStream. - */ - InputStream is; - - /** - * The underlying outputStream. - */ - OutputStream os; - - /** - * The underlying socket. - */ - Socket socket; - - /** - * Create for a given Socket. - */ - public SocketConnection(Socket socket) throws IOException { - this.is = socket.getInputStream(); - this.os = socket.getOutputStream(); - this.socket = socket; - } - - /** - * Disconnect from the server. - */ - public void disconnect() throws IOException { - os.flush(); - socket.close(); - } - - /** - * Flush the outputStream. - */ - public void flush() throws IOException { - os.flush(); - } - - /** - * Read an object from the object input stream. - */ - public Object readObject() throws IOException, ClassNotFoundException { - return getObjectInputStream().readObject(); - } - - /** - * Get the object input stream. - */ - public ObjectInputStream getObjectInputStream() throws IOException { - if (ois == null) { - ois = new ObjectInputStream(is); - } - return ois; - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java index f44dfdefc..018939b9d 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java +++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/BeanPersistIds.java @@ -28,17 +28,15 @@ import java.util.List; * size of data sent around the network. *

*/ -public class BeanPersistIds implements Serializable { +public class BeanPersistIds { - private static final long serialVersionUID = 8389469180931531409L; - - private transient BeanDescriptor beanDescriptor; + private final BeanDescriptor beanDescriptor; private final String descriptorId; - private ArrayList insertIds; - private ArrayList updateIds; - private ArrayList deleteIds; + private List insertIds; + private List updateIds; + private List deleteIds; /** * Create the payload. @@ -62,8 +60,7 @@ public class BeanPersistIds implements Serializable { IdBinder idBinder = beanDescriptor.getIdBinder(); int iudType = dataInput.readInt(); - ArrayList idList = readIdList(dataInput, idBinder); - + List idList = readIdList(dataInput, idBinder); switch (iudType) { case 0: insertIds = idList; @@ -89,24 +86,23 @@ public class BeanPersistIds implements Serializable { * across multiple Packets. *

*/ - public void writeBinaryMessage(BinaryMessageList msgList) throws IOException { + void writeBinaryMessage(BinaryMessageList msgList) throws IOException { writeIdList(beanDescriptor, 0, insertIds, msgList); writeIdList(beanDescriptor, 1, updateIds, msgList); writeIdList(beanDescriptor, 2, deleteIds, msgList); - } - private ArrayList readIdList(DataInput dataInput, IdBinder idBinder) throws IOException { + private List readIdList(DataInput dataInput, IdBinder idBinder) throws IOException { int count = dataInput.readInt(); if (count < 1) { return null; } - ArrayList idList = new ArrayList(count); + List idList = new ArrayList(count); for (int i = 0; i < count; i++) { Object id = idBinder.readData(dataInput); - idList.add((Serializable) id); + idList.add(id); } return idList; } @@ -121,8 +117,7 @@ public class BeanPersistIds implements Serializable { * Packets. *

*/ - private void writeIdList(BeanDescriptor desc, int iudType, ArrayList idList, - BinaryMessageList msgList) throws IOException { + private void writeIdList(BeanDescriptor desc, int iudType, List idList, BinaryMessageList msgList) throws IOException { IdBinder idBinder = desc.getIdBinder(); @@ -144,8 +139,7 @@ public class BeanPersistIds implements Serializable { os.writeInt(count); for (; i < endOfLoop; i++) { - Serializable idValue = idList.get(i); - idBinder.writeData(os, idValue); + idBinder.writeData(os, idList.get(i)); } os.flush(); @@ -174,7 +168,7 @@ public class BeanPersistIds implements Serializable { return sb.toString(); } - public void addId(PersistRequest.Type type, Serializable id) { + void addId(PersistRequest.Type type, Serializable id) { switch (type) { case INSERT: addInsertId(id); @@ -194,21 +188,21 @@ public class BeanPersistIds implements Serializable { private void addInsertId(Serializable id) { if (insertIds == null) { - insertIds = new ArrayList(); + insertIds = new ArrayList(); } insertIds.add(id); } private void addUpdateId(Serializable id) { if (updateIds == null) { - updateIds = new ArrayList(); + updateIds = new ArrayList(); } updateIds.add(id); } private void addDeleteId(Serializable id) { if (deleteIds == null) { - deleteIds = new ArrayList(); + deleteIds = new ArrayList(); } deleteIds.add(id); } @@ -217,19 +211,15 @@ public class BeanPersistIds implements Serializable { return beanDescriptor; } - public List getDeleteIds() { + List getDeleteIds() { return deleteIds; } - public void setBeanDescriptor(BeanDescriptor beanDescriptor) { - this.beanDescriptor = beanDescriptor; - } - /** * Notify the cache and local BeanPersistListener of this event that came * from another server in the cluster. */ - public void notifyCacheAndListener() { + void notifyCacheAndListener() { BeanPersistListener listener = beanDescriptor.getPersistListener(); @@ -238,7 +228,6 @@ public class BeanPersistIds implements Serializable { if (insertIds != null) { if (listener != null) { - // notify listener for (int i = 0; i < insertIds.size(); i++) { listener.remoteInsert(insertIds.get(i)); } @@ -246,25 +235,19 @@ public class BeanPersistIds implements Serializable { } if (updateIds != null) { for (int i = 0; i < updateIds.size(); i++) { - Serializable id = updateIds.get(i); - - // remove from cache + Object id = updateIds.get(i); beanDescriptor.cacheBeanRemove(id); if (listener != null) { - // notify listener - listener.remoteInsert(id); + listener.remoteUpdate(id); } } } if (deleteIds != null) { for (int i = 0; i < deleteIds.size(); i++) { - Serializable id = deleteIds.get(i); - - // remove from cache + Object id = deleteIds.get(i); beanDescriptor.cacheBeanRemove(id); if (listener != null) { - // notify listener - listener.remoteInsert(id); + listener.remoteDelete(id); } } } diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/DeleteByIdMap.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/DeleteByIdMap.java index 69171623f..80dcbb959 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/transaction/DeleteByIdMap.java +++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/DeleteByIdMap.java @@ -27,7 +27,7 @@ public final class DeleteByIdMap { public void notifyCache() { for (BeanPersistIds deleteIds : beanMap.values()) { BeanDescriptor d = deleteIds.getBeanDescriptor(); - List idValues = deleteIds.getDeleteIds(); + List idValues = deleteIds.getDeleteIds(); if (idValues != null) { d.queryCacheClear(); for (int i = 0; i < idValues.size(); i++) { @@ -79,7 +79,7 @@ public final class DeleteByIdMap { /** * Add the deletes to the DocStoreUpdates. */ - public void addDocStoreUpdates(DocStoreUpdates docStoreUpdates, DocStoreMode txnIndexMode) { + void addDocStoreUpdates(DocStoreUpdates docStoreUpdates, DocStoreMode txnIndexMode) { for (BeanPersistIds deleteIds : beanMap.values()) { BeanDescriptor desc = deleteIds.getBeanDescriptor(); DocStoreMode mode = desc.getDocStoreMode(PersistRequest.Type.DELETE, txnIndexMode); @@ -87,7 +87,7 @@ public final class DeleteByIdMap { // Add to queue or bulk update entries boolean queue = (DocStoreMode.QUEUE == mode); String queueId = desc.getDocStoreQueueId(); - List idValues = deleteIds.getDeleteIds(); + List idValues = deleteIds.getDeleteIds(); if (idValues != null) { for (int i = 0; i < idValues.size(); i++) { if (queue) { diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/PostCommitProcessing.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/PostCommitProcessing.java index be337e85f..7eb19291d 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/transaction/PostCommitProcessing.java +++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/PostCommitProcessing.java @@ -78,12 +78,11 @@ public final class PostCommitProcessing { this.remoteTransactionEvent = createRemoteTransactionEvent(); } - public void notifyLocalCacheIndex() { - - // notify cache with bulk insert/update/delete statements + /** + * Notify the local part of L2 cache. + */ + void notifyLocalCache() { processTableEvents(event.getEventTables()); - - // notify cache with bean changes event.notifyCache(); } @@ -105,7 +104,7 @@ public final class PostCommitProcessing { /** * Process any document store updates. */ - protected void processDocStoreUpdates() { + private void processDocStoreUpdates() { if (isDocStoreUpdate()) { // collect 'bulk update' and 'queue' events @@ -129,7 +128,7 @@ public final class PostCommitProcessing { return manager.isDocStoreActive() && (txnDocStoreMode == null || txnDocStoreMode != DocStoreMode.IGNORE); } - public void notifyCluster() { + private void notifyCluster() { if (remoteTransactionEvent != null && !remoteTransactionEvent.isEmpty()) { // send the interesting events to the cluster if (logger.isDebugEnabled()) { @@ -140,10 +139,14 @@ public final class PostCommitProcessing { } } - public Runnable notifyPersistListeners() { + /** + * In background notify persist listeners, cluster and document store. + */ + Runnable backgroundNotify() { return new Runnable() { public void run() { localPersistListenersNotify(); + notifyCluster(); processDocStoreUpdates(); } }; diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/RemoteTransactionEvent.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/RemoteTransactionEvent.java index 060cfda5c..42b79bd3f 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/transaction/RemoteTransactionEvent.java +++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/RemoteTransactionEvent.java @@ -33,11 +33,16 @@ public class RemoteTransactionEvent implements Runnable { } public String toString() { - StringBuilder sb = new StringBuilder(); - sb.append(beanPersistList); + StringBuilder sb = new StringBuilder(100); + if (!beanPersistList.isEmpty()) { + sb.append(beanPersistList); + } if (tableList != null) { sb.append(tableList); } + if (deleteByIdMap != null) { + sb.append(deleteByIdMap.values()); + } return sb.toString(); } @@ -55,15 +60,15 @@ public class RemoteTransactionEvent implements Runnable { } } - if (beanPersistList != null) { - for (int i = 0; i < beanPersistList.size(); i++) { - beanPersistList.get(i).writeBinaryMessage(msgList); - } + for (int i = 0; i < beanPersistList.size(); i++) { + beanPersistList.get(i).writeBinaryMessage(msgList); } } public boolean isEmpty() { - return beanPersistList.isEmpty() && (tableList == null || tableList.isEmpty()); + return beanPersistList.isEmpty() + && (tableList == null || tableList.isEmpty()) + && (deleteByIdMap == null || deleteByIdMap.isEmpty()); } public void addBeanPersistIds(BeanPersistIds beanPersist) { @@ -89,11 +94,7 @@ public class RemoteTransactionEvent implements Runnable { this.server = server; } - public DeleteByIdMap getDeleteByIdMap() { - return deleteByIdMap; - } - - public void setDeleteByIdMap(DeleteByIdMap deleteByIdMap) { + void setDeleteByIdMap(DeleteByIdMap deleteByIdMap) { this.deleteByIdMap = deleteByIdMap; } diff --git a/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java b/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java index 0f466c649..b68d97b78 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java +++ b/src/main/java/com/avaje/ebeaninternal/server/transaction/TransactionManager.java @@ -39,6 +39,8 @@ public class TransactionManager { private static final Logger logger = LoggerFactory.getLogger(TransactionManager.class); + public static final Logger clusterLogger = LoggerFactory.getLogger("org.avaje.ebean.Cluster"); + public static final Logger SQL_LOGGER = LoggerFactory.getLogger("org.avaje.ebean.SQL"); public static final Logger SUM_LOGGER = LoggerFactory.getLogger("org.avaje.ebean.SUM"); @@ -396,12 +398,8 @@ public class TransactionManager { } PostCommitProcessing postCommit = new PostCommitProcessing(clusterManager, this, transaction); - - postCommit.notifyLocalCacheIndex(); - postCommit.notifyCluster(); - - // cluster and text indexing - backgroundExecutor.execute(postCommit.notifyPersistListeners()); + postCommit.notifyLocalCache(); + backgroundExecutor.execute(postCommit.backgroundNotify()); for (TransactionEventListener listener : transactionEventListeners) { listener.postTransactionCommit(transaction); @@ -425,11 +423,8 @@ public class TransactionManager { event.add(tableEvents); PostCommitProcessing postCommit = new PostCommitProcessing(clusterManager, this, event); - - // invalidate parts of local cache and index - postCommit.notifyLocalCacheIndex(); - - backgroundExecutor.execute(postCommit.notifyPersistListeners()); + postCommit.notifyLocalCache(); + backgroundExecutor.execute(postCommit.backgroundNotify()); } /** @@ -437,8 +432,8 @@ public class TransactionManager { */ public void remoteTransactionEvent(RemoteTransactionEvent remoteEvent) { - if (logger.isDebugEnabled()) { - logger.debug("Cluster Received: " + remoteEvent.toString()); + if (clusterLogger.isDebugEnabled()) { + clusterLogger.debug("processing {}", toString()); } List tableIUDList = remoteEvent.getTableIUDList(); @@ -449,11 +444,12 @@ public class TransactionManager { } } + // note DeleteById is written as BeanPersistIds and getBeanPersistList() + // processes both Bean IUD and DeleteById List beanPersistList = remoteEvent.getBeanPersistList(); if (beanPersistList != null) { for (int i = 0; i < beanPersistList.size(); i++) { - BeanPersistIds beanPersist = beanPersistList.get(i); - beanPersist.notifyCacheAndListener(); + beanPersistList.get(i).notifyCacheAndListener(); } } } diff --git a/src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestMcastMemberPackets.java b/src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestMcastMemberPackets.java deleted file mode 100644 index 15355132c..000000000 --- a/src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestMcastMemberPackets.java +++ /dev/null @@ -1,59 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import java.util.List; - -import org.junit.Assert; -import org.junit.Test; - -import com.avaje.ebean.BaseTestCase; -import com.avaje.ebeaninternal.server.cluster.mcast.IncomingPacketsProcessed.GotAllPoint; - -public class TestMcastMemberPackets extends BaseTestCase { - - @Test - public void test() { - - GotAllPoint member = new GotAllPoint("129.12.23.12:9089", 3); - - Assert.assertTrue(member.processPacket(1234)); - Assert.assertTrue(member.processPacket(1235)); - Assert.assertTrue(member.processPacket(1236)); - - Assert.assertEquals(1236l, member.getGotAllPoint()); - Assert.assertEquals(0, member.getMissingPackets().size()); - - Assert.assertFalse(member.processPacket(1234)); - - Assert.assertTrue(member.processPacket(1239)); - List missingPackets = member.getMissingPackets(); - Assert.assertEquals(2, missingPackets.size()); - - Assert.assertTrue(missingPackets.contains(1237l)); - Assert.assertTrue(missingPackets.contains(1238l)); - Assert.assertFalse(missingPackets.contains(1239l)); - Assert.assertFalse(missingPackets.contains(1236l)); - - missingPackets = member.getMissingPackets(); - Assert.assertEquals(2, missingPackets.size()); - Assert.assertTrue(missingPackets.contains(1237l)); - Assert.assertTrue(missingPackets.contains(1238l)); - - Assert.assertEquals(1236l, member.getGotAllPoint()); - - // get a missing packet - Assert.assertTrue(member.processPacket(1237)); - Assert.assertEquals(1237l, member.getGotAllPoint()); - - missingPackets = member.getMissingPackets(); - Assert.assertEquals(1, missingPackets.size()); - Assert.assertTrue(missingPackets.contains(1238l)); - - // but we now hit maxResendIncoming - missingPackets = member.getMissingPackets(); - Assert.assertEquals(0, missingPackets.size()); - // gave up on 1238 .. - Assert.assertEquals(1239l, member.getGotAllPoint()); - - } - -} diff --git a/src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestPacketsAcked.java b/src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestPacketsAcked.java deleted file mode 100644 index 7b886622e..000000000 --- a/src/test/java/com/avaje/ebeaninternal/server/cluster/mcast/TestPacketsAcked.java +++ /dev/null @@ -1,32 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.mcast; - -import org.junit.Assert; -import org.junit.Test; - -import com.avaje.ebean.BaseTestCase; - -public class TestPacketsAcked extends BaseTestCase { - - @Test - public void test() { - - OutgoingPacketsAcked packetsAcked = new OutgoingPacketsAcked(); - - Assert.assertEquals(0l, packetsAcked.getMinimumGotAllPacketId()); - - long receivedAck = packetsAcked.receivedAck("A", new MessageAck("A", 1020l)); - Assert.assertEquals(1020l, packetsAcked.getMinimumGotAllPacketId()); - Assert.assertEquals(1020l, receivedAck); - - receivedAck = packetsAcked.receivedAck("B", new MessageAck("B", 1030l)); - Assert.assertEquals(1020l, packetsAcked.getMinimumGotAllPacketId()); - Assert.assertEquals(0l, receivedAck); - - receivedAck = packetsAcked.receivedAck("C", new MessageAck("C", 1025l)); - Assert.assertEquals(0l, receivedAck); - - receivedAck = packetsAcked.receivedAck("A", new MessageAck("A", 1040l)); - Assert.assertEquals(1025l, receivedAck); - - } -} diff --git a/src/test/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcastTest.java b/src/test/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcastTest.java deleted file mode 100644 index d68635764..000000000 --- a/src/test/java/com/avaje/ebeaninternal/server/cluster/socket/SocketClusterBroadcastTest.java +++ /dev/null @@ -1,79 +0,0 @@ -package com.avaje.ebeaninternal.server.cluster.socket; - -import com.avaje.ebean.config.ContainerConfig; -import com.avaje.ebeaninternal.api.TDSpiEbeanServer; -import com.avaje.ebeaninternal.api.TransactionEventTable; -import com.avaje.ebeaninternal.server.cluster.ClusterManager; -import com.avaje.ebeaninternal.server.transaction.RemoteTransactionEvent; -import org.junit.Test; - -import java.util.Arrays; - -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; - -public class SocketClusterBroadcastTest { - - class TestServer extends TDSpiEbeanServer { - - RemoteTransactionEvent event; - - TestServer(String name) { - super(name); - } - - @Override - public void remoteTransactionEvent(RemoteTransactionEvent event) { - this.event = event; - } - } - - private ContainerConfig createContainerConfig(String local, String threadPoolName) { - ContainerConfig container0 = new ContainerConfig(); - container0.setMode(ContainerConfig.ClusterMode.SOCKET); - - ContainerConfig.SocketConfig socketConfig = new ContainerConfig.SocketConfig(); - socketConfig.setLocalHostPort(local); - socketConfig.setThreadPoolName(threadPoolName); - socketConfig.setMembers(Arrays.asList("127.0.0.1:9876", "127.0.0.1:9866")); - - container0.setSocketConfig(socketConfig); - return container0; - } - - - - @Test - public void testStartup() throws Exception { - - ContainerConfig container0 = createContainerConfig("127.0.0.1:9876", "pool0"); - ClusterManager mgr0 = new ClusterManager(container0); - - TestServer server0 = new TestServer("s001"); - mgr0.registerServer(server0); - - ContainerConfig container1 = createContainerConfig("127.0.0.1:9866", "pool1"); - ClusterManager mgr1 = new ClusterManager(container1); - - TestServer server1 = new TestServer("s001"); - mgr1.registerServer(server1); - - Thread.sleep(1000); - - RemoteTransactionEvent evt = new RemoteTransactionEvent("s001"); - TransactionEventTable.TableIUD tableIUD = new TransactionEventTable.TableIUD("noSuchTable", true, false, false); - evt.addTableIUD(tableIUD); - - assertNull(server1.event); - - mgr0.broadcast(evt); - Thread.sleep(100); - - assertNotNull(server1.event); - - Thread.sleep(1000); - mgr0.shutdown(); - mgr1.shutdown(); - } - -} \ No newline at end of file