cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [3/8] git commit: Revert "Don't shut ExpiringMap down."
Date Wed, 02 Apr 2014 19:57:35 GMT
Revert "Don't shut ExpiringMap down."

This reverts commit d4ec31f21eb83502bd523c057bce2d3b249b3d80.

Conflicts:
	CHANGES.txt


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/4f47a44d
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/4f47a44d
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/4f47a44d

Branch: refs/heads/trunk
Commit: 4f47a44d05a1c9b8e284c2d8c084edb7c0376d1d
Parents: 56d84a7
Author: Brandon Williams <brandonwilliams@apache.org>
Authored: Wed Apr 2 14:53:25 2014 -0500
Committer: Brandon Williams <brandonwilliams@apache.org>
Committed: Wed Apr 2 14:53:25 2014 -0500

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 -
 .../apache/cassandra/net/MessagingService.java  | 13 +++++-
 .../org/apache/cassandra/utils/ExpiringMap.java | 47 +++++++++++++++-----
 .../apache/cassandra/service/RemoveTest.java    |  1 +
 4 files changed, 47 insertions(+), 15 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/4f47a44d/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d999e88..9003309 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,6 +1,5 @@
 2.0.7
  * Allow compaction of system tables during startup (CASSANDRA-6913)
- * Don't shut ExpiringMap down (CASSANDRA-6948)
  * Restrict Windows to parallel repairs (CASSANDRA-6907)
  * (Hadoop) Allow manually specifying start/end tokens in CFIF (CASSANDRA-6436)
  * Fix NPE in MeteredFlusher (CASSANDRA-6820)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/4f47a44d/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index 094e861..ad86bbd 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -392,6 +392,7 @@ public final class MessagingService implements MessagingServiceMBean
      */
     public void listen(InetAddress localEp) throws ConfigurationException
     {
+        callbacks.reset(); // hack to allow tests to stop/restart MS
         for (ServerSocket ss : getServerSockets(localEp))
         {
             SocketThread th = new SocketThread(ss, "ACCEPT-" + localEp);
@@ -535,7 +536,7 @@ public final class MessagingService implements MessagingServiceMBean
     {
         assert message.verb != Verb.MUTATION; // mutations need to call the overload with
a ConsistencyLevel
         int messageId = nextId();
-        ExpiringMap.CacheableObject<CallbackInfo> previous = callbacks.put(messageId,
new CallbackInfo(to, cb, callbackDeserializers.get(message.verb)), timeout);
+        CallbackInfo previous = callbacks.put(messageId, new CallbackInfo(to, cb, callbackDeserializers.get(message.verb)),
timeout);
         assert previous == null : String.format("Callback already exists for id %d! (%s)",
messageId, previous);
         return messageId;
     }
@@ -544,7 +545,7 @@ public final class MessagingService implements MessagingServiceMBean
     {
         assert message.verb == Verb.MUTATION || message.verb == Verb.COUNTER_MUTATION;
         int messageId = nextId();
-        ExpiringMap.CacheableObject<CallbackInfo> previous = callbacks.put(messageId,
new WriteCallbackInfo(to, cb, message, callbackDeserializers.get(message.verb), consistencyLevel),
timeout);
+        CallbackInfo previous = callbacks.put(messageId, new WriteCallbackInfo(to, cb, message,
callbackDeserializers.get(message.verb), consistencyLevel), timeout);
         assert previous == null : String.format("Callback already exists for id %d! (%s)",
messageId, previous);
         return messageId;
     }
@@ -653,6 +654,11 @@ public final class MessagingService implements MessagingServiceMBean
         subscribers.add(subcriber);
     }
 
+    public void clearCallbacksUnsafe()
+    {
+        callbacks.reset();
+    }
+
     /**
      * Wait for callbacks and don't allow any more to be created (since they could require
writing hints)
      */
@@ -662,6 +668,9 @@ public final class MessagingService implements MessagingServiceMBean
         // We may need to schedule hints on the mutation stage, so it's erroneous to shut
down the mutation stage first
         assert !StageManager.getStage(Stage.MUTATION).isShutdown();
 
+        // the important part
+        callbacks.shutdownBlocking();
+
         // attempt to humor tests that try to stop and restart MS
         try
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/4f47a44d/src/java/org/apache/cassandra/utils/ExpiringMap.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/utils/ExpiringMap.java b/src/java/org/apache/cassandra/utils/ExpiringMap.java
index cbec808..7eec40e 100644
--- a/src/java/org/apache/cassandra/utils/ExpiringMap.java
+++ b/src/java/org/apache/cassandra/utils/ExpiringMap.java
@@ -21,7 +21,6 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
@@ -36,6 +35,7 @@ import org.apache.cassandra.concurrent.DebuggableScheduledThreadPoolExecutor;
 public class ExpiringMap<K, V>
 {
     private static final Logger logger = LoggerFactory.getLogger(ExpiringMap.class);
+    private volatile boolean shutdown;
 
     public static class CacheableObject<T>
     {
@@ -55,17 +55,12 @@ public class ExpiringMap<K, V>
         {
             return atNano - createdAt > TimeUnit.MILLISECONDS.toNanos(timeout);
         }
-
-        public String toString()
-        {
-            return "CacheableObject(obj=" + value.toString() + ", deltaFromTimeout=" + (System.nanoTime()
- (createdAt + TimeUnit.MILLISECONDS.toNanos(timeout))) + "ns)";
-        }
     }
 
     // if we use more ExpiringMaps we may want to add multiple threads to this executor
     private static final ScheduledExecutorService service = new DebuggableScheduledThreadPoolExecutor("EXPIRING-MAP-REAPER");
 
-    private final ConcurrentMap<K, CacheableObject<V>> cache = new ConcurrentHashMap<>();
+    private final ConcurrentMap<K, CacheableObject<V>> cache = new ConcurrentHashMap<K,
CacheableObject<V>>();
     private final long defaultExpiration;
 
     public ExpiringMap(long defaultExpiration)
@@ -86,7 +81,7 @@ public class ExpiringMap<K, V>
             throw new IllegalArgumentException("Argument specified must be a positive number");
         }
 
-        Runnable reaperTask = new Runnable()
+        Runnable runnable = new Runnable()
         {
             public void run()
             {
@@ -105,17 +100,45 @@ public class ExpiringMap<K, V>
                 logger.trace("Expired {} entries", n);
             }
         };
-        service.scheduleWithFixedDelay(reaperTask, defaultExpiration / 2, defaultExpiration
/ 2, TimeUnit.MILLISECONDS);
+        service.scheduleWithFixedDelay(runnable, defaultExpiration / 2, defaultExpiration
/ 2, TimeUnit.MILLISECONDS);
     }
 
