cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jasobr...@apache.org
Subject [1/6] cassandra git commit: Fix regression of lagging commitlog flush log message
Date Tue, 05 Jun 2018 20:55:47 GMT
Repository: cassandra
Updated Branches:
  refs/heads/cassandra-3.0 38096da25 -> 214a3abfc
  refs/heads/cassandra-3.11 b92d90dc1 -> 77a12053b
  refs/heads/trunk 5d8767765 -> 843a5fdf2


Fix regression of lagging commitlog flush log message

patch by jasobrown, reviewed by Jordan West for CASSANDRA-14451


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/214a3abf
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/214a3abf
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/214a3abf

Branch: refs/heads/cassandra-3.0
Commit: 214a3abfcc25460af50805b543a5833697a1b341
Parents: 38096da
Author: Jason Brown <jasedbrown@gmail.com>
Authored: Fri Jun 1 05:45:23 2018 -0700
Committer: Jason Brown <jasedbrown@gmail.com>
Committed: Tue Jun 5 13:47:37 2018 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |  1 +
 .../db/commitlog/AbstractCommitLogService.java  | 85 +++++++++++++-------
 .../commitlog/AbstractCommitLogServiceTest.java | 49 ++++++++++-
 3 files changed, 104 insertions(+), 31 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/214a3abf/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 16fe6d1..dfdfbfd 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 3.0.17
+ * Fix regression of lagging commitlog flush log message (CASSANDRA-14451)
  * Add Missing dependencies in pom-all (CASSANDRA-14422)
  * Cleanup StartupClusterConnectivityChecker and PING Verb (CASSANDRA-14447)
  * Fix deprecated repair error notifications from 3.x clusters to legacy JMX clients (CASSANDRA-13121)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/214a3abf/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
index 1cee55d..0845bd5 100644
--- a/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
+++ b/src/java/org/apache/cassandra/db/commitlog/AbstractCommitLogService.java
@@ -29,6 +29,8 @@ import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import com.google.common.annotations.VisibleForTesting;
+
 public abstract class AbstractCommitLogService
 {
     /**
@@ -165,13 +167,15 @@ public abstract class AbstractCommitLogService
 
                 // sync and signal
                 long pollStarted = clock.currentTimeMillis();
-                if (lastSyncedAt + syncIntervalMillis <= pollStarted || shutdown || syncRequested)
+                boolean flushToDisk = lastSyncedAt + syncIntervalMillis <= pollStarted
|| shutdown || syncRequested;
+                if (flushToDisk)
                 {
                     // in this branch, we want to flush the commit log to disk
                     syncRequested = false;
                     commitLog.sync(shutdown, true);
                     lastSyncedAt = pollStarted;
                     syncComplete.signalAll();
+                    syncCount++;
                 }
                 else
                 {
@@ -179,41 +183,15 @@ public abstract class AbstractCommitLogService
                     commitLog.sync(false, false);
                 }
 
-                // sleep any time we have left before the next one is due
                 long now = clock.currentTimeMillis();
-                long sleep = pollStarted + markerIntervalMillis - now;
-                if (sleep < 0)
-                {
-                    // if we have lagged noticeably, update our lag counter
-                    if (firstLagAt == 0)
-                    {
-                        firstLagAt = now;
-                        totalSyncDuration = syncExceededIntervalBy = syncCount = lagCount
= 0;
-                    }
-                    syncExceededIntervalBy -= sleep;
-                    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, (now - firstLagAt) / 1000, String.format("%.2f", (double)
totalSyncDuration / syncCount), lagCount, String.format("%.2f", (double) syncExceededIntervalBy
/ lagCount));
-                    if (logged)
-                        firstLagAt = 0;
-                }
+                if (flushToDisk)
+                    maybeLogFlushLag(pollStarted, now);
 
                 if (!run)
                     return false;
 
                 // if we have lagged this round, we probably have work to do already so we
don't sleep
+                long sleep = pollStarted + markerIntervalMillis - now;
                 if (sleep < 0)
                     return true;
 
@@ -244,6 +222,53 @@ public abstract class AbstractCommitLogService
             }
             return true;
         }
+
+        /**
+         * Add a log entry whenever the time to flush the commit log to disk exceeds {@link
#syncIntervalMillis}.
+         */
+        @VisibleForTesting
+        boolean maybeLogFlushLag(long pollStarted, long now)
+        {
+            long flushDuration = now - pollStarted;
+            totalSyncDuration += flushDuration;
+
+            // this is the timestamp by which we should have completed the flush
+            long maxFlushTimestamp = pollStarted + syncIntervalMillis;
+            if (maxFlushTimestamp > now)
+                return false;
+
+            // if we have lagged noticeably, update our lag counter
+            if (firstLagAt == 0)
+            {
+                firstLagAt = now;
+                syncExceededIntervalBy = lagCount = 0;
+                syncCount = 1;
+                totalSyncDuration = flushDuration;
+            }
+            syncExceededIntervalBy += now - maxFlushTimestamp;
+            lagCount++;
+
+            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, (now - firstLagAt) / 1000, String.format("%.2f", (double) totalSyncDuration
/ syncCount), lagCount, String.format("%.2f", (double) syncExceededIntervalBy / lagCount));
+                if (logged)
+                    firstLagAt = 0;
+            }
+            return true;
+        }
+
+        @VisibleForTesting
+        long getTotalSyncDuration()
+        {
+            return totalSyncDuration;
+        }
     }
 
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/214a3abf/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java
b/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java
index 5a46e5f..6f51eaf 100644
--- a/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/AbstractCommitLogServiceTest.java
@@ -28,6 +28,7 @@ import org.junit.Test;
 
 import org.apache.cassandra.config.Config;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.commitlog.AbstractCommitLogService.SyncRunnable;
 import org.apache.cassandra.utils.Clock;
 import org.apache.cassandra.utils.FreeRunningClock;
 
