cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jbel...@apache.org
Subject svn commit: r1087919 - in /cassandra/trunk/src/java/org/apache/cassandra/db: ColumnFamilyStore.java Memtable.java Table.java
Date Fri, 01 Apr 2011 22:22:31 GMT
Author: jbellis
Date: Fri Apr  1 22:22:30 2011
New Revision: 1087919

URL: http://svn.apache.org/viewvc?rev=1087919&view=rev
Log:
revert #1954 for now

Modified:
    cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
    cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java
    cassandra/trunk/src/java/org/apache/cassandra/db/Table.java

Modified: cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=1087919&r1=1087918&r2=1087919&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Fri Apr  1 22:22:30
2011
@@ -40,7 +40,6 @@ import org.slf4j.LoggerFactory;
 import org.apache.cassandra.cache.AutoSavingCache;
 import org.apache.cassandra.cache.AutoSavingKeyCache;
 import org.apache.cassandra.cache.AutoSavingRowCache;
-import org.apache.cassandra.cache.JMXInstrumentedCache;
 import org.apache.cassandra.concurrent.JMXEnabledThreadPoolExecutor;
 import org.apache.cassandra.concurrent.NamedThreadFactory;
 import org.apache.cassandra.concurrent.StageManager;
@@ -588,18 +587,33 @@ public class ColumnFamilyStore implement
     /** flush the given memtable and swap in a new one for its CFS, if it hasn't been frozen
already.  threadsafe. */
     Future<?> maybeSwitchMemtable(Memtable oldMemtable, final boolean writeCommitLog)
     {
-        // Only one thread will succeed in marking it as pending flush; the others can go
back to processing writes
-        if (!oldMemtable.markPendingFlush())
+        if (oldMemtable.isFrozen())
         {
-            logger.debug("memtable is already pending flush; another thread must be flushing
it");
+            logger.debug("memtable is already frozen; another thread must be flushing it");
             return null;
         }
-        assert getMemtableThreadSafe() == oldMemtable;
 
-        // global synchronization ensures that we schedule discardCompletedSegments calls
in the same order as their
-        // contexts (commitlog position) were read, even though the flush executor is multithreaded.
-        synchronized (ColumnFamilyStore.class)
+        /*
+         * If we can get the writelock, that means no new updates can come in and
+         * all ongoing updates to memtables have completed. We can get the tail
+         * of the log and use it as the starting position for log replay on recovery.
+         *
+         * This is why we Table.flusherLock needs to be global instead of per-Table:
+         * we need to schedule discardCompletedSegments calls in the same order as their
+         * contexts (commitlog position) were read, even though the flush executor
+         * is multithreaded.
+         */
+        Table.switchLock.writeLock().lock();
+        try
         {
+            if (oldMemtable.isFrozen())
+            {
+                logger.debug("memtable is already frozen; another thread must be flushing
it");
+                return null;
+            }
+
+            assert getMemtableThreadSafe() == oldMemtable;
+            oldMemtable.freeze();
             final CommitLogSegment.CommitLogContext ctx = writeCommitLog ? CommitLog.instance.getContext()
: null;
 
             // submit the memtable for any indexed sub-cfses, and our own.
@@ -612,9 +626,8 @@ public class ColumnFamilyStore implement
             }
             final CountDownLatch latch = new CountDownLatch(icc.size());
             for (ColumnFamilyStore cfs : icc)
-            {
                 submitFlush(cfs.data.switchMemtable(), latch);
-            }
+
             // we marked our memtable as frozen as part of the concurrency control,
             // so even if there was nothing to flush we need to switch it out
             if (!icc.contains(this))
@@ -641,6 +654,10 @@ public class ColumnFamilyStore implement
                 }
             });
         }
+        finally
+        {
+            Table.switchLock.writeLock().unlock();
+        }
     }
 
     public boolean isDropped()
