Apache/activemq
1
0
Fork 0
mirror of https://github.com/apache/activemq.git synced 2025-02-09 03:25:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-6207

Browse Source 
Perform writes / sync in predicatable order first to the recovery file
and then into the main index file.
(cherry picked from commit b39ab7867c208a967090eff189594c2b06dafb34)
This commit is contained in:
Timothy Bish 2016-03-11 09:57:53 -05:00
parent 442fadac87
commit 3dc769d1f6
1 changed files with 29 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
activemq-kahadb-store/src/main/java/org/apache/activemq/store/kahadb/disk/page/PageFile.java
View File

@ -1048,12 +1048,13 @@ public class PageFile {
this.checkpointLatch = null; this.checkpointLatch = null;
} }
Checksum checksum = new Adler32(); // First land the writes in the recovery file
if (enableRecoveryFile) { if (enableRecoveryFile) {
Checksum checksum = new Adler32();
recoveryFile.seek(RECOVERY_FILE_HEADER_SIZE); recoveryFile.seek(RECOVERY_FILE_HEADER_SIZE);
}
for (PageWrite w : batch) { for (PageWrite w : batch) {
if (enableRecoveryFile) {
try { try {
checksum.update(w.getDiskBound(), 0, pageSize); checksum.update(w.getDiskBound(), 0, pageSize);
} catch (Throwable t) { } catch (Throwable t) {
@ -1063,36 +1064,36 @@ public class PageFile {
recoveryFile.write(w.getDiskBound(), 0, pageSize); recoveryFile.write(w.getDiskBound(), 0, pageSize);
} }
writeFile.seek(toOffset(w.page.getPageId())); // Can we shrink the recovery buffer??
writeFile.write(w.getDiskBound(), 0, pageSize); if (recoveryPageCount > recoveryFileMaxPageCount) {
w.done(); int t = Math.max(recoveryFileMinPageCount, batch.size());
recoveryFile.setLength(recoveryFileSizeForPages(t));
}
// Record the page writes in the recovery buffer.
recoveryFile.seek(0);
// Store the next tx id...
recoveryFile.writeLong(nextTxid.get());
// Store the checksum for thw write batch so that on recovery we
// know if we have a consistent
// write batch on disk.
recoveryFile.writeLong(checksum.getValue());
// Write the # of pages that will follow
recoveryFile.writeInt(batch.size());
if (enableDiskSyncs) {
recoveryFile.sync();
}
} }
try { try {
if (enableRecoveryFile) { for (PageWrite w : batch) {
// Can we shrink the recovery buffer?? writeFile.seek(toOffset(w.page.getPageId()));
if (recoveryPageCount > recoveryFileMaxPageCount) { writeFile.write(w.getDiskBound(), 0, pageSize);
int t = Math.max(recoveryFileMinPageCount, batch.size()); w.done();
recoveryFile.setLength(recoveryFileSizeForPages(t));
}
// Record the page writes in the recovery buffer.
recoveryFile.seek(0);
// Store the next tx id...
recoveryFile.writeLong(nextTxid.get());
// Store the checksum for thw write batch so that on recovery we
// know if we have a consistent
// write batch on disk.
recoveryFile.writeLong(checksum.getValue());
// Write the # of pages that will follow
recoveryFile.writeInt(batch.size());
} }
if (enableDiskSyncs) { if (enableDiskSyncs) {
// Sync to make sure recovery buffer writes land on disk..
if (enableRecoveryFile) {
recoveryFile.sync();
}
writeFile.sync(); writeFile.sync();
} }
} finally { } finally {