You are viewing a plain text version of this content. The canonical link for it is here.
Posted to s4-commits@incubator.apache.org by mm...@apache.org on 2012/03/02 16:32:00 UTC
[2/3] git commit: S4-7 Robust TCP-based communication - guaranteed
message delivery - pair-wise ordering of messages with potential duplicate
messages - tolerates network glitches - uses an asynchronous,
ordered-per-partition, memory-aware, thread pool t
S4-7 Robust TCP-based communication - guaranteed message delivery - pair-wise ordering of messages with potential duplicate messages - tolerates network glitches - uses an asynchronous, ordered-per-partition, memory-aware, thread pool to send messages using Netty - synchronizes with toplogy to infer partition-to-node mapping; the cluster (partition-to-node mapping) is stored as a HashBiMap - single- and multi-partition tests to check guaranteed and ordered message-delivery and tolerating network glitches - updated module files for the newly added configuration parameters - documented the TCPEmitter design
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/8174f751
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/8174f751
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/8174f751
Branch: refs/heads/S4-7-FIX
Commit: 8174f75137c16613789809d9b2eac54abb04dcca
Parents: 27a4f9e
Author: Karthik Kambatla <kk...@cs.purdue.edu>
Authored: Wed Jan 25 19:00:00 2012 -0500
Committer: Matthieu Morel <mm...@apache.org>
Committed: Fri Mar 2 16:57:39 2012 +0100
----------------------------------------------------------------------
.../src/main/java/org/apache/s4/base/Emitter.java | 18 +-
.../src/main/java/org/apache/s4/base/Listener.java | 20 +-
.../java/org/apache/s4/comm/QueueingEmitter.java | 177 +++---
.../java/org/apache/s4/comm/QueueingListener.java | 16 +-
.../apache/s4/comm/loopback/LoopBackEmitter.java | 10 +-
.../apache/s4/comm/loopback/LoopBackListener.java | 17 +-
.../java/org/apache/s4/comm/tcp/TCPEmitter.java | 497 +++++++++++----
.../java/org/apache/s4/comm/tcp/TCPListener.java | 64 ++-
.../java/org/apache/s4/comm/tools/TaskSetup.java | 13 +-
.../java/org/apache/s4/comm/topology/Cluster.java | 47 +-
.../java/org/apache/s4/comm/udp/UDPEmitter.java | 16 +-
.../java/org/apache/s4/comm/udp/UDPListener.java | 14 +-
.../java/org/apache/s4/comm/DeliveryTestUtil.java | 143 -----
.../s4/comm/tcp/MultiPartitionDeliveryTest.java | 10 +
.../org/apache/s4/comm/tcp/NetworkGlitchTest.java | 31 +
.../org/apache/s4/comm/tcp/SimpleDeliveryTest.java | 10 +
.../java/org/apache/s4/comm/tcp/TCPBasedTest.java | 48 ++
.../java/org/apache/s4/comm/tcp/TCPCommTest.java | 69 --
.../s4/comm/topology/TopologyFromZKTest.java | 1 +
.../s4/comm/udp/MultiPartitionDeliveryTest.java | 7 +
.../org/apache/s4/comm/udp/SimpleDeliveryTest.java | 7 +
.../java/org/apache/s4/comm/udp/UDPBasedTest.java | 50 ++
.../java/org/apache/s4/comm/udp/UDPCommTest.java | 65 --
.../org/apache/s4/comm/util/PartitionInfo.java | 165 +++++
.../org/apache/s4/comm/util/ProtocolTestUtil.java | 81 +++
.../FileBasedClusterManagementTestModule.java | 14 +-
.../ZkBasedClusterManagementTestModule.java | 25 +-
.../java/org/apache/s4/fixtures/ZkBasedTest.java | 36 +-
.../src/main/java/org/apache/s4/core/Main.java | 28 +-
.../test/java/org/apache/s4/deploy/TestModule.java | 6 +
.../src/test/resources/default.s4.properties | 1 +
.../resources/org.apache.s4.deploy.s4.properties | 3 +-
32 files changed, 1079 insertions(+), 630 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java b/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
index 0570c14..ad511c9 100644
--- a/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
+++ b/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
@@ -1,13 +1,17 @@
package org.apache.s4.base;
public interface Emitter {
-
- /*
- * @param partitionId - destination partition
- * @param message - message payload that needs to be sent
- * @return - true - if message is sent across successfully
- * - false - if send fails
- */
+
+ /*
+ * @param partitionId - destination partition
+ *
+ * @param message - message payload that needs to be sent
+ *
+ * @return - true - if message is sent across successfully - false - if send fails
+ */
boolean send(int partitionId, byte[] message);
+
int getPartitionCount();
+
+ void close();
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-base/src/main/java/org/apache/s4/base/Listener.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-base/src/main/java/org/apache/s4/base/Listener.java b/subprojects/s4-base/src/main/java/org/apache/s4/base/Listener.java
index c0b63ef..be60597 100644
--- a/subprojects/s4-base/src/main/java/org/apache/s4/base/Listener.java
+++ b/subprojects/s4-base/src/main/java/org/apache/s4/base/Listener.java
@@ -7,15 +7,15 @@ package org.apache.s4.base;
*/
public interface Listener {
- /*
- * Perform blocking receive on the appropriate communication channel
- *
- * @return
- * <ul><li> byte[] message returned by the channel </li>
- * <li> null if the associated blocking thread is interrupted </li>
- * </ul>
- */
- byte[] recv();
+ /*
+ * Perform blocking receive on the appropriate communication channel
+ *
+ * @return <ul><li> byte[] message returned by the channel </li> <li> null if the associated blocking thread is
+ * interrupted </li> </ul>
+ */
+ byte[] recv();
- public int getPartitionId();
+ public int getPartitionId();
+
+ void close();
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingEmitter.java
index 089503d..7218afe 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingEmitter.java
@@ -9,92 +9,95 @@ import com.google.inject.Inject;
import com.google.inject.name.Named;
public class QueueingEmitter implements Emitter, Runnable {
- private Emitter emitter;
- private BlockingQueue<MessageHolder> queue;
- private long dropCount = 0;
- private volatile Thread thread;
-
- @Inject
- public QueueingEmitter(@Named("ll") Emitter emitter,
- @Named("comm.queue_emmiter_size") int queueSize) {
- this.emitter = emitter;
- queue = new LinkedBlockingQueue<MessageHolder>(queueSize);
- }
-
- public long getDropCount() {
- return dropCount;
- }
-
- public void start() {
- if (thread != null) {
- throw new IllegalStateException(
- "QueueingEmitter is already started");
- }
- thread = new Thread(this, "QueueingEmitter");
- thread.start();
- }
-
- public void stop() {
- if (thread == null) {
- throw new IllegalStateException(
- "QueueingEmitter is already stopped");
- }
- thread.interrupt();
- thread = null;
- }
-
- @Override
- public boolean send(int partitionId, byte[] message) {
- MessageHolder mh = new MessageHolder(partitionId, message);
- if (!queue.offer(mh)) {
- dropCount++;
- return true;
- } else {
- return false;
- }
- }
-
- public void run() {
- while (!Thread.interrupted()) {
- try {
- MessageHolder mh = queue.take();
- // System.out.println("QueueingEmitter: Sending message on low-level emitter");
- emitter.send(mh.getPartitionId(), mh.getMessage());
- } catch (InterruptedException ie) {
- Thread.currentThread().interrupt();
- }
- }
- }
-
- public int getPartitionCount() {
- return emitter.getPartitionCount();
- }
-
- class MessageHolder {
- private int partitionId;
- private byte[] message;
-
- public int getPartitionId() {
- return partitionId;
- }
-
- public void setPartitionId(int partitionId) {
- this.partitionId = partitionId;
- }
-
- public byte[] getMessage() {
- return message;
- }
-
- public void setMessage(byte[] message) {
- this.message = message;
- }
-
- public MessageHolder(int partitionId, byte[] message) {
- super();
- this.partitionId = partitionId;
- this.message = message;
- }
- }
+ private Emitter emitter;
+ private BlockingQueue<MessageHolder> queue;
+ private long dropCount = 0;
+ private volatile Thread thread;
+
+ @Inject
+ public QueueingEmitter(@Named("ll") Emitter emitter, @Named("comm.queue_emmiter_size") int queueSize) {
+ this.emitter = emitter;
+ queue = new LinkedBlockingQueue<MessageHolder>(queueSize);
+ }
+
+ public long getDropCount() {
+ return dropCount;
+ }
+
+ public void start() {
+ if (thread != null) {
+ throw new IllegalStateException("QueueingEmitter is already started");
+ }
+ thread = new Thread(this, "QueueingEmitter");
+ thread.start();
+ }
+
+ public void stop() {
+ if (thread == null) {
+ throw new IllegalStateException("QueueingEmitter is already stopped");
+ }
+ thread.interrupt();
+ thread = null;
+ }
+
+ @Override
+ public boolean send(int partitionId, byte[] message) {
+ MessageHolder mh = new MessageHolder(partitionId, message);
+ if (!queue.offer(mh)) {
+ dropCount++;
+ return true;
+ } else {
+ return false;
+ }
+ }
+
+ public void run() {
+ while (!Thread.interrupted()) {
+ try {
+ MessageHolder mh = queue.take();
+ // System.out.println("QueueingEmitter: Sending message on low-level emitter");
+ emitter.send(mh.getPartitionId(), mh.getMessage());
+ } catch (InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ public int getPartitionCount() {
+ return emitter.getPartitionCount();
+ }
+
+ class MessageHolder {
+ private int partitionId;
+ private byte[] message;
+
+ public int getPartitionId() {
+ return partitionId;
+ }
+
+ public void setPartitionId(int partitionId) {
+ this.partitionId = partitionId;
+ }
+
+ public byte[] getMessage() {
+ return message;
+ }
+
+ public void setMessage(byte[] message) {
+ this.message = message;
+ }
+
+ public MessageHolder(int partitionId, byte[] message) {
+ super();
+ this.partitionId = partitionId;
+ this.message = message;
+ }
+ }
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingListener.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingListener.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingListener.java
index 8d57a9e..46f0a8e 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingListener.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/QueueingListener.java
@@ -1,6 +1,5 @@
package org.apache.s4.comm;
-
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
@@ -16,8 +15,7 @@ public class QueueingListener implements Listener, Runnable {
private volatile Thread thread;
@Inject
- public QueueingListener(@Named("ll") Listener listener,
- @Named("comm.queue_listener_size") int queueSize) {
+ public QueueingListener(@Named("ll") Listener listener, @Named("comm.queue_listener_size") int queueSize) {
this.listener = listener;
queue = new LinkedBlockingQueue<byte[]>(queueSize);
}
@@ -28,8 +26,7 @@ public class QueueingListener implements Listener, Runnable {
public void start() {
if (thread != null) {
- throw new IllegalStateException(
- "QueueingListener is already started");
+ throw new IllegalStateException("QueueingListener is already started");
}
thread = new Thread(this, "QueueingListener");
thread.start();
@@ -37,8 +34,7 @@ public class QueueingListener implements Listener, Runnable {
public void stop() {
if (thread == null) {
- throw new IllegalStateException(
- "QueueingListener is already stopped");
+ throw new IllegalStateException("QueueingListener is already stopped");
}
thread.interrupt();
thread = null;
@@ -69,4 +65,10 @@ public class QueueingListener implements Listener, Runnable {
}
}
}
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackEmitter.java
index d59eabe..74b7345 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackEmitter.java
@@ -4,11 +4,11 @@ import org.apache.s4.base.Emitter;
public class LoopBackEmitter implements Emitter {
private LoopBackListener listener;
-
+
public LoopBackEmitter(LoopBackListener listener) {
this.listener = listener;
}
-
+
@Override
public boolean send(int partitionId, byte[] message) {
listener.put(message);
@@ -20,4 +20,10 @@ public class LoopBackEmitter implements Emitter {
return 1;
}
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackListener.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackListener.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackListener.java
index b312485..8077231 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackListener.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/loopback/LoopBackListener.java
@@ -1,19 +1,18 @@
package org.apache.s4.comm.loopback;
-
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;
import org.apache.s4.base.Listener;
public class LoopBackListener implements Listener {
-
+
private BlockingQueue<byte[]> handoffQueue = new SynchronousQueue<byte[]>();
@Override
public byte[] recv() {
try {
- //System.out.println("LoopBackListener: Taking message from handoff queue");
+ // System.out.println("LoopBackListener: Taking message from handoff queue");
return handoffQueue.take();
} catch (InterruptedException e) {
throw new RuntimeException(e);
@@ -24,14 +23,20 @@ public class LoopBackListener implements Listener {
public int getPartitionId() {
return 0;
}
-
+
public void put(byte[] message) {
try {
- //System.out.println("LoopBackListener: putting message into handoffqueue");
+ // System.out.println("LoopBackListener: putting message into handoffqueue");
handoffQueue.put(message);
} catch (InterruptedException ie) {
Thread.currentThread().interrupt();
- }
+ }
+ }
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
index 5960fc1..196e949 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
@@ -1,11 +1,16 @@
package org.apache.s4.comm.tcp;
+import java.lang.Thread.UncaughtExceptionHandler;
import java.net.ConnectException;
import java.net.InetSocketAddress;
-import java.util.ArrayDeque;
import java.util.Hashtable;
import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
import org.apache.s4.base.Emitter;
import org.apache.s4.comm.topology.ClusterNode;
@@ -15,6 +20,7 @@ import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
@@ -25,83 +31,105 @@ import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
+import org.jboss.netty.channel.group.ChannelGroup;
+import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.LengthFieldPrepender;
+import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.collect.HashBiMap;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.inject.Inject;
-
-public class TCPEmitter implements Emitter, ChannelFutureListener, TopologyChangeListener {
+import com.google.inject.name.Named;
+
+/**
+ * <p>
+ * TCPEmitter - Uses TCP to send messages across partitions. It
+ * <ul>
+ * <li>guarantees message delivery</li>
+ * <li>preserves pair-wise message ordering; might end up sending duplicates to ensure the order</li>
+ * <li>tolerates topology changes, partition re-mapping and network glitches</li>
+ * </ul>
+ * </p>
+ *
+ * <p>
+ * TCPEmitter is designed as follows:
+ * <ul>
+ * <li>maintains per-partition queue of {@code Message}s</li>
+ * <li> <code>send(p, m)</code> queues the message 'm' to partition 'p'</li>
+ * <li>a thread-pool from {@link http
+ * ://docs.jboss.org/netty/3.2/api/org/jboss/netty/handler/execution/OrderedMemoryAwareThreadPoolExecutor.html} is used
+ * to send the messages asynchronously to the appropriate partitions; send operations between a pair of partitions are
+ * serialized</li>
+ * <li>Each {@code Message} implements the {@link ChannelFutureListener} and listens on the {@link ChannelFuture}
+ * corresponding to the send operation</li>
+ * <li>On success, the message marks itself as sent; messages marked sent at the head of the queue are removed</li>
+ * <li>On failure of a message m, 'm' and all the messages queued after 'm' are resent to preserve message ordering</li>
+ * </ul>
+ * </p>
+ */
+
+public class TCPEmitter implements Emitter, TopologyChangeListener {
private static final Logger logger = LoggerFactory.getLogger(TCPEmitter.class);
- private static final int BUFFER_SIZE = 10;
- private static final int NUM_RETRIES = 10;
+
+ private final int numRetries;
+ private final int retryDelayMs;
+ private final int nettyTimeout;
+ private final int bufferSize;
private Topology topology;
private final ClientBootstrap bootstrap;
- static class MessageQueuesPerPartition {
- private Hashtable<Integer, Queue<byte[]>> queues = new Hashtable<Integer, Queue<byte[]>>();
- private boolean bounded;
-
- MessageQueuesPerPartition(boolean bounded) {
- this.bounded = bounded;
- }
-
- private boolean add(int partitionId, byte[] message) {
- Queue<byte[]> messages = queues.get(partitionId);
-
- if (messages == null) {
- messages = new ArrayDeque<byte[]>();
- queues.put(partitionId, messages);
- }
-
- if (bounded && messages.size() >= BUFFER_SIZE) {
- // Too many messages already queued
- return false;
- }
+ /*
+ * debug information
+ */
+ private volatile int instanceId = 0;
+ private volatile long sentMsgCount = 0;
+ private volatile long ackRecvdCount = 0;
- messages.offer(message);
- return true;
- }
-
- private byte[] peek(int partitionId) {
- Queue<byte[]> messages = queues.get(partitionId);
-
- try {
- return messages.peek();
- } catch (NullPointerException npe) {
- return null;
- }
- }
-
- private void remove(int partitionId) {
- Queue<byte[]> messages = queues.get(partitionId);
+ /*
+ * Channel used to send messages to each partition
+ */
+ private HashBiMap<Integer, Channel> partitionChannelMap;
- if (messages.isEmpty()) {
- logger.error("Trying to remove messages from an empty queue for partition" + partitionId);
- return;
- }
+ /*
+ * Node hosting each partition
+ */
+ private HashBiMap<Integer, ClusterNode> partitionNodeMap;
- if (messages != null)
- messages.remove();
- }
- }
+ /*
+ * Messages to be sent, stored per partition
+ */
+ private Hashtable<Integer, SendQueue> sendQueues;
- private HashBiMap<Integer, Channel> partitionChannelMap;
- private HashBiMap<Integer, ClusterNode> partitionNodeMap;
- private MessageQueuesPerPartition queuedMessages = new MessageQueuesPerPartition(true);
+ /*
+ * Thread pool to actually send messages
+ */
+ private PartitionBasedOMATPE sendService;
@Inject
- public TCPEmitter(Topology topology) throws InterruptedException {
+ public TCPEmitter(Topology topology, @Named("tcp.partition.queue_size") int bufferSize,
+ @Named("comm.retries") int retries, @Named("comm.retry_delay") int retryDelay,
+ @Named("comm.timeout") int timeout) throws InterruptedException {
+ this.numRetries = retries;
+ this.retryDelayMs = retryDelay;
+ this.nettyTimeout = timeout;
+ this.bufferSize = bufferSize;
+
this.topology = topology;
topology.addListener(this);
int clusterSize = this.topology.getTopology().getNodes().size();
partitionChannelMap = HashBiMap.create(clusterSize);
partitionNodeMap = HashBiMap.create(clusterSize);
+ sendQueues = new Hashtable<Integer, SendQueue>(clusterSize);
+ sendService = new PartitionBasedOMATPE();
+ /*
+ * Initialize netty related structures
+ */
ChannelFactory factory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),
Executors.newCachedThreadPool());
@@ -112,131 +140,322 @@ public class TCPEmitter implements Emitter, ChannelFutureListener, TopologyChang
public ChannelPipeline getPipeline() {
ChannelPipeline p = Channels.pipeline();
p.addLast("1", new LengthFieldPrepender(4));
- p.addLast("2", new TestHandler());
+ p.addLast("2", new NotifyChannelInterestChange());
return p;
}
});
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("keepAlive", true);
+ bootstrap.setOption("reuseAddress", true);
+ bootstrap.setOption("connectTimeoutMillis", this.nettyTimeout);
}
- private boolean connectTo(Integer partitionId) {
- ClusterNode clusterNode = partitionNodeMap.get(partitionId);
+ private class PartitionBasedOMATPE extends OrderedMemoryAwareThreadPoolExecutor {
+ public PartitionBasedOMATPE() {
+ /*
+ * Create ordered thread pool - memory limit is enforced by the per-partition queues
+ */
+ super(getPartitionCount(), 0, 0, nettyTimeout, TimeUnit.MILLISECONDS, new ThreadFactoryBuilder()
+ .setNameFormat("TCPEmitterSendServiceThread-#" + instanceId++ + "-%d")
+ .setUncaughtExceptionHandler(new UncaughtExceptionHandler() {
+ @Override
+ public void uncaughtException(Thread paramThread, Throwable paramThrowable) {
+ logger.error("Cannot send message", paramThrowable);
+ }
+ }).build());
+ }
- if (clusterNode == null) {
- clusterNode = topology.getTopology().getNodes().get(partitionId);
- partitionNodeMap.forcePut(partitionId, clusterNode);
+ @Override
+ protected ConcurrentMap<Object, Executor> newChildExecutorMap() {
+ return new ConcurrentHashMap<Object, Executor>();
+ }
+
+ @Override
+ protected Object getChildExecutorKey(ChannelEvent e) {
+ return partitionChannelMap.inverse().get(e.getChannel());
+ }
+
+ @Override
+ public boolean removeChildExecutor(Object key) {
+ return super.removeChildExecutor(key);
+ }
+ }
+
+ private class Message implements ChannelFutureListener {
+ private final SendQueue sendQ;
+ private final byte[] message;
+ private boolean sendSuccess = false;
+
+ Message(SendQueue sendQ, byte[] message) {
+ this.sendQ = sendQ;
+ this.message = message;
+ }
+
+ private void sendMessage() {
+ sendQ.emitter.sendMessage(sendQ.partitionId, this);
+ sentMsgCount++;
+ }
+
+ private void messageSendFailure() {
+ synchronized (sendQ.failureFound) {
+ sendQ.failureFound = true;
+ }
+ closeAndRemoveChannel(sendQ.partitionId);
+ sendQ.spawnSendTask();
+ }
+
+ @Override
+ public synchronized void operationComplete(ChannelFuture future) throws Exception {
+ if (future.isSuccess()) {
+ ackRecvdCount++;
+ sendSuccess = true;
+ sendQ.clearWire();
+ return;
+ }
+
+ if (future.isCancelled()) {
+ logger.error("Send I/O cancelled to " + future.getChannel().getRemoteAddress());
+ }
+
+ // failed operation
+ messageSendFailure();
+ }
+ }
+
+ private class SendQueue {
+ private final TCPEmitter emitter;
+ private final int partitionId;
+ private final int bufferCapacity;
+ private final Queue<Message> pending; // messages to be sent
+ private final Queue<Message> wire; // messages in transit
+
+ private Integer bufferSize = 0;
+ private Boolean failureFound = false;
+
+ SendQueue(TCPEmitter emitter, int partitionId, int bufferCapacity) {
+ this.emitter = emitter;
+ this.partitionId = partitionId;
+ this.bufferCapacity = bufferCapacity;
+ this.pending = new ConcurrentLinkedQueue<Message>();
+ this.wire = new ConcurrentLinkedQueue<Message>();
+ }
+
+ private boolean offer(byte[] message) {
+ Message m = new Message(this, message);
+ synchronized (bufferSize) {
+ if (bufferSize >= bufferCapacity) {
+ return false;
+ }
+
+ pending.add(m);
+ bufferSize++;
+ spawnSendTask();
+ return true;
+ }
+ }
+
+ public void clearWire() {
+ while (!wire.isEmpty()) {
+ Message msg = wire.peek();
+ if (!msg.sendSuccess)
+ return;
+ wire.remove();
+ synchronized (bufferSize) {
+ bufferSize--;
+ }
+ }
}
+ private void spawnSendTask() {
+ emitter.sendService.execute(new SendTask(this));
+ }
+
+ private void resendWiredMessages() {
+ clearWire();
+ for (Message msg : wire) {
+ msg.sendMessage();
+ }
+ }
+
+ private void sendPendingMessages() {
+ while (!pending.isEmpty()) {
+ Message msg = pending.remove();
+ msg.sendMessage();
+ wire.add(msg);
+ }
+ }
+
+ private void sendMessages() {
+ while (true) {
+ boolean resend = false;
+ synchronized (failureFound) {
+ if (failureFound) {
+ resend = true;
+ failureFound = false;
+ } else
+ break;
+ }
+
+ if (resend)
+ resendWiredMessages();
+ }
+
+ sendPendingMessages();
+ }
+ }
+
+ private static class SendTask implements Runnable {
+ private final SendQueue sendQ;
+
+ SendTask(SendQueue sendQ) {
+ this.sendQ = sendQ;
+ }
+
+ @Override
+ public void run() {
+ sendQ.sendMessages();
+ }
+ }
+
+ private boolean connectTo(Integer partitionId) {
+ ClusterNode clusterNode = partitionNodeMap.get(partitionId);
+
if (clusterNode == null) {
logger.error("No ClusterNode exists for partitionId " + partitionId);
+ onChange();
return false;
}
- for (int retries = 0; retries < NUM_RETRIES; retries++) {
+ try {
ChannelFuture f = this.bootstrap.connect(new InetSocketAddress(clusterNode.getMachineName(), clusterNode
.getPort()));
- f.awaitUninterruptibly();
+ f.await(nettyTimeout);
if (f.isSuccess()) {
partitionChannelMap.forcePut(partitionId, f.getChannel());
return true;
}
- try {
- Thread.sleep(10);
- } catch (InterruptedException ie) {
- logger.error(String.format("Interrupted while connecting to %s:%d", clusterNode.getMachineName(),
- clusterNode.getPort()));
- }
+ Thread.sleep(retryDelayMs);
+ } catch (InterruptedException ie) {
+ logger.error(String.format("Interrupted while connecting to %s:%d", clusterNode.getMachineName(),
+ clusterNode.getPort()));
}
-
return false;
}
- private void writeMessageToChannel(Channel channel, int partitionId, byte[] message) {
- ChannelBuffer buffer = ChannelBuffers.buffer(message.length);
- buffer.writeBytes(message);
- ChannelFuture f = channel.write(buffer);
- f.addListener(this);
- }
-
- private final Object sendLock = new Object();
+ private void sendMessage(int partitionId, Message m) {
+ boolean messageSent = false;
+ ChannelBuffer buffer = ChannelBuffers.buffer(m.message.length);
+ buffer.writeBytes(m.message);
- @Override
- public boolean send(int partitionId, byte[] message) {
- Channel channel = partitionChannelMap.get(partitionId);
- if (channel == null) {
- if (connectTo(partitionId)) {
- channel = partitionChannelMap.get(partitionId);
- } else {
- // could not connect, queue to the partitionBuffer
- return queuedMessages.add(partitionId, message);
+ for (int retries = 0; retries < numRetries; retries++) {
+ if (!partitionChannelMap.containsKey(partitionId)) {
+ if (!connectTo(partitionId)) {
+ continue;
+ }
}
- }
- /*
- * Try limiting the size of the send queue inside Netty
- */
- if (!channel.isWritable()) {
- synchronized (sendLock) {
- // check again now that we have the lock
- while (!channel.isWritable()) {
+ SendQueue sendQ = sendQueues.get(partitionId);
+ synchronized (sendQ) {
+ if (!partitionChannelMap.get(partitionId).isWritable()) {
try {
- sendLock.wait();
- } catch (InterruptedException ie) {
- return false;
+ logger.debug("Waiting for channel to partition {} to become writable", partitionId);
+ sendQ.wait();
+ } catch (InterruptedException e) {
+ continue;
}
}
}
- }
- /*
- * Channel is available. Write messages in the following order: (1) Messages already on wire, (2) Previously
- * buffered messages, and (3) the Current Message
- *
- * Once the channel returns success delete from the messagesOnTheWire
- */
- byte[] messageBeingSent = null;
- // while ((messageBeingSent = messagesOnTheWire.peek(partitionId)) != null) {
- // writeMessageToChannel(channel, partitionId, messageBeingSent, false);
- // }
-
- while ((messageBeingSent = queuedMessages.peek(partitionId)) != null) {
- writeMessageToChannel(channel, partitionId, messageBeingSent);
- queuedMessages.remove(partitionId);
+ Channel c = partitionChannelMap.get(partitionId);
+ if (c != null && c.isWritable()) {
+ c.write(buffer).addListener(m);
+ messageSent = true;
+ break;
+ }
}
- writeMessageToChannel(channel, partitionId, message);
- return true;
+ if (!messageSent) {
+ m.messageSendFailure();
+ }
}
@Override
- public void operationComplete(ChannelFuture f) {
- int partitionId = partitionChannelMap.inverse().get(f.getChannel());
- if (f.isSuccess()) {
- // messagesOnTheWire.remove(partitionId);
+ public boolean send(int partitionId, byte[] message) {
+ if (!sendQueues.containsKey(partitionId)) {
+ SendQueue sendQueue = new SendQueue(this, partitionId, this.bufferSize);
+ sendQueues.put(partitionId, sendQueue);
}
- if (f.isCancelled()) {
- logger.error("Send I/O was cancelled!! " + f.getChannel().getRemoteAddress());
- } else if (!f.isSuccess()) {
- logger.error("Exception on I/O operation", f.getCause());
- logger.error(String.format("I/O on partition %d failed!", partitionId));
- partitionChannelMap.remove(partitionId);
+ SendQueue sendQueue = sendQueues.get(partitionId);
+ return sendQueue.offer(message);
+ }
+
+ public void close() {
+ // debug
+ logger.debug("Sent {} messages, and received ACKs for {} messages", sentMsgCount, ackRecvdCount);
+
+ for (SendQueue sendQ : sendQueues.values()) {
+ if (!sendQ.wire.isEmpty()) {
+ logger.error("TCPEmitter could not deliver {} messages to partition {}", sendQ.wire.size(),
+ sendQ.partitionId);
+ sendQ.wire.clear();
+ }
+
+ if (!sendQ.pending.isEmpty()) {
+ logger.error("TCPEmitter could not send {} messages to partition {}", sendQ.pending.size(),
+ sendQ.partitionId);
+ sendQ.pending.clear();
+ }
+ }
+
+ ChannelGroup cg = new DefaultChannelGroup();
+ synchronized (partitionChannelMap) {
+ cg.addAll(partitionChannelMap.values());
+ partitionChannelMap.clear();
+ }
+ try {
+ cg.close().await(nettyTimeout);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ // bootstrap.releaseExternalResources();
+ }
+
+ protected void closeAndRemoveChannel(int partition) {
+ SendQueue sendQ = sendQueues.get(partition);
+
+ if (sendQ == null)
+ return;
+
+ synchronized (sendQ) {
+ Channel c = partitionChannelMap.remove(partition);
+ sendService.removeChildExecutor(partition);
+ if (c != null) {
+ try {
+ c.close().await(nettyTimeout);
+ c.disconnect().await(nettyTimeout);
+ } catch (InterruptedException e) {
+ logger.error("Could not close channel to partition {} gracefully", partition);
+ e.printStackTrace();
+ }
+ }
}
}
@Override
public void onChange() {
- /*
- * Close the channels that correspond to changed partitions and update partitionNodeMap
- */
for (ClusterNode clusterNode : topology.getTopology().getNodes()) {
Integer partition = clusterNode.getPartition();
+ if (partition == null) {
+ logger.error("onChange(): Illegal partition for clusterNode - " + clusterNode);
+ return;
+ }
+
ClusterNode oldNode = partitionNodeMap.get(partition);
if (oldNode != null && !oldNode.equals(clusterNode)) {
- partitionChannelMap.remove(partition).close();
+ closeAndRemoveChannel(partition);
}
partitionNodeMap.forcePut(partition, clusterNode);
@@ -249,14 +468,14 @@ public class TCPEmitter implements Emitter, ChannelFutureListener, TopologyChang
return topology.getTopology().getPartitionCount();
}
- class TestHandler extends SimpleChannelHandler {
+ class NotifyChannelInterestChange extends SimpleChannelHandler {
@Override
public void channelInterestChanged(ChannelHandlerContext ctx, ChannelStateEvent e) {
- // logger.info(String.format("%08x %08x %08x", e.getValue(),
- // e.getChannel().getInterestOps(), Channel.OP_WRITE));
- synchronized (sendLock) {
- if (e.getChannel().isWritable()) {
- sendLock.notify();
+ Channel c = e.getChannel();
+ SendQueue sendQ = sendQueues.get(partitionChannelMap.inverse().get(c));
+ synchronized (sendQ) {
+ if (c.isWritable()) {
+ sendQ.notify();
}
}
ctx.sendUpstream(e);
@@ -264,11 +483,19 @@ public class TCPEmitter implements Emitter, ChannelFutureListener, TopologyChang
@Override
public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) {
- Integer partitionId = partitionChannelMap.inverse().get(context.getChannel());
+ Channel c = context.getChannel();
+ Integer partitionId = partitionChannelMap.inverse().get(c);
if (partitionId == null) {
- logger.error("Error on mystery channel!!");
+ return;
}
logger.error("Error on channel to partition " + partitionId);
+ closeAndRemoveChannel(partitionId);
+
+ SendQueue sendQ = sendQueues.get(partitionId);
+ synchronized (sendQ.failureFound) {
+ sendQ.failureFound = true;
+ }
+ sendQ.spawnSendTask();
try {
throw event.getCause();
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
index b3776e1..b21dccf 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
@@ -10,6 +10,7 @@ import org.apache.s4.comm.topology.Assignment;
import org.apache.s4.comm.topology.ClusterNode;
import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
@@ -18,6 +19,8 @@ import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
+import org.jboss.netty.channel.group.ChannelGroup;
+import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.codec.frame.LengthFieldBasedFrameDecoder;
import org.slf4j.Logger;
@@ -25,61 +28,78 @@ import org.slf4j.LoggerFactory;
import com.google.inject.Inject;
-
public class TCPListener implements Listener {
+ private static final Logger logger = LoggerFactory.getLogger(TCPListener.class);
private BlockingQueue<byte[]> handoffQueue = new SynchronousQueue<byte[]>();
private ClusterNode node;
- private static final Logger logger = LoggerFactory.getLogger(TCPListener.class);
-
+ private ServerBootstrap bootstrap;
+ private final ChannelGroup channelGroup = new DefaultChannelGroup();
+ private final int nettyTimeout = 1000;
+
@Inject
public TCPListener(Assignment assignment) {
// wait for an assignment
node = assignment.assignClusterNode();
-
- ChannelFactory factory =
- new NioServerSocketChannelFactory(
- Executors.newCachedThreadPool(),
- Executors.newCachedThreadPool());
- ServerBootstrap bootstrap = new ServerBootstrap(factory);
+ ChannelFactory factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
+ Executors.newCachedThreadPool());
+
+ bootstrap = new ServerBootstrap(factory);
bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
public ChannelPipeline getPipeline() {
ChannelPipeline p = Channels.pipeline();
p.addLast("1", new LengthFieldBasedFrameDecoder(999999, 0, 4, 0, 4));
p.addLast("2", new ChannelHandler(handoffQueue));
-
+
return p;
}
});
bootstrap.setOption("child.tcpNoDelay", true);
bootstrap.setOption("child.keepAlive", true);
-
- bootstrap.bind(new InetSocketAddress(node.getPort()));
+ bootstrap.setOption("child.reuseAddress", true);
+ bootstrap.setOption("child.connectTimeoutMillis", 100);
+ bootstrap.setOption("readWriteFair", true);
+
+ Channel c = bootstrap.bind(new InetSocketAddress(node.getPort()));
+ channelGroup.add(c);
}
-
+
public byte[] recv() {
try {
- return handoffQueue.take();
+ byte[] msg = handoffQueue.take();
+
+ return msg;
} catch (InterruptedException e) {
- return null;
+ return null;
}
}
-
+
public int getPartitionId() {
return node.getPartition();
}
-
+
+ public void close() {
+ // debug
+ // logger.debug("{}: Received and delivered {} messages", getPartitionId(), deliveredMsgCount);
+ try {
+ channelGroup.close().await(nettyTimeout);
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ // bootstrap.releaseExternalResources();
+ }
+
public class ChannelHandler extends SimpleChannelHandler {
private BlockingQueue<byte[]> handoffQueue;
-
+
public ChannelHandler(BlockingQueue<byte[]> handOffQueue) {
this.handoffQueue = handOffQueue;
}
-
- public void messageReceived(ChannelHandlerContext ctx,
- MessageEvent e) {
+
+ public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
+ channelGroup.add(e.getChannel());
ChannelBuffer buffer = (ChannelBuffer) e.getMessage();
try {
handoffQueue.put(buffer.array()); // this holds up the Netty upstream I/O thread if
@@ -88,7 +108,7 @@ public class TCPListener implements Listener {
Thread.currentThread().interrupt();
}
}
-
+
public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) {
logger.error("Error", event.getCause());
if (context.getChannel().isOpen()) {
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
index ab14449..f3928b8 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
@@ -33,8 +33,7 @@ public class TaskSetup {
record.putSimpleField("port", String.valueOf(1300 + i));
record.putSimpleField("partition", String.valueOf(i));
record.putSimpleField("cluster", clusterName);
- zkclient.createPersistent("/" + clusterName + "/tasks/" + taskId,
- record);
+ zkclient.createPersistent("/" + clusterName + "/tasks/" + taskId, record);
}
}
@@ -45,13 +44,11 @@ public class TaskSetup {
taskSetup.setup(clusterName, 10);
String zookeeperAddress = "localhost:2181";
for (int i = 0; i < 10; i++) {
- AssignmentFromZK assignmentFromZK = new AssignmentFromZK(
- clusterName, zookeeperAddress, 30000, 30000);
- ClusterNode assignClusterNode = assignmentFromZK
- .assignClusterNode();
- System.out.println(i+"-->"+assignClusterNode);
+ AssignmentFromZK assignmentFromZK = new AssignmentFromZK(clusterName, zookeeperAddress, 30000, 30000);
+ ClusterNode assignClusterNode = assignmentFromZK.assignClusterNode();
+ System.out.println(i + "-->" + assignClusterNode);
}
- TopologyFromZK topologyFromZK=new TopologyFromZK(clusterName, zookeeperAddress, 30000, 30000);
+ TopologyFromZK topologyFromZK = new TopologyFromZK(clusterName, zookeeperAddress, 30000, 30000);
Thread.sleep(3000);
Cluster topology = topologyFromZK.getTopology();
System.out.println(topology.getNodes().size());
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
index e4ebb4f..17f4037 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
@@ -4,10 +4,12 @@ import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.common.collect.HashBiMap;
import com.google.inject.Inject;
import com.google.inject.Singleton;
import com.google.inject.name.Named;
@@ -25,7 +27,8 @@ public class Cluster {
private static final Logger logger = LoggerFactory.getLogger(Cluster.class);
- List<ClusterNode> nodes = new ArrayList<ClusterNode>();
+ // List<ClusterNode> nodes = new ArrayList<ClusterNode>();
+ HashBiMap<Integer, ClusterNode> nodes = HashBiMap.create();
String mode = "unicast";
String name = "unknown";
@@ -33,12 +36,14 @@ public class Cluster {
final private String[] ports;
final private int numNodes;
private int numPartitions;
- public Cluster(int numPartitions){
+
+ public Cluster(int numPartitions) {
this.hosts = new String[] {};
this.ports = new String[] {};
this.numNodes = 0;
this.numPartitions = numPartitions;
}
+
/**
* Define the hosts and corresponding ports in the cluster.
*
@@ -50,26 +55,22 @@ public class Cluster {
* if number of hosts and ports don't match.
*/
@Inject
- Cluster(@Named("cluster.hosts") String hosts,
- @Named("cluster.ports") String ports) throws IOException {
- if (hosts != null && hosts.length() > 0 && ports != null
- && ports.length() > 0) {
+ Cluster(@Named("cluster.hosts") String hosts, @Named("cluster.ports") String ports) throws IOException {
+ if (hosts != null && hosts.length() > 0 && ports != null && ports.length() > 0) {
this.ports = ports.split(",");
this.hosts = hosts.split(",");
if (this.ports.length != this.hosts.length) {
- logger.error("Number of hosts should match number of ports in properties file. hosts: "
- + hosts + " ports: " + ports);
+ logger.error("Number of hosts should match number of ports in properties file. hosts: " + hosts
+ + " ports: " + ports);
throw new IOException();
}
numNodes = this.hosts.length;
for (int i = 0; i < numNodes; i++) {
- ClusterNode node = new ClusterNode(i,
- Integer.parseInt(this.ports[i]), this.hosts[i], "");
- nodes.add(node);
- logger.info("Added cluster node: " + this.hosts[i] + ":"
- + this.ports[i]);
+ ClusterNode node = new ClusterNode(i, Integer.parseInt(this.ports[i]), this.hosts[i], "");
+ addNode(node);
+ logger.info("Added cluster node: " + this.hosts[i] + ":" + this.ports[i]);
}
numPartitions = numNodes;
} else {
@@ -82,6 +83,7 @@ public class Cluster {
/**
* Number of partitions in the cluster.
+ *
* @return
*/
public int getPartitionCount() {
@@ -92,14 +94,21 @@ public class Cluster {
* @param node
*/
public void addNode(ClusterNode node) {
- nodes.add(node);
+ // nodes.add(node);
+ if (nodes.containsKey(node.getPartition())) {
+ logger.debug("Partition {} already has a corresponding ClusterNode", node.getPartition());
+ }
+ if (nodes.containsValue(node)) {
+ logger.error("ClusterNode is already mapped to partition {}", nodes.inverse().get(node));
+ }
+ nodes.forcePut(node.getPartition(), node);
}
/**
* @return a list of {@link ClusterNode} objects available in the cluster.
*/
- public List<ClusterNode> getNodes() {
- return Collections.unmodifiableList(nodes);
+ public Set<ClusterNode> getNodes() {
+ return Collections.unmodifiableSet(nodes.values());
}
// TODO: do we need mode and name? Making provate for now.
@@ -126,9 +135,9 @@ public class Cluster {
public String toString() {
StringBuffer sb = new StringBuffer();
- sb.append("{name=").append(name).append(",mode=").append(mode)
- .append(",type=").append(",nodes=").append(nodes).append("}");
+ sb.append("{name=").append(name).append(",mode=").append(mode).append(",type=").append(",nodes=").append(nodes)
+ .append("}");
return sb.toString();
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
index 85b164c..3d5a199 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
@@ -12,6 +12,7 @@ import org.apache.s4.base.Emitter;
import org.apache.s4.comm.topology.ClusterNode;
import org.apache.s4.comm.topology.Topology;
import org.apache.s4.comm.topology.TopologyChangeListener;
+import org.slf4j.LoggerFactory;
import com.google.common.collect.HashBiMap;
import com.google.inject.Inject;
@@ -29,6 +30,8 @@ public class UDPEmitter implements Emitter, TopologyChangeListener {
@Inject
public UDPEmitter(Topology topology) {
+ LoggerFactory.getLogger(getClass()).debug("UDPEmitter with topology {}",
+ topology.getTopology().getPartitionCount());
this.topology = topology;
topology.addListener(this);
nodes = HashBiMap.create(topology.getTopology().getNodes().size());
@@ -48,7 +51,10 @@ public class UDPEmitter implements Emitter, TopologyChangeListener {
try {
ClusterNode node = nodes.get(partitionId);
if (node == null) {
- throw new RuntimeException(String.format("Bad partition id %d", partitionId));
+ LoggerFactory.getLogger(getClass()).error(
+ "Cannot send message to partition {} because this partition is not visible to this emitter",
+ partitionId);
+ return false;
}
byte[] byteBuffer = new byte[message.length];
System.arraycopy(message, 0, byteBuffer, 0, message.length);
@@ -77,8 +83,14 @@ public class UDPEmitter implements Emitter, TopologyChangeListener {
synchronized (nodes) {
for (ClusterNode clusterNode : topology.getTopology().getNodes()) {
Integer partition = clusterNode.getPartition();
- nodes.put(partition, clusterNode);
+ nodes.forcePut(partition, clusterNode);
}
}
}
+
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPListener.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPListener.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPListener.java
index 84aa7fa..5cd62de 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPListener.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPListener.java
@@ -12,11 +12,10 @@ import org.apache.s4.comm.topology.ClusterNode;
import com.google.inject.Inject;
-
/**
*
* Implementation of a simple UDP listener.
- *
+ *
*/
public class UDPListener implements Listener, Runnable {
@@ -54,8 +53,7 @@ public class UDPListener implements Listener, Runnable {
while (!Thread.interrupted()) {
socket.receive(datagram);
byte[] data = new byte[datagram.getLength()];
- System.arraycopy(datagram.getData(), datagram.getOffset(),
- data, 0, data.length);
+ System.arraycopy(datagram.getData(), datagram.getOffset(), data, 0, data.length);
datagram.setLength(BUFFER_LENGTH);
try {
handoffQueue.put(data);
@@ -72,7 +70,7 @@ public class UDPListener implements Listener, Runnable {
try {
return handoffQueue.take();
} catch (InterruptedException e) {
- return null;
+ return null;
}
}
@@ -80,4 +78,10 @@ public class UDPListener implements Listener, Runnable {
return node.getPartition();
}
+ @Override
+ public void close() {
+ // TODO Auto-generated method stub
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
deleted file mode 100644
index 19b4de1..0000000
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
+++ /dev/null
@@ -1,143 +0,0 @@
-package org.apache.s4.comm;
-
-import org.apache.s4.base.Emitter;
-import org.apache.s4.base.Listener;
-
-import com.google.inject.Inject;
-import com.google.inject.name.Named;
-
-/*
- * Test util for communication protocols.
- *
- * <ul>
- * <li> The util defines Send and Receive Threads </li>
- * <li> SendThread sends out a pre-defined number of messages to all the partitions </li>
- * <li> ReceiveThread receives all/most of these messages </li>
- * <li> To avoid the receiveThread waiting for ever, it spawns a TimerThread that would
- * interrupt after a pre-defined but long enough interval </li>
- * </ul>
- *
- */
-public class DeliveryTestUtil {
-
- private final Emitter emitter;
- private final Listener listener;
- private final int interval;
- private int numMessages;
- private int sleepCount;
-
- // public Thread sendThread, receiveThread;
- private final int messagesExpected;
-
- @Inject
- public DeliveryTestUtil(Emitter emitter, Listener listener, @Named("emitter.send.interval") int interval,
- @Named("emitter.send.numMessages") int numMessages, @Named("listener.recv.sleepCount") int sleepCount) {
- this.emitter = emitter;
- this.listener = listener;
- this.interval = interval;
- this.numMessages = numMessages;
- this.sleepCount = sleepCount;
- this.messagesExpected = numMessages * this.emitter.getPartitionCount();
-
- // this.sendThread = new SendThread();
- // this.receiveThread = new ReceiveThread();
- }
-
- public class SendThread extends Thread {
- @Override
- public void run() {
- try {
- for (int partition = 0; partition < emitter.getPartitionCount(); partition++) {
- for (int i = 0; i < numMessages; i++) {
- byte[] message = (new String("message-" + i)).getBytes();
- emitter.send(partition, message);
- Thread.sleep(interval);
- }
- }
- } catch (Exception e) {
- e.printStackTrace();
- return;
- }
- }
- }
-
- /*
- * TimerThread - interrupts the passed thread, after specified time-interval.
- */
- class TimerThread extends Thread {
- private final Thread watchThread;
- private Integer sleepCounter;
-
- TimerThread(Thread watchThread) {
- this.watchThread = watchThread;
- this.sleepCounter = new Integer(sleepCount);
- }
-
- public void resetSleepCounter() {
- synchronized (this.sleepCounter) {
- this.sleepCounter = sleepCount;
- }
- }
-
- public void clearSleepCounter() {
- synchronized (this.sleepCounter) {
- this.sleepCounter = 0;
- }
- }
-
- private int getCounter() {
- synchronized (this.sleepCounter) {
- return this.sleepCounter--;
- }
- }
-
- @Override
- public void run() {
- try {
- while (getCounter() > 0) {
- Thread.sleep(interval);
- }
- watchThread.interrupt();
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
- }
-
- class ReceiveThread extends Thread {
- private int messagesReceived = 0;
-
- @Override
- public void run() {
-
- // start the timer thread to interrupt if blocked for too long
- TimerThread timer = new TimerThread(this);
- timer.start();
- while (messagesReceived < messagesExpected) {
- byte[] message = listener.recv();
- timer.resetSleepCounter();
- if (message != null)
- messagesReceived++;
- else
- break;
- }
- timer.clearSleepCounter();
- }
-
- private boolean moreMessages() {
- return ((messagesExpected - messagesReceived) > 0);
- }
- }
-
- public Thread newSendThread() {
- return new SendThread();
- }
-
- public Thread newReceiveThread() {
- return new ReceiveThread();
- }
-
- public boolean moreMessages(Thread recvThread) {
- return ((ReceiveThread) recvThread).moreMessages();
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/MultiPartitionDeliveryTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/MultiPartitionDeliveryTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/MultiPartitionDeliveryTest.java
new file mode 100644
index 0000000..8e94c44
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/MultiPartitionDeliveryTest.java
@@ -0,0 +1,10 @@
+package org.apache.s4.comm.tcp;
+
+import org.slf4j.LoggerFactory;
+
+public class MultiPartitionDeliveryTest extends TCPBasedTest {
+ public MultiPartitionDeliveryTest() {
+ super(6);
+ logger = LoggerFactory.getLogger(MultiPartitionDeliveryTest.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/NetworkGlitchTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/NetworkGlitchTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/NetworkGlitchTest.java
new file mode 100644
index 0000000..f17c330
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/NetworkGlitchTest.java
@@ -0,0 +1,31 @@
+package org.apache.s4.comm.tcp;
+
+import org.apache.s4.comm.util.PartitionInfo;
+import org.junit.Assert;
+import org.slf4j.LoggerFactory;
+
+public class NetworkGlitchTest extends TCPBasedTest {
+ public NetworkGlitchTest() {
+ super(2);
+ logger = LoggerFactory.getLogger(NetworkGlitchTest.class);
+ }
+
+ @Override
+ public void testDelivery() throws InterruptedException {
+ PartitionInfo util = partitions[0];
+
+ startThreads();
+
+ for (int i = 1; i < 10; i++) {
+ Thread.sleep(100);
+ ((TCPEmitter) util.emitter).closeAndRemoveChannel(0);
+ }
+
+ waitForThreads();
+
+ Assert.assertTrue("Message delivery", messageDelivery());
+
+ logger.info("Message ordering - " + messageOrdering());
+ Assert.assertTrue("Pairwise message ordering", messageOrdering());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/SimpleDeliveryTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/SimpleDeliveryTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/SimpleDeliveryTest.java
new file mode 100644
index 0000000..64e5c73
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/SimpleDeliveryTest.java
@@ -0,0 +1,10 @@
+package org.apache.s4.comm.tcp;
+
+import org.slf4j.LoggerFactory;
+
+public class SimpleDeliveryTest extends TCPBasedTest {
+ public SimpleDeliveryTest() {
+ super();
+ logger = LoggerFactory.getLogger(SimpleDeliveryTest.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPBasedTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPBasedTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPBasedTest.java
new file mode 100644
index 0000000..6c3c242
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPBasedTest.java
@@ -0,0 +1,48 @@
+package org.apache.s4.comm.tcp;
+
+import org.apache.s4.comm.util.ProtocolTestUtil;
+import org.apache.s4.fixtures.ZkBasedClusterManagementTestModule;
+import org.junit.Assert;
+import org.slf4j.Logger;
+
+import com.google.inject.Guice;
+import com.google.inject.name.Names;
+
+public abstract class TCPBasedTest extends ProtocolTestUtil {
+ protected Logger logger;
+
+ protected TCPBasedTest() {
+ super();
+ super.injector = Guice.createInjector(new TCPTestModule());
+ }
+
+ protected TCPBasedTest(int numTasks) {
+ super(numTasks);
+ super.injector = Guice.createInjector(new TCPTestModule());
+ }
+
+ class TCPTestModule extends ZkBasedClusterManagementTestModule {
+ TCPTestModule() {
+ super(TCPEmitter.class, TCPListener.class);
+ }
+
+ @Override
+ protected void configure() {
+ super.configure();
+ bind(Integer.class).annotatedWith(Names.named("tcp.partition.queue_size")).toInstance(256);
+ bind(Integer.class).annotatedWith(Names.named("emitter.send.interval")).toInstance(2);
+ bind(Integer.class).annotatedWith(Names.named("emitter.send.numMessages")).toInstance(250);
+ }
+ }
+
+ @Override
+ public void testDelivery() throws InterruptedException {
+ startThreads();
+ waitForThreads();
+
+ Assert.assertTrue("Message Delivery", messageDelivery());
+
+ logger.info("Message ordering - " + messageOrdering());
+ Assert.assertTrue("Pairwise message ordering", messageOrdering());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
deleted file mode 100644
index f02bf64..0000000
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
+++ /dev/null
@@ -1,69 +0,0 @@
-package org.apache.s4.comm.tcp;
-
-import java.io.IOException;
-import org.apache.s4.comm.DeliveryTestUtil;
-import org.apache.s4.comm.tcp.TCPEmitter;
-import org.apache.s4.comm.tcp.TCPListener;
-import org.apache.s4.fixtures.ZkBasedClusterManagementTestModule;
-import org.apache.s4.fixtures.ZkBasedTest;
-import org.apache.zookeeper.KeeperException;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import com.google.inject.name.Names;
-
-public class TCPCommTest extends ZkBasedTest {
- DeliveryTestUtil util;
-
- @Before
- public void setup() throws IOException, InterruptedException, KeeperException {
- Injector injector = Guice.createInjector(new TCPCommTestModule());
- util = injector.getInstance(DeliveryTestUtil.class);
- }
-
- class TCPCommTestModule extends ZkBasedClusterManagementTestModule {
- TCPCommTestModule() {
- super(TCPEmitter.class, TCPListener.class);
- }
-
- @Override
- protected void configure() {
- super.configure();
- bind(Integer.class).annotatedWith(Names.named("emitter.send.interval")).toInstance(100);
- bind(Integer.class).annotatedWith(Names.named("emitter.send.numMessages")).toInstance(200);
- bind(Integer.class).annotatedWith(Names.named("listener.recv.sleepCount")).toInstance(10);
- }
- }
-
- /**
- * Tests the protocol. If all components function without throwing exceptions, the test passes. The test also
- * reports the loss of messages, if any.
- *
- * @throws InterruptedException
- */
- @Test
- public void testTCPDelivery() throws InterruptedException {
- try {
- Thread sendThread = util.newSendThread();
- Thread receiveThread = util.newReceiveThread();
-
- // start send and receive threads
- sendThread.start();
- receiveThread.start();
-
- // wait for them to finish
- sendThread.join();
- receiveThread.join();
-
- Assert.assertTrue("Guaranteed message delivery", !util.moreMessages(receiveThread));
- } catch (Exception e) {
- e.printStackTrace();
- Assert.fail("TCP has failed basic functionality test");
- }
-
- System.out.println("Done");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/topology/TopologyFromZKTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/topology/TopologyFromZKTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/topology/TopologyFromZKTest.java
index eb5f42c..1b47285 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/topology/TopologyFromZKTest.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/topology/TopologyFromZKTest.java
@@ -8,6 +8,7 @@ import java.util.concurrent.locks.ReentrantLock;
import static org.junit.Assert.*;
import org.apache.s4.comm.tools.TaskSetup;
+import org.apache.s4.fixtures.ZkBasedTest;
import org.junit.Test;
public class TopologyFromZKTest extends ZKBaseTest {
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/MultiPartitionDeliveryTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/MultiPartitionDeliveryTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/MultiPartitionDeliveryTest.java
new file mode 100644
index 0000000..0559b41
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/MultiPartitionDeliveryTest.java
@@ -0,0 +1,7 @@
+package org.apache.s4.comm.udp;
+
+public class MultiPartitionDeliveryTest extends UDPBasedTest {
+ public MultiPartitionDeliveryTest() {
+ super(2);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/SimpleDeliveryTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/SimpleDeliveryTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/SimpleDeliveryTest.java
new file mode 100644
index 0000000..c29f0ca
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/SimpleDeliveryTest.java
@@ -0,0 +1,7 @@
+package org.apache.s4.comm.udp;
+
+public class SimpleDeliveryTest extends UDPBasedTest {
+ public SimpleDeliveryTest() {
+ super();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPBasedTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPBasedTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPBasedTest.java
new file mode 100644
index 0000000..9a68bd6
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPBasedTest.java
@@ -0,0 +1,50 @@
+package org.apache.s4.comm.udp;
+
+import org.apache.s4.comm.util.ProtocolTestUtil;
+import org.apache.s4.fixtures.ZkBasedClusterManagementTestModule;
+import org.junit.Assert;
+
+import com.google.inject.Guice;
+import com.google.inject.name.Names;
+
+public abstract class UDPBasedTest extends ProtocolTestUtil {
+ protected UDPBasedTest() {
+ super();
+ super.injector = Guice.createInjector(new UDPTestModule());
+ }
+
+ protected UDPBasedTest(int numTasks) {
+ super(numTasks);
+ super.injector = Guice.createInjector(new UDPTestModule());
+ }
+
+ class UDPTestModule extends ZkBasedClusterManagementTestModule {
+ UDPTestModule() {
+ super(UDPEmitter.class, UDPListener.class);
+ }
+
+ @Override
+ protected void configure() {
+ super.configure();
+ bind(Integer.class).annotatedWith(Names.named("emitter.send.interval")).toInstance(2);
+ bind(Integer.class).annotatedWith(Names.named("emitter.send.numMessages")).toInstance(250);
+ }
+ }
+
+ /**
+ * Tests the protocol. If all components function without throwing exceptions, the test passes.
+ *
+ * @throws InterruptedException
+ */
+ @Override
+ public void testDelivery() {
+ try {
+ Thread.sleep(1000);
+ startThreads();
+ waitForThreads();
+ messageDelivery();
+ } catch (Exception e) {
+ Assert.fail("UDP DeliveryTest");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPCommTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPCommTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPCommTest.java
deleted file mode 100644
index 808a357..0000000
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/udp/UDPCommTest.java
+++ /dev/null
@@ -1,65 +0,0 @@
-package org.apache.s4.comm.udp;
-
-import java.io.IOException;
-
-import org.apache.s4.comm.DeliveryTestUtil;
-import org.apache.s4.fixtures.ZkBasedClusterManagementTestModule;
-import org.apache.s4.fixtures.ZkBasedTest;
-import org.apache.zookeeper.KeeperException;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import com.google.inject.name.Names;
-
-public class UDPCommTest extends ZkBasedTest {
- DeliveryTestUtil util;
-
- @Before
- public void setup() throws IOException, InterruptedException, KeeperException {
- Injector injector = Guice.createInjector(new UDPCommTestModule());
- util = injector.getInstance(DeliveryTestUtil.class);
- }
-
- class UDPCommTestModule extends ZkBasedClusterManagementTestModule {
- UDPCommTestModule() {
- super(UDPEmitter.class, UDPListener.class);
- }
-
- @Override
- protected void configure() {
- super.configure();
- bind(Integer.class).annotatedWith(Names.named("emitter.send.interval")).toInstance(100);
- bind(Integer.class).annotatedWith(Names.named("emitter.send.numMessages")).toInstance(200);
- bind(Integer.class).annotatedWith(Names.named("listener.recv.sleepCount")).toInstance(10);
- }
- }
-
- /**
- * Tests the protocol. If all components function without throwing exceptions, the test passes.
- *
- * @throws InterruptedException
- */
- @Test
- public void testUDPDelivery() throws InterruptedException {
- try {
- Thread sendThread = util.newSendThread();
- Thread receiveThread = util.newReceiveThread();
-
- // start send and receive threads
- sendThread.start();
- receiveThread.start();
-
- // wait for them to finish
- sendThread.join();
- receiveThread.join();
-
- } catch (Exception e) {
- e.printStackTrace();
- Assert.fail("UDP has failed basic functionality test");
- }
- System.out.println("Done");
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
new file mode 100644
index 0000000..0e59dc5
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
@@ -0,0 +1,165 @@
+package org.apache.s4.comm.util;
+
+import java.util.ArrayList;
+import java.util.Hashtable;
+import java.util.List;
+
+import org.apache.s4.base.Emitter;
+import org.apache.s4.base.Listener;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.name.Named;
+
+/**
+ * Test util for communication protocols.
+ *
+ * <ul>
+ * <li>The util defines Send and Receive Threads</li>
+ * <li>SendThread sends out a pre-defined number of messages to all the partitions</li>
+ * <li>ReceiveThread receives all/most of these messages</li>
+ * <li>To avoid the receiveThread waiting for ever, it spawns a TimerThread that would interrupt after a pre-defined but
+ * long enough interval</li>
+ * </ul>
+ *
+ */
+public class PartitionInfo {
+ private static final Logger logger = LoggerFactory.getLogger(PartitionInfo.class);
+ public Emitter emitter;
+ public Listener listener;
+ private final int numRetries;
+ private final int retryDelayMs;
+ private int numMessages;
+ private int partitionId;
+
+ public SendThread sendThread;
+ public ReceiveThread receiveThread;
+ private int messagesExpected;
+
+ @Inject
+ public PartitionInfo(Emitter emitter, Listener listener, @Named("comm.retries") int retries,
+ @Named("comm.retry_delay") int retryDelay, @Named("emitter.send.numMessages") int numMessages) {
+ this.emitter = emitter;
+ this.listener = listener;
+ this.partitionId = this.listener.getPartitionId();
+ logger.debug("# Partitions = {}; Current partition = {}", this.emitter.getPartitionCount(),
+ this.listener.getPartitionId());
+
+ this.numRetries = retries;
+ this.retryDelayMs = retryDelay;
+ this.numMessages = numMessages;
+ this.messagesExpected = numMessages * this.emitter.getPartitionCount();
+
+ this.sendThread = new SendThread();
+ this.receiveThread = new ReceiveThread();
+ }
+
+ public class SendThread extends Thread {
+ public SendThread() {
+ super("SendThread");
+ }
+
+ @Override
+ public void run() {
+ logger.debug("SendThread {}: started", partitionId);
+ try {
+ for (int i = 0; i < numMessages; i++) {
+ for (int partition = 0; partition < emitter.getPartitionCount(); partition++) {
+ byte[] message = new String(partitionId + " " + i).getBytes();
+ for (int retries = 0; retries < numRetries; retries++) {
+ if (emitter.send(partition, message))
+ break;
+ logger.debug("SendThread {}: Resending message to {}", partitionId, partition);
+ Thread.sleep(retryDelayMs);
+ }
+ }
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ return;
+ }
+
+ logger.debug("SendThread {}: Exiting", partitionId);
+ }
+ }
+
+ public class ReceiveThread extends Thread {
+ private int messagesReceived = 0;
+ private Hashtable<Integer, List<Integer>> receivedMessages;
+
+ ReceiveThread() {
+ super("ReceiveThread");
+ receivedMessages = new Hashtable<Integer, List<Integer>>();
+ }
+
+ @Override
+ public void run() {
+ logger.debug("ReceiveThread {}: started", partitionId);
+ while (messagesReceived < messagesExpected) {
+ byte[] message = listener.recv();
+ if (message == null) {
+ logger.error("ReceiveThread {}: received a null message", partitionId);
+ break;
+ }
+
+ // process and store the message
+ String msgString = new String(message);
+ String[] msgTokens = msgString.split(" ");
+ Integer senderPartition = Integer.parseInt(msgTokens[0]);
+ Integer receivedMsg = Integer.parseInt(msgTokens[1]);
+
+ if (!receivedMessages.containsKey(senderPartition)) {
+ receivedMessages.put(senderPartition, new ArrayList<Integer>());
+ }
+
+ List<Integer> messagesList = receivedMessages.get(senderPartition);
+
+ if (messagesList.contains(receivedMsg)) {
+ // logger.debug("ReceiveThread {}: Already received message - {}", partitionId, msgString);
+ messagesList.remove(receivedMsg);
+ } else {
+ messagesReceived++;
+ }
+ messagesList.add(receivedMsg);
+ }
+
+ logger.debug("ReceiveThread {}: Exiting with {} messages left", partitionId, moreMessages());
+ }
+
+ public boolean orderedDelivery() {
+ for (List<Integer> messagesList : receivedMessages.values()) {
+ int lastMsg = -1;
+ for (Integer msg : messagesList) {
+ if (msg <= lastMsg) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ public boolean messageDelivery() {
+ for (List<Integer> messagesList : receivedMessages.values()) {
+ if (messagesList.size() < numMessages) {
+ printRecvdCounts();
+ return false;
+ }
+ }
+ return true;
+ }
+
+ public void printRecvdCounts() {
+ int recvdCount[] = new int[emitter.getPartitionCount()];
+ for (Integer sender : receivedMessages.keySet()) {
+ recvdCount[sender] = receivedMessages.get(sender).size();
+ }
+
+ logger.debug("ReceiveThread {}: recvdCounts: {}", partitionId, recvdCount);
+ }
+
+ public int moreMessages() {
+ return (messagesExpected - messagesReceived);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/ProtocolTestUtil.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/ProtocolTestUtil.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/ProtocolTestUtil.java
new file mode 100644
index 0000000..d02d222
--- /dev/null
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/ProtocolTestUtil.java
@@ -0,0 +1,81 @@
+package org.apache.s4.comm.util;
+
+import java.io.IOException;
+
+import org.apache.s4.fixtures.ZkBasedTest;
+import org.apache.zookeeper.KeeperException;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import com.google.inject.Injector;
+
+public abstract class ProtocolTestUtil extends ZkBasedTest {
+ protected PartitionInfo[] partitions;
+ protected Injector injector;
+
+ protected ProtocolTestUtil() {
+ super();
+ }
+
+ protected ProtocolTestUtil(int numTasks) {
+ super(numTasks);
+ }
+
+ @Before
+ public void setup() throws IOException, InterruptedException, KeeperException {
+ partitions = new PartitionInfo[super.numTasks];
+ for (int i = 0; i < this.numTasks; i++) {
+ partitions[i] = injector.getInstance(PartitionInfo.class);
+ }
+ }
+
+ protected void startThreads() {
+ for (PartitionInfo partition : partitions) {
+ partition.sendThread.start();
+ partition.receiveThread.start();
+ }
+ }
+
+ protected void waitForThreads() throws InterruptedException {
+ for (PartitionInfo partition : partitions) {
+ partition.sendThread.join();
+ partition.receiveThread.join();
+ }
+ }
+
+ protected boolean messageDelivery() {
+ for (PartitionInfo partition : partitions) {
+ if (!partition.receiveThread.messageDelivery())
+ return false;
+ }
+ return true;
+ }
+
+ protected boolean messageOrdering() {
+ for (PartitionInfo partition : partitions) {
+ if (!partition.receiveThread.orderedDelivery())
+ return false;
+ }
+ return true;
+ }
+
+ @After
+ public void tearDown() {
+ for (PartitionInfo partition : partitions) {
+ // debug
+ partition.receiveThread.printRecvdCounts();
+ if (partition.emitter != null) {
+ partition.emitter.close();
+ partition.emitter = null;
+ }
+ if (partition.listener != null) {
+ partition.listener.close();
+ partition.listener = null;
+ }
+ }
+ }
+
+ @Test(timeout = 60000)
+ public abstract void testDelivery() throws InterruptedException;
+}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/FileBasedClusterManagementTestModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/FileBasedClusterManagementTestModule.java b/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/FileBasedClusterManagementTestModule.java
index ebcf388..67d4c91 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/FileBasedClusterManagementTestModule.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/FileBasedClusterManagementTestModule.java
@@ -14,13 +14,13 @@ import org.apache.s4.base.Listener;
import org.apache.s4.base.SerializerDeserializer;
import org.apache.s4.comm.DefaultHasher;
import org.apache.s4.comm.serialize.KryoSerDeser;
+import org.apache.s4.comm.tcp.TCPEmitter;
+import org.apache.s4.comm.tcp.TCPListener;
import org.apache.s4.comm.topology.Assignment;
import org.apache.s4.comm.topology.AssignmentFromFile;
import org.apache.s4.comm.topology.Cluster;
import org.apache.s4.comm.topology.Topology;
import org.apache.s4.comm.topology.TopologyFromFile;
-import org.apache.s4.comm.udp.UDPEmitter;
-import org.apache.s4.comm.udp.UDPListener;
import com.google.inject.AbstractModule;
import com.google.inject.Binder;
@@ -69,9 +69,13 @@ public abstract class FileBasedClusterManagementTestModule<T> extends AbstractMo
bind(SerializerDeserializer.class).to(KryoSerDeser.class);
bind(Assignment.class).to(AssignmentFromFile.class);
bind(Topology.class).to(TopologyFromFile.class);
- bind(Emitter.class).to(UDPEmitter.class);
- bind(Listener.class).to(UDPListener.class);
+ bind(Emitter.class).to(TCPEmitter.class);
+ bind(Listener.class).to(TCPListener.class);
- }
+ bind(Integer.class).annotatedWith(Names.named("comm.retries")).toInstance(10);
+ bind(Integer.class).annotatedWith(Names.named("comm.retry_delay")).toInstance(10);
+ bind(Integer.class).annotatedWith(Names.named("comm.timeout")).toInstance(1000);
+ bind(Integer.class).annotatedWith(Names.named("tcp.partition.queue_size")).toInstance(256);
+ }
}