IntenseFinance/nosqlbench
3
0
Fork 0
mirror of https://github.com/nosqlbench/nosqlbench.git synced 2026-09-03 20:53:07 -05:00
Code Issues Packages Projects Releases Wiki Activity

implement closeable spaces via decorator interface

Browse Source
This commit is contained in:
Jonathan Shook
2022-11-18 23:18:32 -06:00
parent 8b76d1fcf7
commit 4471d90ccb
3 changed files with 51 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/BaseDriverAdapter.java
+14 -25
View File
@@ -17,7 +17,6 @@
package io.nosqlbench.engine.api.activityimpl.uniform;
import io.nosqlbench.api.config.standard.*;
import io.nosqlbench.engine.api.activityapi.core.Shutdownable;
import io.nosqlbench.engine.api.activityimpl.uniform.fieldmappers.FieldDestructuringMapper;
import io.nosqlbench.engine.api.activityimpl.uniform.flowtypes.Op;
import io.nosqlbench.engine.api.templating.ParsedOp;
@@ -32,7 +31,7 @@ import java.util.function.Function;
import java.util.function.LongFunction;
import java.util.stream.Collectors;
public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter<R,S>, NBConfigurable, NBReconfigurable, Shutdownable {
public abstract class BaseDriverAdapter<R extends Op, S> implements DriverAdapter<R, S>, NBConfigurable, NBReconfigurable {
private final static Logger logger = LogManager.getLogger("ADAPTER");
private DriverSpaceCache<? extends S> spaceCache;
@@ -47,22 +46,22 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
*/
@Override
public final Function<Map<String, Object>, Map<String, Object>> getPreprocessor() {
List<Function<Map<String,Object>,Map<String,Object>>> mappers = new ArrayList<>();
List<Function<Map<String,Object>,Map<String,Object>>> stmtRemappers =
List<Function<Map<String, Object>, Map<String, Object>>> mappers = new ArrayList<>();
List<Function<Map<String, Object>, Map<String, Object>>> stmtRemappers =
getOpStmtRemappers().stream()
.map(m -> new FieldDestructuringMapper("stmt",m))
.map(m -> new FieldDestructuringMapper("stmt", m))
.collect(Collectors.toList());
mappers.addAll(stmtRemappers);
mappers.addAll(getOpFieldRemappers());
if (mappers.size()==0) {
if (mappers.size() == 0) {
return (i) -> i;
}
Function<Map<String,Object>,Map<String,Object>> remapper = null;
Function<Map<String, Object>, Map<String, Object>> remapper = null;
for (int i = 0; i < mappers.size(); i++) {
if (i==0) {
remapper=mappers.get(i);
if (i == 0) {
remapper = mappers.get(i);
} else {
remapper = remapper.andThen(mappers.get(i));
}
@@ -106,7 +105,7 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
*
* @return A list of optionally applied remapping functions.
*/
public List<Function<String, Optional<Map<String,Object>>>> getOpStmtRemappers() {
public List<Function<String, Optional<Map<String, Object>>>> getOpStmtRemappers() {
return List.of();
}
@@ -116,14 +115,14 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
* @return
*/
@Override
public List<Function<Map<String,Object>,Map<String,Object>>> getOpFieldRemappers() {
public List<Function<Map<String, Object>, Map<String, Object>>> getOpFieldRemappers() {
return List.of();
}
@Override
public synchronized final DriverSpaceCache<? extends S> getSpaceCache() {
if (spaceCache==null) {
spaceCache=new DriverSpaceCache<>(getSpaceInitializer(getConfiguration()));
if (spaceCache == null) {
spaceCache = new DriverSpaceCache<>(getSpaceInitializer(getConfiguration()));
}
return spaceCache;
}
@@ -153,7 +152,7 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
public NBConfigModel getConfigModel() {
return ConfigModel.of(BaseDriverAdapter.class)
.add(Param.optional("alias"))
.add(Param.defaultTo("strict",true,"strict op field mode, which requires that provided op fields are recognized and used"))
.add(Param.defaultTo("strict", true, "strict op field mode, which requires that provided op fields are recognized and used"))
.add(Param.optional(List.of("op", "stmt", "statement"), String.class, "op template in statement form"))
.add(Param.optional("tags", String.class, "tags to be used to filter operations"))
.add(Param.defaultTo("errors", "stop", "error handler configuration"))
@@ -166,7 +165,7 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
.add(Param.optional("seq", String.class, "sequencing algorithm"))
.add(Param.optional("instrument", Boolean.class))
.add(Param.optional(List.of("workload", "yaml"), String.class, "location of workload yaml file"))
.add(Param.optional("driver",String.class))
.add(Param.optional("driver", String.class))
.asReadOnly();
}
@@ -185,14 +184,4 @@ public abstract class BaseDriverAdapter<R extends Op,S> implements DriverAdapter
DriverSpaceCache<? extends S> cache = getSpaceCache();
return l -> getSpaceCache().get(spaceNameF.apply(l));
}
@Override
public void shutdown() {
spaceCache.getElements().forEach((spacename,space) -> {
if (space instanceof Shutdownable shutdownable) {
logger.trace("Shutting down space '" + spacename +"'");
shutdownable.shutdown();
}
});
}
}
adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/DriverAdapter.java
+9 -1
View File
@@ -148,7 +148,15 @@ public interface DriverAdapter<OPTYPE extends Op, SPACETYPE> {
DriverSpaceCache<? extends SPACETYPE> getSpaceCache();
/**
* @return A function which can initialize a new S
* This method allows each driver adapter to create named state which is automatically
* cached and re-used by name. For each (driver,space) combination in an activity,
* a distinct space instance will be created. In general, adapter developers will
* use the space type associated with an adapter to wrap native driver instances
* one-to-one. As such, if the space implementation is a {@link AutoCloseable},
* it will be explicitly shutdown as part of the activity shutdown.
*
* @return A function which can initialize a new Space, which is a place to hold
* object state related to retained objects for the lifetime of a native driver.
*/
default Function<String, ? extends SPACETYPE> getSpaceInitializer(NBConfiguration cfg) {
return n -> null;
engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java
+28 -16
View File
@@ -19,7 +19,6 @@ package io.nosqlbench.engine.api.activityimpl.uniform;
import io.nosqlbench.api.config.standard.*;
import io.nosqlbench.api.engine.activityimpl.ActivityDef;
import io.nosqlbench.api.errors.OpConfigError;
import io.nosqlbench.engine.api.activityapi.core.Shutdownable;
import io.nosqlbench.engine.api.activityapi.planning.OpSequence;
import io.nosqlbench.engine.api.activityconfig.StatementsLoader;
import io.nosqlbench.engine.api.activityconfig.yaml.OpTemplate;
@@ -59,12 +58,11 @@ public class StandardActivity<R extends Op, S> extends SimpleActivity implements
Optional<String> yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload");
if (yaml_loc.isPresent()) {
Map<String,Object> disposable = new LinkedHashMap<>(activityDef.getParams());
Map<String, Object> disposable = new LinkedHashMap<>(activityDef.getParams());
StmtsDocList workload = StatementsLoader.loadPath(logger, yaml_loc.get(), disposable, "activities");
yamlmodel = workload.getConfigModel();
}
else {
yamlmodel= ConfigModel.of(StandardActivity.class).asReadOnly();
} else {
yamlmodel = ConfigModel.of(StandardActivity.class).asReadOnly();
}
ServiceLoader<DriverAdapter> adapterLoader = ServiceLoader.load(DriverAdapter.class);
@@ -78,7 +76,7 @@ public class StandardActivity<R extends Op, S> extends SimpleActivity implements
List<DriverAdapter> adapterlist = new ArrayList<>();
for (OpTemplate ot : opTemplates) {
ParsedOp incompleteOpDef = new ParsedOp(ot, NBConfiguration.empty(), List.of());
String driverName = incompleteOpDef.takeOptionalStaticValue("driver",String.class)
String driverName = incompleteOpDef.takeOptionalStaticValue("driver", String.class)
.or(() -> activityDef.getParams().getOptionalString("driver"))
.orElseThrow(() -> new OpConfigError("Unable to identify driver name for op template:\n" + ot));
@@ -100,13 +98,13 @@ public class StandardActivity<R extends Op, S> extends SimpleActivity implements
combinedConfig = combinedModel.matchConfig(activityDef.getParams());
configurable.applyConfig(combinedConfig);
}
adapters.put(driverName,adapter);
mappers.put(driverName,adapter.getOpMapper());
adapters.put(driverName, adapter);
mappers.put(driverName, adapter.getOpMapper());
}
DriverAdapter adapter = adapters.get(driverName);
adapterlist.add(adapter);
ParsedOp pop = new ParsedOp(ot,adapter.getConfiguration(),List.of(adapter.getPreprocessor()));
ParsedOp pop = new ParsedOp(ot, adapter.getConfiguration(), List.of(adapter.getPreprocessor()));
Optional<String> discard = pop.takeOptionalStaticValue("driver", String.class);
pops.add(pop);
}
@@ -153,13 +151,13 @@ public class StandardActivity<R extends Op, S> extends SimpleActivity implements
if (adapter instanceof NBReconfigurable configurable) {
NBConfigModel cfgModel = configurable.getReconfigModel();
NBConfiguration cfg = cfgModel.matchConfig(activityDef.getParams());
NBReconfigurable.applyMatching(cfg,List.of(configurable));
NBReconfigurable.applyMatching(cfg, List.of(configurable));
}
}
}
@Override
public List<OpTemplate> getSyntheticOpTemplates(StmtsDocList stmtsDocList, Map<String,Object> cfg) {
public List<OpTemplate> getSyntheticOpTemplates(StmtsDocList stmtsDocList, Map<String, Object> cfg) {
List<OpTemplate> opTemplates = new ArrayList<>();
for (DriverAdapter adapter : adapters.values()) {
if (adapter instanceof SyntheticOpTemplateProvider sotp) {
@@ -170,12 +168,26 @@ public class StandardActivity<R extends Op, S> extends SimpleActivity implements
return opTemplates;
}
/**
* This is done here since driver adapters are intended to keep all of their state within
* dedicated <em>state space</em> types. Any space which implements {@link io.nosqlbench.engine.api.activityapi.core.Shutdownable}
* will be closed when this activity shuts down.
*/
@Override
public void shutdownActivity() {
adapters.forEach((name, adapter) -> {
if (adapter instanceof Shutdownable shutdownable) {
shutdownable.shutdown();
}
});
for (Map.Entry<String, DriverAdapter> entry : adapters.entrySet()) {
String adapterName = entry.getKey();
DriverAdapter<?,?> adapter = entry.getValue();
adapter.getSpaceCache().getElements().forEach((spaceName, space) -> {
if (space instanceof AutoCloseable autocloseable) {
try {
autocloseable.close();
} catch (Exception e) {
throw new RuntimeException("Error while shutting down state space for " +
"adapter=" + adapterName + ", space=" + spaceName + ": " + e, e);
}
}
});
}
}
}