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

Merge branch 'cassandra-2.2' into trunk

This commit is contained in:
Tyler Hobbs 2015-05-26 14:23:21 -05:00
parent e4da2e28da 91187b542e
commit 3123e88e54
12 changed files with 222 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
build.xml
View File

@ -1215,6 +1215,7 @@
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Djava.io.tmpdir=${tmp.dir}"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
</testmacro>
<fileset dir="${test.unit.src}">
<exclude name="**/pig/*.java" />
@ -1230,6 +1231,7 @@
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Dcassandra.config.loader=org.apache.cassandra.OffsetAwareConfigurationLoader"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
</testmacro>
</target>
@ -1257,6 +1259,7 @@
<jvmarg value="-Dmigration-sstable-root=${test.data}/migration-sstables"/>
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
</testmacro>
</target>
@ -1276,6 +1279,7 @@
<jvmarg value="-Dcassandra.ring_delay_ms=1000"/>
<jvmarg value="-Dcassandra.tolerate_sstable_size=true"/>
<jvmarg value="-Dcassandra.config=file:///${compressed_yaml}"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
</testmacro>
<fileset dir="${test.unit.src}">
<exclude name="**/pig/*.java" />
@ -1336,6 +1340,7 @@
<jvmarg value="-Xss256k"/>
<jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
<jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
<classpath>
<path refid="cassandra.classpath" />
<pathelement location="${test.classes}"/>
@ -1379,6 +1384,7 @@
<jvmarg value="-Xss256k"/>
<jvmarg value="-Dcassandra.test.use_prepared=${cassandra.test.use_prepared}"/>
<jvmarg value="-Dcassandra.memtable_row_overhead_computation_step=100"/>
<jvmarg value="-Dcassandra.skip_sync=true" />
<classpath>
<path refid="cassandra.classpath" />
<pathelement location="${test.classes}"/>

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

