diff --git a/CHANGES.txt b/CHANGES.txt index 7949cbecb3..d89fca35d5 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -182,6 +182,7 @@ * Update jackson JSON jars (CASSANDRA-13949) * Avoid locks when checking LCS fanout and if we should defrag (CASSANDRA-13930) Merged from 3.0: + * Improve commit log chain marker updating (CASSANDRA-14108) * Extra range tombstone bound creates double rows (CASSANDRA-14008) * Fix SStable ordering by max timestamp in SinglePartitionReadCommand (CASSANDRA-14010) * Accept role names containing forward-slash (CASSANDRA-14088) diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml index 02cec12198..80e4515431 100644 --- a/conf/cassandra.yaml +++ b/conf/cassandra.yaml @@ -379,14 +379,6 @@ counter_cache_save_period: 7200 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 4fa3bedb27..1db8217a27 100644 --- a/src/java/org/apache/cassandra/config/Config.java +++ b/src/java/org/apache/cassandra/config/Config.java @@ -192,7 +192,6 @@ 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 807f0821f1..fa496d43e7 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -1812,16 +1812,6 @@ 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/AbstractCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java index 041065066e..4571b547f4 100644 --- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java @@ -28,13 +28,19 @@ 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.Clock; import org.apache.cassandra.utils.NoSpamLogger; import org.apache.cassandra.utils.concurrent.WaitQueue; public abstract class AbstractCommitLogService { + /** + * When in {@link Config.CommitLogSync#periodic} mode, the default number of milliseconds to wait between updating + * the commit log chained markers. + */ + static final long DEFAULT_MARKER_INTERVAL_MILLIS = 100; + private Thread thread; private volatile boolean shutdown = false; @@ -54,13 +60,13 @@ public abstract class AbstractCommitLogService /** * The duration between syncs to disk. */ - private final long syncIntervalNanos; + 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; + final long markerIntervalNanos; /** * A flag that callers outside of the sync thread can use to signal they want the commitlog segments @@ -77,9 +83,9 @@ 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 syncIntervalMillis) + AbstractCommitLogService(final CommitLog commitLog, final String name, long syncIntervalMillis) { - this(commitLog, name, syncIntervalMillis, syncIntervalMillis); + this (commitLog, name, syncIntervalMillis, false); } /** @@ -87,27 +93,36 @@ public abstract class AbstractCommitLogService * Batch or Periodic contract. * * Subclasses may be notified when a sync finishes by using the syncComplete WaitQueue. + * + * @param markHeadersFaster true if the chained markers should be updated more frequently than on the disk sync bounds. */ - AbstractCommitLogService(final CommitLog commitLog, final String name, final long syncIntervalMillis, long markerIntervalMillis) + AbstractCommitLogService(final CommitLog commitLog, final String name, long syncIntervalMillis, boolean markHeadersFaster) { this.commitLog = commitLog; this.name = name; - 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) + final long markerIntervalMillis; + if (markHeadersFaster && syncIntervalMillis > DEFAULT_MARKER_INTERVAL_MILLIS) + { + markerIntervalMillis = DEFAULT_MARKER_INTERVAL_MILLIS; + long modulo = syncIntervalMillis % markerIntervalMillis; + if (modulo != 0) + { + // quantize syncIntervalMillis to a multiple of markerIntervalMillis + syncIntervalMillis -= modulo; + + if (modulo >= markerIntervalMillis / 2) + syncIntervalMillis += markerIntervalMillis; + } + logger.debug("Will update the commitlog markers every {}ms and flush every {}ms", markerIntervalMillis, syncIntervalMillis); + } + else { - logger.debug("commit log marker interval {} is less than zero or above the sync interval {}; setting value to sync interval", - markerIntervalMillis, syncIntervalMillis); markerIntervalMillis = syncIntervalMillis; } - + assert syncIntervalMillis % markerIntervalMillis == 0; this.markerIntervalNanos = TimeUnit.NANOSECONDS.convert(markerIntervalMillis, TimeUnit.MILLISECONDS); + this.syncIntervalNanos = TimeUnit.NANOSECONDS.convert(syncIntervalMillis, TimeUnit.MILLISECONDS); } // Separated into individual method to ensure relevant objects are constructed before this is started. @@ -116,97 +131,111 @@ public abstract class AbstractCommitLogService if (syncIntervalNanos < 1) throw new IllegalArgumentException(String.format("Commit log flush interval must be positive: %fms", syncIntervalNanos * 1e-6)); - - Runnable runnable = new Runnable() - { - public void run() - { - long firstLagAt = 0; - long totalSyncDuration = 0; // total time spent syncing since firstLagAt - long syncExceededIntervalBy = 0; // time that syncs exceeded pollInterval since firstLagAt - int lagCount = 0; - int syncCount = 0; - - while (true) - { - // always run once after shutdown signalled - boolean shutdownRequested = shutdown; - - try - { - // sync and signal - 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 = pollStarted + markerIntervalNanos; - if (wakeUpAt < now) - { - // if we have lagged noticeably, update our lag counter - if (firstLagAt == 0) - { - firstLagAt = now; - totalSyncDuration = syncExceededIntervalBy = syncCount = lagCount = 0; - } - syncExceededIntervalBy += now - wakeUpAt; - lagCount++; - } - syncCount++; - totalSyncDuration += now - pollStarted; - - if (firstLagAt > 0) - { - //Only reset the lag tracking if it actually logged this time - boolean logged = NoSpamLogger.log(logger, - NoSpamLogger.Level.WARN, - 5, - TimeUnit.MINUTES, - "Out of {} commit log syncs over the past {}s with average duration of {}ms, {} have exceeded the configured commit interval by an average of {}ms", - syncCount, - String.format("%.2f", (now - firstLagAt) * 1e-9d), - String.format("%.2f", totalSyncDuration * 1e-6d / syncCount), - lagCount, - String.format("%.2f", syncExceededIntervalBy * 1e-6d / lagCount)); - if (logged) - firstLagAt = 0; - } - - if (shutdownRequested) - return; - - if (wakeUpAt > now) - LockSupport.parkNanos(wakeUpAt - now); - } - catch (Throwable t) - { - if (!CommitLog.handleCommitError("Failed to persist commits to disk", t)) - break; - - // sleep for full poll-interval after an error, so we don't spam the log file - LockSupport.parkNanos(markerIntervalNanos); - } - } - } - }; - shutdown = false; + Runnable runnable = new SyncRunnable(new Clock()); thread = NamedThreadFactory.createThread(runnable, name); thread.start(); } + class SyncRunnable implements Runnable + { + private final Clock clock; + private long firstLagAt = 0; + private long totalSyncDuration = 0; // total time spent syncing since firstLagAt + private long syncExceededIntervalBy = 0; // time that syncs exceeded pollInterval since firstLagAt + private int lagCount = 0; + private int syncCount = 0; + + SyncRunnable(Clock clock) + { + this.clock = clock; + } + + public void run() + { + while (true) + { + if (!sync()) + break; + } + } + + boolean sync() + { + // always run once after shutdown signalled + boolean shutdownRequested = shutdown; + + try + { + // sync and signal + long pollStarted = clock.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(); + syncCount++; + } + 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 = clock.nanoTime(); + long wakeUpAt = pollStarted + markerIntervalNanos; + if (wakeUpAt < now) + { + // if we have lagged noticeably, update our lag counter + if (firstLagAt == 0) + { + firstLagAt = now; + totalSyncDuration = syncExceededIntervalBy = syncCount = lagCount = 0; + } + syncExceededIntervalBy += now - wakeUpAt; + lagCount++; + } + totalSyncDuration += now - pollStarted; + + if (firstLagAt > 0) + { + //Only reset the lag tracking if it actually logged this time + boolean logged = NoSpamLogger.log(logger, + NoSpamLogger.Level.WARN, + 5, + TimeUnit.MINUTES, + "Out of {} commit log syncs over the past {}s with average duration of {}ms, {} have exceeded the configured commit interval by an average of {}ms", + syncCount, + String.format("%.2f", (now - firstLagAt) * 1e-9d), + String.format("%.2f", totalSyncDuration * 1e-6d / syncCount), + lagCount, + String.format("%.2f", syncExceededIntervalBy * 1e-6d / lagCount)); + if (logged) + firstLagAt = 0; + } + + if (shutdownRequested) + return false; + + if (wakeUpAt > now) + LockSupport.parkNanos(wakeUpAt - now); + } + catch (Throwable t) + { + if (!CommitLog.handleCommitError("Failed to persist commits to disk", t)) + return false; + + // sleep for full poll-interval after an error, so we don't spam the log file + LockSupport.parkNanos(markerIntervalNanos); + } + + return true; + } + } + /** * Block for @param alloc to be sync'd as necessary, and handle bookkeeping */ diff --git a/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java index bf6eb4911f..efd3394988 100644 --- a/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java +++ b/src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java @@ -25,7 +25,8 @@ class PeriodicCommitLogService extends AbstractCommitLogService public PeriodicCommitLogService(final CommitLog commitLog) { - super(commitLog, "PERIODIC-COMMIT-LOG-SYNCER", DatabaseDescriptor.getCommitLogSyncPeriod(), DatabaseDescriptor.getCommitLogMarkerPeriod()); + super(commitLog, "PERIODIC-COMMIT-LOG-SYNCER", DatabaseDescriptor.getCommitLogSyncPeriod(), + !(commitLog.configuration.useCompression() || commitLog.configuration.useEncryption())); } protected void maybeWaitForSync(CommitLogSegment.Allocation alloc) diff --git a/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java b/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java new file mode 100644 index 0000000000..18f15fac8d --- /dev/null +++ b/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java @@ -0,0 +1,166 @@ +/* + * 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.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.Assert; +import org.junit.BeforeClass; +import org.junit.Test; + +import org.apache.cassandra.config.Config; +import org.apache.cassandra.config.DatabaseDescriptor; +import org.apache.cassandra.utils.Clock; +import org.apache.cassandra.utils.FreeRunningClock; + +import static org.apache.cassandra.db.commitlog.AbstractCommitLogService.DEFAULT_MARKER_INTERVAL_MILLIS; + +public class AbstractCommitLogServiceTest +{ + @BeforeClass + public static void before() + { + DatabaseDescriptor.daemonInitialization(); + DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic); + DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000); + } + + @Test + public void testConstructorSyncIsQuantized() + { + long syncTimeMillis = 10 * 1000; + FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis); + Assert.assertEquals(toNanos(DEFAULT_MARKER_INTERVAL_MILLIS), commitLogService.markerIntervalNanos); + Assert.assertEquals(toNanos(syncTimeMillis), commitLogService.syncIntervalNanos); + } + + @Test + public void testConstructorSyncEqualsMarkerDefault() + { + long syncTimeMillis = 100; + FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis); + Assert.assertEquals(toNanos(DEFAULT_MARKER_INTERVAL_MILLIS), commitLogService.markerIntervalNanos); + Assert.assertEquals(toNanos(syncTimeMillis), commitLogService.syncIntervalNanos); + Assert.assertEquals(commitLogService.markerIntervalNanos, commitLogService.syncIntervalNanos); + } + + @Test + public void testConstructorSyncShouldRoundUp() + { + long syncTimeMillis = 151; + long expectedMillis = 200; + FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis); + Assert.assertEquals(toNanos(DEFAULT_MARKER_INTERVAL_MILLIS), commitLogService.markerIntervalNanos); + Assert.assertEquals(toNanos(expectedMillis), commitLogService.syncIntervalNanos); + } + + @Test + public void testConstructorSyncShouldRoundDown() + { + long syncTimeMillis = 121; + long expectedMillis = 100; + FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis); + Assert.assertEquals(toNanos(DEFAULT_MARKER_INTERVAL_MILLIS), commitLogService.markerIntervalNanos); + Assert.assertEquals(toNanos(expectedMillis), commitLogService.syncIntervalNanos); + } + + @Test + public void testConstructorSyncTinyValue() + { + long syncTimeMillis = 10; + long expectedNanos = toNanos(syncTimeMillis); + FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis); + Assert.assertEquals(expectedNanos, commitLogService.markerIntervalNanos); + Assert.assertEquals(expectedNanos, commitLogService.syncIntervalNanos); + } + + private static long toNanos(long millis) + { + return TimeUnit.MILLISECONDS.toNanos(millis); + } + + private static class FakeCommitLogService extends AbstractCommitLogService + { + FakeCommitLogService(long syncIntervalMillis) + { + super(new FakeCommitLog(), "This is not a real commit log", syncIntervalMillis, true); + lastSyncedAt = 0; + } + + protected void maybeWaitForSync(CommitLogSegment.Allocation alloc) + { + // nop + } + } + + @Test + public void testSync() + { + long syncTimeMillis = AbstractCommitLogService.DEFAULT_MARKER_INTERVAL_MILLIS * 2; + FreeRunningClock clock = new FreeRunningClock(); + FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis); + AbstractCommitLogService.SyncRunnable syncRunnable = commitLogService.new SyncRunnable(clock); + FakeCommitLog commitLog = (FakeCommitLog) commitLogService.commitLog; + + // at time 0 + Assert.assertTrue(syncRunnable.sync()); + Assert.assertEquals(1, commitLog.markCount.get()); + Assert.assertEquals(0, commitLog.syncCount.get()); + + // at time DEFAULT_MARKER_INTERVAL_MILLIS + clock.advance(DEFAULT_MARKER_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); + Assert.assertTrue(syncRunnable.sync()); + Assert.assertEquals(2, commitLog.markCount.get()); + Assert.assertEquals(0, commitLog.syncCount.get()); + + // at time DEFAULT_MARKER_INTERVAL_MILLIS * 2 + clock.advance(DEFAULT_MARKER_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); + Assert.assertTrue(syncRunnable.sync()); + Assert.assertEquals(2, commitLog.markCount.get()); + Assert.assertEquals(1, commitLog.syncCount.get()); + + // at time DEFAULT_MARKER_INTERVAL_MILLIS * 3, but with shutdown! + clock.advance(DEFAULT_MARKER_INTERVAL_MILLIS, TimeUnit.MILLISECONDS); + commitLogService.shutdown(); + Assert.assertFalse(syncRunnable.sync()); + Assert.assertEquals(2, commitLog.markCount.get()); + Assert.assertEquals(2, commitLog.syncCount.get()); + } + + private static class FakeCommitLog extends CommitLog + { + private final AtomicInteger markCount = new AtomicInteger(); + private final AtomicInteger syncCount = new AtomicInteger(); + + FakeCommitLog() + { + super(null); + } + + @Override + public void sync(boolean flush) + { + if (flush) + syncCount.incrementAndGet(); + else + markCount.incrementAndGet(); + } + } +} diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java index be44ec3f10..fb90d59462 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java @@ -64,7 +64,6 @@ public class CommitLogChainedMarkersTest DatabaseDescriptor.setCommitLogSegmentSize(5); DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic); DatabaseDescriptor.setCommitLogSyncPeriod(10000 * 1000); - DatabaseDescriptor.setCommitLogMarkerPeriod(1); SchemaLoader.prepareServer(); SchemaLoader.createKeyspace(KEYSPACE1, KeyspaceParams.simple(1), diff --git a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java index 085f7cde4e..6b167b2500 100644 --- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java +++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java @@ -44,7 +44,6 @@ 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.apache.cassandra.schema.TableId; import org.jboss.byteman.contrib.bmunit.BMRule; import org.jboss.byteman.contrib.bmunit.BMRules; import org.jboss.byteman.contrib.bmunit.BMUnitRunner; @@ -66,13 +65,13 @@ public class CommitLogSegmentBackpressureTest @Test @BMRules(rules = {@BMRule(name = "Acquire Semaphore before sync", - targetClass = "AbstractCommitLogService$1", - targetMethod = "run", + targetClass = "AbstractCommitLogService$SyncRunnable", + targetMethod = "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", + targetClass = "AbstractCommitLogService$SyncRunnable", + targetMethod = "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