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

incremental progress

Browse Source
This commit is contained in:
Jonathan Shook
2023-10-09 11:59:34 -05:00
parent 9fa711b7ab
commit 9e91a6201d
44 changed files with 710 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivitiesExceptionHandler.java
View File

@@ -16,7 +16,7 @@
package io.nosqlbench.engine.core.lifecycle.activity;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ScenarioActivitiesController;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -24,9 +24,9 @@ public class ActivitiesExceptionHandler implements Thread.UncaughtExceptionHandl
private static final Logger logger = LogManager.getLogger(ActivitiesExceptionHandler.class);
private final ActivitiesController controller;
private final ScenarioActivitiesController controller;
public ActivitiesExceptionHandler(ActivitiesController controller) {
public ActivitiesExceptionHandler(ScenarioActivitiesController controller) {
this.controller = controller;
logger.debug(() -> "Activities exception handler starting up for executor '" + this.controller + "'");
}

6
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivitiesProgressIndicator.java
View File

@@ -22,7 +22,7 @@ import io.nosqlbench.engine.api.activityapi.core.progress.StateCapable;
import io.nosqlbench.engine.api.metrics.IndicatorMode;
import io.nosqlbench.api.engine.metrics.PeriodicRunnable;
import io.nosqlbench.api.engine.util.Unit;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ScenarioActivitiesController;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -35,14 +35,14 @@ public class ActivitiesProgressIndicator implements Runnable {
private final static Logger logger = LogManager.getLogger("PROGRESS");
private final String indicatorSpec;
private final ActivitiesController sc;
private final ScenarioActivitiesController sc;
private PeriodicRunnable<ActivitiesProgressIndicator> runnable;
private IndicatorMode indicatorMode = IndicatorMode.console;
private final Set<String> seen = new HashSet<>();
private long intervalMillis = 1L;
public ActivitiesProgressIndicator(ActivitiesController sc, String indicatorSpec) {
public ActivitiesProgressIndicator(ScenarioActivitiesController sc, String indicatorSpec) {
this.sc = sc;
this.indicatorSpec = indicatorSpec;
start();

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

@@ -57,7 +57,7 @@ import java.util.stream.Collectors;
* This allows the state tracking to work consistently for all observers.</p>
*/
public class ActivityExecutor implements NBLabeledElement, ActivityController, ParameterMap.Listener, ProgressCapable, Callable<ExecutionResult> {
public class ActivityExecutor implements NBLabeledElement, ParameterMap.Listener, ProgressCapable, Callable<ExecutionResult> {
// TODO Encapsulate valid state transitions to be only modifiable within the appropriate type view.
@@ -81,7 +81,6 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
this.activity = activity;
this.activityDef = activity.getActivityDef();
activity.getActivityDef().getParams().addListener(this);
activity.setActivityController(this);
this.tally = activity.getRunStateTally();
}
@@ -257,16 +256,24 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
private void increaseActiveMotorCountUpToThreadParam(ActivityDef activityDef) {
// Create motor slots
while (motors.size() < activityDef.getThreads()) {
Motor motor = activity.getMotorDispenserDelegate().getMotor(activityDef, motors.size());
logger.trace(() -> "Starting cycle motor thread:" + motor);
motors.add(motor);
try {
while (motors.size() < activityDef.getThreads()) {
Motor motor = activity.getMotorDispenserDelegate().getMotor(activityDef, motors.size());
logger.trace(() -> "Starting cycle motor thread:" + motor);
motors.add(motor);
}
} catch (Exception e) {
System.out.print("critical error while starting motors: " + e);
logger.error("critical error while starting motors:" + e,e);
throw new RuntimeException(e);
}
}
private void reduceActiveMotorCountDownToThreadParam(ActivityDef activityDef) {
// Stop and remove extra motor slots
if (activityDef.getThreads()==0) {
logger.warn("setting threads to zero is not advised. At least one thread has to be active to keep the activity alive.");
}
while (motors.size() > activityDef.getThreads()) {
Motor motor = motors.get(motors.size() - 1);
logger.trace(() -> "Stopping cycle motor thread:" + motor);
@@ -364,29 +371,6 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
this.requestStopMotors();
}
@Override
public synchronized void stopActivityWithReasonAsync(String reason) {
logger.info(() -> "Stopping activity " + this.activityDef.getAlias() + ": " + reason);
this.exception = new RuntimeException("Stopping activity " + this.activityDef.getAlias() + ": " + reason);
logger.error("stopping with reason: " + exception);
requestStopMotors();
}
@Override
public synchronized void stopActivityWithErrorAsync(Throwable throwable) {
if (exception == null) {
this.exception = new RuntimeException(throwable);
logger.error("stopping on error: " + throwable.toString(), throwable);
} else {
if (activityDef.getParams().getOptionalBoolean("fullerrors").orElse(false)) {
logger.error("additional error: " + throwable.toString(), throwable);
} else {
logger.warn("summarized error (fullerrors=false): " + throwable.toString());
}
}
requestStopMotors();
}
@Override
public ProgressMeterDisplay getProgressMeter() {
return this.activity.getProgressMeter();
@@ -589,6 +573,10 @@ public class ActivityExecutor implements NBLabeledElement, ActivityController, P
.build());
}
public void awaitMotorsRunningOrTerminalState() {
awaitMotorsAtLeastRunning();
}
private class ThreadsGauge implements Gauge<Double> {
public ThreadsGauge(ActivityExecutor activityExecutor) {
ActivityExecutor ae = activityExecutor;

6
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/ActivityBindings.java
View File

@@ -28,11 +28,11 @@ import java.util.stream.Collectors;
*/
public class ActivityBindings implements Bindings, ProxyObject {
private final ActivitiesController scenario;
private final ScenarioActivitiesController scenario;
private final Map<String, Bindings> elementMap = new HashMap<String, Bindings>();
public ActivityBindings(ActivitiesController activitiesController) {
this.scenario = activitiesController;
public ActivityBindings(ScenarioActivitiesController scenarioActivitiesController) {
this.scenario = scenarioActivitiesController;
}
@Override

8
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/NBDefaultSceneFixtures.java
View File

@@ -56,7 +56,7 @@ public class NBDefaultSceneFixtures implements NBSceneFixtures {
* a given scenario. A scenario doesn't complete unless until all activities
* are complete or errored.
*/
private ActivitiesController controller;
private ScenarioActivitiesController controller;
/*
* Extensions provide additional scripting capabilities which are not provided by the
* scripting or other runtimes, or new ways of tapping into extant features.
@@ -68,7 +68,7 @@ public class NBDefaultSceneFixtures implements NBSceneFixtures {
private Reader in;
public NBDefaultSceneFixtures(ScenarioParams params, NBComponent parent, ActivitiesController controller, Extensions extensions, PrintWriter out, PrintWriter err, Reader in) {
public NBDefaultSceneFixtures(ScenarioParams params, NBComponent parent, ScenarioActivitiesController controller, Extensions extensions, PrintWriter out, PrintWriter err, Reader in) {
this.params = params;
this.session = parent;
this.controller = controller;
@@ -84,7 +84,7 @@ public class NBDefaultSceneFixtures implements NBSceneFixtures {
new NBSession(
new TestComponent("scene", name), "scene~"+name
),
new ActivitiesController(),
new ScenarioActivitiesController(),
Extensions.ofNone(),
new PrintWriter(System.out),
new PrintWriter(System.err),
@@ -104,7 +104,7 @@ public class NBDefaultSceneFixtures implements NBSceneFixtures {
}
@Override
public ActivitiesController controller() {
public ScenarioActivitiesController controller() {
return controller;
}

2
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/NBSceneBuffer.java
View File

@@ -80,7 +80,7 @@ public class NBSceneBuffer implements NBSceneFixtures {
}
@Override
public ActivitiesController controller() {
public ScenarioActivitiesController controller() {
return fixtures.controller();
}

2
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/NBSceneFixtures.java
View File

@@ -27,7 +27,7 @@ public interface NBSceneFixtures {
NBComponent component();
ActivitiesController controller();
ScenarioActivitiesController controller();
Extensions extensions();

23
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/ActivitiesController.java → engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/ScenarioActivitiesController.java
View File

@@ -41,31 +41,31 @@ import java.util.stream.Collectors;
* A ScenarioController provides a way to start Activities,
* modify them while running, and forceStopMotors, pause or restart them.
*/
public class ActivitiesController extends NBBaseComponent {
public class ScenarioActivitiesController extends NBBaseComponent {
private static final Logger logger = LogManager.getLogger(ActivitiesController.class);
private static final Logger logger = LogManager.getLogger(ScenarioActivitiesController.class);
private static final Logger scenariologger = LogManager.getLogger("SCENARIO");
private final ActivityLoader activityLoader;
private final Map<String, ActivityRuntimeInfo> activityInfoMap = new ConcurrentHashMap<>();
private final ExecutorService activitiesExecutor;
private final ExecutorService executorService;
public ActivitiesController() {
public ScenarioActivitiesController() {
super(new TestComponent("test","test"));
this.activityLoader = new ActivityLoader();
ActivitiesExceptionHandler exceptionHandler = new ActivitiesExceptionHandler(this);
IndexedThreadFactory indexedThreadFactory = new IndexedThreadFactory("ACTIVITY", exceptionHandler);
this.activitiesExecutor = Executors.newVirtualThreadPerTaskExecutor();
this.executorService = Executors.newVirtualThreadPerTaskExecutor();
}
public ActivitiesController(NBComponent parent) {
public ScenarioActivitiesController(NBComponent parent) {
super(parent);
this.activityLoader = new ActivityLoader();
ActivitiesExceptionHandler exceptionHandler = new ActivitiesExceptionHandler(this);
IndexedThreadFactory indexedThreadFactory = new IndexedThreadFactory("ACTIVITY", exceptionHandler);
this.activitiesExecutor = Executors.newCachedThreadPool(indexedThreadFactory);
this.executorService = Executors.newVirtualThreadPerTaskExecutor();
}
/**
@@ -84,8 +84,9 @@ public class ActivitiesController extends NBBaseComponent {
if (!this.activityInfoMap.containsKey(activityDef.getAlias())) {
Activity activity = this.activityLoader.loadActivity(activityDef, this);
ActivityExecutor executor = new ActivityExecutor(activity);
Future<ExecutionResult> startedActivity = activitiesExecutor.submit(executor);
Future<ExecutionResult> startedActivity = executorService.submit(executor);
ActivityRuntimeInfo activityRuntimeInfo = new ActivityRuntimeInfo(activity, startedActivity, executor);
activityRuntimeInfo.getActivityExecutor().awaitMotorsRunningOrTerminalState();
this.activityInfoMap.put(activity.getAlias(), activityRuntimeInfo);
}
@@ -181,6 +182,10 @@ public class ActivitiesController extends NBBaseComponent {
runtimeInfo.stopActivity();
}
public synchronized void stop(Activity activity) {
stop(activity.getActivityDef());
}
/**
* <p>Stop an activity, given an activity def map. The only part of the map that is important is the
* alias parameter. This method retains the map signature to provide convenience for scripting.</p>
@@ -452,7 +457,7 @@ public class ActivitiesController extends NBBaseComponent {
public void shutdown() {
logger.debug(() -> "Requesting ScenarioController shutdown.");
this.activitiesExecutor.shutdownNow();
this.executorService.shutdownNow();
// try {
// if (!this.activitiesExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
// logger.info(() -> "Scenario is being forced to shutdown after waiting 5 seconds for graceful shutdown.");

20
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/SceneBuilder.java
View File

@@ -2,13 +2,13 @@ package io.nosqlbench.engine.core.lifecycle.scenario.context;
/*
* Copyright (c) 2022 nosqlbench
*
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -21,17 +21,19 @@ package io.nosqlbench.engine.core.lifecycle.scenario.context;
import io.nosqlbench.components.NBComponent;
import io.nosqlbench.engine.core.lifecycle.scenario.execution.Extensions;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.io.Reader;
import java.util.Map;
public class SceneBuilder implements SceneBuilderFacets.ALL {
private Map<String,String> params;
private ActivitiesController controller;
private ScenarioActivitiesController controller;
private Extensions extensions;
private PrintWriter out;
private PrintWriter err;
private Reader in;
private PrintWriter out = new PrintWriter(System.out);
private PrintWriter err = new PrintWriter(System.err);
private Reader in = new InputStreamReader(System.in);
private NBComponent component;
private NBSceneBuffer.IOType iotype;
@@ -47,7 +49,7 @@ public class SceneBuilder implements SceneBuilderFacets.ALL {
new NBDefaultSceneFixtures(
ScenarioParams.of(this.params),
this.component,
((this.controller!=null) ? this.controller : new ActivitiesController(component)),
((this.controller!=null) ? this.controller : new ScenarioActivitiesController(component)),
this.extensions,
this.out,
this.err,
@@ -57,7 +59,7 @@ public class SceneBuilder implements SceneBuilderFacets.ALL {
@Override
public SceneBuilder controller(ActivitiesController controller) {
public SceneBuilder controller(ScenarioActivitiesController controller) {
this.controller = controller;
return this;
}

8
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/context/SceneBuilderFacets.java
View File

@@ -2,13 +2,13 @@ package io.nosqlbench.engine.core.lifecycle.scenario.context;
/*
* Copyright (c) 2022 nosqlbench
*
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*
* http://www.apache.org/licenses/LICENSE-2.0
*
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -44,7 +44,7 @@ public interface SceneBuilderFacets {
}
public interface WantsController extends WantsStdin, WantsIoType {
public WantsStdin controller(ActivitiesController controller);
public WantsStdin controller(ScenarioActivitiesController controller);
}

6
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/direct/SCBaseScenario.java
View File

@@ -17,7 +17,7 @@
package io.nosqlbench.engine.core.lifecycle.scenario.direct;
import io.nosqlbench.components.NBComponent;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ScenarioActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.NBSceneFixtures;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ScenarioParams;
import io.nosqlbench.engine.core.lifecycle.scenario.execution.Extensions;
@@ -32,7 +32,7 @@ public abstract class SCBaseScenario extends NBScenario {
protected Reader stdin;
protected PrintWriter stdout;
protected Writer stderr;
protected ActivitiesController controller;
protected ScenarioActivitiesController controller;
protected ScenarioParams params;
protected Extensions extensions;
@@ -60,7 +60,7 @@ public abstract class SCBaseScenario extends NBScenario {
* <LI>component, an {@link NBComponent} - The NB component upon which all metrics or other services are attached.</LI>
* <LI>stdin - a {@link Reader} representing the input buffer which would normally be {@link System#in}
* <LI>stdout, stderr</LI>- a {@link PrintWriter}; This can be buffered virtually, attached to {@link System#out} and {@link System#err} or both for IO tracing.</LI>
* <LI>controller - A dedicated {@link ActivitiesController} which can be used to define, start, top, and interact with activities.</LI>
* <LI>controller - A dedicated {@link ScenarioActivitiesController} which can be used to define, start, top, and interact with activities.</LI>
* <LI>params - The {@link ScenarioParams} which have been passed to this scenario.</LI>
* <LI>extensions - A dedicated ahndle to the {@link Extensions} service.</LI>
* <LI><EM>all component services</EM> as this scenario IS a component. This includes all implemented methods in any of the {@link NBComponent} sub-interfaces.</EM>

22
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/execution/NBScenario.java
View File

@@ -26,7 +26,7 @@ import io.nosqlbench.components.NBBaseComponent;
import io.nosqlbench.components.NBComponentErrorHandler;
import io.nosqlbench.engine.core.annotation.Annotators;
import io.nosqlbench.engine.core.lifecycle.activity.ActivitiesProgressIndicator;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ScenarioActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.NBSceneBuffer;
import io.nosqlbench.engine.core.lifecycle.scenario.context.NBSceneFixtures;
import io.nosqlbench.engine.core.lifecycle.scenario.script.NBScriptedScenario;
@@ -54,7 +54,7 @@ public abstract class NBScenario extends NBBaseComponent
private ScenarioMetadata scenarioMetadata;
private ActivitiesController activitiesController;
private ScenarioActivitiesController scenarioActivitiesController;
private Exception error;
private String progressInterval = "console:10s";
private ActivitiesProgressIndicator activitiesProgressIndicator;
@@ -68,15 +68,15 @@ public abstract class NBScenario extends NBBaseComponent
}
public void forceStopScenario(int i, boolean b) {
activitiesController.forceStopScenario(i,b);
scenarioActivitiesController.forceStopScenario(i,b);
}
// public Map<String, String> getParams() {
// return this.params;
// }
public ActivitiesController getActivitiesController() {
return this.activitiesController;
public ScenarioActivitiesController getActivitiesController() {
return this.scenarioActivitiesController;
}
public enum State {
@@ -111,7 +111,7 @@ public abstract class NBScenario extends NBBaseComponent
*/
@Override
public final ScenarioResult apply(NBSceneBuffer sctx) {
this.activitiesController=sctx.controller();
this.scenarioActivitiesController =sctx.controller();
this.scenarioShutdownHook = new ScenarioShutdownHook(this);
Runtime.getRuntime().addShutdownHook(this.scenarioShutdownHook);
@@ -127,24 +127,24 @@ public abstract class NBScenario extends NBBaseComponent
);
if (!"disabled".equals(progressInterval) && progressInterval!=null && !progressInterval.isEmpty())
this.activitiesProgressIndicator = new ActivitiesProgressIndicator(activitiesController, this.progressInterval);
this.activitiesProgressIndicator = new ActivitiesProgressIndicator(scenarioActivitiesController, this.progressInterval);
ScenarioResult result = null;
try {
runScenario(sctx.asFixtures());
final long awaitCompletionTime = 86400 * 365 * 1000L;
this.logger.debug("Awaiting completion of scenario and activities for {} millis.", awaitCompletionTime);
this.activitiesController.awaitCompletion(awaitCompletionTime);
this.scenarioActivitiesController.awaitCompletion(awaitCompletionTime);
} catch (Exception e) {
try {
activitiesController.forceStopScenario(3000, false);
scenarioActivitiesController.forceStopScenario(3000, false);
} catch (final Exception eInner) {
this.logger.debug("Found inner exception while forcing stop with rethrow=false: {}", eInner);
throw new RuntimeException(e);
}
this.error = e;
} finally {
this.activitiesController.shutdown();
this.scenarioActivitiesController.shutdown();
this.endedAtMillis = System.currentTimeMillis();
result = new ScenarioResult(
sctx,
@@ -183,7 +183,7 @@ public abstract class NBScenario extends NBBaseComponent
} else
this.logger.info(
"Scenario completed successfully, with {} logical activities.",
activitiesController.getActivityExecutorMap().size()
scenarioActivitiesController.getActivityExecutorMap().size()
);
this.logger.info(() -> "scenario state: " + state);

6
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/script/bindings/PolyglotScenarioController.java
View File

@@ -17,7 +17,7 @@
package io.nosqlbench.engine.core.lifecycle.scenario.script.bindings;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ActivitiesController;
import io.nosqlbench.engine.core.lifecycle.scenario.context.ScenarioActivitiesController;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.graalvm.polyglot.Value;
@@ -29,9 +29,9 @@ public class PolyglotScenarioController {
private static final Logger logger = LogManager.getLogger("SCENARIO/POLYGLOT");
private final ActivitiesController controller;
private final ScenarioActivitiesController controller;
public PolyglotScenarioController(ActivitiesController inner) {
public PolyglotScenarioController(ScenarioActivitiesController inner) {
this.controller = inner;
}