@@ -117,7 +118,7 @@ public class AbstractCommitLogServiceTest
         long syncTimeMillis = AbstractCommitLogService.DEFAULT_MARKER_INTERVAL_MILLIS * 2;
         FreeRunningClock clock = new FreeRunningClock();
         FakeCommitLogService commitLogService = new FakeCommitLogService(syncTimeMillis);
-        AbstractCommitLogService.SyncRunnable syncRunnable = commitLogService.new SyncRunnable(clock);
+        SyncRunnable syncRunnable = commitLogService.new SyncRunnable(clock);
         FakeCommitLog commitLog = (FakeCommitLog) commitLogService.commitLog;
 
         // at time 0
@@ -173,4 +174,50 @@ public class AbstractCommitLogServiceTest
                 markCount.incrementAndGet();
         }
     }
+
+    @Test
+    public void maybeLogFlushLag_MustLog()
+    {
+        long syncTimeMillis = 10;
+        SyncRunnable syncRunnable = new FakeCommitLogService(syncTimeMillis).new SyncRunnable(new
FreeRunningClock());
+        long pollStarted = 1;
+        long now = pollStarted + (syncTimeMillis * 2);
+        Assert.assertTrue(syncRunnable.maybeLogFlushLag(pollStarted, now));
+        Assert.assertEquals(now - pollStarted, syncRunnable.getTotalSyncDuration());
+    }
+
+    @Test
+    public void maybeLogFlushLag_NoLog()
+    {
+        long syncTimeMillis = 10;
+        SyncRunnable syncRunnable = new FakeCommitLogService(syncTimeMillis).new SyncRunnable(new
FreeRunningClock());
+        long pollStarted = 1;
+        long now = pollStarted + (syncTimeMillis - 1);
+        Assert.assertFalse(syncRunnable.maybeLogFlushLag(pollStarted, now));
+        Assert.assertEquals(now - pollStarted, syncRunnable.getTotalSyncDuration());
+    }
+
+    /**
+     * Mostly tests that {@link SyncRunnable#totalSyncDuration} is handled correctly
+     */
+    @Test
+    public void maybeLogFlushLag_MultipleOperations()
+    {
+        long syncTimeMillis = 10;
+        SyncRunnable syncRunnable = new FakeCommitLogService(syncTimeMillis).new SyncRunnable(new
FreeRunningClock());
+
+        long pollStarted = 1;
+        long now = pollStarted + (syncTimeMillis - 1);
+
+        int runCount = 12;
+        for (int i = 1; i <= runCount; i++)
+        {
+            Assert.assertFalse(syncRunnable.maybeLogFlushLag(pollStarted, now));
+            Assert.assertEquals(i * (now - pollStarted), syncRunnable.getTotalSyncDuration());
+        }
+
+        now = pollStarted + (syncTimeMillis * 2);
+        Assert.assertTrue(syncRunnable.maybeLogFlushLag(pollStarted, now));
+        Assert.assertEquals(now - pollStarted, syncRunnable.getTotalSyncDuration());
+    }
 }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message