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

Make jmh jar and add RW + Compaction tests 

Patch by tjake; reviewed by Nitsan Wakart for CASSANDRA-12586
This commit is contained in:
T Jake Luciani 2016-09-01 10:36:11 -04:00
parent 7d857b46fb
commit 1d7466425b
5 changed files with 309 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.10
* Add JMH benchmarks.jar (CASSANDRA-12586)
* Add row offset support to SASI (CASSANDRA-11990)
* Cleanup uses of AlterTableStatementColumn (CASSANDRA-12567)
* Add keep-alive to streaming (CASSANDRA-11841)

29
build.xml
View File

@ -156,12 +156,12 @@
<pathelement location="${build.classes.main}" />
<pathelement location="${build.classes.thrift}" />
<fileset dir="${build.lib}">
<include name="**/*.jar" />
<exclude name="**/*-sources.jar"/>
<include name="**/*.jar" />
<exclude name="**/*-sources.jar"/>
</fileset>
<fileset dir="${build.dir.lib}">
<include name="**/*.jar" />
<exclude name="**/*-sources.jar"/>
<include name="**/*.jar" />
<exclude name="**/*-sources.jar"/>
</fileset>
</path>
@ -1152,6 +1152,25 @@
</rat:report>
</target>
<target name="build-jmh" depends="build-test" description="Create JMH uber jar">
<jar jarfile="${build.test.dir}/deps.jar">
<zipgroupfileset dir="${build.dir.lib}/jars">
<include name="*jmh*.jar"/>
<include name="jopt*.jar"/>
<include name="commons*.jar"/>
</zipgroupfileset>
<zipgroupfileset dir="${build.lib}" includes="*.jar"/>
</jar>
<jar jarfile="${build.test.dir}/benchmarks.jar">
<manifest>
<attribute name="Main-Class" value="org.openjdk.jmh.Main"/>
</manifest>
<zipfileset src="${build.test.dir}/deps.jar" excludes="META-INF/*.SF" />
<fileset dir="${build.classes.main}"/>
<fileset dir="${test.classes}"/>
</jar>
</target>
<target name="build-test" depends="build" description="Compile test classes">
<javac
compiler="modern"
@ -1701,7 +1720,7 @@
</target>
<!-- run microbenchmarks suite -->
<target name="microbench" depends="build-test">
<target name="microbench" depends="build-jmh">
<java classname="org.openjdk.jmh.Main"
fork="true"
failonerror="true">

3
src/java/org/apache/cassandra/io/sstable/format/SSTableReader.java
View File

@ -42,6 +42,7 @@ import org.apache.cassandra.cache.ChunkCache;
import org.apache.cassandra.cache.InstrumentingCache;
import org.apache.cassandra.cache.KeyCacheKey;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.cassandra.concurrent.NamedThreadFactory;
import org.apache.cassandra.concurrent.ScheduledExecutors;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
@ -145,7 +146,7 @@ public abstract class SSTableReader extends SSTable implements SelfRefCounted<SS
// Do NOT start this thread pool in client mode
ScheduledThreadPoolExecutor syncExecutor = new ScheduledThreadPoolExecutor(1);
ScheduledThreadPoolExecutor syncExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("read-hotness-tracker"));
// Immediately remove readMeter sync task when cancelled.
syncExecutor.setRemoveOnCancelPolicy(true);
return syncExecutor;

167
test/microbench/org/apache/cassandra/test/microbench/CompactionBench.java Normal file
View File

