Author: jbellis
Date: Tue Jan 11 23:39:33 2011
New Revision: 1057932
URL: http://svn.apache.org/viewvc?rev=1057932&view=rev
Log:
backport CASSANDRA-1959 from 0.7
Modified:
cassandra/branches/cassandra-0.6/CHANGES.txt
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/net/MessagingService.java
Modified: cassandra/branches/cassandra-0.6/CHANGES.txt
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/CHANGES.txt?rev=1057932&r1=1057931&r2=1057932&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.6/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.6/CHANGES.txt Tue Jan 11 23:39:33 2011
@@ -1,6 +1,8 @@
0.6.10
- * buffer network stack to avoid inefficient small TCP messages while avoiding
- the nagle/delayed ack problem (CASSANDRA-1896)
+ * buffer network stack to avoid inefficient small TCP messages while avoiding
+ the nagle/delayed ack problem (CASSANDRA-1896)
+ * fix race condition in MessagingService.targets (CASSANDRA-1959)
+
0.6.9
* add clustertool, config-converter, sstablekeys, and schematool
Modified:
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/net/MessagingService.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/net/MessagingService.java?rev=1057932&r1=1057931&r2=1057932&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/net/MessagingService.java
(original)
+++
cassandra/branches/cassandra-0.6/src/java/org/apache/cassandra/net/MessagingService.java
Tue Jan 11 23:39:33 2011
@@ -29,6 +29,7 @@ import java.nio.channels.AsynchronousClo
import java.nio.channels.ServerSocketChannel;
import java.security.MessageDigest;
import java.util.*;
+import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
@@ -53,6 +54,7 @@ import org.apache.cassandra.utils.Expiri
import org.apache.cassandra.utils.GuidGenerator;
import org.apache.cassandra.utils.SimpleCondition;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
+import org.cliffc.high_scale_lib.NonBlockingHashSet;
public class MessagingService implements ILatencyPublisher
{
@@ -65,7 +67,7 @@ public class MessagingService implements
/* This records all the results mapped by message Id */
private static ExpiringMap<String, IMessageCallback> callbacks;
- private static Multimap<String, InetAddress> targets;
+ private static ConcurrentMap<String, Collection<InetAddress>> targets =
new NonBlockingHashMap<String, Collection<InetAddress>>();
/* Lookup table for registering message handlers based on the verb. */
private static Map<StorageService.Verb, IVerbHandler> verbHandlers_;
@@ -109,7 +111,7 @@ public class MessagingService implements
{
public Object apply(String messageId)
{
- Collection<InetAddress> addresses =
targets.removeAll(messageId);
+ Collection<InetAddress> addresses = targets.remove(messageId);
if (addresses == null)
return null;
@@ -122,7 +124,6 @@ public class MessagingService implements
return null;
}
};
- targets = ArrayListMultimap.create();
callbacks = new ExpiringMap<String, IMessageCallback>((long) (1.1 *
DatabaseDescriptor.getRpcTimeout()), timeoutReporter);
defaultExecutor_ = new
JMXEnabledThreadPoolExecutor("MISCELLANEOUS-POOL");
@@ -240,12 +241,33 @@ public class MessagingService implements
addCallback(cb, messageId);
for (InetAddress endpoint : to)
{
- targets.put(messageId, endpoint);
+ putTarget(messageId, endpoint);
sendOneWay(message, endpoint);
}
return messageId;
}
+ private static void putTarget(String messageId, InetAddress endpoint)
+ {
+ Collection<InetAddress> addresses = targets.get(messageId);
+ if (addresses == null)
+ {
+ addresses = new NonBlockingHashSet<InetAddress>();
+ Collection<InetAddress> oldAddresses =
targets.putIfAbsent(messageId, addresses);
+ if (oldAddresses != null)
+ addresses = oldAddresses;
+ }
+ addresses.add(endpoint);
+ }
+
+ private static void removeTarget(String messageId, InetAddress from)
+ {
+ Collection<InetAddress> addresses = targets.get(messageId);
+ // null is expected if we removed the callback or we got a reply after
its timeout expired
+ if (addresses != null)
+ addresses.remove(from);
+ }
+
public void addCallback(IAsyncCallback cb, String messageId)
{
callbacks.put(messageId, cb);
@@ -265,7 +287,7 @@ public class MessagingService implements
{
String messageId = message.getMessageId();
addCallback(cb, messageId);
- targets.put(messageId, to);
+ putTarget(messageId, to);
sendOneWay(message, to);
return messageId;
}
@@ -294,7 +316,7 @@ public class MessagingService implements
for ( int i = 0; i < messages.length; ++i )
{
messages[i].setMessageId(groupId);
- targets.put(groupId, to[i]);
+ putTarget(groupId, to[i]);
sendOneWay(messages[i], to[i]);
}
return groupId;
@@ -348,7 +370,7 @@ public class MessagingService implements
{
IAsyncResult iar = new AsyncResult();
callbacks.put(message.getMessageId(), iar);
- targets.put(message.getMessageId(), to);
+ putTarget(message.getMessageId(), to);
sendOneWay(message, to);
return iar;
}
@@ -429,7 +451,7 @@ public class MessagingService implements
public static IMessageCallback removeRegisteredCallback(String messageId)
{
- targets.removeAll(messageId); // TODO fix this when we clean up quorum
reads to do proper RR
+ targets.remove(messageId); // TODO fix this when we clean up quorum
reads to do proper RR
return callbacks.remove(messageId);
}
@@ -440,7 +462,7 @@ public class MessagingService implements
public static void responseReceivedFrom(String messageId, InetAddress from)
{
- targets.remove(messageId, from);
+ removeTarget(messageId, from);
}
public static void validateMagic(int magic) throws IOException