@ -27,6 +27,7 @@ import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.compress.BufferType;
import org.apache.cassandra.io.compress.ICompressor;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.SyncUtil;
/*
* Compressed commit log segment. Provides an in-memory buffer for the mutation threads. On sync compresses the written
@ -120,7 +121,7 @@ public class CompressedSegment extends CommitLogSegment
// Protected by synchronization on CommitLogSegment.sync().
writeSyncMarker(compressedBuffer, 0, (int) channel.position(), (int) channel.position() + compressedBuffer.remaining());
channel.write(compressedBuffer);
channel.force(true);
SyncUtil.force(channel, true);
}
catch (Exception e)
{

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

@ -27,6 +27,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.io.FSWriteError;
import org.apache.cassandra.io.util.FileUtils;
import org.apache.cassandra.utils.CLibrary;
import org.apache.cassandra.utils.SyncUtil;
/*
* Memory-mapped segment. Maps the destination channel into an appropriately-sized memory-mapped buffer in which the
@ -91,7 +92,7 @@ public class MemoryMappedSegment extends CommitLogSegment
writeSyncMarker(buffer, startMarker, startMarker, nextMarker);
try {
((MappedByteBuffer) buffer).force();
SyncUtil.force((MappedByteBuffer) buffer);
}
catch (Exception e) // MappedByteBuffer.force() does not declare IOException but can actually throw it
{

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

@ -51,6 +51,7 @@ import org.apache.cassandra.utils.StreamingHistogram;
import org.apache.cassandra.utils.concurrent.Transactional;
import static org.apache.cassandra.utils.Throwables.merge;
import org.apache.cassandra.utils.SyncUtil;
public class BigTableWriter extends SSTableWriter
{
@ -513,7 +514,7 @@ public class BigTableWriter extends SSTableWriter
DataOutputStreamPlus stream = new BufferedDataOutputStreamPlus(fos);
FilterFactory.serialize(bf, stream);
stream.flush();
fos.getFD().sync();
SyncUtil.sync(fos);
stream.close();
}
catch (IOException e)

5
src/java/org/apache/cassandra/io/util/SequentialWriter.java
View File

@ -36,6 +36,7 @@ import org.apache.cassandra.utils.CLibrary;
import org.apache.cassandra.utils.concurrent.Transactional;
import static org.apache.cassandra.utils.Throwables.merge;
import org.apache.cassandra.utils.SyncUtil;
/**
* Adds buffering, mark, and fsyncing to OutputStream. We always fsync on close; we may also
@ -227,7 +228,7 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
{
try
{
channel.force(false);
SyncUtil.force(channel, false);
}
catch (IOException e)
{
@ -244,7 +245,7 @@ public class SequentialWriter extends OutputStream implements WritableByteChanne
if (!directorySynced)
{
CLibrary.trySync(directoryFD);
SyncUtil.trySync(directoryFD);
directorySynced = true;
}

165
src/java/org/apache/cassandra/utils/SyncUtil.java Normal file
View File

@ -0,0 +1,165 @@
package org.apache.cassandra.utils;
import java.io.FileDescriptor;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.io.SyncFailedException;
import java.lang.reflect.Field;
import java.nio.MappedByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.FileChannel;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.cassandra.config.Config;
import com.google.common.base.Preconditions;
/*
* A wrapper around various mechanisms for syncing files that makes it possible it intercept
* and skip syncing. Useful for unit tests in certain environments where syncs can have outliers
* bad enough to causes tests to run 10s of seconds longer.
*/
public class SyncUtil
{
public static boolean SKIP_SYNC = Boolean.getBoolean(Config.PROPERTY_PREFIX + "skip_sync");
private static final Field mbbFDField;
private static final Field fdClosedField;
private static final Field fdUseCountField;
static
{
Field mbbFDFieldTemp = null;
try
{
mbbFDFieldTemp = MappedByteBuffer.class.getDeclaredField("fd");
mbbFDFieldTemp.setAccessible(true);
}
catch (NoSuchFieldException e)
{
}
mbbFDField = mbbFDFieldTemp;
//Java 8
Field fdClosedFieldTemp = null;
try
{
fdClosedFieldTemp = FileDescriptor.class.getDeclaredField("closed");
fdClosedFieldTemp.setAccessible(true);
}
catch (NoSuchFieldException e)
{
}
fdClosedField = fdClosedFieldTemp;
//Java 7
Field fdUseCountTemp = null;
try
{
fdUseCountTemp = FileDescriptor.class.getDeclaredField("useCount");
fdUseCountTemp.setAccessible(true);
}
catch (NoSuchFieldException e)
{
}
fdUseCountField = fdUseCountTemp;
}
public static MappedByteBuffer force(MappedByteBuffer buf)
{
Preconditions.checkNotNull(buf);
if (SKIP_SYNC)
{
Object fd = null;
try
{
if (mbbFDField != null)
{
fd = mbbFDField.get(buf);
}
}
catch (Exception e)
{
throw new RuntimeException(e);
}
//This is what MappedByteBuffer.force() throws if a you call force() on an umapped buffer
if (mbbFDField != null && fd == null)
throw new UnsupportedOperationException();
return buf;
}
else
{
return buf.force();
}
}
public static void sync(FileDescriptor fd) throws SyncFailedException
{
Preconditions.checkNotNull(fd);
if (SKIP_SYNC)
{
boolean closed = false;
try
{
if (fdClosedField != null)
closed = fdClosedField.getBoolean(fd);
}
catch (Exception e)
{
throw new RuntimeException(e);
}
int useCount = 1;
try
{
if (fdUseCountField != null)
useCount = ((AtomicInteger)fdUseCountField.get(fd)).get();
}
catch (Exception e)
{
throw new RuntimeException(e);
}
if (closed || !fd.valid() || useCount < 0)
throw new SyncFailedException("Closed " + closed + " valid " + fd.valid() + " useCount " + useCount);
}
else
{
fd.sync();
}
}
public static void force(FileChannel fc, boolean metaData) throws IOException
{
Preconditions.checkNotNull(fc);
if (SKIP_SYNC)
{
if (!fc.isOpen())
throw new ClosedChannelException();
}
else
{
fc.force(metaData);
}
}
public static void sync(RandomAccessFile ras) throws IOException
{
Preconditions.checkNotNull(ras);
sync(ras.getFD());
}
public static void sync(FileOutputStream fos) throws IOException
{
Preconditions.checkNotNull(fos);
sync(fos.getFD());
}
public static void trySync(int fd)
{
if (SKIP_SYNC)
return;
else
CLibrary.trySync(fd);
}
}

15
test/unit/org/apache/cassandra/Util.java
View File

@ -28,6 +28,7 @@ import java.nio.channels.FileChannel;
import java.util.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.function.Supplier;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
@ -67,6 +68,7 @@ import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.CounterId;
import org.apache.hadoop.fs.FileUtil;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
public class Util
@ -379,4 +381,17 @@ public class Util
Composite endName = CellNames.simpleDense(ByteBufferUtil.bytes(finish));
return new RangeTombstone(startName, endName, timestamp , localtime);
}
public static void spinAssertEquals(Object expected, Supplier<Object> s, int timeoutInSeconds)
{
long now = System.currentTimeMillis();
while (System.currentTimeMillis() - now < now + (1000 * timeoutInSeconds))
{
if (s.get().equals(expected))
break;
Thread.yield();
}
assertEquals(expected, s.get());
}
}

8
test/unit/org/apache/cassandra/db/ColumnFamilyMetricTest.java
View File

@ -19,11 +19,12 @@ package org.apache.cassandra.db;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.function.Supplier;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
import org.apache.cassandra.config.KSMetaData;
import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.locator.SimpleStrategy;
@ -78,10 +79,9 @@ public class ColumnFamilyMetricTest
store.truncateBlocking();
// after truncate, size metrics should be down to 0
assertEquals(0, store.metric.liveDiskSpaceUsed.getCount());
assertEquals(0, store.metric.totalDiskSpaceUsed.getCount());
Util.spinAssertEquals(0L, () -> store.metric.liveDiskSpaceUsed.getCount(), 30);
Util.spinAssertEquals(0L, () -> store.metric.totalDiskSpaceUsed.getCount(), 30);
store.enableAutoCompaction();
}
}

6
test/unit/org/apache/cassandra/io/compress/CompressedRandomAccessReaderTest.java
View File

