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

more merge fixes

Browse Source
This commit is contained in:
Jonathan Shook
2023-09-29 18:58:32 -05:00
parent 02f69d0038
commit c0a3ec2509
14 changed files with 157 additions and 132 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/ActivityLoader.java
View File

@@ -16,7 +16,7 @@
package io.nosqlbench.engine.core.lifecycle.activity;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.api.config.NBComponent;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.engine.api.activityapi.core.Activity;
import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivityType;
@@ -41,9 +41,9 @@ public class ActivityLoader {
this.scenario = scenario;
}
public synchronized Activity loadActivity(ActivityDef activityDef, final NBLabeledElement labels) {
public synchronized Activity loadActivity(ActivityDef activityDef, final NBComponent parent) {
activityDef= activityDef.deprecate("yaml","workload").deprecate("type","driver");
final Activity activity = new StandardActivityType(activityDef, labels).getAssembledActivity(activityDef, this.activityMap, labels);
final Activity activity = new StandardActivityType(activityDef, parent).getAssembledActivity(activityDef, this.activityMap, parent);
this.activityMap.put(activity.getAlias(),activity);
ActivityLoader.logger.debug("Resolved activity for alias '{}'", activityDef.getAlias());
return activity;

20
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/activity/ActivityTypeLoader.java
View File

@@ -16,17 +16,17 @@
package io.nosqlbench.engine.core.lifecycle.activity;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.engine.api.activityapi.core.ActivityType;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.adapters.api.activityimpl.uniform.DriverAdapter;
import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivityType;
import io.nosqlbench.nb.annotations.Maturity;
import io.nosqlbench.api.system.NBEnvironment;
import io.nosqlbench.api.config.NBComponent;
import io.nosqlbench.api.content.Content;
import io.nosqlbench.api.content.NBIO;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.api.errors.BasicError;
import io.nosqlbench.api.spi.SimpleServiceLoader;
import io.nosqlbench.api.system.NBEnvironment;
import io.nosqlbench.engine.api.activityapi.core.ActivityType;
import io.nosqlbench.engine.api.activityimpl.uniform.StandardActivityType;
import io.nosqlbench.nb.annotations.Maturity;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -118,7 +118,7 @@ public class ActivityTypeLoader {
return urlsToAdd;
}
public Optional<ActivityType> load(final ActivityDef activityDef, final NBLabeledElement labels) {
public Optional<ActivityType> load(final ActivityDef activityDef, final NBComponent parent) {
String driverName = activityDef.getParams()
.getOptionalString("driver", "type")
@@ -135,18 +135,18 @@ public class ActivityTypeLoader {
})
.ifPresent(this::extendClassLoader);
return getDriverAdapter(driverName,activityDef,labels)
return getDriverAdapter(driverName,activityDef,parent)
.or(() -> this.ACTIVITYTYPE_SPI_FINDER.getOptionally(driverName));
}
private Optional<ActivityType> getDriverAdapter(final String activityTypeName, final ActivityDef activityDef, final NBLabeledElement labels) {
private Optional<ActivityType> getDriverAdapter(final String activityTypeName, final ActivityDef activityDef, final NBComponent parent) {
final Optional<DriverAdapter> oda = this.DRIVERADAPTER_SPI_FINDER.getOptionally(activityTypeName);
if (oda.isPresent()) {
final DriverAdapter<?, ?> driverAdapter = oda.get();
final ActivityType activityType = new StandardActivityType<>(driverAdapter, activityDef, labels);
final ActivityType activityType = new StandardActivityType<>(driverAdapter, activityDef, parent);
return Optional.of(activityType);
}
return Optional.empty();

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

@@ -74,7 +74,7 @@ public class ScenarioController implements NBLabeledElement {
private synchronized ActivityRuntimeInfo doStartActivity(ActivityDef activityDef) {
if (!this.activityInfoMap.containsKey(activityDef.getAlias())) {
Activity activity = this.activityLoader.loadActivity(activityDef, this);
Activity activity = this.activityLoader.loadActivity(activityDef, scenario);
ActivityExecutor executor = new ActivityExecutor(activity, this.scenario.getScenarioName());
Future<ExecutionResult> startedActivity = activitiesExecutor.submit(executor);
ActivityRuntimeInfo activityRuntimeInfo = new ActivityRuntimeInfo(activity, startedActivity, executor);

14
engine-core/src/main/java/io/nosqlbench/engine/core/lifecycle/scenario/script/MetricsMapper.java
View File

@@ -16,20 +16,20 @@
package io.nosqlbench.engine.core.lifecycle.scenario.script;
import com.codahale.metrics.*;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.engine.api.activityapi.core.Activity;
import io.nosqlbench.engine.api.activityapi.core.ActivityType;
import io.nosqlbench.api.config.standard.TestComponent;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.api.engine.metrics.ActivityMetrics;
import io.nosqlbench.engine.api.activityapi.core.Activity;
import io.nosqlbench.engine.api.activityapi.core.ActivityType;
import io.nosqlbench.engine.core.lifecycle.activity.ActivityTypeLoader;
import io.nosqlbench.engine.core.lifecycle.scenario.script.bindings.PolyglotMetricRegistryBindings;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.lang.reflect.Method;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.*;
import java.util.Timer;
import java.util.Map.Entry;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -68,11 +68,11 @@ public enum MetricsMapper {
final ActivityDef activityDef = ActivityDef.parseActivityDef(activitySpec);
MetricsMapper.logger.info(() -> "introspecting metric names for " + activitySpec);
final Optional<ActivityType> activityType = new ActivityTypeLoader().load(activityDef, NBLabeledElement.EMPTY);
final Optional<ActivityType> activityType = new ActivityTypeLoader().load(activityDef, TestComponent.INSTANCE);
if (!activityType.isPresent())
throw new RuntimeException("Activity type '" + activityDef.getActivityType() + "' does not exist in this runtime.");
final Activity activity = activityType.get().getAssembledActivity(activityDef, new HashMap<>(), NBLabeledElement.EMPTY);
final Activity activity = activityType.get().getAssembledActivity(activityDef, new HashMap<>(), TestComponent.INSTANCE);
final PolyglotMetricRegistryBindings nashornMetricRegistryBindings = new PolyglotMetricRegistryBindings(ActivityMetrics.getMetricRegistry());
activity.initActivity();
activity.getInputDispenserDelegate().getInput(0);

12
engine-core/src/main/java/io/nosqlbench/engine/core/metadata/MarkdownFinder.java
View File

@@ -16,14 +16,14 @@
package io.nosqlbench.engine.core.metadata;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.engine.api.activityapi.core.ActivityType;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.engine.core.lifecycle.activity.ActivityTypeLoader;
import io.nosqlbench.nb.annotations.Service;
import io.nosqlbench.api.config.standard.TestComponent;
import io.nosqlbench.api.content.Content;
import io.nosqlbench.api.content.NBIO;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.api.errors.BasicError;
import io.nosqlbench.engine.api.activityapi.core.ActivityType;
import io.nosqlbench.engine.core.lifecycle.activity.ActivityTypeLoader;
import io.nosqlbench.nb.annotations.Service;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -64,7 +64,7 @@ public class MarkdownFinder {
}
public String forActivityInstance(final String s) {
final ActivityType activityType = new ActivityTypeLoader().load(ActivityDef.parseActivityDef("driver="+s), NBLabeledElement.EMPTY).orElseThrow(
final ActivityType activityType = new ActivityTypeLoader().load(ActivityDef.parseActivityDef("driver="+s), TestComponent.INSTANCE).orElseThrow(
() -> new BasicError("Unable to find driver for '" + s + '\'')
);
return this.forResourceMarkdown(activityType.getClass().getAnnotation(Service.class)

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

@@ -18,16 +18,16 @@ package io.nosqlbench.engine.core;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.engine.api.activityapi.core.ActionDispenser;
import io.nosqlbench.engine.api.activityapi.core.Activity;
import io.nosqlbench.engine.api.activityapi.core.MotorDispenser;
import io.nosqlbench.engine.api.activityapi.core.SyncAction;
import io.nosqlbench.engine.api.activityapi.core.*;
import io.nosqlbench.engine.api.activityapi.input.Input;
import io.nosqlbench.engine.api.activityapi.input.InputDispenser;
import io.nosqlbench.engine.api.activityapi.output.OutputDispenser;
import io.nosqlbench.engine.api.activityimpl.CoreServices;
import io.nosqlbench.engine.api.activityimpl.SimpleActivity;
import io.nosqlbench.engine.api.activityimpl.action.CoreActionDispenser;
import io.nosqlbench.engine.api.activityimpl.input.AtomicInput;
import io.nosqlbench.engine.api.activityimpl.input.CoreInputDispenser;
import io.nosqlbench.engine.api.activityimpl.motor.CoreMotor;
import io.nosqlbench.engine.api.activityimpl.motor.CoreMotorDispenser;
import io.nosqlbench.engine.core.lifecycle.ExecutionResult;
import io.nosqlbench.engine.core.lifecycle.activity.ActivityExecutor;
@@ -36,9 +36,22 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.junit.jupiter.api.Test;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.Future;
import io.nosqlbench.api.config.standard.TestComponent;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.engine.api.activityapi.core.ActionDispenser;
import io.nosqlbench.engine.api.activityapi.core.Activity;
import io.nosqlbench.engine.api.activityapi.core.MotorDispenser;
import io.nosqlbench.engine.api.activityapi.core.SyncAction;
import java.util.Map;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.ForkJoinTask;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import static org.assertj.core.api.Assertions.fail;
@@ -83,49 +96,49 @@ 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, TestComponent.INSTANCE);
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() {
final ActivityDef activityDef = ActivityDef.parseActivityDef("driver=diag;alias=test-dynamic-params;cycles=1000;initdelay=5000;");
new ActivityTypeLoader().load(activityDef,NBLabeledElement.EMPTY);
new ActivityTypeLoader().load(activityDef,TestComponent.INSTANCE);
Activity simpleActivity = new SimpleActivity(activityDef, NBLabeledElement.forMap(Map.of()));
Activity simpleActivity = new SimpleActivity(TestComponent.INSTANCE,activityDef);
// this.getActivityMotorFactory(this.motorActionDelay(999), new AtomicInput(simpleActivity,activityDef));
@@ -171,6 +184,18 @@ class ActivityExecutorTest {
}
}
private MotorDispenser<?> getActivityMotorFactory(final Action lc, Input ls) {
return new MotorDispenser<>() {
@Override
public Motor getMotor(final ActivityDef activityDef, final int slotId) {
final Activity activity = new SimpleActivity(TestComponent.INSTANCE,activityDef);
final Motor<?> cm = new CoreMotor<>(activity, slotId, ls);
cm.setAction(lc);
return cm;
}
};
}
private SyncAction motorActionDelay(long delay) {
return new SyncAction() {
@Override
@@ -190,7 +215,7 @@ class ActivityExecutorTest {
private static final Logger logger = LogManager.getLogger(DelayedInitActivity.class);
public DelayedInitActivity(final ActivityDef activityDef) {
super(activityDef, NBLabeledElement.EMPTY);
super(TestComponent.INSTANCE,activityDef);
}
@Override

31
engine-core/src/test/java/io/nosqlbench/engine/core/CoreMotorTest.java
View File

@@ -16,7 +16,7 @@
package io.nosqlbench.engine.core;
import io.nosqlbench.api.labels.NBLabeledElement;
import io.nosqlbench.api.config.standard.TestComponent;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.engine.api.activityapi.core.Action;
import io.nosqlbench.engine.api.activityapi.core.Activity;
@@ -27,7 +27,6 @@ import io.nosqlbench.engine.api.activityimpl.motor.CoreMotor;
import io.nosqlbench.engine.core.fortesting.BlockingSegmentInput;
import org.junit.jupiter.api.Test;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicLongArray;
import java.util.function.Predicate;
@@ -39,9 +38,9 @@ public class CoreMotorTest {
@Test
public void testBasicActivityMotor() {
final Activity activity = new SimpleActivity(
ActivityDef.parseActivityDef("alias=foo"),
NBLabeledElement.forMap(Map.of("testing","coremotor"))
);
new TestComponent("testing", "coremotor"),
ActivityDef.parseActivityDef("alias=foo")
);
final BlockingSegmentInput lockstepper = new BlockingSegmentInput();
final Motor cm = new CoreMotor(activity, 5L, lockstepper);
final AtomicLong observableAction = new AtomicLong(-3L);
@@ -51,18 +50,19 @@ public class CoreMotorTest {
t.start();
try {
Thread.sleep(1000); // allow action time to be waiting in monitor for test fixture
} catch (final InterruptedException ignored) {}
} catch (final InterruptedException ignored) {
}
lockstepper.publishSegment(5L);
final boolean result = this.awaitCondition(atomicInteger -> 5L == atomicInteger.get(),observableAction,5000,100);
final boolean result = this.awaitCondition(atomicInteger -> 5L == atomicInteger.get(), observableAction, 5000, 100);
assertThat(observableAction.get()).isEqualTo(5L);
}
@Test
public void testIteratorStride() {
SimpleActivity activity = new SimpleActivity("stride=3", NBLabeledElement.EMPTY);
SimpleActivity activity = new SimpleActivity(TestComponent.INSTANCE, "stride=3");
final BlockingSegmentInput lockstepper = new BlockingSegmentInput();
final Motor cm1 = new CoreMotor(activity,1L, lockstepper);
final Motor cm1 = new CoreMotor(activity, 1L, lockstepper);
final AtomicLongArray ary = new AtomicLongArray(10);
final Action a1 = this.getTestArrayConsumer(ary);
cm1.setAction(a1);
@@ -72,11 +72,12 @@ public class CoreMotorTest {
t1.start();
try {
Thread.sleep(500); // allow action time to be waiting in monitor for test fixture
} catch (final InterruptedException ignored) {}
} catch (final InterruptedException ignored) {
}
lockstepper.publishSegment(11L,12L,13L);
lockstepper.publishSegment(11L, 12L, 13L);
final boolean result = this.awaitAryCondition(ala -> 13L == ala.get(2),ary,5000,100);
final boolean result = this.awaitAryCondition(ala -> 13L == ala.get(2), ary, 5000, 100);
assertThat(ary.get(0)).isEqualTo(11L);
assertThat(ary.get(1)).isEqualTo(12L);
assertThat(ary.get(2)).isEqualTo(13L);
@@ -87,6 +88,7 @@ public class CoreMotorTest {
private SyncAction getTestArrayConsumer(AtomicLongArray ary) {
return new SyncAction() {
private int offset;
@Override
public int runCycle(final long cycle) {
ary.set(this.offset, cycle);
@@ -95,6 +97,7 @@ public class CoreMotorTest {
}
};
}
private SyncAction getTestConsumer(AtomicLong atomicLong) {
return new SyncAction() {
@Override
@@ -108,7 +111,7 @@ public class CoreMotorTest {
private boolean awaitAryCondition(final Predicate<AtomicLongArray> atomicLongAryPredicate, final AtomicLongArray ary, final long millis, final long retry) {
final long start = System.currentTimeMillis();
long now=start;
long now = start;
while (now < (start + millis)) {
final boolean result = atomicLongAryPredicate.test(ary);
if (result) return true;
@@ -123,7 +126,7 @@ public class CoreMotorTest {
private boolean awaitCondition(final Predicate<AtomicLong> atomicPredicate, final AtomicLong atomicInteger, final long millis, final long retry) {
final long start = System.currentTimeMillis();
long now=start;
long now = start;
while (now < (start + millis)) {
final boolean result = atomicPredicate.test(atomicInteger);
if (result) return true;