From 4471d90ccb515b08e0ffef536b75f55865271248 Mon Sep 17 00:00:00 2001 From: Jonathan Shook Date: Fri, 18 Nov 2022 23:18:32 -0600 Subject: [PATCH] implement closeable spaces via decorator interface --- .../uniform/BaseDriverAdapter.java | 39 ++++++---------- .../activityimpl/uniform/DriverAdapter.java | 10 ++++- .../uniform/StandardActivity.java | 44 ++++++++++++------- 3 files changed, 51 insertions(+), 42 deletions(-) diff --git a/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/BaseDriverAdapter.java b/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/BaseDriverAdapter.java index 0137d5b07..3e4ffef9d 100644 --- a/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/BaseDriverAdapter.java +++ b/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/BaseDriverAdapter.java @@ -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 implements DriverAdapter, NBConfigurable, NBReconfigurable, Shutdownable { +public abstract class BaseDriverAdapter implements DriverAdapter, NBConfigurable, NBReconfigurable { private final static Logger logger = LogManager.getLogger("ADAPTER"); private DriverSpaceCache spaceCache; @@ -47,22 +46,22 @@ public abstract class BaseDriverAdapter implements DriverAdapter */ @Override public final Function, Map> getPreprocessor() { - List,Map>> mappers = new ArrayList<>(); - List,Map>> stmtRemappers = + List, Map>> mappers = new ArrayList<>(); + List, Map>> 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> remapper = null; + Function, Map> 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 implements DriverAdapter * * @return A list of optionally applied remapping functions. */ - public List>>> getOpStmtRemappers() { + public List>>> getOpStmtRemappers() { return List.of(); } @@ -116,14 +115,14 @@ public abstract class BaseDriverAdapter implements DriverAdapter * @return */ @Override - public List,Map>> getOpFieldRemappers() { + public List, Map>> getOpFieldRemappers() { return List.of(); } @Override public synchronized final DriverSpaceCache 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 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 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 implements DriverAdapter DriverSpaceCache 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(); - } - }); - } } diff --git a/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/DriverAdapter.java b/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/DriverAdapter.java index 6f42f8cf4..14bfae422 100644 --- a/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/DriverAdapter.java +++ b/adapters-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/DriverAdapter.java @@ -148,7 +148,15 @@ public interface DriverAdapter { DriverSpaceCache 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 getSpaceInitializer(NBConfiguration cfg) { return n -> null; diff --git a/engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java b/engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java index a15a66664..0b47d30c2 100644 --- a/engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java +++ b/engine-api/src/main/java/io/nosqlbench/engine/api/activityimpl/uniform/StandardActivity.java @@ -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 extends SimpleActivity implements Optional yaml_loc = activityDef.getParams().getOptionalString("yaml", "workload"); if (yaml_loc.isPresent()) { - Map disposable = new LinkedHashMap<>(activityDef.getParams()); + Map 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 adapterLoader = ServiceLoader.load(DriverAdapter.class); @@ -78,7 +76,7 @@ public class StandardActivity extends SimpleActivity implements List 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 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 discard = pop.takeOptionalStaticValue("driver", String.class); pops.add(pop); } @@ -153,13 +151,13 @@ public class StandardActivity 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 getSyntheticOpTemplates(StmtsDocList stmtsDocList, Map cfg) { + public List getSyntheticOpTemplates(StmtsDocList stmtsDocList, Map cfg) { List opTemplates = new ArrayList<>(); for (DriverAdapter adapter : adapters.values()) { if (adapter instanceof SyntheticOpTemplateProvider sotp) { @@ -170,12 +168,26 @@ public class StandardActivity extends SimpleActivity implements return opTemplates; } + /** + * This is done here since driver adapters are intended to keep all of their state within + * dedicated state space 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 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); + } + } + }); + } } }