Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

AMQ-7082 - ensure flush sees free list, use lazySet to avoid store barrier, thanks cshannon

This commit is contained in:
gtully 2018-10-22 15:16:26 +01:00 committed by Jeff Genender
parent c6103415b9
commit a9b81ace62
1 changed files with 6 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -39,6 +39,7 @@ import java.util.TreeMap;
import java.util.concurrent.CountDownLatch; import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.Adler32; import java.util.zip.Adler32;
import java.util.zip.Checksum; import java.util.zip.Checksum;
@ -134,7 +135,7 @@ public class PageFile {
// Keeps track of free pages. // Keeps track of free pages.
private final AtomicLong nextFreePageId = new AtomicLong(); private final AtomicLong nextFreePageId = new AtomicLong();
private SequenceSet freeList = new SequenceSet(); private SequenceSet freeList = new SequenceSet();
private SequenceSet recoveredFreeList = null; private AtomicReference<SequenceSet> recoveredFreeList = new AtomicReference<SequenceSet>();
private final AtomicLong nextTxid = new AtomicLong(); private final AtomicLong nextTxid = new AtomicLong();
// Persistent settings stored in the page file. // Persistent settings stored in the page file.
@ -469,8 +470,8 @@ public class PageFile {
LOG.info(toString() + ". Recovered pageFile free list of size: " + newFreePages.rangeSize()); LOG.info(toString() + ". Recovered pageFile free list of size: " + newFreePages.rangeSize());
if (!newFreePages.isEmpty()) { if (!newFreePages.isEmpty()) {
// allow flush (with index lock held) to merge // allow flush (with index lock held) to merge eventually
recoveredFreeList = newFreePages; recoveredFreeList.lazySet(newFreePages);
} }
// all set for clean shutdown // all set for clean shutdown
needsFreePageRecovery = false; needsFreePageRecovery = false;
@ -561,9 +562,9 @@ public class PageFile {
throw new IOException("Page file already stopped: checkpointing is not allowed"); throw new IOException("Page file already stopped: checkpointing is not allowed");
} }
SequenceSet toMerge = recoveredFreeList; SequenceSet toMerge = recoveredFreeList.get();
if (toMerge != null) { if (toMerge != null) {
recoveredFreeList = null; recoveredFreeList.lazySet(null);
Sequence seq = toMerge.getHead(); Sequence seq = toMerge.getHead();
while (seq != null) { while (seq != null) {
freeList.add(seq); freeList.add(seq);