diff --git a/CHANGES.txt b/CHANGES.txt index 81da11db10..51f896110c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -175,6 +175,7 @@ * Update jackson JSON jars (CASSANDRA-13949) * Avoid locks when checking LCS fanout and if we should defrag (CASSANDRA-13930) Merged from 3.0: + * More frequent commitlog chained markers (CASSANDRA-13987) * Fix serialized size of DataLimits (CASSANDRA-14057) * Add flag to allow dropping oversized read repair mutations (CASSANDRA-13975) * Fix SSTableLoader logger message (CASSANDRA-14003) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index ba478e7fc4..02cec12198 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -375,10 +375,18 @@ counter_cache_save_period: 7200 # # the default option is "periodic" where writes may be acked immediately # and the CommitLog is simply synced every commitlog_sync_period_in_ms -# milliseconds. +# milliseconds. commitlog_sync: periodic commitlog_sync_period_in_ms: 10000 +# Time interval in millis at which we should update the chained markers in the commitlog. +# This allows more of the commitlog to be replayed from the mmapped file +# if the cassandra process crashes; this does not help in durability for surviving a host fail. +# This value only makes sense if it is significantly less that commitlog_sync_period_in_ms, +# and only applies to periodic mode when not using commitlog compression or encryption. +# commitlog_marker_period_in_ms: 100 + + # The size of the individual commitlog file segments. A commitlog # segment may be archived, deleted, or recycled once all the data # in it (potentially from each columnfamily in the system) has been diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java index 1db8217a27..4fa3bedb27 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -192,6 +192,7 @@ public class Config public double commitlog_sync_batch_window_in_ms = Double.NaN; public double commitlog_sync_group_window_in_ms = Double.NaN; public int commitlog_sync_period_in_ms; + public int commitlog_marker_period_in_ms = 0; public int commitlog_segment_size_in_mb = 32; public ParameterizedClass commitlog_compression; public int commitlog_max_compression_buffers_in_pool = 3; diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 843bca2e5b..acfd60fdb0 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1812,6 +1812,16 @@ public class DatabaseDescriptor conf.commitlog_sync_period_in_ms = periodMillis; } + public static void setCommitLogMarkerPeriod(int markerPeriod) + { + conf.commitlog_marker_period_in_ms = markerPeriod; + } + + public static int getCommitLogMarkerPeriod() + { + return conf.commitlog_marker_period_in_ms; + } + public static Config.CommitLogSync getCommitLogSync() { return conf.commitlog_sync; diff --git a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java index 42fffc4fb9..e93d72aaee 100755 --- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java +++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogSegmentManager.java @@ -512,9 +512,11 @@ public abstract class AbstractCommitLogSegmentManager } /** - * Forces a disk flush on the commit log files that need it. Blocking. + * Requests commit log files sync themselves, if needed. This may or may not involve flushing to disk. + * + * @param flush Request that the sync operation flush the file to disk. */ - public void sync() throws IOException + public void sync(boolean flush) throws IOException { CommitLogSegment current = allocatingFrom; for (CommitLogSegment segment : getActiveSegments()) @@ -522,7 +524,7 @@ public abstract class AbstractCommitLogSegmentManager // Do not sync segments that became active after sync started. if (segment.id > current.id) return; - segment.sync(); + segment.sync(flush); } } diff --git a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java index 71100a3e42..041065066e 100644 --- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java @@ -27,6 +27,8 @@ import org.slf4j.LoggerFactory; import com.codahale.metrics.Timer.Context; import org.apache.cassandra.concurrent.NamedThreadFactory; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.commitlog.CommitLogSegment.Allocation; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.concurrent.WaitQueue; @@ -48,7 +50,24 @@ public abstract class AbstractCommitLogService final CommitLog commitLog; private final String name; - private final long pollIntervalNanos; + + /** + * The duration between syncs to disk. + */ + private final long syncIntervalNanos; + + /** + * The duration between updating the chained markers in the the commit log file. This value should be + * 0 < {@link #markerIntervalNanos} <= {@link #syncIntervalNanos}. + */ + private final long markerIntervalNanos; + + /** + * A flag that callers outside of the sync thread can use to signal they want the commitlog segments + * to be flushed to disk. Note: this flag is primarily to support commit log's batch mode, which requires + * an immediate flush to disk on every mutation; see {@link BatchCommitLogService#maybeWaitForSync(Allocation)}. + */ + private volatile boolean syncRequested; private static final Logger logger = LoggerFactory.getLogger(AbstractCommitLogService.class); @@ -58,19 +77,45 @@ public abstract class AbstractCommitLogService * * Subclasses may be notified when a sync finishes by using the syncComplete WaitQueue. */ - AbstractCommitLogService(final CommitLog commitLog, final String name, final long pollIntervalMillis) + AbstractCommitLogService(final CommitLog commitLog, final String name, final long syncIntervalMillis) + { + this(commitLog, name, syncIntervalMillis, syncIntervalMillis); + } + + /** + * CommitLogService provides a fsync service for Allocations, fulfilling either the + * Batch or Periodic contract. + * + * Subclasses may be notified when a sync finishes by using the syncComplete WaitQueue. + */ + AbstractCommitLogService(final CommitLog commitLog, final String name, final long syncIntervalMillis, long markerIntervalMillis) { this.commitLog = commitLog; this.name = name; - this.pollIntervalNanos = TimeUnit.NANOSECONDS.convert(pollIntervalMillis, TimeUnit.MILLISECONDS); + this.syncIntervalNanos = TimeUnit.NANOSECONDS.convert(syncIntervalMillis, TimeUnit.MILLISECONDS); + + // if we are not using periodic mode, or we using compression/encryption, we shouldn't update the chained markers + // faster than the sync interval + if (DatabaseDescriptor.getCommitLogSync() != Config.CommitLogSync.periodic || commitLog.configuration.useCompression() || commitLog.configuration.useEncryption()) + markerIntervalMillis = syncIntervalMillis; + + // apply basic bounds checking on the marker interval + if (markerIntervalMillis <= 0 || markerIntervalMillis > syncIntervalMillis) + { + logger.debug("commit log marker interval {} is less than zero or above the sync interval {}; setting value to sync interval", + markerIntervalMillis, syncIntervalMillis); + markerIntervalMillis = syncIntervalMillis; + } + + this.markerIntervalNanos = TimeUnit.NANOSECONDS.convert(markerIntervalMillis, TimeUnit.MILLISECONDS); } // Separated into individual method to ensure relevant objects are constructed before this is started. void start() { - if (pollIntervalNanos < 1) + if (syncIntervalNanos < 1) throw new IllegalArgumentException(String.format("Commit log flush interval must be positive: %fms", - pollIntervalNanos * 1e-6)); + syncIntervalNanos * 1e-6)); Runnable runnable = new Runnable() { @@ -90,16 +135,24 @@ public abstract class AbstractCommitLogService try { // sync and signal - long syncStarted = System.nanoTime(); - // This is a target for Byteman in CommitLogSegmentManagerTest - commitLog.sync(); - lastSyncedAt = syncStarted; - syncComplete.signalAll(); - + long pollStarted = System.nanoTime(); + if (lastSyncedAt + syncIntervalNanos <= pollStarted || shutdownRequested || syncRequested) + { + // in this branch, we want to flush the commit log to disk + commitLog.sync(true); + syncRequested = false; + lastSyncedAt = pollStarted; + syncComplete.signalAll(); + } + else + { + // in this branch, just update the commit log sync headers + commitLog.sync(false); + } // sleep any time we have left before the next one is due long now = System.nanoTime(); - long wakeUpAt = syncStarted + pollIntervalNanos; + long wakeUpAt = pollStarted + markerIntervalNanos; if (wakeUpAt < now) { // if we have lagged noticeably, update our lag counter @@ -112,7 +165,7 @@ public abstract class AbstractCommitLogService lagCount++; } syncCount++; - totalSyncDuration += now - syncStarted; + totalSyncDuration += now - pollStarted; if (firstLagAt > 0) { @@ -143,7 +196,7 @@ public abstract class AbstractCommitLogService break; // sleep for full poll-interval after an error, so we don't spam the log file - LockSupport.parkNanos(pollIntervalNanos); + LockSupport.parkNanos(markerIntervalNanos); } } } @@ -168,8 +221,9 @@ public abstract class AbstractCommitLogService /** * Request an additional sync cycle without blocking. */ - public void requestExtraSync() + void requestExtraSync() { + syncRequested = true; LockSupport.unpark(thread); } diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java index 68075ac5ee..6537adcd3e 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLog.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLog.java @@ -233,9 +233,9 @@ public class CommitLog implements CommitLogMBean /** * Forces a disk flush on the commit log files that need it. Blocking. */ - public void sync() throws IOException + public void sync(boolean flush) throws IOException { - segmentManager.sync(); + segmentManager.sync(flush); } /** diff --git a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java index 3a57139b56..016dbc117c 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/CommitLogSegment.java @@ -97,6 +97,12 @@ public abstract class CommitLogSegment @VisibleForTesting volatile int lastSyncedOffset; + /** + * Everything before this offset has it's markers written into the {@link #buffer}, but has not necessarily + * been flushed to disk. This value should be greater than or equal to {@link #lastSyncedOffset}. + */ + private volatile int lastMarkerOffset; + // The end position of the buffer. Initially set to its capacity and updated to point to the last written position // as the segment is being closed. // No need to be volatile as writes are protected by appendOrder barrier. @@ -184,7 +190,8 @@ public abstract class CommitLogSegment { CommitLogDescriptor.writeHeader(buffer, descriptor, additionalHeaderParameters()); endOfBuffer = buffer.capacity(); - lastSyncedOffset = buffer.position(); + + lastSyncedOffset = lastMarkerOffset = buffer.position(); allocatePosition.set(lastSyncedOffset + SYNC_MARKER_SIZE); headerWritten = true; } @@ -261,7 +268,7 @@ public abstract class CommitLogSegment // ensures no more of this segment is writeable, by allocating any unused section at the end and marking it discarded void discardUnusedTail() { - // We guard this with the OpOrdering instead of synchronised due to potential dead-lock with CLSM.advanceAllocatingFrom() + // We guard this with the OpOrdering instead of synchronised due to potential dead-lock with ACLSM.advanceAllocatingFrom() // Ensures endOfBuffer update is reflected in the buffer end position picked up by sync(). // This actually isn't strictly necessary, as currently all calls to discardUnusedTail are executed either by the thread // running sync or within a mutation already protected by this OpOrdering, but to prevent future potential mistakes, @@ -300,15 +307,20 @@ public abstract class CommitLogSegment } /** - * Forces a disk flush for this segment file. + * Update the chained markers in the commit log buffer and possibly force a disk flush for this segment file. + * + * @param flush true if the segment should flush to disk; else, false for just updating the chained markers. */ - synchronized void sync() + synchronized void sync(boolean flush) { if (!headerWritten) throw new IllegalStateException("commit log header has not been written"); - boolean close = false; + assert lastMarkerOffset >= lastSyncedOffset : String.format("commit log segment positions are incorrect: last marked = %d, last synced = %d", + lastMarkerOffset, lastSyncedOffset); // check we have more work to do - if (allocatePosition.get() <= lastSyncedOffset + SYNC_MARKER_SIZE) + final boolean needToMarkData = allocatePosition.get() > lastMarkerOffset + SYNC_MARKER_SIZE; + final boolean hasDataToFlush = lastSyncedOffset != lastMarkerOffset; + if (!(needToMarkData || hasDataToFlush)) return; // Note: Even if the very first allocation of this sync section failed, we still want to enter this // to ensure the segment is closed. As allocatePosition is set to 1 beyond the capacity of the buffer, @@ -316,34 +328,50 @@ public abstract class CommitLogSegment // succeeded in the previous sync. assert buffer != null; // Only close once. - int startMarker = lastSyncedOffset; - // Allocate a new sync marker; this is both necessary in itself, but also serves to demarcate - // the point at which we can safely consider records to have been completely written to. - int nextMarker = allocate(SYNC_MARKER_SIZE); - if (nextMarker < 0) + boolean close = false; + int startMarker = lastMarkerOffset; + int nextMarker, sectionEnd; + if (needToMarkData) { - // Ensure no more of this CLS is writeable, and mark ourselves for closing. - discardUnusedTail(); - close = true; + // Allocate a new sync marker; this is both necessary in itself, but also serves to demarcate + // the point at which we can safely consider records to have been completely written to. + nextMarker = allocate(SYNC_MARKER_SIZE); + if (nextMarker < 0) + { + // Ensure no more of this CLS is writeable, and mark ourselves for closing. + discardUnusedTail(); + close = true; - // We use the buffer size as the synced position after a close instead of the end of the actual data - // to make sure we only close the buffer once. - // The endOfBuffer position may be incorrect at this point (to be written by another stalled thread). - nextMarker = buffer.capacity(); + // We use the buffer size as the synced position after a close instead of the end of the actual data + // to make sure we only close the buffer once. + // The endOfBuffer position may be incorrect at this point (to be written by another stalled thread). + nextMarker = buffer.capacity(); + } + // Wait for mutations to complete as well as endOfBuffer to have been written. + waitForModifications(); + sectionEnd = close ? endOfBuffer : nextMarker; + + // Possibly perform compression or encryption and update the chained markers + write(startMarker, sectionEnd); + lastMarkerOffset = sectionEnd; + } + else + { + // note: we don't need to waitForModifications() as, once we get to this block, we are only doing the flush + // and any mutations have already been fully written into the segment (as we wait for it in the previous block). + nextMarker = lastMarkerOffset; + sectionEnd = nextMarker; } - // Wait for mutations to complete as well as endOfBuffer to have been written. - waitForModifications(); - int sectionEnd = close ? endOfBuffer : nextMarker; - // Possibly perform compression or encryption, writing to file and flush. - write(startMarker, sectionEnd); + if (flush || close) + { + flush(startMarker, sectionEnd); + if (cdcState == CDCState.CONTAINS) + writeCDCIndexFile(descriptor, sectionEnd, close); + lastSyncedOffset = lastMarkerOffset = nextMarker; + } - if (cdcState == CDCState.CONTAINS) - writeCDCIndexFile(descriptor, sectionEnd, close); - - // Signal the sync as complete. - lastSyncedOffset = nextMarker; if (close) internalClose(); syncComplete.signalAll(); @@ -394,6 +422,8 @@ public abstract class CommitLogSegment abstract void write(int lastSyncedOffset, int nextMarker); + abstract void flush(int startMarker, int nextMarker); + public boolean isStillAllocating() { return allocatePosition.get() < endOfBuffer; @@ -488,7 +518,7 @@ public abstract class CommitLogSegment synchronized void close() { discardUnusedTail(); - sync(); + sync(true); assert buffer == null; } diff --git a/src/java/org/apache/cassandra/db/commitlog/CompressedSegment.java b/src/java/org/apache/cassandra/db/commitlog/CompressedSegment.java index 967db157fa..d5e61137d8 100644 --- a/src/java/org/apache/cassandra/db/commitlog/CompressedSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/CompressedSegment.java @@ -21,7 +21,6 @@ import java.nio.ByteBuffer; import org.apache.cassandra.io.FSWriteError; import org.apache.cassandra.io.compress.ICompressor; -import org.apache.cassandra.utils.SyncUtil; /** * Compressed commit log segment. Provides an in-memory buffer for the mutation threads. On sync compresses the written @@ -30,7 +29,7 @@ import org.apache.cassandra.utils.SyncUtil; * The format of the compressed commit log is as follows: * - standard commit log header (as written by {@link CommitLogDescriptor#writeHeader(ByteBuffer, CommitLogDescriptor)}) * - a series of 'sync segments' that are written every time the commit log is sync()'ed - * -- a sync section header, see {@link CommitLogSegment#writeSyncMarker(ByteBuffer, int, int, int)} + * -- a sync section header, see {@link CommitLogSegment#writeSyncMarker(long, ByteBuffer, int, int, int)} * -- total plain text length for this section * -- a block of compressed data */ @@ -82,7 +81,6 @@ public class CompressedSegment extends FileDirectSegment channel.write(compressedBuffer); assert channel.position() - lastWrittenPos == compressedBuffer.limit(); lastWrittenPos = channel.position(); - SyncUtil.force(channel, true); } catch (Exception e) { diff --git a/src/java/org/apache/cassandra/db/commitlog/EncryptedSegment.java b/src/java/org/apache/cassandra/db/commitlog/EncryptedSegment.java index 87825ab0e6..21b7c11fb0 100644 --- a/src/java/org/apache/cassandra/db/commitlog/EncryptedSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/EncryptedSegment.java @@ -32,7 +32,6 @@ import org.apache.cassandra.io.compress.ICompressor; import org.apache.cassandra.security.EncryptionUtils; import org.apache.cassandra.security.EncryptionContext; import org.apache.cassandra.utils.Hex; -import org.apache.cassandra.utils.SyncUtil; import static org.apache.cassandra.security.EncryptionUtils.ENCRYPTED_BLOCK_HEADER_SIZE; @@ -143,8 +142,6 @@ public class EncryptedSegment extends FileDirectSegment channel.position(syncMarkerPosition); channel.write(buffer); - - SyncUtil.force(channel, true); } catch (Exception e) { diff --git a/src/java/org/apache/cassandra/db/commitlog/FileDirectSegment.java b/src/java/org/apache/cassandra/db/commitlog/FileDirectSegment.java index 55084be092..d5431f875b 100644 --- a/src/java/org/apache/cassandra/db/commitlog/FileDirectSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/FileDirectSegment.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.nio.ByteBuffer; import org.apache.cassandra.io.FSWriteError; +import org.apache.cassandra.utils.SyncUtil; /** * Writes to the backing commit log file only on sync, allowing transformations of the mutations, @@ -63,4 +64,17 @@ public abstract class FileDirectSegment extends CommitLogSegment manager.notifyBufferFreed(); } } + + @Override + protected void flush(int startMarker, int nextMarker) + { + try + { + SyncUtil.force(channel, true); + } + catch (Exception e) + { + throw new FSWriteError(e, getPath()); + } + } } diff --git a/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java b/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java index bbf9ad23ab..e79278d9dd 100644 --- a/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java +++ b/src/java/org/apache/cassandra/db/commitlog/MemoryMappedSegment.java @@ -77,7 +77,11 @@ public class MemoryMappedSegment extends CommitLogSegment // write previous sync marker to point to next sync marker // we don't chain the crcs here to ensure this method is idempotent if it fails writeSyncMarker(id, buffer, startMarker, startMarker, nextMarker); + } + @Override + protected void flush(int startMarker, int nextMarker) + { try { SyncUtil.force((MappedByteBuffer) buffer); diff --git a/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java index bde832b38d..bf6eb4911f 100644 --- a/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java @@ -25,7 +25,7 @@ class PeriodicCommitLogService extends AbstractCommitLogService public PeriodicCommitLogService(final CommitLog commitLog) { - super(commitLog, "PERIODIC-COMMIT-LOG-SYNCER", DatabaseDescriptor.getCommitLogSyncPeriod()); + super(commitLog, "PERIODIC-COMMIT-LOG-SYNCER", DatabaseDescriptor.getCommitLogSyncPeriod(), DatabaseDescriptor.getCommitLogMarkerPeriod()); } protected void maybeWaitForSync(CommitLogSegment.Allocation alloc) diff --git a/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java b/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java index 3695da802e..18bc6e0a95 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CDCTestReplayer.java @@ -39,7 +39,7 @@ public class CDCTestReplayer extends CommitLogReplayer public CDCTestReplayer() throws IOException { super(CommitLog.instance, CommitLogPosition.NONE, null, ReplayFilter.create()); - CommitLog.instance.sync(); + CommitLog.instance.sync(true); commitLogReader = new CommitLogTestReader(); } diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java new file mode 100644 index 0000000000..be44ec3f10 --- /dev/null +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.cassandra.db.commitlog; + +import java.io.File; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Random; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.apache.cassandra.SchemaLoader; +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.db.ColumnFamilyStore; +import org.apache.cassandra.db.Keyspace; +import org.apache.cassandra.db.Mutation; +import org.apache.cassandra.db.RowUpdateBuilder; +import org.apache.cassandra.db.compaction.CompactionManager; +import org.apache.cassandra.db.marshal.AsciiType; +import org.apache.cassandra.db.marshal.BytesType; +import org.apache.cassandra.schema.KeyspaceParams; +import org.jboss.byteman.contrib.bmunit.BMRule; +import org.jboss.byteman.contrib.bmunit.BMUnitRunner; + +/** + * Tests the commitlog to make sure we can replay it - explicitly for the case where we update the chained markers + * in the commit log segment but do not flush the file to disk. + */ +@RunWith(BMUnitRunner.class) +public class CommitLogChainedMarkersTest +{ + private static final String KEYSPACE1 = "CommitLogTest"; + private static final String STANDARD1 = "CommitLogChainedMarkersTest"; + + @Test + @BMRule(name = "force all calls to sync() to not flush to disk", + targetClass = "CommitLogSegment", + targetMethod = "sync(boolean)", + action = "$flush = false") + public void replayCommitLogWithoutFlushing() throws IOException + { + // this method is blend of CommitLogSegmentBackpressureTest & CommitLogReaderTest methods + DatabaseDescriptor.daemonInitialization(); + DatabaseDescriptor.setCommitLogSegmentSize(5); + DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic); + DatabaseDescriptor.setCommitLogSyncPeriod(10000 * 1000); + DatabaseDescriptor.setCommitLogMarkerPeriod(1); + SchemaLoader.prepareServer(); + SchemaLoader.createKeyspace(KEYSPACE1, + KeyspaceParams.simple(1), + SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance, BytesType.instance)); + + CompactionManager.instance.disableAutoCompaction(); + + ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1); + + byte[] entropy = new byte[1024]; + new Random().nextBytes(entropy); + final Mutation m = new RowUpdateBuilder(cfs1.metadata.get(), 0, "k") + .clustering("bytes") + .add("val", ByteBuffer.wrap(entropy)) + .build(); + + int samples = 10000; + for (int i = 0; i < samples; i++) + CommitLog.instance.add(m); + + CommitLog.instance.sync(false); + + ArrayList toCheck = CommitLogReaderTest.getCommitLogs(); + CommitLogReader reader = new CommitLogReader(); + CommitLogReaderTest.TestCLRHandler testHandler = new CommitLogReaderTest.TestCLRHandler(cfs1.metadata.get()); + for (File f : toCheck) + reader.readCommitLogSegment(testHandler, f, CommitLogReader.ALL_MUTATIONS, false); + + Assert.assertEquals(samples, testHandler.seenMutationCount()); + } +} diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java index f4a63a9ae6..085f7cde4e 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java @@ -68,12 +68,12 @@ public class CommitLogSegmentBackpressureTest @BMRules(rules = {@BMRule(name = "Acquire Semaphore before sync", targetClass = "AbstractCommitLogService$1", targetMethod = "run", - targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync", + targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync(boolean)", action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.acquire()"), @BMRule(name = "Release Semaphore after sync", targetClass = "AbstractCommitLogService$1", targetMethod = "run", - targetLocation = "AFTER INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync", + targetLocation = "AFTER INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync(boolean)", action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.release()")}) public void testCompressedCommitLogBackpressure() throws Throwable { diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java index 80dfd015b2..7417cd3ca3 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentManagerCDCTest.java @@ -167,7 +167,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester .add("data", randomizeBuffer(DatabaseDescriptor.getCommitLogSegmentSize() / 3)) .build().apply(); - CommitLog.instance.sync(); + CommitLog.instance.sync(true); CommitLogSegment currentSegment = CommitLog.instance.segmentManager.allocatingFrom(); int syncOffset = currentSegment.lastSyncedOffset; @@ -233,7 +233,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester Assert.assertFalse("Expected index file to not be created but found: " + cdcIndexFile, cdcIndexFile.exists()); // Sync and confirm no index written as index is written on flush - CommitLog.instance.sync(); + CommitLog.instance.sync(true); Assert.assertTrue("File does not exist: " + linked, Files.exists(linked)); Assert.assertFalse("Expected index file to not be created but found: " + cdcIndexFile, cdcIndexFile.exists()); @@ -260,7 +260,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester Assert.assertTrue("File does not exist: " + linked, Files.exists(linked)); // Sync and confirm index written as index is written on flush - CommitLog.instance.sync(); + CommitLog.instance.sync(true); Assert.assertTrue("File does not exist: " + linked, Files.exists(linked)); Assert.assertTrue("Expected cdc index file after flush but found none: " + cdcIndexFile, cdcIndexFile.exists()); @@ -293,7 +293,7 @@ public class CommitLogSegmentManagerCDCTest extends CQLTester // pass } - CommitLog.instance.sync(); + CommitLog.instance.sync(true); CommitLog.instance.stopUnsafe(false); // Build up a list of expected index files after replay and then clear out cdc_raw diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java index 8d04eccd40..da895a026a 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTest.java @@ -341,7 +341,7 @@ public abstract class CommitLogTest // "Flush": this won't delete anything TableId id1 = rm.getTableIds().iterator().next(); - CommitLog.instance.sync(); + CommitLog.instance.sync(true); CommitLog.instance.discardCompletedSegments(id1, CommitLogPosition.NONE, CommitLog.instance.getCurrentPosition()); assertEquals(1, CommitLog.instance.segmentManager.getActiveSegments().size()); @@ -676,7 +676,7 @@ public abstract class CommitLogTest cellCount += 1; CommitLog.instance.add(rm2); - CommitLog.instance.sync(); + CommitLog.instance.sync(true); SimpleCountingReplayer replayer = new SimpleCountingReplayer(CommitLog.instance, CommitLogPosition.NONE, cfs.metadata()); List activeSegments = CommitLog.instance.getActiveSegmentNames(); @@ -713,7 +713,7 @@ public abstract class CommitLogTest } } - CommitLog.instance.sync(); + CommitLog.instance.sync(true); SimpleCountingReplayer replayer = new SimpleCountingReplayer(CommitLog.instance, commitLogPosition, cfs.metadata()); List activeSegments = CommitLog.instance.getActiveSegmentNames(); @@ -806,7 +806,7 @@ public abstract class CommitLogTest DatabaseDescriptor.setDiskFailurePolicy(oldPolicy); } - CommitLog.instance.sync(); + CommitLog.instance.sync(true); System.setProperty("cassandra.replayList", KEYSPACE1 + "." + STANDARD1); // Currently we don't attempt to re-flush a memtable that failed, thus make sure data is replayed by commitlog. // If retries work subsequent flushes should clear up error and this should change to expect 0. @@ -839,7 +839,7 @@ public abstract class CommitLogTest for (SSTableReader reader : cfs.getLiveSSTables()) reader.reloadSSTableMetadata(); - CommitLog.instance.sync(); + CommitLog.instance.sync(true); System.setProperty("cassandra.replayList", KEYSPACE1 + "." + STANDARD1); // In the absence of error, this should be 0 because forceBlockingFlush/forceRecycleAllSegments would have // persisted all data in the commit log. Because we know there was an error, there must be something left to diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java index 7f43378b0c..9a22b04d2e 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogTestReplayer.java @@ -40,7 +40,7 @@ public class CommitLogTestReplayer extends CommitLogReplayer public CommitLogTestReplayer(Predicate processor) throws IOException { super(CommitLog.instance, CommitLogPosition.NONE, null, ReplayFilter.create()); - CommitLog.instance.sync(); + CommitLog.instance.sync(true); this.processor = processor; commitLogReader = new CommitLogTestReader();