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

Refactor TransactionLog code and fix order of cleanup bug on Windows 

patch by stefania; reviewed by benedict for CASSANDRA-10286
This commit is contained in:
Stefania Alborghetti 2015-09-09 11:26:14 +08:00 committed by Benedict Elliott Smith
parent ff27eb304c
commit c163d0bc36
11 changed files with 20 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/java/org/apache/cassandra/io/compress/CompressedSequentialWriter.java
View File

@ -263,7 +263,7 @@ public class CompressedSequentialWriter extends SequentialWriter
@Override
protected Throwable doCommit(Throwable accumulate)
{
return metadataWriter.commit(accumulate);
return super.doCommit(metadataWriter.commit(accumulate));
}
@Override
@ -278,7 +278,6 @@ public class CompressedSequentialWriter extends SequentialWriter
syncInternal();
if (descriptor != null)
crcMetadata.writeFullChecksum(descriptor);
releaseFileHandle();
sstableMetadataCollector.addCompressionRatio(compressedSize, uncompressedSize);
metadataWriter.finalizeLength(current(), chunkCount).prepareToCommit();
}

4
src/java/org/apache/cassandra/io/compress/CompressionMetadata.java
View File

@ -410,7 +410,7 @@ public class CompressionMetadata
count = chunkIndex;
}
protected Throwable doPreCleanup(Throwable failed)
protected Throwable doPostCleanup(Throwable failed)
{
return offsets.close(failed);
}
@ -422,7 +422,7 @@ public class CompressionMetadata
protected Throwable doAbort(Throwable accumulate)
{
return FileUtils.deleteWithConfirm(filePath, false, accumulate);
return accumulate;
}
}

4
src/java/org/apache/cassandra/io/sstable/SSTableTxnWriter.java
View File

@ -69,13 +69,13 @@ public class SSTableTxnWriter extends Transactional.AbstractTransactional implem
protected Throwable doAbort(Throwable accumulate)
{
return writer.abort(txn.abort(accumulate));
return txn.abort(writer.abort(accumulate));
}
protected void doPrepare()
{
txn.prepareToCommit();
writer.prepareToCommit();
txn.prepareToCommit();
}
public Collection<SSTableReader> finish(boolean openResult)

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

@ -81,10 +81,6 @@ public class BigTableWriter extends SSTableWriter
dbuilder = SegmentedFile.getBuilder(DatabaseDescriptor.getDiskAccessMode(), false);
}
iwriter = new IndexWriter(keyCount, dataFile);
// txnLogs will delete if safe to do so (early readers)
iwriter.indexFile.deleteFile(false);
dataFile.deleteFile(false);
}
public void mark()
@ -322,7 +318,7 @@ public class BigTableWriter extends SSTableWriter
}
@Override
protected Throwable doPreCleanup(Throwable accumulate)
protected Throwable doPostCleanup(Throwable accumulate)
{
accumulate = dbuilder.close(accumulate);
return accumulate;
@ -485,7 +481,7 @@ public class BigTableWriter extends SSTableWriter
}
@Override
protected Throwable doPreCleanup(Throwable accumulate)
protected Throwable doPostCleanup(Throwable accumulate)
{
accumulate = summary.close(accumulate);
accumulate = bf.close(accumulate);

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

@ -50,7 +50,7 @@ public class ChecksummedSequentialWriter extends SequentialWriter
@Override
protected Throwable doCommit(Throwable accumulate)
{
return crcWriter.commit(accumulate);
return super.doCommit(crcWriter.commit(accumulate));
}
@Override
@ -66,9 +66,6 @@ public class ChecksummedSequentialWriter extends SequentialWriter
if (descriptor != null)
crcMetadata.writeFullChecksum(descriptor);
crcWriter.setDescriptor(descriptor).prepareToCommit();
// we must cleanup our file handles during prepareCommit for Windows compatibility as we cannot rename an open file;
// TODO: once we stop file renaming, remove this for clarity
releaseFileHandle();
}
}

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

@ -68,8 +68,6 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
// due to lack of multiple-inheritance, we proxy our transactional implementation
protected class TransactionalProxy extends AbstractTransactional
{
private boolean deleteFile = true;
@Override
protected Throwable doPreCleanup(Throwable accumulate)
{
@ -90,9 +88,6 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
protected void doPrepare()
{
syncInternal();
// we must cleanup our file handles during prepareCommit for Windows compatibility as we cannot rename an open file;
// TODO: once we stop file renaming, remove this for clarity
releaseFileHandle();
}
protected Throwable doCommit(Throwable accumulate)
@ -102,10 +97,7 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
protected Throwable doAbort(Throwable accumulate)
{
if (deleteFile)
return FileUtils.deleteWithConfirm(filePath, false, accumulate);
else
return accumulate;
return accumulate;
}
}
@ -409,23 +401,6 @@ public class SequentialWriter extends BufferedDataOutputStreamPlus implements Tr
return new TransactionalProxy();
}
public void deleteFile(boolean val)
{
txnProxy.deleteFile = val;
}
public void releaseFileHandle()
{
try
{
channel.close();
}
catch (IOException e)
{
throw new FSWriteError(e, filePath);
}
}
/**
* Class to hold a mark to the position of the file
*/

