natas_hw
/
dubbo
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix DubboMergingDigest concurrent issue (#12913)

This commit is contained in:
Albumen Kevin 2023-08-16 15:42:05 +08:00 committed by GitHub
parent 85b03bca56
commit 90cae426ab
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 16 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
dubbo-metrics/dubbo-metrics-api/src/main/java/org/apache/dubbo/metrics/aggregate/DubboMergingDigest.java
View File

@ -18,11 +18,12 @@
package org.apache.dubbo.metrics.aggregate;
import org.apache.dubbo.metrics.exception.MetricsNeverHappenException;
import com.tdunning.math.stats.Centroid;
import com.tdunning.math.stats.ScaleFunction;
import com.tdunning.math.stats.Sort;
import com.tdunning.math.stats.TDigest;
import org.apache.dubbo.metrics.exception.MetricsNeverHappenException;
import java.nio.ByteBuffer;
import java.util.AbstractCollection;
@ -285,16 +286,17 @@ public class DubboMergingDigest extends DubboAbstractTDigest {
throw new IllegalArgumentException("Cannot add NaN to t-digest");
}
int where;
synchronized (this) {
// There is a small probability of entering here
if (tempUsed.get() >= tempWeight.length - lastUsedCell.get() - 1) {
mergeNewValues();
}
where = tempUsed.getAndIncrement();
tempWeight[where] = w;
tempMean[where] = x;
unmergedWeight.addAndGet(w);
}
int where = tempUsed.getAndIncrement();
tempWeight[where] = w;
tempMean[where] = x;
unmergedWeight.addAndGet(w);
if (x < min) {
min = x;
}

14
dubbo-metrics/dubbo-metrics-api/src/test/java/org/apache/dubbo/metrics/aggregate/TimeWindowQuantileTest.java
View File

@ -18,6 +18,7 @@
package org.apache.dubbo.metrics.aggregate;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.RepeatedTest;
import org.junit.jupiter.api.Test;
import java.util.concurrent.ExecutorService;
@ -37,6 +38,7 @@ class TimeWindowQuantileTest {
}
@Test
@RepeatedTest(100)
void testMulti() {
ExecutorService executorService = Executors.newFixedThreadPool(200);
@ -51,12 +53,14 @@ class TimeWindowQuantileTest {
quantile.add(finalI));
}
index++;
try {
Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
// try {
// Thread.sleep(1);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
}
executorService.shutdown();
}
}