IntenseFinance/nosqlbench
3
0
Fork 0
mirror of https://github.com/nosqlbench/nosqlbench.git synced 2025-02-25 18:55:28 -06:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #369 from lhotari/lh-rate-limit-per-thread

Browse Source 
Make ratelimiter instance thread specific for PulsarActivity
This commit is contained in:
Jonathan Shook 2021-10-19 14:33:33 -05:00 committed by GitHub
commit b82b4b1a43
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 20 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
driver-pulsar/src/main/java/io/nosqlbench/driver/pulsar/PulsarActivity.java
View File

@ -10,6 +10,8 @@ import io.nosqlbench.driver.pulsar.util.PulsarNBClientConf;
import io.nosqlbench.engine.api.activityapi.core.ActivityDefObserver;
import io.nosqlbench.engine.api.activityapi.errorhandling.modular.NBErrorHandler;
import io.nosqlbench.engine.api.activityapi.planning.OpSequence;
import io.nosqlbench.engine.api.activityapi.ratelimits.RateLimiter;
import io.nosqlbench.engine.api.activityapi.ratelimits.RateLimiters;
import io.nosqlbench.engine.api.activityimpl.ActivityDef;
import io.nosqlbench.engine.api.activityimpl.OpDispenser;
import io.nosqlbench.engine.api.activityimpl.SimpleActivity;
@ -57,6 +59,7 @@ public class PulsarActivity extends SimpleActivity implements ActivityDefObserve
private NBErrorHandler errorHandler;
private OpSequence<OpDispenser<PulsarOp>> sequencer;
private volatile Throwable asyncOperationFailure;
private boolean cycleratePerThread;
public PulsarActivity(ActivityDef activityDef) {
super(activityDef);
@ -108,11 +111,26 @@ public class PulsarActivity extends SimpleActivity implements ActivityDefObserve
() -> activityDef.getParams().getOptionalString("errors").orElse("stop"),
this::getExceptionMetrics
);
cycleratePerThread = activityDef.getParams().takeBoolOrDefault("cyclerate_per_thread", false);
}
private final ThreadLocal<RateLimiter> cycleLimiterThreadLocal = ThreadLocal.withInitial(() -> {
if (super.getCycleLimiter() != null) {
return RateLimiters.createOrUpdate(this.getActivityDef(), "cycles", null,
super.getCycleLimiter().getRateSpec());
} else {
return null;
}
});
@Override
public synchronized void onActivityDefUpdate(ActivityDef activityDef) {
super.onActivityDefUpdate(activityDef);
public RateLimiter getCycleLimiter() {
if (cycleratePerThread) {
return cycleLimiterThreadLocal.get();
} else {
return super.getCycleLimiter();
}
}
public NBErrorHandler getErrorHandler() { return errorHandler; }