p/4443/050_process_queued_xfers

Read queued token ranges transfers from a column family, and initiate
relocations.

Patch by eevans; reviewed by Brandon Williams for CASSANDRA-4559

src/java/org/apache/cassandra/config/CFMetaData.java

@@ -200,6 +200,11 @@ public final class CFMetaData
                                                             + "data blob"
                                                             + ") WITH COMMENT='uncommited batches' AND gc_grace_seconds=0");
 
+    public static final CFMetaData RangeXfersCf = compile(17, "CREATE TABLE " + SystemTable.RANGE_XFERS_CF + " ("
+                                                              + "token_bytes blob PRIMARY KEY,"
+                                                              + "requested_at timestamp"
+                                                              + ") WITH COMMENT='ranges requested for transfer here'");
+
     public enum Caching
     {
         ALL, KEYS_ONLY, ROWS_ONLY, NONE;

src/java/org/apache/cassandra/config/KSMetaData.java

@@ -79,6 +79,7 @@ public final class KSMetaData
     public static KSMetaData systemKeyspace()
     {
         List<CFMetaData> cfDefs = Arrays.asList(CFMetaData.BatchlogCF,
+                                                CFMetaData.RangeXfersCf,
                                                 CFMetaData.LocalCf,
                                                 CFMetaData.PeersCf,
                                                 CFMetaData.HintsCf,

src/java/org/apache/cassandra/cql3/UntypedResultSet.java

@@ -20,6 +20,7 @@ package org.apache.cassandra.cql3;
 
 import java.net.InetAddress;
 import java.nio.ByteBuffer;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
@@ -124,6 +125,11 @@ public class UntypedResultSet implements Iterable<UntypedResultSet.Row>
             return UUIDType.instance.compose(data.get(column));
         }
 
+        public Date getTimestamp(String column)
+        {
+            return DateType.instance.compose(data.get(column));
+        }
+
         @Override
         public String toString()
         {

src/java/org/apache/cassandra/db/SystemTable.java

@@ -58,6 +58,7 @@ public class SystemTable
     public static final String INDEX_CF = "IndexInfo";
     public static final String COUNTER_ID_CF = "NodeIdInfo";
     public static final String HINTS_CF = "hints";
+    public static final String RANGE_XFERS_CF = "range_xfers";
     public static final String BATCHLOG_CF = "batchlog";
     // see layout description in the DefsTable class header
     public static final String SCHEMA_KEYSPACES_CF = "schema_keyspaces";

src/java/org/apache/cassandra/service/ScheduledRangeTransferExecutorService.java

@@ -0,0 +1,122 @@
+package org.apache.cassandra.service;
+
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Date;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.SystemTable;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.apache.cassandra.cql3.QueryProcessor.processInternal;
+
+public class ScheduledRangeTransferExecutorService
+{
+    private static final Logger LOG = LoggerFactory.getLogger(ScheduledRangeTransferExecutorService.class);
+    private static final int INTERVAL = 10;
+    private ScheduledExecutorService scheduler;
+
+    public void setup()
+    {
+        if (DatabaseDescriptor.getNumTokens() == 1)
+        {
+            LOG.warn("Cannot start range transfer scheduler: endpoint is not virtual nodes-enabled");
+            return;
+        }
+
+        scheduler = Executors.newSingleThreadScheduledExecutor(new RangeTransferThreadFactory());
+        scheduler.scheduleWithFixedDelay(new RangeTransfer(), 0, INTERVAL, TimeUnit.SECONDS);
+        LOG.info("Enabling scheduled transfers of token ranges");
+    }
+
+    public void tearDown()
+    {
+        if (scheduler == null)
+        {
+            LOG.warn("Unabled to shutdown; Scheduler never enabled");
+            return;
+        }
+ 
+        LOG.info("Shutting down range transfer scheduler");
+        scheduler.shutdownNow();
+    }
+}
+
+class RangeTransfer implements Runnable
+{
+    private static final Logger LOG = LoggerFactory.getLogger(RangeTransfer.class);
+
+    public void run()
+    {
+        UntypedResultSet res = processInternal("SELECT * FROM system." + SystemTable.RANGE_XFERS_CF + " LIMIT 1");
+
+        if (res.size() < 1)
+        {
+            LOG.debug("No queued ranges to transfer");
+            return;
+        }
+
+        if (!isReady())
+            return;
+
+        UntypedResultSet.Row row = res.iterator().next();
+
+        Date requestedAt = row.getTimestamp("requested_at");
+        ByteBuffer tokenBytes = row.getBytes("token_bytes");
+        Token token = StorageService.getPartitioner().getTokenFactory().fromByteArray(tokenBytes);
+
+        LOG.info("Initiating transfer of {} (scheduled at {})", token, requestedAt.toString());
+        try
+        {
+            StorageService.instance.relocateTokens(Collections.singleton(token));
+        }
+        catch (Exception e)
+        {
+            LOG.error("Error removing {}: {}", token, e);
+        }
+        finally
+        {
+            LOG.debug("Removing queued entry for transfer of {}", token);
+            processInternal(String.format("DELETE FROM system.%s WHERE token_bytes = '%s'",
+                                          SystemTable.RANGE_XFERS_CF,
+                                          ByteBufferUtil.bytesToHex(tokenBytes)));
+        }
+    }
+
+    private boolean isReady()
+    {
+        int targetTokens = DatabaseDescriptor.getNumTokens();
+        int highMark = (int)Math.ceil(targetTokens + (targetTokens * .10));
+        int actualTokens = StorageService.instance.getTokens().size();
+
+        if (actualTokens >= highMark)
+        {
+            LOG.warn("Pausing until token count stabilizes (target={}, actual={})", targetTokens, actualTokens);
+            return false;
+        }
+
+        return true;
+    }
+}
+
+class RangeTransferThreadFactory implements ThreadFactory
+{
+    private AtomicInteger count = new AtomicInteger(0);
+
+    public Thread newThread(Runnable r)
+    {
+        Thread rangeXferThread = new Thread(r);
+        rangeXferThread.setName(String.format("ScheduledRangeXfers:%d", count.getAndIncrement()));
+        return rangeXferThread;
+    }
+}

src/java/org/apache/cassandra/service/StorageService.java

@@ -173,6 +173,8 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
 
     private static final AtomicInteger nextRepairCommand = new AtomicInteger();
 
+    private static ScheduledRangeTransferExecutorService rangeXferExecutor = new ScheduledRangeTransferExecutorService();
+
     public void finishBootstrapping()
     {
         isBootstrapMode = false;
@@ -2752,7 +2754,7 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
         relocateTokens(tokens);
     }
 
-    private void relocateTokens(Collection<Token> srcTokens)
+    void relocateTokens(Collection<Token> srcTokens)
     {
         assert srcTokens != null;
         InetAddress localAddress = FBUtilities.getBroadcastAddress();
@@ -3505,4 +3507,14 @@ public class StorageService implements IEndpointStateChangeSubscriber, StorageSe
     {
         return tracingProbability;
     }
+
+    public void enableScheduledRangeXfers()
+    {
+        rangeXferExecutor.setup();
+    }
+
+    public void disableScheduledRangeXfers()
+    {
+        rangeXferExecutor.tearDown();
+    }
 }

src/java/org/apache/cassandra/service/StorageServiceMBean.java

@@ -457,4 +457,9 @@ public interface StorageServiceMBean
      * Returns the configured tracing probability.
      */
     public double getTracingProbability();
+
+    /** Begin processing of queued range transfers. */
+    public void enableScheduledRangeXfers();
+    /** Disable processing of queued range transfers. */
+    public void disableScheduledRangeXfers();
 }