@ -0,0 +1,167 @@
/*
* 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.test.microbench;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.cassandra.UpdateBuilder;
import org.apache.cassandra.concurrent.StageManager;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.statements.ParsedStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.sstable.Descriptor;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.service.CassandraDaemon;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.hadoop.util.bloom.Key;
import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.profile.StackProfiler;
import org.openjdk.jmh.results.Result;
import org.openjdk.jmh.results.RunResult;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
@BenchmarkMode(Mode.AverageTime)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 25, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 5, time = 2, timeUnit = TimeUnit.SECONDS)
@Fork(value = 1)
@Threads(1)
@State(Scope.Benchmark)
public class CompactionBench extends CQLTester
{
static String keyspace;
String table;
String writeStatement;
String readStatement;
ColumnFamilyStore cfs;
List<File> snapshotFiles;
List<Descriptor> liveFiles;
@Setup(Level.Trial)
public void setup() throws Throwable
{
CQLTester.prepareServer();
keyspace = createKeyspace("CREATE KEYSPACE %s with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 } and durable_writes = false");
table = createTable(keyspace, "CREATE TABLE %s ( userid bigint, picid bigint, commentid bigint, PRIMARY KEY(userid, picid))");
execute("use "+keyspace+";");
writeStatement = "INSERT INTO "+table+"(userid,picid,commentid)VALUES(?,?,?)";
readStatement = "SELECT * from "+table+" limit 100";
Keyspace.system().forEach(k -> k.getColumnFamilyStores().forEach(c -> c.disableAutoCompaction()));
cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
cfs.disableAutoCompaction();
//Warm up
System.err.println("Writing 50k");
for (long i = 0; i < 50000; i++)
execute(writeStatement, i, i, i );
cfs.forceBlockingFlush();
System.err.println("Writing 50k again...");
for (long i = 0; i < 50000; i++)
execute(writeStatement, i, i, i );
cfs.forceBlockingFlush();
cfs.snapshot("originals");
snapshotFiles = cfs.getDirectories().sstableLister(Directories.OnTxnErr.IGNORE).snapshots("originals").listFiles();
}
@TearDown(Level.Trial)
public void teardown() throws IOException, ExecutionException, InterruptedException
{
int active = Thread.currentThread().getThreadGroup().activeCount();
Thread[] threads = new Thread[active];
Thread.currentThread().getThreadGroup().enumerate(threads);
for (Thread t : threads)
{
if (!t.isDaemon())
System.err.println("Thread "+t.getName());
}
CQLTester.cleanup();
}
@TearDown(Level.Invocation)
public void resetSnapshot()
{
cfs.truncateBlocking();
List<File> directories = cfs.getDirectories().getCFDirectories();
for (File file : directories)
{
for (File f : file.listFiles())
{
if (f.isDirectory())
continue;
FileUtils.delete(f);
}
}
for (File file : snapshotFiles)
FileUtils.createHardLink(file, new File(file.toPath().getParent().getParent().getParent().toFile(), file.getName()));
cfs.loadNewSSTables();
}
@Benchmark
public void compactTest() throws Throwable
{
cfs.forceMajorCompaction();
}
}

115
test/microbench/org/apache/cassandra/test/microbench/ReadWriteTest.java Normal file
View File

@ -0,0 +1,115 @@
/*
* 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.test.microbench;
import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.*;
import org.apache.cassandra.UpdateBuilder;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Config;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.statements.ParsedStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Directories;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.io.util.DataInputBuffer;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.io.util.DataOutputBufferFixed;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.net.MessageIn;
import org.apache.cassandra.net.MessageOut;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceMetadata;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.service.CassandraDaemon;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.profile.StackProfiler;
import org.openjdk.jmh.results.Result;
import org.openjdk.jmh.results.RunResult;
import org.openjdk.jmh.runner.Runner;
import org.openjdk.jmh.runner.options.Options;
import org.openjdk.jmh.runner.options.OptionsBuilder;
@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.MILLISECONDS)
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 5, time = 2, timeUnit = TimeUnit.SECONDS)
@Fork(value = 1)
@Threads(1)
@State(Scope.Benchmark)
public class ReadWriteTest extends CQLTester
{
static String keyspace;
String table;
String writeStatement;
String readStatement;
long numRows = 0;
ColumnFamilyStore cfs;
@Setup(Level.Trial)
public void setup() throws Throwable
{
CQLTester.setUpClass();
keyspace = createKeyspace("CREATE KEYSPACE %s with replication = { 'class' : 'SimpleStrategy', 'replication_factor' : 1 } and durable_writes = false");
table = createTable(keyspace, "CREATE TABLE %s ( userid bigint, picid bigint, commentid bigint, PRIMARY KEY(userid, picid))");
execute("use "+keyspace+";");
writeStatement = "INSERT INTO "+table+"(userid,picid,commentid)VALUES(?,?,?)";
readStatement = "SELECT * from "+table+" limit 100";
cfs = Keyspace.open(keyspace).getColumnFamilyStore(table);
cfs.disableAutoCompaction();
//Warm up
System.err.println("Writing 50k");
for (long i = 0; i < 5000; i++)
execute(writeStatement, i, i, i );
}
@TearDown(Level.Trial)
public void teardown() throws IOException, ExecutionException, InterruptedException
{
CQLTester.cleanup();
}
@Benchmark
public Object write() throws Throwable
{
numRows++;
return execute(writeStatement, numRows, numRows, numRows );
}
@Benchmark
public Object read() throws Throwable
{
return execute(readStatement);
}
}