@@ -696,6 +713,8 @@ public class ColumnFamilyStore implement
 
     /**
      * Insert/Update the column family for this key.
+     * Caller is responsible for acquiring Table.flusherLock!
+     * param @ lock - lock that needs to be used.
      * param @ key - key for update/insert
      * param @ columnFamily - columnFamily changes
      */
@@ -947,11 +966,14 @@ public class ColumnFamilyStore implement
     }
 
     /**
-     * get the current memtable in a threadsafe fashion. 
-     * Returning memtable is ok because memtable is volatile, and thus
-     * introduce a happens-before ordering.
+     * get the current memtable in a threadsafe fashion.  note that simply "return memtable_"
is
+     * incorrect; you need to lock to introduce a thread safe happens-before ordering.
+     *
+     * do NOT use this method to do either a put or get on the memtable object, since it
could be
+     * flushed in the meantime (and its executor terminated).
      *
-     * do NOT make this method public or it will really get impossible to reason about these
things.
+     * also do NOT make this method public or it will really get impossible to reason about
these things.
+     * @return
      */
     private Memtable getMemtableThreadSafe()
     {
@@ -998,10 +1020,10 @@ public class ColumnFamilyStore implement
         return readStats.getTotalLatencyMicros();
     }
 
-    // TODO this actually isn't a good meature of pending tasks
+// TODO this actually isn't a good meature of pending tasks
     public int getPendingTasks()
     {
-        return 0;
+        return Table.switchLock.getQueueLength();
     }
 
     public long getWriteCount()

Modified: cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java?rev=1087919&r1=1087918&r2=1087919&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/Memtable.java Fri Apr  1 22:22:30 2011
@@ -52,8 +52,7 @@ public class Memtable implements Compara
 {
     private static final Logger logger = LoggerFactory.getLogger(Memtable.class);
 
-    private final AtomicBoolean isPendingFlush = new AtomicBoolean(false);
-    private final AtomicInteger activeWriters = new AtomicInteger(0);
+    private boolean isFrozen;
 
     private final AtomicLong currentThroughput = new AtomicLong(0);
     private final AtomicLong currentOperations = new AtomicLong(0);
@@ -106,30 +105,25 @@ public class Memtable implements Compara
         return currentThroughput.get() >= this.THRESHOLD || currentOperations.get() >=
this.THRESHOLD_COUNT;
     }
 
-    boolean isPendingFlush()
+    boolean isFrozen()
     {
-        return isPendingFlush.get();
+        return isFrozen;
     }
 
-    boolean markPendingFlush()
+    void freeze()
     {
-        return isPendingFlush.compareAndSet(false, true);
+        isFrozen = true;
     }
 
     /**
      * Should only be called by ColumnFamilyStore.apply.  NOT a public API.
+     * (CFS handles locking to avoid submitting an op
+     *  to a flushing memtable.  Any other way is unsafe.)
     */
     void put(DecoratedKey key, ColumnFamily columnFamily)
     {
-        try
-        {
-            activeWriters.incrementAndGet();
-            resolve(key, columnFamily);
-        }
-        finally
-        {
-            activeWriters.decrementAndGet();
-        }
+        assert !isFrozen; // not 100% foolproof but hell, it's an assert
+        resolve(key, columnFamily);
     }
 
     private void resolve(DecoratedKey key, ColumnFamily cf)
@@ -178,7 +172,6 @@ public class Memtable implements Compara
         {
             public void runMayThrow() throws IOException
             {
-                waitForWriters();
                 cfs.flushLock.lock();
                 try
                 {
@@ -197,25 +190,6 @@ public class Memtable implements Compara
         });
     }
 
