You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by bs...@apache.org on 2015/12/18 17:33:47 UTC
incubator-geode git commit: implementation of
AvailablePort.isPortAvailable for multicast
Repository: incubator-geode
Updated Branches:
refs/heads/develop 5d1f38ed3 -> b826b4b00
implementation of AvailablePort.isPortAvailable for multicast
The AvailablePort utility has not had support for finding a free multicast
port since feature/GEODE-77 was merged to the develop branch. This change-
set enables the old multicast code in AvailablePort and implements the
check & response for its datagrams in JGroupsMessenger. The JGroups
transport method for receiving packets is overridden to look for these
datagrams and respond to them.
I also renamed AvailablePort.JGroups to AvailablePort.MULTICAST, which
describes the kind of port that's being tested a little better.
Project: http://git-wip-us.apache.org/repos/asf/incubator-geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-geode/commit/b826b4b0
Tree: http://git-wip-us.apache.org/repos/asf/incubator-geode/tree/b826b4b0
Diff: http://git-wip-us.apache.org/repos/asf/incubator-geode/diff/b826b4b0
Branch: refs/heads/develop
Commit: b826b4b0077dc159cbcee68775cc197b025d5586
Parents: 5d1f38e
Author: Bruce Schuchardt <bs...@pivotal.io>
Authored: Fri Dec 18 08:31:52 2015 -0800
Committer: Bruce Schuchardt <bs...@pivotal.io>
Committed: Fri Dec 18 08:31:52 2015 -0800
----------------------------------------------------------------------
.../membership/gms/messenger/Transport.java | 26 +++++
.../gemfire/internal/AvailablePort.java | 103 +++++++++----------
.../distributed/DistributedSystemDUnitTest.java | 6 +-
.../gemfire/distributed/LocatorDUnitTest.java | 2 +-
.../messenger/JGroupsMessengerJUnitTest.java | 4 +-
.../gemfire/internal/AvailablePortHelper.java | 2 +-
.../internal/AvailablePortJUnitTest.java | 2 +
7 files changed, 86 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/Transport.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/Transport.java b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/Transport.java
index 1687261..fbb4b28 100755
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/Transport.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/Transport.java
@@ -138,5 +138,31 @@ public class Transport extends UDP {
}
}
+ // overridden to implement AvailablePort response
+ @Override
+ public void receive(Address sender, byte[] data, int offset, int length, boolean copy_buffer) {
+ if(data == null) return;
+
+ // drop message from self; it has already been looped back up (https://issues.jboss.org/browse/JGRP-1765)
+ if(local_physical_addr != null && local_physical_addr.equals(sender))
+ return;
+
+ if (length-offset == 4
+ && data[offset] == 'p'
+ && data[offset+1] == 'i'
+ && data[offset+2] == 'n'
+ && data[offset+3] == 'g') {
+ // AvailablePort check
+ data[offset+1] = 'o';
+ try {
+ sendToSingleMember(sender, data, offset, length);
+ } catch (Exception e) {
+ log.fatal("Unable to respond to available-port check", e);
+ }
+ return;
+ }
+
+ super.receive(sender, data, offset, length, copy_buffer);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
index c4907ad..93c6bc2 100644
--- a/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
+++ b/gemfire-core/src/main/java/com/gemstone/gemfire/internal/AvailablePort.java
@@ -35,8 +35,8 @@ public class AvailablePort {
/** Is the port available for a Socket (TCP) connection? */
public static final int SOCKET = 0;
- /** Is the port available for a JGroups (UDP) connection */
- public static final int JGROUPS = 1;
+ /** Is the port available for a JGroups (UDP) multicast connection */
+ public static final int MULTICAST = 1;
/////////////////////// Static Methods ///////////////////////
@@ -50,7 +50,7 @@ public class AvailablePort {
if (protocol == SOCKET) {
name = System.getProperty("gemfire.bind-address");
}
- else if (protocol == JGROUPS) {
+ else if (protocol == MULTICAST) {
name = System.getProperty("gemfire.mcast-address");
}
if (name != null) {
@@ -72,7 +72,7 @@ public class AvailablePort {
* The port to check
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
*
* @throws IllegalArgumentException
* <code>protocol</code> is unknown
@@ -90,7 +90,7 @@ public class AvailablePort {
* The port to check
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
* @param addr the bind address (or mcast address) to use
*
* @throws IllegalArgumentException
@@ -106,47 +106,44 @@ public class AvailablePort {
}
}
- else if (protocol == JGROUPS) {
+ else if (protocol == MULTICAST) {
DatagramSocket socket = null;
try {
- // TODO - need to find out if anyone is listening on this port
- return true;
-
-// socket = new MulticastSocket();
-// socket.setSoTimeout(Integer.getInteger("AvailablePort.timeout", 2000).intValue());
-// byte[] buffer = new byte[4];
-// buffer[0] = (byte)'p';
-// buffer[1] = (byte)'i';
-// buffer[2] = (byte)'n';
-// buffer[3] = (byte)'g';
-// SocketAddress mcaddr = new InetSocketAddress(
-// addr==null? DistributionConfig.DEFAULT_MCAST_ADDRESS : addr, port);
-// DatagramPacket packet = new DatagramPacket(buffer, 0, buffer.length, mcaddr);
-// socket.send(packet);
-// try {
-// socket.receive(packet);
-// packet.getData(); // make sure there's data, but no need to process it
-// return false;
-// }
-// catch (SocketTimeoutException ste) {
-// //System.out.println("socket read timed out");
-// return true;
-// }
-// catch (Exception e) {
-// e.printStackTrace();
-// return false;
-// }
-// }
-// catch (java.io.IOException ioe) {
-// if (ioe.getMessage().equals("Network is unreachable")) {
-// throw new RuntimeException(LocalizedStrings.AvailablePort_NETWORK_IS_UNREACHABLE.toLocalizedString(), ioe);
-// }
-// ioe.printStackTrace();
-// return false;
-// }
-// catch (Exception e) {
-// e.printStackTrace();
-// return false;
+ socket = new MulticastSocket();
+ socket.setSoTimeout(Integer.getInteger("AvailablePort.timeout", 2000).intValue());
+ byte[] buffer = new byte[4];
+ buffer[0] = (byte)'p';
+ buffer[1] = (byte)'i';
+ buffer[2] = (byte)'n';
+ buffer[3] = (byte)'g';
+ SocketAddress mcaddr = new InetSocketAddress(
+ addr==null? DistributionConfig.DEFAULT_MCAST_ADDRESS : addr, port);
+ DatagramPacket packet = new DatagramPacket(buffer, 0, buffer.length, mcaddr);
+ socket.send(packet);
+ try {
+ socket.receive(packet);
+ packet.getData(); // make sure there's data, but no need to process it
+ return false;
+ }
+ catch (SocketTimeoutException ste) {
+ //System.out.println("socket read timed out");
+ return true;
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ return false;
+ }
+ }
+ catch (java.io.IOException ioe) {
+ if (ioe.getMessage().equals("Network is unreachable")) {
+ throw new RuntimeException(LocalizedStrings.AvailablePort_NETWORK_IS_UNREACHABLE.toLocalizedString(), ioe);
+ }
+ ioe.printStackTrace();
+ return false;
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ return false;
}
finally {
if (socket != null) {
@@ -173,7 +170,7 @@ public class AvailablePort {
} else {
return keepOneInterface(addr, port);
}
- } else if (protocol == JGROUPS) {
+ } else if (protocol == MULTICAST) {
throw new IllegalArgumentException("You can not keep the JGROUPS protocol");
} else {
throw new IllegalArgumentException(LocalizedStrings.AvailablePort_UNKNOWN_PROTOCOL_0.toLocalizedString(Integer.valueOf(protocol)));
@@ -297,7 +294,7 @@ public class AvailablePort {
*
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
*
* @throws IllegalArgumentException
* <code>protocol</code> is unknown
@@ -314,7 +311,7 @@ public class AvailablePort {
*
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
*
* @throws IllegalArgumentException
* <code>protocol</code> is unknown
@@ -329,7 +326,7 @@ public class AvailablePort {
*
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
*
* @throws IllegalArgumentException
* <code>protocol</code> is unknown
@@ -345,7 +342,7 @@ public class AvailablePort {
*
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
* @param addr the bind-address or mcast address to use
*
* @throws IllegalArgumentException
@@ -356,7 +353,7 @@ public class AvailablePort {
int port = getRandomWildcardBindPortNumber();
if (isPortAvailable(port, protocol, addr)) {
// don't return the products default multicast port
- if ( !(protocol == JGROUPS && port == DistributionConfig.DEFAULT_MCAST_PORT) ){
+ if ( !(protocol == MULTICAST && port == DistributionConfig.DEFAULT_MCAST_PORT) ){
return port;
}
}
@@ -377,7 +374,7 @@ public class AvailablePort {
*
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
* @param addr the bind-address or mcast address to use
*
* @throws IllegalArgumentException
@@ -399,7 +396,7 @@ public class AvailablePort {
*
* @param protocol
* The protocol to check (either {@link #SOCKET} or {@link
- * #JGROUPS}).
+ * #MULTICAST}).
* @param addr the bind-address or mcast address to use
*
* @throws IllegalArgumentException
@@ -543,7 +540,7 @@ public class AvailablePort {
} else if (protocolString.equalsIgnoreCase("javagroups") ||
protocolString.equalsIgnoreCase("jgroups")) {
- protocol = JGROUPS;
+ protocol = MULTICAST;
} else {
usage("Unknown protocol: " + protocolString);
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/DistributedSystemDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/DistributedSystemDUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/DistributedSystemDUnitTest.java
index a3ff8df..9f1e90b 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/DistributedSystemDUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/DistributedSystemDUnitTest.java
@@ -317,7 +317,7 @@ public class DistributedSystemDUnitTest extends DistributedTestCase {
// TODO this needs to use a locator
public void _testMembershipPortRange() throws Exception {
Properties config = new Properties();
- int mcastPort = AvailablePort.getRandomAvailablePort(AvailablePort.JGROUPS);
+ int mcastPort = AvailablePort.getRandomAvailablePort(AvailablePort.MULTICAST);
int unicastPort = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
config.setProperty("mcast-port", String.valueOf(mcastPort));
config.setProperty("locators", "");
@@ -337,7 +337,7 @@ public class DistributedSystemDUnitTest extends DistributedTestCase {
// TODO this needs to use a locator
public void _testMembershipPortRangeWithExactThreeValues() throws Exception {
Properties config = new Properties();
- int mcastPort = AvailablePort.getRandomAvailablePort(AvailablePort.JGROUPS);
+ int mcastPort = AvailablePort.getRandomAvailablePort(AvailablePort.MULTICAST);
config.setProperty("mcast-port", String.valueOf(mcastPort));
config.setProperty("locators", "");
config.setProperty(DistributionConfig.MEMBERSHIP_PORT_RANGE_NAME, ""
@@ -358,7 +358,7 @@ public class DistributedSystemDUnitTest extends DistributedTestCase {
// TODO this needs to use a locator
public void _testConflictingUDPPort() throws Exception {
final Properties config = new Properties();
- final int mcastPort = AvailablePort.getRandomAvailablePort(AvailablePort.JGROUPS);
+ final int mcastPort = AvailablePort.getRandomAvailablePort(AvailablePort.MULTICAST);
final int unicastPort = AvailablePort.getRandomAvailablePort(AvailablePort.SOCKET);
config.setProperty("mcast-port", String.valueOf(mcastPort));
config.setProperty("locators", "");
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
index 7d34ab1..80f527b 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/LocatorDUnitTest.java
@@ -1454,7 +1454,7 @@ public class LocatorDUnitTest extends DistributedTestCase {
final int port2 = freeTCPPorts[1];
this.port2 = port2;
deleteLocatorStateFile(port1, port2);
- final int mcastport = AvailablePort.getRandomAvailablePort(AvailablePort.JGROUPS);
+ final int mcastport = AvailablePort.getRandomAvailablePort(AvailablePort.MULTICAST);
final String host0 = getServerHostName(host);
final String locators = host0 + "[" + port1 + "]," +
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
index c844583..9d3dbac 100755
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/distributed/internal/membership/gms/messenger/JGroupsMessengerJUnitTest.java
@@ -68,6 +68,7 @@ import com.gemstone.gemfire.distributed.internal.membership.gms.messages.JoinReq
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.JoinResponseMessage;
import com.gemstone.gemfire.distributed.internal.membership.gms.messages.LeaveRequestMessage;
import com.gemstone.gemfire.distributed.internal.membership.gms.messenger.JGroupsMessenger.JGroupsReceiver;
+import com.gemstone.gemfire.internal.AvailablePort;
import com.gemstone.gemfire.internal.AvailablePortHelper;
import com.gemstone.gemfire.internal.DataSerializableFixedID;
import com.gemstone.gemfire.internal.HeapDataOutputStream;
@@ -812,8 +813,9 @@ public class JGroupsMessengerJUnitTest {
public void testMulticastTest() throws Exception {
initMocks(true);
boolean result = messenger.testMulticast(50);
- // this shouldln't succeed
+ // this shouldln't succeed because there's no-one to respond
assertFalse(result);
+ assertFalse(AvailablePort.isPortAvailable(services.getConfig().getDistributionConfig().getMcastPort(), AvailablePort.MULTICAST));
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortHelper.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortHelper.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortHelper.java
index 5230ff9..246e6c7 100755
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortHelper.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortHelper.java
@@ -110,7 +110,7 @@ public class AvailablePortHelper {
public static int[] getRandomAvailableUDPPorts(int count) {
Set set = new HashSet();
while (set.size() < count) {
- int port = AvailablePort.getRandomAvailablePort(AvailablePort.JGROUPS);
+ int port = AvailablePort.getRandomAvailablePort(AvailablePort.MULTICAST);
set.add(new Integer(port));
}
int[] ports = new int[set.size()];
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/b826b4b0/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortJUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortJUnitTest.java b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortJUnitTest.java
index 7244822..6ba91b8 100644
--- a/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortJUnitTest.java
+++ b/gemfire-core/src/test/java/com/gemstone/gemfire/internal/AvailablePortJUnitTest.java
@@ -69,4 +69,6 @@ public class AvailablePortJUnitTest {
socket.bind(new InetSocketAddress((InetAddress)null, port));
assertFalse(AvailablePort.isPortAvailable(port, AvailablePort.SOCKET));
}
+
+ // multicast availability is tested in JGroupsMessengerJUnitTest
}