github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-25 07:07:54 -04:00
Code Issues Projects Releases Packages Wiki Activity

#7549 fix unsafe concurreny in PQ count getters

Browse source 
Fixes #8336
This commit is contained in:
Armin 2017-09-20 15:30:21 +02:00 • committed by Armin Braun
parent 6962fc2a06
commit 68b22280bc
1 changed files with 17 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

16
logstash-core/src/main/java/org/logstash/ackedqueue/Queue.java
View file

@ -768,16 +768,26 @@ public class Queue implements Closeable {
} }
public long getAckedCount() { public long getAckedCount() {
lock.lock();
try {
return headPage.ackedSeqNums.cardinality() + tailPages.stream() return headPage.ackedSeqNums.cardinality() + tailPages.stream()
.mapToLong(page -> page.ackedSeqNums.cardinality()) .mapToLong(page -> page.ackedSeqNums.cardinality()).sum();
.sum(); } finally {
lock.unlock();
}
} }
public long getUnackedCount() { public long getUnackedCount() {
lock.lock();
try {
long headPageCount = (headPage.getElementCount() - headPage.ackedSeqNums.cardinality()); long headPageCount = (headPage.getElementCount() - headPage.ackedSeqNums.cardinality());
long tailPagesCount = tailPages.stream() long tailPagesCount = tailPages.stream()
.mapToLong(page -> (page.getElementCount() - page.ackedSeqNums.cardinality())).sum(); .mapToLong(page -> (page.getElementCount() - page.ackedSeqNums.cardinality()))
.sum();
return headPageCount + tailPagesCount; return headPageCount + tailPagesCount;
} finally {
lock.unlock();
}
} }
protected long nextSeqNum() { protected long nextSeqNum() {