You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by gd...@apache.org on 2010/03/11 18:53:34 UTC
svn commit: r921963 - in /incubator/cassandra/trunk:
src/java/org/apache/cassandra/db/DefsTable.java
src/java/org/apache/cassandra/db/Table.java
test/unit/org/apache/cassandra/db/DefsTest.java
Author: gdusbabek
Date: Thu Mar 11 17:53:34 2010
New Revision: 921963
URL: http://svn.apache.org/viewvc?rev=921963&view=rev
Log:
improve concurrency during defn mutations. Patch by Gary Dusbabek, reviewed by Jonthan Ellis. CASSANDRA-840
Modified:
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DefsTable.java
incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/DefsTest.java
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DefsTable.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DefsTable.java?rev=921963&r1=921962&r2=921963&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DefsTable.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/DefsTable.java Thu Mar 11 17:53:34 2010
@@ -18,6 +18,7 @@
package org.apache.cassandra.db;
+import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.KSMetaData;
@@ -27,6 +28,7 @@ import org.apache.cassandra.db.filter.Sl
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.UUIDGen;
+import org.apache.cassandra.utils.WrappedRunnable;
import static org.apache.cassandra.config.DatabaseDescriptor.ConfigurationException;
@@ -36,171 +38,165 @@ import java.io.FileFilter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
public class DefsTable
{
+ private static final ExecutorService executor = new JMXEnabledThreadPoolExecutor("DEFINITIONS-UPDATER");
+
public static final String MIGRATIONS_CF = "Migrations";
public static final String SCHEMA_CF = "Schema";
/** add a column family. */
- public static synchronized void add(CFMetaData cfm) throws IOException, ConfigurationException
+ public static Future add(final CFMetaData cfm)
{
- Table.openLock.lock();
- try
+ return executor.submit(new WrappedRunnable()
{
- // make sure the ks is real and the cf doesn't already exist.
- KSMetaData ksm = DatabaseDescriptor.getTableDefinition(cfm.tableName);
- if (ksm == null)
- throw new ConfigurationException("Keyspace does not already exist.");
- else if (ksm.cfMetaData().containsKey(cfm.cfName))
- throw new ConfigurationException("CF is already defined in that keyspace.");
-
- // clone ksm but include the new cf def.
- List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
- newCfs.add(cfm);
- ksm = new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, ksm.snitch, newCfs.toArray(new CFMetaData[newCfs.size()]));
-
- // store it.
- UUID newVersion = UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress());
- RowMutation rm = new RowMutation(Table.DEFINITIONS, newVersion.toString());
- rm.add(new QueryPath(SCHEMA_CF, null, ksm.name.getBytes()), KSMetaData.serialize(ksm), System.currentTimeMillis());
- rm.apply();
-
- // reinitialize the table.
- DatabaseDescriptor.setTableDefinition(ksm, newVersion);
- Table.close(ksm.name);
- Table.open(ksm.name);
-
- // force creation of a new commit log segment.
- CommitLog.instance().forceNewSegment();
- }
- finally
- {
- Table.openLock.unlock();
- }
+ protected void runMayThrow() throws Exception
+ {
+ // make sure the ks is real and the cf doesn't already exist.
+ KSMetaData ksm = DatabaseDescriptor.getTableDefinition(cfm.tableName);
+ if (ksm == null)
+ throw new ConfigurationException("Keyspace does not already exist.");
+ else if (ksm.cfMetaData().containsKey(cfm.cfName))
+ throw new ConfigurationException("CF is already defined in that keyspace.");
+
+ // clone ksm but include the new cf def.
+ List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
+ newCfs.add(cfm);
+ ksm = new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, ksm.snitch, newCfs.toArray(new CFMetaData[newCfs.size()]));
+
+ // store it.
+ UUID newVersion = UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress());
+ RowMutation rm = new RowMutation(Table.DEFINITIONS, newVersion.toString());
+ rm.add(new QueryPath(SCHEMA_CF, null, ksm.name.getBytes()), KSMetaData.serialize(ksm), System.currentTimeMillis());
+ rm.apply();
+
+ // reinitialize the table.
+ Table.open(ksm.name).addCf(cfm.cfName);
+ DatabaseDescriptor.setTableDefinition(ksm, newVersion);
+
+ // force creation of a new commit log segment.
+ CommitLog.instance().forceNewSegment();
+ }
+ });
}
/**
* drop a column family. blockOnDeletion was added to make testing simpler.
*/
- public static synchronized void drop(CFMetaData cfm, boolean blockOnDeletion) throws IOException, ConfigurationException
+ public static Future drop(final CFMetaData cfm, final boolean blockOnDeletion)
{
- Table.openLock.lock();
- try
- {
- KSMetaData ksm = DatabaseDescriptor.getTableDefinition(cfm.tableName);
- if (ksm == null)
- throw new ConfigurationException("Keyspace does not already exist.");
- else if (!ksm.cfMetaData().containsKey(cfm.cfName))
- throw new ConfigurationException("CF is not defined in that keyspace.");
-
- // clone ksm but do not include the new def
- List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
- newCfs.remove(cfm);
- assert newCfs.size() == ksm.cfMetaData().size() - 1;
- ksm = new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, ksm.snitch, newCfs.toArray(new CFMetaData[newCfs.size()]));
-
- // store it.
- UUID newVersion = UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress());
- RowMutation rm = new RowMutation(Table.DEFINITIONS, newVersion.toString());
- rm.add(new QueryPath(SCHEMA_CF, null, ksm.name.getBytes()), KSMetaData.serialize(ksm), System.currentTimeMillis());
- rm.apply();
-
- // reinitialize the table.
- CFMetaData.purge(cfm);
- DatabaseDescriptor.setTableDefinition(ksm, newVersion);
- Table.close(ksm.name);
- Table.open(ksm.name);
-
- // indicate that some files need to be deleted (eventually)
- SystemTable.markForRemoval(cfm);
-
- // we don't really need a new segment, but let's force it to be consistent with other operations.
- CommitLog.instance().forceNewSegment();
- }
- finally
+ return executor.submit(new WrappedRunnable()
{
- Table.openLock.unlock();
- }
-
- if (blockOnDeletion)
- {
- // notify the compaction manager that it needs to clean up the dropped cf files.
- try
+ protected void runMayThrow() throws Exception
{
- CompactionManager.instance.submitGraveyardCleanup().get();
- }
- catch (InterruptedException e)
- {
- throw new RuntimeException(e);
- }
- catch (ExecutionException e)
- {
- throw new RuntimeException(e);
+ KSMetaData ksm = DatabaseDescriptor.getTableDefinition(cfm.tableName);
+ if (ksm == null)
+ throw new ConfigurationException("Keyspace does not already exist.");
+ else if (!ksm.cfMetaData().containsKey(cfm.cfName))
+ throw new ConfigurationException("CF is not defined in that keyspace.");
+
+ // clone ksm but do not include the new def
+ List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
+ newCfs.remove(cfm);
+ assert newCfs.size() == ksm.cfMetaData().size() - 1;
+ ksm = new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, ksm.snitch, newCfs.toArray(new CFMetaData[newCfs.size()]));
+
+ // store it.
+ UUID newVersion = UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress());
+ RowMutation rm = new RowMutation(Table.DEFINITIONS, newVersion.toString());
+ rm.add(new QueryPath(SCHEMA_CF, null, ksm.name.getBytes()), KSMetaData.serialize(ksm), System.currentTimeMillis());
+ rm.apply();
+
+ // reinitialize the table.
+ CFMetaData.purge(cfm);
+ DatabaseDescriptor.setTableDefinition(ksm, newVersion);
+ Table.open(ksm.name).dropCf(cfm.cfName);
+
+ // indicate that some files need to be deleted (eventually)
+ SystemTable.markForRemoval(cfm);
+
+ // we don't really need a new segment, but let's force it to be consistent with other operations.
+ CommitLog.instance().forceNewSegment();
+
+
+ if (blockOnDeletion)
+ {
+ // notify the compaction manager that it needs to clean up the dropped cf files.
+ try
+ {
+ CompactionManager.instance.submitGraveyardCleanup().get();
+ }
+ catch (InterruptedException e)
+ {
+ throw new RuntimeException(e);
+ }
+ catch (ExecutionException e)
+ {
+ throw new RuntimeException(e);
+ }
+ }
}
- }
+ });
}
- public static synchronized void rename(CFMetaData oldCfm, String newName) throws IOException, ConfigurationException
+ /** rename a column family */
+ public static Future rename(final CFMetaData oldCfm, final String newName)
{
- Table.openLock.lock();
- try
+ return executor.submit(new WrappedRunnable()
{
- KSMetaData ksm = DatabaseDescriptor.getTableDefinition(oldCfm.tableName);
- if (ksm == null)
- throw new ConfigurationException("Keyspace does not already exist.");
- if (!ksm.cfMetaData().containsKey(oldCfm.cfName))
- throw new ConfigurationException("CF is not defined in that keyspace.");
- if (ksm.cfMetaData().containsKey(newName))
- throw new ConfigurationException("CF is already defined in that keyspace.");
-
- // clone the ksm, replacing cfm with the new one.
- List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
- newCfs.remove(oldCfm);
- assert newCfs.size() == ksm.cfMetaData().size() - 1;
- CFMetaData newCfm = CFMetaData.rename(oldCfm, newName);
- newCfs.add(newCfm);
- ksm = new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, ksm.snitch, newCfs.toArray(new CFMetaData[newCfs.size()]));
-
- // store it
- UUID newVersion = UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress());
- RowMutation rm = new RowMutation(Table.DEFINITIONS, newVersion.toString());
- rm.add(new QueryPath(SCHEMA_CF, null, ksm.name.getBytes()), KSMetaData.serialize(ksm), System.currentTimeMillis());
- rm.apply();
-
- // reset defs.
- DatabaseDescriptor.setTableDefinition(ksm, newVersion);
- Table.close(ksm.name);
-
- // rename the files.
- try
- {
- renameStorageFiles(ksm.name, oldCfm.cfName, newCfm.cfName);
- }
- catch (IOException e)
+ protected void runMayThrow() throws Exception
{
- // todo: is this a big enough problem to bring the entire node down? For sure, it is something that needs to be addressed immediately.
- ConfigurationException cex = new ConfigurationException("Critical: encountered IOException while attempting to rename CF storage files for " + oldCfm.cfName);
- cex.initCause(e);
- throw cex;
+ KSMetaData ksm = DatabaseDescriptor.getTableDefinition(oldCfm.tableName);
+ if (ksm == null)
+ throw new ConfigurationException("Keyspace does not already exist.");
+ if (!ksm.cfMetaData().containsKey(oldCfm.cfName))
+ throw new ConfigurationException("CF is not defined in that keyspace.");
+ if (ksm.cfMetaData().containsKey(newName))
+ throw new ConfigurationException("CF is already defined in that keyspace.");
+
+ // clone the ksm, replacing cfm with the new one.
+ List<CFMetaData> newCfs = new ArrayList<CFMetaData>(ksm.cfMetaData().values());
+ newCfs.remove(oldCfm);
+ assert newCfs.size() == ksm.cfMetaData().size() - 1;
+ CFMetaData newCfm = CFMetaData.rename(oldCfm, newName);
+ newCfs.add(newCfm);
+ ksm = new KSMetaData(ksm.name, ksm.strategyClass, ksm.replicationFactor, ksm.snitch, newCfs.toArray(new CFMetaData[newCfs.size()]));
+
+ // store it
+ UUID newVersion = UUIDGen.makeType1UUIDFromHost(FBUtilities.getLocalAddress());
+ RowMutation rm = new RowMutation(Table.DEFINITIONS, newVersion.toString());
+ rm.add(new QueryPath(SCHEMA_CF, null, ksm.name.getBytes()), KSMetaData.serialize(ksm), System.currentTimeMillis());
+ rm.apply();
+
+ // leave it up to operators to ensure there are no writes going on durng the file rename. Just know that
+ // attempting row mutations on oldcfName right now would be really bad.
+ try
+ {
+ renameStorageFiles(ksm.name, oldCfm.cfName, newCfm.cfName);
+ }
+ catch (IOException e)
+ {
+ // todo: is this a big enough problem to bring the entire node down? For sure, it is something that needs to be addressed immediately.
+ ConfigurationException cex = new ConfigurationException("Critical: encountered IOException while attempting to rename CF storage files for " + oldCfm.cfName);
+ cex.initCause(e);
+ throw cex;
+ }
+ // reset defs.
+ DatabaseDescriptor.setTableDefinition(ksm, newVersion);
+ Table.open(ksm.name).renameCf(oldCfm.cfName, newName);
+
+ CommitLog.instance().forceNewSegment();
}
-
-
- Table.open(ksm.name);
-
- CommitLog.instance().forceNewSegment();
- }
- finally
- {
- Table.openLock.unlock();
- }
+ });
}
+ /** dumps current keyspace definitions to storage */
public static synchronized void dumpToStorage(UUID version) throws IOException
{
String versionKey = version.toString();
@@ -214,6 +210,7 @@ public class DefsTable
rm.apply();
}
+ /** loads a version of keyspace definitions from storage */
public static synchronized Collection<KSMetaData> loadFromStorage(UUID version) throws IOException
{
Table defs = Table.open(Table.DEFINITIONS);
@@ -230,6 +227,7 @@ public class DefsTable
return tables;
}
+ /** gets all the files that belong to a given column family. */
static Collection<File> getFiles(String table, final String cf)
{
List<File> found = new ArrayList<File>();
@@ -239,8 +237,7 @@ public class DefsTable
{
public boolean accept(File pathname)
{
- return pathname.getName().startsWith(cf + "-") && pathname.getName().endsWith(".db") && pathname.exists();
-
+ return pathname.getName().startsWith(cf + "-") && pathname.getName().endsWith(".db") && pathname.exists();
}
});
for (File f : dbFiles)
@@ -269,6 +266,5 @@ public class DefsTable
}
if (mostRecentProblem != null)
throw new IOException("One or more IOExceptions encountered while renaming files. Most recent problem is included.", mostRecentProblem);
-
}
}
Modified: incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java?rev=921963&r1=921962&r2=921963&view=diff
==============================================================================
--- incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java (original)
+++ incubator/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java Thu Mar 11 17:53:34 2010
@@ -83,9 +83,6 @@ public class Table
private final Map<String, ColumnFamilyStore> columnFamilyStores = new HashMap<String, ColumnFamilyStore>();
// cache application CFs since Range queries ask for them a _lot_
private SortedSet<String> applicationColumnFamilies;
-
- // this lock blocks other threads from opening a table during critical operations.
- public static final Lock openLock = new ReentrantLock();
public static Table open(String table) throws IOException
{
@@ -94,8 +91,7 @@ public class Table
{
// instantiate the Table. we could use putIfAbsent but it's important to making sure it is only done once
// per keyspace, so we synchronize and re-check before doing it.
- openLock.lock();
- try
+ synchronized (Table.class)
{
tableInstance = instances.get(table);
if (tableInstance == null)
@@ -104,23 +100,10 @@ public class Table
instances.put(table, tableInstance);
}
}
- finally
- {
- openLock.unlock();
- }
}
return tableInstance;
}
- // prepares a table to be replaced
- public static void close(String table) throws IOException
- {
- // yes, I see the irony.
- Table t = open(table);
- if (t != null)
- open(table).release();
- }
-
public Set<String> getColumnFamilies()
{
return DatabaseDescriptor.getTableDefinition(name).cfMetaData().keySet();
@@ -275,38 +258,40 @@ public class Table
}, checkMs, checkMs);
}
- // undoes the constructor.
- private void release() throws IOException
+ /** removes a cf from internal structures (doesn't change disk files). */
+ public void dropCf(String cfName) throws IOException
{
- openLock.lock();
- try
+ assert columnFamilyStores.containsKey(cfName);
+ ColumnFamilyStore cfs = columnFamilyStores.remove(cfName);
+ if (cfs != null)
{
- instances.remove(name);
- for (String cfName : getColumnFamilies())
+ try
{
- // clear out memtables.
- ColumnFamilyStore cfs = columnFamilyStores.remove(cfName);
- if (cfs != null)
- {
- try
- {
- cfs.forceBlockingFlush();
- }
- catch (ExecutionException e)
- {
- throw new IOException(e);
- }
- catch (InterruptedException e)
- {
- throw new IOException(e);
- }
- }
+ cfs.forceBlockingFlush();
+ }
+ catch (ExecutionException e)
+ {
+ throw new IOException(e);
+ }
+ catch (InterruptedException e)
+ {
+ throw new IOException(e);
}
}
- finally
- {
- openLock.unlock();
- }
+ }
+
+ /** adds a cf to internal structures, ends up creating disk files). */
+ public void addCf(String cfName) throws IOException
+ {
+ assert !columnFamilyStores.containsKey(cfName) : cfName;
+ columnFamilyStores.put(cfName, ColumnFamilyStore.createColumnFamilyStore(name, cfName));
+ }
+
+ /** basically a combined drop and add */
+ public void renameCf(String oldName, String newName) throws IOException
+ {
+ dropCf(oldName);
+ addCf(newName);
}
/**
@@ -360,12 +345,19 @@ public class Table
{
Memtable memtableToFlush;
ColumnFamilyStore cfs = columnFamilyStores.get(columnFamily.name());
- if ((memtableToFlush=cfs.apply(mutation.key(), columnFamily)) != null)
- memtablesToFlush.put(cfs, memtableToFlush);
-
- ColumnFamily cachedRow = cfs.getRawCachedRow(mutation.key());
- if (cachedRow != null)
- cachedRow.addAll(columnFamily);
+ if (cfs == null)
+ {
+ logger.error("Attempting to mutate non-existant column family " + columnFamily.name());
+ }
+ else
+ {
+ if ((memtableToFlush=cfs.apply(mutation.key(), columnFamily)) != null)
+ memtablesToFlush.put(cfs, memtableToFlush);
+
+ ColumnFamily cachedRow = cfs.getRawCachedRow(mutation.key());
+ if (cachedRow != null)
+ cachedRow.addAll(columnFamily);
+ }
}
}
finally
Modified: incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/DefsTest.java
URL: http://svn.apache.org/viewvc/incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/DefsTest.java?rev=921963&r1=921962&r2=921963&view=diff
==============================================================================
--- incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/DefsTest.java (original)
+++ incubator/cassandra/trunk/test/unit/org/apache/cassandra/db/DefsTest.java Thu Mar 11 17:53:34 2010
@@ -66,9 +66,22 @@ public class DefsTest extends CleanupHel
}
}
+ @Test
+ public void addNewCfToBogusTable() throws InterruptedException
+ {
+ CFMetaData newCf = new CFMetaData("MadeUpKeyspace", "NewCF", "Standard", new UTF8Type(), null, "new cf", 0, 0);
+ try
+ {
+ DefsTable.add(newCf).get();
+ throw new AssertionError("You should't be able to do anything to a keyspace that doesn't exist.");
+ }
+ catch (ExecutionException expected)
+ {
+ }
+ }
@Test
- public void addNewCF() throws IOException, ConfigurationException, ExecutionException, InterruptedException
+ public void addNewCF() throws IOException, ExecutionException, InterruptedException
{
final String ks = "Keyspace1";
final String cf = "BrandNewCf";
@@ -76,7 +89,8 @@ public class DefsTest extends CleanupHel
CFMetaData newCf = new CFMetaData(original.name, cf, "Standard", new UTF8Type(), null, "A New Column Family", 0, 0);
int clSegments = CommitLog.instance().getSegmentCount();
- DefsTable.add(newCf);
+ assert !DatabaseDescriptor.getTableDefinition(ks).cfMetaData().containsKey(newCf.cfName);
+ DefsTable.add(newCf).get();
assert CommitLog.instance().getSegmentCount() == clSegments + 1;
assert DatabaseDescriptor.getTableDefinition(ks).cfMetaData().containsKey(newCf.cfName);
@@ -97,7 +111,7 @@ public class DefsTest extends CleanupHel
}
@Test
- public void removeCf() throws IOException, ConfigurationException, ExecutionException, InterruptedException
+ public void removeCf() throws IOException, ExecutionException, InterruptedException
{
// sanity
final KSMetaData ks = DatabaseDescriptor.getTableDefinition("Keyspace1");
@@ -116,7 +130,7 @@ public class DefsTest extends CleanupHel
store.getFlushPath();
assert DefsTable.getFiles(cfm.tableName, cfm.cfName).size() > 0;
- DefsTable.drop(cfm, true);
+ DefsTable.drop(cfm, true).get();
assert !DatabaseDescriptor.getTableDefinition(ks.name).cfMetaData().containsKey(cfm.cfName);
@@ -138,7 +152,7 @@ public class DefsTest extends CleanupHel
}
@Test
- public void renameCf() throws IOException, ConfigurationException, ExecutionException, InterruptedException
+ public void renameCf() throws IOException, ExecutionException, InterruptedException
{
final KSMetaData ks = DatabaseDescriptor.getTableDefinition("Keyspace2");
assert ks != null;
@@ -157,7 +171,7 @@ public class DefsTest extends CleanupHel
assert fileCount > 0;
final String newCfmName = "St4ndard1Replacement";
- DefsTable.rename(oldCfm, newCfmName);
+ DefsTable.rename(oldCfm, newCfmName).get();
assert !DatabaseDescriptor.getTableDefinition(ks.name).cfMetaData().containsKey(oldCfm.cfName);
assert DatabaseDescriptor.getTableDefinition(ks.name).cfMetaData().containsKey(newCfmName);