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
[1/3] git commit: S4-7 fixed race condition - also increased sent
messages
Updated Branches:
refs/heads/S4-7-FIX [created] 875dd4f60
S4-7 fixed race condition
- also increased sent messages
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/875dd4f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/875dd4f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/875dd4f6
Branch: refs/heads/S4-7-FIX
Commit: 875dd4f608a827109aa53900b1eadd3a0df4a5be
Parents: 8174f75
Author: Matthieu Morel <mm...@apache.org>
Authored: Fri Mar 2 17:30:56 2012 +0100
Committer: Matthieu Morel <mm...@apache.org>
Committed: Fri Mar 2 17:30:56 2012 +0100
----------------------------------------------------------------------
.../java/org/apache/s4/comm/tcp/TCPEmitter.java | 4 ++--
.../java/org/apache/s4/comm/tcp/TCPBasedTest.java | 2 +-
.../org/apache/s4/comm/util/PartitionInfo.java | 3 ++-
3 files changed, 5 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/875dd4f6/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 196e949..ae3fc5e 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
@@ -279,8 +279,8 @@ public class TCPEmitter implements Emitter, TopologyChangeListener {
}
private void sendPendingMessages() {
- while (!pending.isEmpty()) {
- Message msg = pending.remove();
+ Message msg = null;
+ while ((msg = pending.poll()) != null) {
msg.sendMessage();
wire.add(msg);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/875dd4f6/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
index 6c3c242..1d502b7 100644
--- 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
@@ -31,7 +31,7 @@ public abstract class TCPBasedTest extends ProtocolTestUtil {
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);
+ bind(Integer.class).annotatedWith(Names.named("emitter.send.numMessages")).toInstance(5000);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/875dd4f6/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
index 0e59dc5..1669b31 100644
--- 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
@@ -68,8 +68,9 @@ public class PartitionInfo {
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))
+ if (emitter.send(partition, message)) {
break;
+ }
logger.debug("SendThread {}: Resending message to {}", partitionId, partition);
Thread.sleep(retryDelayMs);
}