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

provide threads metric

Browse Source
This commit is contained in:
Jonathan Shook
2023-09-27 12:47:06 -05:00
parent 96269785e2
commit a57e967930
4 changed files with 65 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityExecutor.java
View File

@@ -413,6 +413,7 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
// before threads start running such as metrics instruments
activity.initActivity();
startMotorExecutorService();
registerMetrics();
startRunningActivityThreads();
awaitMotorsAtLeastRunning();
logger.debug("STARTED " + activityDef.getAlias());
@@ -421,6 +422,7 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
this.exception = e;
} finally {
stoppedAt=System.currentTimeMillis();
unregisterMetrics();
activity.shutdownActivity();
activity.closeAutoCloseables();
ExecutionResult result = new ExecutionResult(startedAt, stoppedAt, "", exception);
@@ -446,19 +448,18 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
}
}
public synchronized void startActivity() {
RunStateImage startable = tally.awaitNoneOther(1000L, RunState.Uninitialized, RunState.Stopped);
if (startable.isTimeout()) {
throw new RuntimeException("Unable to start activity '" + getActivity().getAlias() + "' which is in state " + startable);
}
startMotorExecutorService();
startRunningActivityThreads();
awaitMotorsAtLeastRunning();
registerMetrics();
}
// public synchronized void startActivity() {
// RunStateImage startable = tally.awaitNoneOther(1000L, RunState.Uninitialized, RunState.Stopped);
// if (startable.isTimeout()) {
// throw new RuntimeException("Unable to start activity '" + getActivity().getAlias() + "' which is in state " + startable);
// }
// startMotorExecutorService();
// startRunningActivityThreads();
// awaitMotorsAtLeastRunning();
// }
private void registerMetrics() {
this.threadsGauge= threadsGauge = ActivityMetrics.register(new NBFunctionGauge(activity, () -> (double) this.motors.size(), "threads"));
this.threadsGauge = ActivityMetrics.register(new NBFunctionGauge(activity, () -> (double) this.motors.size(), "threads"));
}
private void unregisterMetrics() {
ActivityMetrics.unregister(this.threadsGauge);
@@ -485,7 +486,6 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
} finally {
logger.trace(() -> "finally shutting down activity " + this.getActivity().getAlias());
this.stoppedAt = System.currentTimeMillis();
unregisterMetrics();
activity.setRunState(RunState.Stopped);
}

83
engine-core/src/test/java/io/nosqlbench/engine/core/ActivityExecutorTest.java
View File

@@ -16,8 +16,8 @@
package io.nosqlbench.engine.core;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.engine.api.activityapi.core.*;
import io.nosqlbench.engine.api.activityapi.input.Input;
import io.nosqlbench.engine.api.activityapi.input.InputDispenser;
@@ -37,11 +37,9 @@ import org.apache.logging.log4j.Logger;
import org.junit.jupiter.api.Test;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
class ActivityExecutorTest {
@@ -85,41 +83,41 @@ class ActivityExecutorTest {
//
// }
@Test
synchronized void testDelayedStartSanity() {
ActivityDef activityDef = ActivityDef.parseActivityDef("driver=diag;alias=test-delayed-start;cycles=1000;initdelay=2000;");
new ActivityTypeLoader().load(activityDef, NBLabeledElement.EMPTY);
Activity activity = new DelayedInitActivity(activityDef);
final InputDispenser inputDispenser = new CoreInputDispenser(activity);
final ActionDispenser actionDispenser = new CoreActionDispenser(activity);
final OutputDispenser outputDispenser = CoreServices.getOutputDispenser(activity).orElse(null);
MotorDispenser<?> motorDispenser = new CoreMotorDispenser(activity, inputDispenser, actionDispenser, outputDispenser);
activity.setActionDispenserDelegate(actionDispenser);
activity.setOutputDispenserDelegate(outputDispenser);
activity.setInputDispenserDelegate(inputDispenser);
activity.setMotorDispenserDelegate(motorDispenser);
ActivityExecutor activityExecutor = new ActivityExecutor(activity, "test-delayed-start");
ExecutorService testExecutor = Executors.newCachedThreadPool();
Future<ExecutionResult> future = testExecutor.submit(activityExecutor);
try {
activityDef.setThreads(1);
activityExecutor.startActivity();
future.get();
testExecutor.shutdownNow();
} catch (final Exception e) {
fail("Unexpected exception", e);
}
assertThat(inputDispenser.getInput(10).getInputSegment(3)).isNull();
}
// @Test
// synchronized void testDelayedStartSanity() {
//
// ActivityDef activityDef = ActivityDef.parseActivityDef("driver=diag;alias=test-delayed-start;cycles=1000;initdelay=2000;");
// new ActivityTypeLoader().load(activityDef, NBLabeledElement.EMPTY);
//
// Activity activity = new DelayedInitActivity(activityDef);
// final InputDispenser inputDispenser = new CoreInputDispenser(activity);
// final ActionDispenser actionDispenser = new CoreActionDispenser(activity);
// final OutputDispenser outputDispenser = CoreServices.getOutputDispenser(activity).orElse(null);
//
// MotorDispenser<?> motorDispenser = new CoreMotorDispenser(activity, inputDispenser, actionDispenser, outputDispenser);
// activity.setActionDispenserDelegate(actionDispenser);
// activity.setOutputDispenserDelegate(outputDispenser);
// activity.setInputDispenserDelegate(inputDispenser);
// activity.setMotorDispenserDelegate(motorDispenser);
//
// ActivityExecutor activityExecutor = new ActivityExecutor(activity, "test-delayed-start");
//
// ExecutorService testExecutor = Executors.newCachedThreadPool();
// Future<ExecutionResult> future = testExecutor.submit(activityExecutor);
//
//
// try {
// activityDef.setThreads(1);
// activityExecutor.startActivity();
// future.get();
// testExecutor.shutdownNow();
//
// } catch (final Exception e) {
// fail("Unexpected exception", e);
// }
//
// assertThat(inputDispenser.getInput(10).getInputSegment(3)).isNull();
// }
@Test
synchronized void testNewActivityExecutor() {
@@ -144,7 +142,9 @@ class ActivityExecutorTest {
ActivityExecutor activityExecutor = new ActivityExecutor(simpleActivity, "test-new-executor");
activityDef.setThreads(5);
activityExecutor.startActivity();
ForkJoinTask<ExecutionResult> executionResultForkJoinTask = ForkJoinPool.commonPool().submit(activityExecutor);
// activityExecutor.startActivity();
final int[] speeds = {1, 50, 5, 50, 2, 50};
for (int offset = 0; offset < speeds.length; offset += 2) {
@@ -160,6 +160,7 @@ class ActivityExecutorTest {
fail("Not expecting exception", e);
}
}
executionResultForkJoinTask.cancel(true);
// Used for slowing the roll due to state transitions in test.
try {