cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bdeggles...@apache.org
Subject [6/6] cassandra git commit: Merge branch 'cassandra-3.11' into trunk
Date Thu, 26 Oct 2017 20:43:35 GMT
Merge branch 'cassandra-3.11' into trunk


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

Branch: refs/heads/trunk
Commit: af2c78459ad69b3263620fbb5da484f1186133d1
Parents: 2b507c0 7d26815
Author: Blake Eggleston <bdeggleston@gmail.com>
Authored: Thu Oct 26 13:37:50 2017 -0700
Committer: Blake Eggleston <bdeggleston@gmail.com>
Committed: Thu Oct 26 13:38:18 2017 -0700

----------------------------------------------------------------------
 .../CommitLogSegmentBackpressureTest.java       | 146 ++++++++++++++++
 .../commitlog/CommitLogSegmentManagerTest.java  | 173 -------------------
 .../db/commitlog/CommitlogShutdownTest.java     |  99 +++++++++++
 3 files changed, 245 insertions(+), 173 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/af2c7845/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java
index 0000000,3956de5..f4a63a9
mode 000000,100644..100644
--- a/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/CommitLogSegmentBackpressureTest.java
@@@ -1,0 -1,145 +1,146 @@@
+ /*
+  *
+  * 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.nio.ByteBuffer;
+ import java.util.ArrayList;
+ import java.util.Random;
+ import java.util.concurrent.Semaphore;
+ 
+ import com.google.common.collect.ImmutableMap;
+ 
+ import org.junit.Assert;
+ import org.junit.Test;
+ import org.junit.runner.RunWith;
+ 
+ import org.apache.cassandra.SchemaLoader;
+ import org.apache.cassandra.Util;
+ import org.apache.cassandra.config.Config.CommitLogSync;
+ import org.apache.cassandra.config.DatabaseDescriptor;
+ import org.apache.cassandra.config.ParameterizedClass;
+ 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.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;
+ 
+ /**
+  * Since this test depends on byteman rules being setup during initialization, you shouldn't
add tests to this class
+  */
+ @RunWith(BMUnitRunner.class)
+ public class CommitLogSegmentBackpressureTest
+ {
+     //Block commit log service from syncing
+     private static final Semaphore allowSync = new Semaphore(1);
+ 
+     private static final String KEYSPACE1 = "CommitLogTest";
+     private static final String STANDARD1 = "Standard1";
+     private static final String STANDARD2 = "Standard2";
+ 
+     private final static byte[] entropy = new byte[1024 * 256];
+ 
+     @Test
+     @BMRules(rules = {@BMRule(name = "Acquire Semaphore before sync",
+                               targetClass = "AbstractCommitLogService$1",
+                               targetMethod = "run",
+                               targetLocation = "AT INVOKE org.apache.cassandra.db.commitlog.CommitLog.sync",
+                               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",
+                               action = "org.apache.cassandra.db.commitlog.CommitLogSegmentBackpressureTest.allowSync.release()")})
+     public void testCompressedCommitLogBackpressure() throws Throwable
+     {
+         // Perform all initialization before making CommitLog.Sync blocking
+         // Doing the initialization within the method guarantee that Byteman has performed
its injections when we start
+         new Random().nextBytes(entropy);
+         DatabaseDescriptor.daemonInitialization();
+         DatabaseDescriptor.setCommitLogCompression(new ParameterizedClass("LZ4Compressor",
ImmutableMap.of()));
+         DatabaseDescriptor.setCommitLogSegmentSize(1);
+         DatabaseDescriptor.setCommitLogSync(CommitLogSync.periodic);
+         DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000);
+         DatabaseDescriptor.setCommitLogMaxCompressionBuffersPerPool(3);
+         SchemaLoader.prepareServer();
+         SchemaLoader.createKeyspace(KEYSPACE1,
+                                     KeyspaceParams.simple(1),
+                                     SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance,
BytesType.instance),
+                                     SchemaLoader.standardCFMD(KEYSPACE1, STANDARD2, 0, AsciiType.instance,
BytesType.instance));
+ 
+         CompactionManager.instance.disableAutoCompaction();
+ 
+         ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
+ 
 -        final Mutation m = new RowUpdateBuilder(cfs1.metadata, 0, "k").clustering("bytes")