5
src/java/org/apache/cassandra/utils/concurrent/Transactional.java
View File

@ -88,7 +88,8 @@ public interface Transactional extends AutoCloseable
// Transactional objects will perform cleanup in the commit() or abort() calls
/**
* perform an exception-safe pre-abort cleanup; this will still be run *after* commit
* perform an exception-safe pre-abort/commit cleanup;
* this will be run after prepareToCommit (so before commit), and before abort
*/
protected Throwable doPreCleanup(Throwable accumulate){ return accumulate; }
@ -113,7 +114,6 @@ public interface Transactional extends AutoCloseable
if (state != State.READY_TO_COMMIT)
throw new IllegalStateException("Cannot commit unless READY_TO_COMMIT; state is " + state);
accumulate = doCommit(accumulate);
accumulate = doPreCleanup(accumulate);
accumulate = doPostCleanup(accumulate);
state = State.COMMITTED;
return accumulate;
@ -171,6 +171,7 @@ public interface Transactional extends AutoCloseable
throw new IllegalStateException("Cannot prepare to commit unless IN_PROGRESS; state is " + state);
doPrepare();
maybeFail(doPreCleanup(null));
state = State.READY_TO_COMMIT;
}

1
test/unit/org/apache/cassandra/io/compress/CompressedSequentialWriterTest.java
View File

@ -222,7 +222,6 @@ public class CompressedSequentialWriterTest extends SequentialWriterTest
protected void assertAborted() throws Exception
{
super.assertAborted();
Assert.assertFalse(offsetsFile.exists());
}
void cleanup()

10
test/unit/org/apache/cassandra/io/sstable/SSTableLoaderTest.java
View File

@ -131,11 +131,14 @@ public class SSTableLoaderTest
writer.addRow("key1", "col1", "100");
}
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1);
cfs.forceBlockingFlush(); // wait for sstables to be on disk else we won't be able to stream them
final CountDownLatch latch = new CountDownLatch(1);
SSTableLoader loader = new SSTableLoader(dataDir, new TestClient(), new OutputHandler.SystemOutput(false, false));
loader.stream(Collections.emptySet(), completionStreamListener(latch)).get();
List<FilteredPartition> partitions = Util.getAll(Util.cmd(Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD1)).build());
List<FilteredPartition> partitions = Util.getAll(Util.cmd(cfs).build());
assertEquals(1, partitions.size());
assertEquals("key1", AsciiType.instance.getString(partitions.get(0).partitionKey().getKey()));
@ -175,6 +178,9 @@ public class SSTableLoaderTest
writer.addRow(String.format("key%d", i), String.format("col%d", j), "100");
}
ColumnFamilyStore cfs = Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD2);
cfs.forceBlockingFlush(); // wait for sstables to be on disk else we won't be able to stream them
//make sure we have some tables...
assertTrue(dataDir.listFiles().length > 0);
@ -183,7 +189,7 @@ public class SSTableLoaderTest
SSTableLoader loader = new SSTableLoader(dataDir, new TestClient(), new OutputHandler.SystemOutput(false, false));
loader.stream(Collections.emptySet(), completionStreamListener(latch)).get();
List<FilteredPartition> partitions = Util.getAll(Util.cmd(Keyspace.open(KEYSPACE1).getColumnFamilyStore(CF_STANDARD2)).build());
List<FilteredPartition> partitions = Util.getAll(Util.cmd(cfs).build());
assertTrue(partitions.size() > 0 && partitions.size() < NB_PARTITIONS);

1
test/unit/org/apache/cassandra/io/util/ChecksummedSequentialWriterTest.java
View File

@ -85,7 +85,6 @@ public class ChecksummedSequentialWriterTest extends SequentialWriterTest
protected void assertAborted() throws Exception
{
super.assertAborted();
Assert.assertFalse(crcFile.exists());
}
}

1
test/unit/org/apache/cassandra/io/util/SequentialWriterTest.java
View File

@ -102,7 +102,6 @@ public class SequentialWriterTest extends AbstractTransactionalTest
protected void assertAborted() throws Exception
{
Assert.assertFalse(writer.isOpen());
Assert.assertFalse(file.exists());
}
protected void assertCommitted() throws Exception