-    /*
-     * Wait for all writers to be done with this memtable before flushing.
-     * A busy-wait is probably alright since we'll new wait long.
-     */
-    private void waitForWriters()
-    {
-        while (activeWriters.get() > 0)
-        {
-            try
-            {
-                Thread.sleep(3);
-            }
-            catch (InterruptedException e)
-            {
-                logger.error("Interrupted while waiting on writers.", e);
-            }
-        }
-    }
-
     public String toString()
     {
         return String.format("Memtable-%s@%s(%s bytes, %s operations)",

Modified: cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java?rev=1087919&r1=1087918&r2=1087919&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/Table.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/Table.java Fri Apr  1 22:22:30 2011
@@ -24,7 +24,7 @@ import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.*;
-import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 import com.google.common.base.Function;
 import com.google.common.collect.Iterables;
@@ -57,6 +57,14 @@ public class Table
     private static final Logger logger = LoggerFactory.getLogger(Table.class);
     private static final String SNAPSHOT_SUBDIR_NAME = "snapshots";
 
+    /**
+     * accesses to CFS.memtable should acquire this for thread safety.
+     * Table.maybeSwitchMemtable should aquire the writeLock; see that method for the full
explanation.
+     *
+     * (Enabling fairness in the RRWL is observed to decrease throughput, so we leave it
off.)
+     */
+    static final ReentrantReadWriteLock switchLock = new ReentrantReadWriteLock();
+
     // It is possible to call Table.open without a running daemon, so it makes sense to ensure
     // proper directories here as well as in CassandraDaemon.
     static 
@@ -371,64 +379,72 @@ public class Table
             logger.debug("applying mutation of row {}", ByteBufferUtil.bytesToHex(mutation.key()));
 
         // write the mutation to the commitlog and memtables
-        if (writeCommitLog)
-            CommitLog.instance.add(mutation);
-
-        DecoratedKey<?> key = StorageService.getPartitioner().decorateKey(mutation.key());
-        for (ColumnFamily cf : mutation.getColumnFamilies())
+        switchLock.readLock().lock();
+        try
         {
-            ColumnFamilyStore cfs = columnFamilyStores.get(cf.id());
-            if (cfs == null)
+            if (writeCommitLog)
+                CommitLog.instance.add(mutation);
+        
+            DecoratedKey<?> key = StorageService.getPartitioner().decorateKey(mutation.key());
+            for (ColumnFamily cf : mutation.getColumnFamilies())
             {
-                logger.error("Attempting to mutate non-existant column family " + cf.id());
-                continue;
-            }
+                ColumnFamilyStore cfs = columnFamilyStores.get(cf.id());
+                if (cfs == null)
+                {
+                    logger.error("Attempting to mutate non-existant column family " + cf.id());
+                    continue;
+                }
 
-            SortedSet<ByteBuffer> mutatedIndexedColumns = null;
-            for (ByteBuffer column : cfs.getIndexedColumns())
-            {
-                if (cf.getColumnNames().contains(column) || cf.isMarkedForDelete())
+                SortedSet<ByteBuffer> mutatedIndexedColumns = null;
+                for (ByteBuffer column : cfs.getIndexedColumns())
                 {
-                    if (mutatedIndexedColumns == null)
-                        mutatedIndexedColumns = new TreeSet<ByteBuffer>();
-                    mutatedIndexedColumns.add(column);
-                    if (logger.isDebugEnabled())
+                    if (cf.getColumnNames().contains(column) || cf.isMarkedForDelete())
                     {
-                        // can't actually use validator to print value here, because we overload
value
-                        // for deletion timestamp as well (which may not be a well-formed
value for the column type)
-                        ByteBuffer value = cf.getColumn(column) == null ? null : cf.getColumn(column).value();
// may be null on row-level deletion
-                        logger.debug(String.format("mutating indexed column %s value %s",
-                                    cf.getComparator().getString(column),
-                                    value == null ? "null" : ByteBufferUtil.bytesToHex(value)));
+                        if (mutatedIndexedColumns == null)
+                            mutatedIndexedColumns = new TreeSet<ByteBuffer>();
+                        mutatedIndexedColumns.add(column);
+                        if (logger.isDebugEnabled())
+                        {
+                            // can't actually use validator to print value here, because
we overload value
+                            // for deletion timestamp as well (which may not be a well-formed
value for the column type)
+                            ByteBuffer value = cf.getColumn(column) == null ? null : cf.getColumn(column).value();
// may be null on row-level deletion
+                            logger.debug(String.format("mutating indexed column %s value
%s",
+                                                       cf.getComparator().getString(column),
+                                                       value == null ? "null" : ByteBufferUtil.bytesToHex(value)));
+                        }
                     }
                 }
-            }
 
-            synchronized (indexLockFor(mutation.key()))
-            {
-                ColumnFamily oldIndexedColumns = null;
-                if (mutatedIndexedColumns != null)
+                synchronized (indexLockFor(mutation.key()))
                 {
-                    // with the raw data CF, we can just apply every update in any order
and let
-                    // read-time resolution throw out obsolete versions, thus avoiding read-before-write.
-                    // but for indexed data we need to make sure that we're not creating
index entries
-                    // for obsolete writes.
-                    oldIndexedColumns = readCurrentIndexedColumns(key, cfs, mutatedIndexedColumns);
-                    logger.debug("Pre-mutation index row is {}", oldIndexedColumns);
-                    ignoreObsoleteMutations(cf, mutatedIndexedColumns, oldIndexedColumns);
-                }
+                    ColumnFamily oldIndexedColumns = null;
+                    if (mutatedIndexedColumns != null)
+                    {
+                        // with the raw data CF, we can just apply every update in any order
and let
+                        // read-time resolution throw out obsolete versions, thus avoiding
read-before-write.
+                        // but for indexed data we need to make sure that we're not creating
index entries
+                        // for obsolete writes.
+                        oldIndexedColumns = readCurrentIndexedColumns(key, cfs, mutatedIndexedColumns);
+                        logger.debug("Pre-mutation index row is {}", oldIndexedColumns);
+                        ignoreObsoleteMutations(cf, mutatedIndexedColumns, oldIndexedColumns);
+                    }
 
-                Memtable fullMemtable = cfs.apply(key, cf);
-                if (fullMemtable != null)
-                    memtablesToFlush = addFullMemtable(memtablesToFlush, fullMemtable);
+                    Memtable fullMemtable = cfs.apply(key, cf);
+                    if (fullMemtable != null)
+                        memtablesToFlush = addFullMemtable(memtablesToFlush, fullMemtable);
 
-                if (mutatedIndexedColumns != null)
-                {
-                    // ignore full index memtables -- we flush those when the "master" one
is full
-                    applyIndexUpdates(mutation.key(), cf, cfs, mutatedIndexedColumns, oldIndexedColumns);
+                    if (mutatedIndexedColumns != null)
+                    {
+                        // ignore full index memtables -- we flush those when the "master"
one is full
+                        applyIndexUpdates(mutation.key(), cf, cfs, mutatedIndexedColumns,
oldIndexedColumns);
+                    }
                 }
             }
         }
