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/07/13 14:36:58 UTC
[2/4] git commit: Synchronize cluster updates
Synchronize cluster updates
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/a4449c17
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/a4449c17
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/a4449c17
Branch: refs/heads/piper
Commit: a4449c17bbc278e3f3a198a7c4bde6c2bb59d7ef
Parents: 75706eb
Author: Matthieu Morel <mm...@apache.org>
Authored: Fri Jul 13 16:03:39 2012 +0200
Committer: Matthieu Morel <mm...@apache.org>
Committed: Fri Jul 13 16:03:39 2012 +0200
----------------------------------------------------------------------
.../java/org/apache/s4/comm/tcp/TCPEmitter.java | 35 ++++++++++-----
1 files changed, 23 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/a4449c17/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 69ec5f1..f88d9b7 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
@@ -9,6 +9,8 @@ import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
import org.apache.s4.base.Emitter;
import org.apache.s4.base.EventMessage;
@@ -111,6 +113,9 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
@Inject
SerializerDeserializer serDeser;
+ // lock for synchronizing between cluster updates callbacks and other code
+ private final Lock lock;
+
@Inject
public TCPEmitter(Cluster topology, @Named("tcp.partition.queue_size") int bufferSize,
@Named("comm.retries") int retries, @Named("comm.retry_delay") int retryDelay,
@@ -120,6 +125,7 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
this.nettyTimeout = timeout;
this.bufferCapacity = bufferSize;
this.topology = topology;
+ this.lock = new ReentrantLock();
// Initialize data structures
int clusterSize = this.topology.getPhysicalCluster().getNodes().size();
@@ -161,8 +167,8 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
@Inject
private void init() {
- this.topology.addListener(this);
refreshCluster();
+ this.topology.addListener(this);
}
private class Message implements ChannelFutureListener {
@@ -346,7 +352,7 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
if (clusterNode == null) {
logger.error("No ClusterNode exists for partitionId " + partitionId);
- onChange();
+ refreshCluster();
return false;
}
@@ -465,18 +471,23 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
}
private void refreshCluster() {
- for (ClusterNode clusterNode : topology.getPhysicalCluster().getNodes()) {
- Integer partition = clusterNode.getPartition();
- if (partition == null) {
- logger.error("onChange(): Illegal partition for clusterNode - " + clusterNode);
- return;
- }
+ lock.lock();
+ try {
+ for (ClusterNode clusterNode : topology.getPhysicalCluster().getNodes()) {
+ Integer partition = clusterNode.getPartition();
+ if (partition == null) {
+ logger.error("onChange(): Illegal partition for clusterNode - " + clusterNode);
+ return;
+ }
- ClusterNode oldNode = partitionNodeMap.remove(partition);
- if (oldNode != null && !oldNode.equals(clusterNode)) {
- removeChannel(partition);
+ ClusterNode oldNode = partitionNodeMap.remove(partition);
+ if (oldNode != null && !oldNode.equals(clusterNode)) {
+ removeChannel(partition);
+ }
+ partitionNodeMap.forcePut(partition, clusterNode);
}
- partitionNodeMap.forcePut(partition, clusterNode);
+ } finally {
+ lock.unlock();
}
}