-    public CacheableObject<V> put(K key, V value)
+    public void shutdownBlocking()
+    {
+        service.shutdown();
+        try
+        {
+            service.awaitTermination(defaultExpiration * 2, TimeUnit.MILLISECONDS);
+        }
+        catch (InterruptedException e)
+        {
+            throw new AssertionError(e);
+        }
+    }
+
+    public void reset()
+    {
+        shutdown = false;
+        cache.clear();
+    }
+
+    public V put(K key, V value)
     {
         return put(key, value, this.defaultExpiration);
     }
 
-    public CacheableObject<V> put(K key, V value, long timeout)
+    public V put(K key, V value, long timeout)
     {
-        return cache.put(key, new CacheableObject<V>(value, timeout));
+        if (shutdown)
+        {
+            // StorageProxy isn't equipped to deal with "I'm nominally alive, but I can't
send any messages out."
+            // So we'll just sit on this thread until the rest of the server shutdown completes.
+            //
+            // See comments in CustomTThreadPoolServer.serve, CASSANDRA-3335, and CASSANDRA-3727.
+            Uninterruptibles.sleepUninterruptibly(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
+        }
+        CacheableObject<V> previous = cache.put(key, new CacheableObject<V>(value,
timeout));
+        return (previous == null) ? null : previous.value;
     }
 
     public V get(K key)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/4f47a44d/test/unit/org/apache/cassandra/service/RemoveTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/RemoveTest.java b/test/unit/org/apache/cassandra/service/RemoveTest.java
index 82e5e9e..62dd636 100644
--- a/test/unit/org/apache/cassandra/service/RemoveTest.java
+++ b/test/unit/org/apache/cassandra/service/RemoveTest.java
@@ -92,6 +92,7 @@ public class RemoveTest
     public void tearDown()
     {
         SinkManager.clear();
+        MessagingService.instance().clearCallbacksUnsafe();
         MessagingService.instance().shutdown();
     }
 


Mime
View raw message