cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From brandonwilli...@apache.org
Subject [4/4] git commit: Update CQLSSTableWriter to allow parallel writing of SSTables on the same table within the same JVM
Date Thu, 30 Oct 2014 15:50:22 GMT
Update CQLSSTableWriter to allow parallel writing of SSTables on the same table within the
same JVM

Patch by Carl Yeksigian, reviewed by Benjamin Lerer for CASSANDRA-7463


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

Branch: refs/heads/trunk
Commit: fbb6b8abfa359be3475ba55912d939096637b8f8
Parents: c4c9303
Author: Brandon Williams <brandonwilliams@apache.org>
Authored: Thu Oct 30 10:49:42 2014 -0500
Committer: Brandon Williams <brandonwilliams@apache.org>
Committed: Thu Oct 30 10:49:42 2014 -0500

----------------------------------------------------------------------
 .../io/sstable/AbstractSSTableSimpleWriter.java | 12 ++-
 .../cassandra/io/sstable/CQLSSTableWriter.java  | 41 +++++----
 .../io/sstable/CQLSSTableWriterTest.java        | 93 ++++++++++++++++++++
 3 files changed, 127 insertions(+), 19 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbb6b8ab/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
index 11d6d5e..2854308 100644
--- a/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/AbstractSSTableSimpleWriter.java
@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.cassandra.config.CFMetaData;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -45,6 +46,7 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
     protected ByteBuffer currentSuperColumn;
     protected final CounterId counterid = CounterId.generate();
     private SSTableFormat.Type formatType = DatabaseDescriptor.getSSTableFormat();
+    protected static AtomicInteger generation = new AtomicInteger(0);
 
 
     public AbstractSSTableSimpleWriter(File directory, CFMetaData metadata, IPartitioner
partitioner)
@@ -83,9 +85,15 @@ public abstract class AbstractSSTableSimpleWriter implements Closeable
                 return false;
             }
         });
-        int maxGen = 0;
+        int maxGen = generation.getAndIncrement();
         for (Descriptor desc : existing)
-            maxGen = Math.max(maxGen, desc.generation);
+        {
+            while (desc.generation > maxGen)
+            {
+                maxGen = generation.getAndIncrement();
+            }
+        }
+
         return new Descriptor(directory, keyspace, columnFamily, maxGen + 1, Descriptor.Type.TEMP,
fmt).filenameFor(Component.DATA);
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbb6b8ab/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
index f7d467e..d4b4eab 100644
--- a/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
@@ -28,6 +28,7 @@ import java.util.List;
 import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
 
 import org.apache.cassandra.cql3.statements.*;
 import org.apache.cassandra.cql3.*;
@@ -338,25 +339,31 @@ public class CQLSSTableWriter implements Closeable
         {
             try
             {
-                this.schema = getStatement(schema, CreateTableStatement.class, "CREATE TABLE").left.getCFMetaData().rebuild();
-
-                // We need to register the keyspace/table metadata through Schema, otherwise
we won't be able to properly
-                // build the insert statement in using().
-                if (Schema.instance.getKSMetaData(this.schema.ksName) == null)
-                {
-                    KSMetaData ksm = KSMetaData.newKeyspace(this.schema.ksName,
-                                                            AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
-                                                            ImmutableMap.of("replication_factor",
"1"),
-                                                            true,
-                                                            Collections.singleton(this.schema));
-                    Schema.instance.load(ksm);
-                }
-                else if (Schema.instance.getCFMetaData(this.schema.ksName, this.schema.cfName)
== null)
+                synchronized (CQLSSTableWriter.class)
                 {
-                    Schema.instance.load(this.schema);
-                }
+                    this.schema = getStatement(schema, CreateTableStatement.class, "CREATE
TABLE").left.getCFMetaData().rebuild();
 
-                return this;
+                    // We need to register the keyspace/table metadata through Schema, otherwise
we won't be able to properly
+                    // build the insert statement in using().
+                    KSMetaData ksm = Schema.instance.getKSMetaData(this.schema.ksName);
+                    if (ksm == null)
+                    {
+                        ksm = KSMetaData.newKeyspace(this.schema.ksName,
+                                AbstractReplicationStrategy.getClass("org.apache.cassandra.locator.SimpleStrategy"),
+                                ImmutableMap.of("replication_factor", "1"),
+                                true,
+                                Collections.singleton(this.schema));
+                        Schema.instance.load(ksm);
+                    }
+                    else if (Schema.instance.getCFMetaData(this.schema.ksName, this.schema.cfName)
== null)
+                    {
+                        Schema.instance.load(this.schema);
+                        ksm = KSMetaData.cloneWith(ksm, Iterables.concat(ksm.cfMetaData().values(),
Collections.singleton(this.schema)));
+                        Schema.instance.setKeyspaceDefinition(ksm);
+                        Keyspace.open(ksm.name).initCf(this.schema.cfId, this.schema.cfName,
false);
+                    }
+                    return this;
+                }
             }
             catch (RequestValidationException e)
             {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/fbb6b8ab/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
index 250545a..2a41794 100644
--- a/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
+++ b/test/unit/org/apache/cassandra/io/sstable/CQLSSTableWriterTest.java
@@ -163,4 +163,97 @@ public class CQLSSTableWriterTest
         };
         assert dataDir.list(filterDataFiles).length > 1 : Arrays.toString(dataDir.list(filterDataFiles));
     }
