svn commit: r898414 - /incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java

[jb...@apache.org]

Author: jbellis
Date: Tue Jan 12 16:47:23 2010
New Revision: 898414

URL: http://svn.apache.org/viewvc?rev=898414&view=rev
Log:
avoid creating a new thread for each requested GC.  easier for telemetery (and better design anyway) to re-use the same thread.  patch by jbellis

Modified:
    incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java

Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java?rev=898414&r1=898413&r2=898414&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/CompactionManager.java Tue Jan 12 16:47:23 2010
@@ -29,7 +29,6 @@
 import org.apache.log4j.Logger;
 
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
-import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.io.*;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -37,7 +36,6 @@
 import org.apache.cassandra.service.AntiEntropyService;
 import org.apache.cassandra.io.util.FileUtils;
 import org.apache.cassandra.utils.FBUtilities;
-import org.apache.cassandra.utils.WrappedRunnable;
 import org.cliffc.high_scale_lib.NonBlockingHashMap;
 
 import java.net.InetAddress;
@@ -53,6 +51,8 @@
     private static final Logger logger = Logger.getLogger(CompactionManager.class);
     public static final CompactionManager instance;
 
+    private static volatile boolean gcRequested;
+
     private int minimumCompactionThreshold = 4; // compact this many sstables min at a time
     private int maximumCompactionThreshold = 32; // compact this many sstables max at a time
 
@@ -68,11 +68,40 @@
         {
             throw new RuntimeException(e);
         }
+
+        /**
+         * thread that requests GCs to clean out obsolete sstables, sleeping rpc timeout first so that most in-progress ops can complete
+         * (thus, no longer reference the sstables in question)
+         */
+        new Thread(new Runnable()
+        {
+            final long gcDelay = DatabaseDescriptor.getRpcTimeout();
+
+            public void run()
+            {
+                while (true)
+                {
+                    try
+                    {
+                        Thread.sleep(gcDelay * 10);
+                        if (gcRequested)
+                        {
+                            Thread.sleep(gcDelay);
+                            System.gc();
+                            gcRequested = false;
+                        }
+                    }
+                    catch (InterruptedException e)
+                    {
+                        throw new AssertionError(e);
+                    }
+                }
+            }
+        }, "COMPACTION-GC-INVOKER").start();
     }
 
     private CompactionExecutor executor = new CompactionExecutor();
     private Map<ColumnFamilyStore, Integer> estimatedCompactions = new NonBlockingHashMap<ColumnFamilyStore, Integer>();
-    private static final NamedThreadFactory gcThreadFactory = new NamedThreadFactory("GC-INVOKER");
 
     /**
      * Call this whenever a compaction might be needed on the given columnfamily.
@@ -308,7 +337,7 @@
 
         SSTableReader ssTable = writer.closeAndOpenReader(DatabaseDescriptor.getKeysCachedFraction(table.name, cfs.getColumnFamilyName()));
         cfs.replaceCompactedSSTables(sstables, Arrays.asList(ssTable));
-        gcAfterRpcTimeout();
+        gcRequested = true;
         submitMinorIfNeeded(cfs);
 
         String format = "Compacted to %s.  %d/%d bytes for %d keys.  Time: %dms.";
@@ -409,7 +438,7 @@
         {
             cfs.replaceCompactedSSTables(originalSSTables, sstables);
         }
-        CompactionManager.gcAfterRpcTimeout();
+        gcRequested = true;
     }
 
     /**
@@ -441,22 +470,6 @@
         }
     }
 
-    /**
-     * perform a GC to clean out obsolete sstables, sleeping rpc timeout first so that most in-progress ops can complete
-     * (thus, no longer reference the sstables in question)
-     */
-    static void gcAfterRpcTimeout()
-    {
-        gcThreadFactory.newThread(new WrappedRunnable()
-        {
-            public void runMayThrow() throws InterruptedException
-            {
-                Thread.sleep(DatabaseDescriptor.getRpcTimeout());
-                System.gc();
-            }
-        }).start();
-    }
-
     /*
     * Group files of similar size into buckets.
     */