++        final Mutation m = new RowUpdateBuilder(cfs1.metadata(), 0, "k").clustering("bytes")
+                                                                       .add("val", ByteBuffer.wrap(entropy))
+                                                                       .build();
+ 
+         Thread dummyThread = new Thread(() -> {
+             for (int i = 0; i < 20; i++)
+                 CommitLog.instance.add(m);
+         });
+ 
+         try
+         {
+             // Makes sure any call to CommitLog.sync is blocking
+             allowSync.acquire();
+ 
+             dummyThread.start();
+ 
+             AbstractCommitLogSegmentManager clsm = CommitLog.instance.segmentManager;
+ 
+             Util.spinAssertEquals(3, () -> clsm.getActiveSegments().size(), 5);
+ 
+             Thread.sleep(1000);
+ 
+             // Should only be able to create 3 segments not 7 because it blocks waiting
for truncation that never comes
+             Assert.assertEquals(3, clsm.getActiveSegments().size());
+ 
+             // Discard the currently active segments so allocation can continue.
+             // Take snapshot of the list, otherwise this will also discard newly allocated
segments.
+             new ArrayList<>(clsm.getActiveSegments()).forEach( clsm::archiveAndDiscard
);
+ 
+             // The allocated count should reach the limit again.
+             Util.spinAssertEquals(3, () -> clsm.getActiveSegments().size(), 5);
+         }
+         finally
+         {
+             // Allow the CommitLog.sync to perform normally.
+             allowSync.release();
+         }
+         try
+         {
+             // Wait for the dummy thread to die
+             dummyThread.join();
+         }
+         catch (InterruptedException e)
+         {
+             Thread.currentThread().interrupt();
+         }
+     }
+ }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/af2c7845/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
index 0000000,91a3f02..711cf65
mode 000000,100644..100644
--- a/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
+++ b/test/unit/org/apache/cassandra/db/commitlog/CommitlogShutdownTest.java
@@@ -1,0 -1,99 +1,99 @@@
+ /*
+  * 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.nio.ByteBuffer;
+ import java.util.Random;
+ import java.util.UUID;
+ 
+ import com.google.common.collect.ImmutableMap;
+ 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.config.ParameterizedClass;
+ 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.apache.cassandra.schema.TableId;
+ import org.jboss.byteman.contrib.bmunit.BMRule;
+ import org.jboss.byteman.contrib.bmunit.BMUnitRunner;
+ 
+ /**
+  * Since this test depends on byteman rules being setup during initialization, you shouldn't
add tests to this class
+  */
+ @RunWith(BMUnitRunner.class)
+ public class CommitlogShutdownTest
+ {
+     private static final String KEYSPACE1 = "CommitLogTest";
+     private static final String STANDARD1 = "Standard1";
+ 
+     private final static byte[] entropy = new byte[1024 * 256];
+ 
+     @Test
+     @BMRule(name = "Make removing commitlog segments slow",
+     targetClass = "CommitLogSegment",
+     targetMethod = "discard",
+     action = "Thread.sleep(50)")
+     public void testShutdownWithPendingTasks() throws Exception
+     {
+         new Random().nextBytes(entropy);
+         DatabaseDescriptor.daemonInitialization();
+         DatabaseDescriptor.setCommitLogCompression(new ParameterizedClass("LZ4Compressor",
ImmutableMap.of()));
+         DatabaseDescriptor.setCommitLogSegmentSize(1);
+         DatabaseDescriptor.setCommitLogSync(Config.CommitLogSync.periodic);
+         DatabaseDescriptor.setCommitLogSyncPeriod(10 * 1000);
+         SchemaLoader.prepareServer();
+         SchemaLoader.createKeyspace(KEYSPACE1,
+                                     KeyspaceParams.simple(1),
+                                     SchemaLoader.standardCFMD(KEYSPACE1, STANDARD1, 0, AsciiType.instance,
BytesType.instance));
+ 
+                                     CompactionManager.instance.disableAutoCompaction();
+ 
 -        CommitLog.instance.resetUnsafe(true);
+         ColumnFamilyStore cfs1 = Keyspace.open(KEYSPACE1).getColumnFamilyStore(STANDARD1);
+ 
 -        final Mutation m = new RowUpdateBuilder(cfs1.metadata, 0, "k")
++        final Mutation m = new RowUpdateBuilder(cfs1.metadata.get(), 0, "k")
+                            .clustering("bytes")
+                            .add("val", ByteBuffer.wrap(entropy))
+                            .build();
+ 
+         // force creating several commitlog files
+         for (int i = 0; i < 10; i++)
+         {
+             CommitLog.instance.add(m);
+         }
+ 
+         // schedule discarding completed segments and immediately issue a shutdown
 -        UUID cfid = m.getColumnFamilyIds().iterator().next();
 -        CommitLog.instance.discardCompletedSegments(cfid, CommitLogPosition.NONE, CommitLog.instance.getCurrentPosition());
++        TableId tableId = m.getTableIds().iterator().next();
++        CommitLog.instance.discardCompletedSegments(tableId, CommitLogPosition.NONE, CommitLog.instance.getCurrentPosition());
+         CommitLog.instance.shutdownBlocking();
+ 
+         // the shutdown should block until all logs except the currently active one and
perhaps a new, empty one are gone
+         Assert.assertTrue(new File(DatabaseDescriptor.getCommitLogLocation()).listFiles().length
<= 2);
+     }
+ }


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


Mime
View raw message