+
+
+    private static final int NUMBER_WRITES_IN_RUNNABLE = 10;
+    private class WriterThread extends Thread
+    {
+        private final File dataDir;
+        private final int id;
+        public volatile Exception exception;
+
+        public WriterThread(File dataDir, int id)
+        {
+            this.dataDir = dataDir;
+            this.id = id;
+        }
+
+        @Override
+        public void run()
+        {
+            String schema = "CREATE TABLE cql_keyspace.table2 ("
+                    + "  k int,"
+                    + "  v int,"
+                    + "  PRIMARY KEY (k, v)"
+                    + ")";
+            String insert = "INSERT INTO cql_keyspace.table2 (k, v) VALUES (?, ?)";
+            CQLSSTableWriter writer = CQLSSTableWriter.builder()
+                    .inDirectory(dataDir)
+                    .forTable(schema)
+                    .withPartitioner(StorageService.instance.getPartitioner())
+                    .using(insert).build();
+
+            try
+            {
+                for (int i = 0; i < NUMBER_WRITES_IN_RUNNABLE; i++)
+                {
+                    writer.addRow(id, i);
+                }
+                writer.close();
+            }
+            catch (Exception e)
+            {
+                exception = e;
+            }
+        }
+    }
+
+    @Test
+    public void testConcurrentWriters() throws Exception
+    {
+        String KS = "cql_keyspace";
+        String TABLE = "table2";
+
+        File tempdir = Files.createTempDir();
+        File dataDir = new File(tempdir.getAbsolutePath() + File.separator + KS + File.separator
+ TABLE);
+        assert dataDir.mkdirs();
+
+        WriterThread[] threads = new WriterThread[5];
+        for (int i = 0; i < threads.length; i++)
+        {
+            WriterThread thread = new WriterThread(dataDir, i);
+            threads[i] = thread;
+            thread.start();
+        }
+
+        for (WriterThread thread : threads)
+        {
+            thread.join();
+            assert !thread.isAlive() : "Thread should be dead by now";
+            if (thread.exception != null)
+            {
+                throw thread.exception;
+            }
+        }
+
+        SSTableLoader loader = new SSTableLoader(dataDir, new SSTableLoader.Client()
+        {
+            public void init(String keyspace)
+            {
+                for (Range<Token> range : StorageService.instance.getLocalRanges("cql_keyspace"))
+                    addRangeForEndpoint(range, FBUtilities.getBroadcastAddress());
+                setPartitioner(StorageService.getPartitioner());
+            }
+
+            public CFMetaData getCFMetaData(String keyspace, String cfName)
+            {
+                return Schema.instance.getCFMetaData(keyspace, cfName);
+            }
+        }, new OutputHandler.SystemOutput(false, false));
+
+        loader.stream().get();
+
+        UntypedResultSet rs = QueryProcessor.processInternal("SELECT * FROM cql_keyspace.table2;");
+        assertEquals(threads.length * NUMBER_WRITES_IN_RUNNABLE, rs.size());
+    }
 }


Mime
View raw message