+        finally
+        {
+            switchLock.readLock().unlock();
+        }
 
         // flush memtables that got filled up outside the readlock (maybeSwitchMemtable acquires
writeLock).
         // usually mTF will be empty and this will be a no-op.
@@ -582,12 +598,19 @@ public class Table
                 DecoratedKey<?> key = iter.next();
                 logger.debug("Indexing row {} ", key);
                 List<Memtable> memtablesToFlush = Collections.emptyList();
-
-                synchronized (indexLockFor(key.key))
+                switchLock.readLock().lock();
+                try
+                {
+                    synchronized (indexLockFor(key.key))
+                    {
+                        ColumnFamily cf = readCurrentIndexedColumns(key, cfs, columns);
+                        if (cf != null)
+                            memtablesToFlush = applyIndexUpdates(key.key, cf, cfs, cf.getColumnNames(),
null);
+                    }
+                }
+                finally
                 {
-                    ColumnFamily cf = readCurrentIndexedColumns(key, cfs, columns);
-                    if (cf != null)
-                        memtablesToFlush = applyIndexUpdates(key.key, cf, cfs, cf.getColumnNames(),
null);
+                    switchLock.readLock().unlock();
                 }
 
                 // during index build, we do flush index memtables separately from master;
otherwise we could OOM



Mime
View raw message