natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.11' into trunk

This commit is contained in:
Jason Brown 2017-12-13 19:52:58 -08:00
parent 7a40abb6a5 c000827afd
commit d569c831cb
9 changed files with 305 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

8
conf/cassandra.yaml
View File

@ -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

1
src/java/org/apache/cassandra/config/Config.java
View File

@ -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;

10
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -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;

235
src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
View File

@ -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
*/

3
src/java/org/apache/cassandra/db/commitlog/PeriodicCommitLogService.java
View File

@ -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)

166
test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java Normal file
View File

@ -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();
}
}
}

1
test/unit/org/apache/cassandra/db/commitlog/CommitLogChainedMarkersTest.java
View File

@ -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),

9
test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java
View File

@ -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