github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 06:37:19 -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 5ee5431707
commit e248d3a07d
1 changed files with 17 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

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