@ -25,7 +25,6 @@ import java.util.Collections;
import java.util.Random;
import org.junit.Test;
import org.apache.cassandra.db.composites.SimpleDenseCellNameType;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.exceptions.ConfigurationException;
@ -35,6 +34,7 @@ import org.apache.cassandra.io.util.ChannelProxy;
import org.apache.cassandra.io.util.FileMark;
import org.apache.cassandra.io.util.RandomAccessReader;
import org.apache.cassandra.io.util.SequentialWriter;
import org.apache.cassandra.utils.SyncUtil;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
@ -196,7 +196,7 @@ public class CompressedRandomAccessReaderTest
for (int i = 0; i < checksum.length; i++)
{
checksumModifier.write(random.nextInt());
checksumModifier.getFD().sync(); // making sure that change was synced with disk
SyncUtil.sync(checksumModifier); // making sure that change was synced with disk
final RandomAccessReader r = CompressedRandomAccessReader.open(channel, meta);
@ -238,6 +238,6 @@ public class CompressedRandomAccessReaderTest
{
file.seek(checksumOffset);
file.write(checksum);
file.getFD().sync();
SyncUtil.sync(file);
}
}

3
test/unit/org/apache/cassandra/io/sstable/SSTableRewriterTest.java
View File

@ -262,6 +262,7 @@ public class SSTableRewriterTest extends SchemaLoader
SSTableReader s = writeFile(cfs, 1000);
cfs.addSSTable(s);
long startStorageMetricsLoad = StorageMetrics.load.getCount();
long sBytesOnDisk = s.bytesOnDisk();
Set<SSTableReader> compacting = Sets.newHashSet(s);
SSTableRewriter.overrideOpenInterval(10000000);
@ -292,7 +293,7 @@ public class SSTableRewriterTest extends SchemaLoader
for (SSTableReader x : cfs.getSSTables())
sum += x.bytesOnDisk();
assertEquals(sum, cfs.metric.liveDiskSpaceUsed.getCount());
assertEquals(startStorageMetricsLoad - s.bytesOnDisk() + sum, StorageMetrics.load.getCount());
assertEquals(startStorageMetricsLoad - sBytesOnDisk + sum, StorageMetrics.load.getCount());
assertEquals(files, sstables.size());
assertEquals(files, cfs.getSSTables().size());
SSTableDeletingTask.waitForDeletions();

14
test/unit/org/apache/cassandra/io/util/BufferedRandomAccessFileTest.java
View File

@ -21,6 +21,7 @@ package org.apache.cassandra.io.util;
import org.apache.cassandra.service.FileCacheService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.SyncUtil;
import java.io.File;
import java.io.FileOutputStream;
@ -36,7 +37,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import static org.apache.cassandra.Util.expectEOF;
import static org.apache.cassandra.Util.expectException;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertEquals;
@ -425,11 +425,11 @@ public class BufferedRandomAccessFileTest
try (final RandomAccessReader r = RandomAccessReader.open(new File(tmpFile.getPath())))
{
assert tmpFile.getPath().equals(r.getPath());
// Create a mark and move the rw there.
final FileMark mark = r.mark();
r.reset(mark);
// Expect this call to succeed.
r.bytesPastMark(mark);
}
@ -469,7 +469,7 @@ public class BufferedRandomAccessFileTest
try (RandomAccessReader copy = RandomAccessReader.open(new File(r.getPath())))
{
ByteBuffer contents = copy.readBytes((int) copy.length());
assertEquals(contents.limit(), data.length);
assertEquals(ByteBufferUtil.compare(contents, data), 0);
}
@ -517,12 +517,12 @@ public class BufferedRandomAccessFileTest
{
w.write(new byte[30]);
w.flush();
try (RandomAccessReader r = RandomAccessReader.open(w))
{
r.seek(10);
r.mark();
r.seek(0);
r.bytesPastMark();
}
@ -680,7 +680,7 @@ public class BufferedRandomAccessFileTest
f.deleteOnExit();
FileOutputStream fout = new FileOutputStream(f);
fout.write(data);
fout.getFD().sync();
SyncUtil.sync(fout);
fout.close();
return f;
}

16
test/unit/org/apache/cassandra/schema/DefsTest.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.schema;
import java.io.File;
import java.nio.ByteBuffer;
import java.util.function.Supplier;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
import org.apache.cassandra.SchemaLoader;
@ -256,11 +257,16 @@ public class DefsTest
Assert.assertFalse("This mutation should have failed since the CF no longer exists.", success);
// verify that the files are gone.
for (File file : store.directories.sstableLister().listFiles())
{
if (file.getPath().endsWith("Data.db") && !new File(file.getPath().replace("Data.db", "Compacted")).exists())
throw new AssertionError("undeleted file " + file);
}
Supplier<Object> lambda = () -> {
for (File file : store.directories.sstableLister().listFiles())
{
if (file.getPath().endsWith("Data.db") && !new File(file.getPath().replace("Data.db", "Compacted")).exists())
return false;
}
return true;
};
Util.spinAssertEquals(true, lambda, 30);
}
@Test