diff options
13 files changed, 895 insertions, 211 deletions
diff --git a/opendc/opendc-experiments-sc20/build.gradle.kts b/opendc/opendc-experiments-sc20/build.gradle.kts index df291039..b7440792 100644 --- a/opendc/opendc-experiments-sc20/build.gradle.kts +++ b/opendc/opendc-experiments-sc20/build.gradle.kts @@ -46,6 +46,7 @@ dependencies { exclude(group = "org.slf4j", module = "slf4j-log4j12") exclude(group = "log4j") } + implementation("com.zaxxer:HikariCP:3.4.5") runtimeOnly("org.apache.logging.log4j:log4j-slf4j-impl:2.13.1") runtimeOnly("org.postgresql:postgresql:42.2.12") runtimeOnly(project(":odcsim:odcsim-engine-omega")) diff --git a/opendc/opendc-experiments-sc20/schema.sql b/opendc/opendc-experiments-sc20/schema.sql index 677240fa..515348c6 100644 --- a/opendc/opendc-experiments-sc20/schema.sql +++ b/opendc/opendc-experiments-sc20/schema.sql @@ -1,13 +1,26 @@ --- A portfolio represents a collection of scenarios are tested. -DROP TABLE IF EXISTS portfolios; +-- An experiment represents a collection of portfolios. +DROP TABLE IF EXISTS experiments CASCADE; +CREATE TABLE experiments +( + id BIGSERIAL PRIMARY KEY NOT NULL, + creation_time TIMESTAMP NOT NULL DEFAULT (now()) +); + +-- A portfolio represents a collection of scenarios tested. +DROP TABLE IF EXISTS portfolios CASCADE; CREATE TABLE portfolios ( - id BIGSERIAL PRIMARY KEY NOT NULL, - name TEXT NOT NULL + id BIGSERIAL PRIMARY KEY NOT NULL, + experiment_id BIGINT NOT NULL, + name TEXT NOT NULL, + + FOREIGN KEY (experiment_id) REFERENCES experiments (id) + ON DELETE CASCADE + ON UPDATE CASCADE ); -- A scenario represents a single point in the design space (a unique combination of parameters) -DROP TABLE IF EXISTS scenarios; +DROP TABLE IF EXISTS scenarios CASCADE; CREATE TABLE scenarios ( id BIGSERIAL PRIMARY KEY NOT NULL, @@ -17,27 +30,30 @@ CREATE TABLE scenarios workload_name TEXT NOT NULL, workload_fraction DOUBLE PRECISION NOT NULL, allocation_policy TEXT NOT NULL, - failures BIT NOT NULL, - interference BIT NOT NULL, + failures BOOLEAN NOT NULL, + interference BOOLEAN NOT NULL, FOREIGN KEY (portfolio_id) REFERENCES portfolios (id) ON DELETE CASCADE ON UPDATE CASCADE + ); +DROP TYPE IF EXISTS run_state CASCADE; CREATE TYPE run_state AS ENUM ('wait', 'active', 'fail', 'ok'); -- An experiment run represent a single invocation of a trial and is used to distinguish between repetitions of the -- same set of parameters. -DROP TABLE IF EXISTS runs; +DROP TABLE IF EXISTS runs CASCADE; CREATE TABLE runs ( - id INTEGER NOT NULL, - scenario_id BIGINT NOT NULL, - seed INTEGER NOT NULL, - state run_state NOT NULL DEFAULT 'wait'::run_state, - start_time TIMESTAMP NOT NULL, - end_time TIMESTAMP, + id INTEGER NOT NULL, + scenario_id BIGINT NOT NULL, + seed INTEGER NOT NULL, + state run_state NOT NULL DEFAULT 'wait'::run_state, + submit_time TIMESTAMP NOT NULL DEFAULT (now()), + start_time TIMESTAMP DEFAULT (NULL), + end_time TIMESTAMP DEFAULT (NULL), PRIMARY KEY (scenario_id, id), FOREIGN KEY (scenario_id) REFERENCES scenarios (id) @@ -46,7 +62,7 @@ CREATE TABLE runs ); -- Metrics of the hypervisors reported per slice -DROP TABLE IF EXISTS host_metrics; +DROP TABLE IF EXISTS host_metrics CASCADE; CREATE TABLE host_metrics ( id BIGSERIAL PRIMARY KEY NOT NULL, @@ -70,10 +86,11 @@ CREATE TABLE host_metrics ON UPDATE CASCADE ); +DROP INDEX IF EXISTS host_metrics_idx; CREATE INDEX host_metrics_idx ON host_metrics (scenario_id, run_id, timestamp, host_id); -- Metrics of the VMs reported per slice -DROP TABLE IF EXISTS vm_metrics; +DROP TABLE IF EXISTS vm_metrics CASCADE; CREATE TABLE vm_metrics ( id BIGSERIAL PRIMARY KEY NOT NULL, @@ -96,27 +113,29 @@ CREATE TABLE vm_metrics ON UPDATE CASCADE ); +DROP INDEX IF EXISTS vm_metrics_idx; CREATE INDEX vm_metrics_idx ON vm_metrics (scenario_id, run_id, timestamp, vm_id); -- Metrics of the provisioner reported per change -DROP TABLE IF EXISTS provisioner_metrics; +DROP TABLE IF EXISTS provisioner_metrics CASCADE; CREATE TABLE provisioner_metrics ( - id BIGSERIAL PRIMARY KEY NOT NULL, - scenario_id BIGINT NOT NULL, - run_id INTEGER NOT NULL, - timestamp TIMESTAMP NOT NULL, - host_total_count INTEGER NOT NULL, - host_available_count INTEGER NOT NULL, - vm_total_count INTEGER NOT NULL, - vm_active_count INTEGER NOT NULL, - vm_inactive_count INTEGER NOT NULL, - vm_waiting_count INTEGER NOT NULL, - vm_failed_count INTEGER NOT NULL, + id BIGSERIAL PRIMARY KEY NOT NULL, + scenario_id BIGINT NOT NULL, + run_id INTEGER NOT NULL, + timestamp TIMESTAMP NOT NULL, + host_total_count INTEGER NOT NULL, + host_available_count INTEGER NOT NULL, + vm_total_count INTEGER NOT NULL, + vm_active_count INTEGER NOT NULL, + vm_inactive_count INTEGER NOT NULL, + vm_waiting_count INTEGER NOT NULL, + vm_failed_count INTEGER NOT NULL, FOREIGN KEY (scenario_id, run_id) REFERENCES runs (scenario_id, id) ON DELETE CASCADE ON UPDATE CASCADE ); +DROP INDEX IF EXISTS provisioner_metrics_idx; CREATE INDEX provisioner_metrics_idx ON provisioner_metrics (scenario_id, run_id, timestamp); diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt new file mode 100644 index 00000000..a9429367 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunner.kt @@ -0,0 +1,202 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +import com.atlarge.opendc.compute.core.workload.PerformanceInterferenceModel +import com.atlarge.opendc.compute.core.workload.VmWorkload +import com.atlarge.opendc.experiments.sc20.reporter.ExperimentReporterProvider +import com.atlarge.opendc.experiments.sc20.trace.Sc20ParquetTraceReader +import com.atlarge.opendc.experiments.sc20.util.DatabaseHelper +import com.atlarge.opendc.format.environment.EnvironmentReader +import com.atlarge.opendc.format.environment.sc20.Sc20ClusterEnvironmentReader +import com.atlarge.opendc.format.trace.TraceReader +import kotlinx.coroutines.CoroutineDispatcher +import kotlinx.coroutines.asCoroutineDispatcher +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import mu.KotlinLogging +import java.io.Closeable +import java.io.File +import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicInteger +import javax.sql.DataSource +import kotlin.random.Random +import kotlin.system.measureTimeMillis + +/** + * The logger for the experiment runner. + */ +private val logger = KotlinLogging.logger {} + +/** + * The experiment runner is responsible for orchestrating the simulation runs of an experiment. + * + * @param portfolios The portfolios to consider. + * @param ds The data source to write the experimental results to. + */ +public class ExperimentRunner( + private val portfolios: List<Portfolio>, + private val ds: DataSource, + private val reporterProvider: ExperimentReporterProvider, + private val environmentPath: File, + private val tracePath: File, + private val performanceInterferenceModel: PerformanceInterferenceModel +) : Closeable { + /** + * The database helper to write the execution plan. + */ + private val helper = DatabaseHelper(ds.connection) + + /** + * The experiment identifier. + */ + private var experimentId = -1L + + /** + * The mapping of portfolios to their ids. + */ + private val portfolioIds = mutableMapOf<Portfolio, Long>() + + /** + * The mapping of scenarios to their ids. + */ + private val scenarioIds = mutableMapOf<Scenario, Long>() + + /** + * Create an execution plan + */ + private fun createPlan(): List<Run> { + val runs = mutableListOf<Run>() + + for (portfolio in portfolios) { + val portfolioId = helper.persist(portfolio, experimentId) + portfolioIds[portfolio] = portfolioId + var scenarios = 0 + var runCount = 0 + + for (scenario in portfolio.scenarios) { + val scenarioId = helper.persist(scenario, portfolioId) + scenarioIds[scenario] = scenarioId + scenarios++ + + for (run in scenario.runs) { + helper.persist(run, scenarioId) + runCount++ + runs.add(run) + } + } + + logger.info { "Portfolio $portfolioId: ${portfolio.name} ($scenarios scenarios, $runCount runs total)" } + } + + return runs + } + + /** + * Create a trace reader for the specified trace. + */ + private fun createTraceReader( + name: String, + performanceInterferenceModel: PerformanceInterferenceModel, + seed: Int + ): TraceReader<VmWorkload> { + return Sc20ParquetTraceReader( + File(tracePath, name), + performanceInterferenceModel, + emptyList(), + Random(seed) + ) + } + + /** + * Create the environment reader for the specified environment. + */ + private fun createEnvironmentReader(name: String): EnvironmentReader { + return Sc20ClusterEnvironmentReader(File(environmentPath, "$name.txt")) + } + + /** + * Run the specified run. + */ + private fun run(run: Run) { + val reporter = reporterProvider.createReporter(ds, experimentId) + val traceReader = createTraceReader(run.scenario.workload.name, performanceInterferenceModel, run.seed) + val environmentReader = createEnvironmentReader(run.scenario.topology.name) + run.scenario(run, reporter, environmentReader, traceReader) + } + + /** + * Run the portfolios. + */ + @OptIn(ExperimentalStdlibApi::class) + public fun run() { + experimentId = helper.createExperiment() + logger.info { "Creating execution plan for experiment $experimentId" } + + val plan = createPlan() + val total = plan.size + val finished = AtomicInteger() + val dispatcher = Executors.newWorkStealingPool().asCoroutineDispatcher() + + runBlocking { + val mainDispatcher = coroutineContext[CoroutineDispatcher.Key]!! + for (run in plan) { + val scenarioId = scenarioIds[run.scenario]!! + launch(dispatcher) { + launch(mainDispatcher) { + helper.startRun(scenarioId, run.id) + } + + logger.info { "[${finished.get()}/$total] Starting run ($scenarioId, ${run.id})" } + + try { + + val duration = measureTimeMillis { + run(run) + } + + finished.incrementAndGet() + logger.info { "[${finished.get()}/$total] Finished run ($scenarioId, ${run.id}) in $duration milliseconds" } + + withContext(mainDispatcher) { + helper.finishRun(scenarioId, run.id, hasFailed = false) + } + } catch (e: Throwable) { + logger.error("A run has failed", e) + finished.incrementAndGet() + withContext(mainDispatcher) { + helper.finishRun(scenarioId, run.id, hasFailed = true) + } + } + } + } + } + } + + override fun close() { + helper.close() + } +} diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt index b2fbba39..80a91dec 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/ExperimentRunnerCli.kt @@ -24,52 +24,27 @@ package com.atlarge.opendc.experiments.sc20 -import com.atlarge.odcsim.SimulationEngineProvider -import com.atlarge.opendc.compute.core.workload.PerformanceInterferenceModel -import com.atlarge.opendc.compute.core.workload.VmWorkload -import com.atlarge.opendc.compute.virt.service.allocation.AvailableCoreMemoryAllocationPolicy -import com.atlarge.opendc.compute.virt.service.allocation.AvailableMemoryAllocationPolicy -import com.atlarge.opendc.compute.virt.service.allocation.NumberOfActiveServersAllocationPolicy -import com.atlarge.opendc.compute.virt.service.allocation.ProvisionedCoresAllocationPolicy -import com.atlarge.opendc.compute.virt.service.allocation.RandomAllocationPolicy -import com.atlarge.opendc.compute.virt.service.allocation.ReplayAllocationPolicy import com.atlarge.opendc.experiments.sc20.reporter.ExperimentParquetReporter import com.atlarge.opendc.experiments.sc20.reporter.ExperimentPostgresReporter import com.atlarge.opendc.experiments.sc20.reporter.ExperimentReporter -import com.atlarge.opendc.experiments.sc20.trace.Sc20ParquetTraceReader -import com.atlarge.opendc.format.environment.sc20.Sc20ClusterEnvironmentReader -import com.atlarge.opendc.format.trace.TraceReader +import com.atlarge.opendc.experiments.sc20.reporter.ExperimentReporterProvider import com.atlarge.opendc.format.trace.sc20.Sc20PerformanceInterferenceReader import com.atlarge.opendc.format.trace.sc20.Sc20VmPlacementReader -import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper -import com.fasterxml.jackson.module.kotlin.readValue import com.github.ajalt.clikt.core.CliktCommand import com.github.ajalt.clikt.parameters.groups.OptionGroup -import com.github.ajalt.clikt.parameters.groups.default import com.github.ajalt.clikt.parameters.groups.groupChoice -import com.github.ajalt.clikt.parameters.groups.mutuallyExclusiveOptions import com.github.ajalt.clikt.parameters.groups.required import com.github.ajalt.clikt.parameters.options.convert import com.github.ajalt.clikt.parameters.options.default import com.github.ajalt.clikt.parameters.options.defaultLazy -import com.github.ajalt.clikt.parameters.options.flag import com.github.ajalt.clikt.parameters.options.option import com.github.ajalt.clikt.parameters.options.required -import com.github.ajalt.clikt.parameters.types.choice import com.github.ajalt.clikt.parameters.types.file -import com.github.ajalt.clikt.parameters.types.int -import com.github.ajalt.clikt.parameters.types.long -import kotlinx.coroutines.cancel -import kotlinx.coroutines.channels.Channel -import kotlinx.coroutines.launch -import kotlinx.coroutines.runBlocking +import com.zaxxer.hikari.HikariDataSource import mu.KotlinLogging import java.io.File -import java.io.FileReader import java.io.InputStream -import java.sql.DriverManager -import java.util.ServiceLoader -import kotlin.random.Random +import javax.sql.DataSource /** * The logger for this experiment. @@ -80,10 +55,15 @@ private val logger = KotlinLogging.logger {} * Represents the command for running the experiment. */ class ExperimentCli : CliktCommand(name = "sc20-experiment") { - private val environment by option("--environment-file", help = "path to the environment file") - .file() + private val environmentPath by option("--environment-path", help = "path to the environment directory") + .file(canBeFile = false) + .required() + + private val tracePath by option("--trace-path", help = "path to the traces directory") + .file(canBeFile = false) .required() - private val performanceInterferenceStream by option("--performance-interference-file", help = "path to the performance interference file") + + private val performanceInterferenceStream by option("--performance-interference-model", help = "path to the performance interference file") .file() .convert { it.inputStream() as InputStream } .defaultLazy { ExperimentCli::class.java.getResourceAsStream("/env/performance-interference.json") } @@ -95,180 +75,53 @@ class ExperimentCli : CliktCommand(name = "sc20-experiment") { } .default(emptyMap()) - private val selectedVms by mutuallyExclusiveOptions( - option("--selected-vms", help = "the VMs to run").convert { parseVMs(it) }, - option("--selected-vms-file").file().convert { parseVMs(FileReader(it).readText()) } - ).default(emptyList()) - - private val seed by option(help = "the random seed") - .int() - .default(0) - private val failures by option("-x", "--failures", help = "enable (correlated) machine failures") - .flag() - private val failureInterval by option(help = "expected number of hours between failures") - .int() - .default(24 * 7) // one week - private val allocationPolicy by option(help = "name of VM allocation policy to use") - .choice( - "mem", "mem-inv", - "core-mem", "core-mem-inv", - "active-servers", "active-servers-inv", - "provisioned-cores", "provisioned-cores-inv", - "random", "replay" - ) - .default("core-mem") - - private val trace by option().groupChoice( - "sc20-parquet" to Trace.Sc20Parquet() - ).required() - private val reporter by option().groupChoice( "parquet" to Reporter.Parquet(), "postgres" to Reporter.Postgres() ).required() - private fun parseVMs(string: String): List<String> { - // Handle case where VM list contains a VM name with an (escaped) single-quote in it - val sanitizedString = string.replace("\\'", "\\\\[") - .replace("'", "\"") - .replace("\\\\[", "'") - val vms: List<String> = jacksonObjectMapper().readValue(sanitizedString) - return vms - } + private val jdbcUrl by option("--jdbc-url", help = "JDBC connection url").required() override fun run() { - logger.info("seed: $seed") - logger.info("failures: $failures") - logger.info("allocation-policy: $allocationPolicy") - - val start = System.currentTimeMillis() - val reporter: ExperimentReporter = reporter.createReporter() - - val provider = ServiceLoader.load(SimulationEngineProvider::class.java).first() - val system = provider("test") - val root = system.newDomain("root") - - val chan = Channel<Unit>(Channel.CONFLATED) - val allocationPolicy = when (this.allocationPolicy) { - "mem" -> AvailableMemoryAllocationPolicy() - "mem-inv" -> AvailableMemoryAllocationPolicy(true) - "core-mem" -> AvailableCoreMemoryAllocationPolicy() - "core-mem-inv" -> AvailableCoreMemoryAllocationPolicy(true) - "active-servers" -> NumberOfActiveServersAllocationPolicy() - "active-servers-inv" -> NumberOfActiveServersAllocationPolicy(true) - "provisioned-cores" -> ProvisionedCoresAllocationPolicy() - "provisioned-cores-inv" -> ProvisionedCoresAllocationPolicy(true) - "random" -> RandomAllocationPolicy(Random(seed)) - "replay" -> ReplayAllocationPolicy(vmPlacements) - else -> throw IllegalArgumentException("Unknown policy ${this.allocationPolicy}") - } - - val performanceInterferenceModel = try { - Sc20PerformanceInterferenceReader(performanceInterferenceStream).construct() - } catch (e: Throwable) { - reporter.close() - throw e - } - val environmentReader = Sc20ClusterEnvironmentReader(environment) - val traceReader = try { - trace.createTraceReader(performanceInterferenceModel, selectedVms, seed) - } catch (e: Throwable) { - reporter.close() - throw e - } - - root.launch { - val (bareMetalProvisioner, scheduler) = createProvisioner(root, environmentReader, allocationPolicy) - - val failureDomain = if (failures) { - logger.info("ENABLING failures") - createFailureDomain(seed, failureInterval, bareMetalProvisioner, chan) - } else { - null - } - - attachMonitor(scheduler, reporter) - processTrace(traceReader, scheduler, chan, reporter, vmPlacements) - - logger.debug("SUBMIT=${scheduler.submittedVms}") - logger.debug("FAIL=${scheduler.unscheduledVms}") - logger.debug("QUEUED=${scheduler.queuedVms}") - logger.debug("RUNNING=${scheduler.runningVms}") - logger.debug("FINISHED=${scheduler.finishedVms}") + val ds = HikariDataSource() + ds.jdbcUrl = jdbcUrl + ds.addDataSourceProperty("reWriteBatchedInserts", "true") + + val portfolios = listOf( + HorVerPortfolio // , + // MoreVelocityPortfolio, + // MoreHpcPortfolio, + // OperationalPhenomenaPortfolio + ) - failureDomain?.cancel() - scheduler.terminate() - logger.info("Simulation took ${System.currentTimeMillis() - start} milliseconds") - } + val performanceInterferenceModel = Sc20PerformanceInterferenceReader(performanceInterferenceStream) + .construct() - runBlocking { - system.run() - system.terminate() + try { + val runner = ExperimentRunner(portfolios, ds, reporter, environmentPath, tracePath, performanceInterferenceModel) + runner.run() + } finally { + ds.close() } - - // Explicitly close the monitor to flush its buffer - reporter.close() } } /** * An option for specifying the type of reporter to use. */ -internal sealed class Reporter(name: String) : OptionGroup(name) { - /** - * Create the [ExperimentReporter] for this option. - */ - abstract fun createReporter(): ExperimentReporter - +internal sealed class Reporter(name: String) : OptionGroup(name), ExperimentReporterProvider { class Parquet : Reporter("Options for reporting using Parquet") { - private val path by option("--parquet-path", help = "path to where the output should be stored") + private val path by option("--parquet-directory", help = "path to where the output should be stored") .file() - .defaultLazy { File("data/results-${System.currentTimeMillis()}.parquet") } + .defaultLazy { File("data") } - override fun createReporter(): ExperimentReporter = - ExperimentParquetReporter(path) + override fun createReporter(ds: DataSource, experimentId: Long): ExperimentReporter = + ExperimentParquetReporter(File(path, "results-${System.currentTimeMillis()}.parquet")) } class Postgres : Reporter("Options for reporting using PostgreSQL") { - private val url by option("--postgres-url", help = "JDBC connection url").required() - private val experimentId by option(help = "Experiment ID").long().required() - - override fun createReporter(): ExperimentReporter { - val conn = DriverManager.getConnection(url) - return ExperimentPostgresReporter(conn, experimentId) - } - } -} - -/** - * An option for specifying the type of trace to use. - */ -internal sealed class Trace(type: String) : OptionGroup(type) { - /** - * Create a [TraceReader] for this type of trace. - */ - abstract fun createTraceReader(performanceInterferenceModel: PerformanceInterferenceModel, vms: List<String>, seed: Int): TraceReader<VmWorkload> - - class Sc20Parquet : Trace("SC20 Parquet format") { - /** - * Path to trace directory. - */ - private val path by option("--trace-path", help = "path to the trace directory") - .file(canBeFile = false) - .required() - - override fun createTraceReader( - performanceInterferenceModel: PerformanceInterferenceModel, - vms: List<String>, - seed: Int - ): TraceReader<VmWorkload> { - return Sc20ParquetTraceReader( - path, - performanceInterferenceModel, - vms, - Random(seed) - ) - } + override fun createReporter(ds: DataSource, experimentId: Long): ExperimentReporter = + ExperimentPostgresReporter(ds.connection, experimentId) } } diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolio.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolio.kt new file mode 100644 index 00000000..34505fce --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolio.kt @@ -0,0 +1,35 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +/** + * A portfolio represents a collection of scenarios are tested. + */ +public abstract class Portfolio(val name: String) { + /** + * The scenarios of this portfolio consists of. + */ + abstract val scenarios: Sequence<Scenario> +} diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt new file mode 100644 index 00000000..0e612ae0 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Portfolios.kt @@ -0,0 +1,163 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +abstract class AbstractSc20Portfolio(name: String) : Portfolio(name) { + abstract val topologies: List<Topology> + abstract val workloads: List<Workload> + abstract val operationalPhenomena: List<Pair<Boolean, Boolean>> + abstract val allocationPolicies: List<String> + + open val repetitions = 2 + + override val scenarios: Sequence<Scenario> = sequence { + for (topology in topologies) { + for (workload in workloads) { + for ((hasFailures, hasInterference) in operationalPhenomena) { + for (allocationPolicy in allocationPolicies) { + yield( + Scenario( + this@AbstractSc20Portfolio, + repetitions, + topology, + workload, + allocationPolicy, + hasFailures, + hasInterference + ) + ) + } + } + } + } + } +} + +object HorVerPortfolio : AbstractSc20Portfolio("horizontal_vs_vertical") { + override val topologies = listOf( + Topology("base"), + Topology("rep-vol-hor-hom"), + Topology("rep-vol-hor-het"), + Topology("rep-vol-ver-hom"), + Topology("rep-vol-ver-het"), + Topology("exp-vol-hor-hom"), + Topology("exp-vol-hor-het"), + Topology("exp-vol-ver-hom"), + Topology("exp-vol-ver-het") + ) + + override val workloads = listOf( + // Workload("solvinity", 0.1), + // Workload("solvinity", 0.25), + Workload("small-parquet", 0.5), + Workload("small-parquet", 1.0) + ) + + override val operationalPhenomena = listOf( + true to true + ) + + override val allocationPolicies = listOf( + "active-servers" + ) +} + +object MoreVelocityPortfolio : AbstractSc20Portfolio("more_velocity") { + override val topologies = listOf( + Topology("base"), + Topology("rep-vel-ver-hom"), + Topology("rep-vel-ver-het"), + Topology("exp-vel-ver-hom"), + Topology("exp-vel-ver-het") + ) + + override val workloads = listOf( + // Workload("solvinity", 0.1), + // Workload("solvinity", 0.25), + Workload("solvinity", 0.5), + Workload("solvinity", 1.0) + ) + + override val operationalPhenomena = listOf( + true to true + ) + + override val allocationPolicies = listOf( + "active-servers" + ) +} + +object MoreHpcPortfolio : AbstractSc20Portfolio("more_velocity") { + override val topologies = listOf( + Topology("base"), + Topology("exp-vol-hor-hom"), + Topology("exp-vol-ver-hom"), + Topology("exp-vel-ver-hom") + ) + + override val workloads = listOf( + // Workload("solvinity", 0.1), + // Workload("solvinity", 0.25), + Workload("solvinity", 0.5), + Workload("solvinity", 1.0) + ) + + override val operationalPhenomena = listOf( + true to true + ) + + override val allocationPolicies = listOf( + "active-servers" + ) +} + +object OperationalPhenomenaPortfolio : AbstractSc20Portfolio("more_velocity") { + override val topologies = listOf( + Topology("base") + ) + + override val workloads = listOf( + // Workload("solvinity", 0.1), + // Workload("solvinity", 0.25), + Workload("solvinity", 1.0) + ) + + override val operationalPhenomena = listOf( + true to true, + false to true, + true to false, + true to true + ) + + override val allocationPolicies = listOf( + "mem", + "mem-inv", + "core-mem", + "core-mem-inv", + "active-servers", + "active-servers-inv", + "random" + ) +} diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Run.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Run.kt new file mode 100644 index 00000000..b2151b16 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Run.kt @@ -0,0 +1,31 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +/** + * An experiment run represent a single invocation of a trial and is used to distinguish between repetitions of the + * same set of parameters. + */ +public data class Run(val scenario: Scenario, val id: Int, val seed: Int) diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Scenario.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Scenario.kt new file mode 100644 index 00000000..66a8babf --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Scenario.kt @@ -0,0 +1,129 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +import com.atlarge.odcsim.SimulationEngineProvider +import com.atlarge.opendc.compute.core.workload.VmWorkload +import com.atlarge.opendc.compute.virt.service.allocation.AvailableCoreMemoryAllocationPolicy +import com.atlarge.opendc.compute.virt.service.allocation.AvailableMemoryAllocationPolicy +import com.atlarge.opendc.compute.virt.service.allocation.NumberOfActiveServersAllocationPolicy +import com.atlarge.opendc.compute.virt.service.allocation.ProvisionedCoresAllocationPolicy +import com.atlarge.opendc.compute.virt.service.allocation.RandomAllocationPolicy +import com.atlarge.opendc.compute.virt.service.allocation.ReplayAllocationPolicy +import com.atlarge.opendc.experiments.sc20.reporter.ExperimentReporter +import com.atlarge.opendc.format.environment.EnvironmentReader +import com.atlarge.opendc.format.trace.TraceReader +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import mu.KotlinLogging +import java.util.ServiceLoader +import kotlin.random.Random + +/** + * The logger for the experiment scenario. + */ +private val logger = KotlinLogging.logger {} + +/** + * The provider for the simulation engine to use. + */ +private val provider = ServiceLoader.load(SimulationEngineProvider::class.java).first() + +/** + * A scenario represents a single point in the design space (a unique combination of parameters). + */ +public class Scenario( + val portfolio: Portfolio, + val repetitions: Int, + val topology: Topology, + val workload: Workload, + val allocationPolicy: String, + val hasFailures: Boolean, + val hasInterference: Boolean, + val failureInterval: Int = 24 * 7 +) { + /** + * The runs this scenario consists of. + */ + public val runs: Sequence<Run> = sequence { + repeat(repetitions) { i -> + yield(Run(this@Scenario, i, i)) + } + } + + /** + * Perform a single run of this scenario. + */ + public operator fun invoke(run: Run, reporter: ExperimentReporter, environment: EnvironmentReader, trace: TraceReader<VmWorkload>) { + val system = provider("experiment-${run.id}") + val root = system.newDomain("root") + val seeder = Random(run.seed) + + val chan = Channel<Unit>(Channel.CONFLATED) + val allocationPolicy = when (this.allocationPolicy) { + "mem" -> AvailableMemoryAllocationPolicy() + "mem-inv" -> AvailableMemoryAllocationPolicy(true) + "core-mem" -> AvailableCoreMemoryAllocationPolicy() + "core-mem-inv" -> AvailableCoreMemoryAllocationPolicy(true) + "active-servers" -> NumberOfActiveServersAllocationPolicy() + "active-servers-inv" -> NumberOfActiveServersAllocationPolicy(true) + "provisioned-cores" -> ProvisionedCoresAllocationPolicy() + "provisioned-cores-inv" -> ProvisionedCoresAllocationPolicy(true) + "random" -> RandomAllocationPolicy(Random(seeder.nextInt())) + "replay" -> ReplayAllocationPolicy(emptyMap()) + else -> throw IllegalArgumentException("Unknown policy ${this.allocationPolicy}") + } + + root.launch { + val (bareMetalProvisioner, scheduler) = createProvisioner(root, environment, allocationPolicy) + + val failureDomain = if (hasFailures) { + logger.debug("ENABLING failures") + createFailureDomain(seeder.nextInt(), failureInterval, bareMetalProvisioner, chan) + } else { + null + } + + attachMonitor(scheduler, reporter) + processTrace(trace, scheduler, chan, reporter, emptyMap()) + + logger.debug("SUBMIT=${scheduler.submittedVms}") + logger.debug("FAIL=${scheduler.unscheduledVms}") + logger.debug("QUEUED=${scheduler.queuedVms}") + logger.debug("RUNNING=${scheduler.runningVms}") + logger.debug("FINISHED=${scheduler.finishedVms}") + + failureDomain?.cancel() + scheduler.terminate() + } + + runBlocking { + system.run() + system.terminate() + } + } +} diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Topology.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Topology.kt new file mode 100644 index 00000000..d2be9599 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Topology.kt @@ -0,0 +1,30 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +/** + * The datacenter topology on which we test the workload. + */ +public data class Topology(val name: String) diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Workload.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Workload.kt new file mode 100644 index 00000000..4ab5ec8c --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/Workload.kt @@ -0,0 +1,30 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20 + +/** + * A workload that is considered for a scenario. + */ +public class Workload(val name: String, val fraction: Double) diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporterProvider.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporterProvider.kt new file mode 100644 index 00000000..d0dfd2e8 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/reporter/ExperimentReporterProvider.kt @@ -0,0 +1,31 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20.reporter + +import javax.sql.DataSource + +interface ExperimentReporterProvider { + public fun createReporter(ds: DataSource, experimentId: Long): ExperimentReporter +} diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/Sc20ParquetTraceReader.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/Sc20StreamingParquetTraceReader.kt index 8a204ca3..8a204ca3 100644 --- a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/Sc20ParquetTraceReader.kt +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/trace/Sc20StreamingParquetTraceReader.kt diff --git a/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/util/DatabaseHelper.kt b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/util/DatabaseHelper.kt new file mode 100644 index 00000000..0292a2e3 --- /dev/null +++ b/opendc/opendc-experiments-sc20/src/main/kotlin/com/atlarge/opendc/experiments/sc20/util/DatabaseHelper.kt @@ -0,0 +1,160 @@ +/* + * MIT License + * + * Copyright (c) 2020 atlarge-research + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package com.atlarge.opendc.experiments.sc20.util + +import com.atlarge.opendc.experiments.sc20.Portfolio +import com.atlarge.opendc.experiments.sc20.Run +import com.atlarge.opendc.experiments.sc20.Scenario +import java.io.Closeable +import java.sql.Connection +import java.sql.Statement + +/** + * A helper class for writing to the database. + */ +class DatabaseHelper(val conn: Connection) : Closeable { + /** + * Prepared statement to create experiment. + */ + private val createExperiment = conn.prepareStatement("INSERT INTO experiments DEFAULT VALUES", arrayOf("id")) + + /** + * Prepared statement for creating a portfolio. + */ + private val createPortfolio = conn.prepareStatement("INSERT INTO portfolios (experiment_id, name) VALUES (?, ?)", arrayOf("id")) + + /** + * Prepared statement for creating a scenario + */ + private val createScenario = conn.prepareStatement("INSERT INTO scenarios (portfolio_id, repetitions, topology, workload_name, workload_fraction, allocation_policy, failures, interference) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", arrayOf("id")) + + /** + * Prepared statement for creating a run. + */ + private val createRun = conn.prepareStatement("INSERT INTO runs (id, scenario_id, seed) VALUES (?, ?, ?)", arrayOf("id")) + + /** + * Prepared statement for starting a run. + */ + private val startRun = conn.prepareStatement("UPDATE runs SET state = 'active'::run_state,start_time = now() WHERE id = ? AND scenario_id = ?") + + /** + * Prepared statement for finishing a run. + */ + private val finishRun = conn.prepareStatement("UPDATE runs SET state = ?::run_state,end_time = now() WHERE id = ? AND scenario_id = ?") + + /** + * Create a new experiment and return its id. + */ + fun createExperiment(): Long { + val affectedRows = createExperiment.executeUpdate() + check(affectedRows != 0) + return createExperiment.latestId + } + + /** + * Persist a [Portfolio] and return its id. + */ + fun persist(portfolio: Portfolio, experimentId: Long): Long { + createPortfolio.setLong(1, experimentId) + createPortfolio.setString(2, portfolio.name) + + val affectedRows = createPortfolio.executeUpdate() + check(affectedRows != 0) + return createPortfolio.latestId + } + + /** + * Persist a [Scenario] and return its id. + */ + fun persist(scenario: Scenario, portfolioId: Long): Long { + createScenario.setLong(1, portfolioId) + createScenario.setInt(2, scenario.repetitions) + createScenario.setString(3, scenario.topology.name) + createScenario.setString(4, scenario.workload.name) + createScenario.setDouble(5, scenario.workload.fraction) + createScenario.setString(6, scenario.allocationPolicy) + createScenario.setBoolean(7, scenario.hasFailures) + createScenario.setBoolean(8, scenario.hasInterference) + + val affectedRows = createScenario.executeUpdate() + check(affectedRows != 0) + return createScenario.latestId + } + + /** + * Persist a [Run] and return its id. + */ + fun persist(run: Run, scenarioId: Long): Int { + createRun.setInt(1, run.id) + createRun.setLong(2, scenarioId) + createRun.setInt(3, run.seed) + + val affectedRows = createRun.executeUpdate() + check(affectedRows != 0) + return createRun.latestId.toInt() + } + + /** + * Start run. + */ + fun startRun(scenario: Long, run: Int) { + startRun.setInt(1, run) + startRun.setLong(2, scenario) + + val affectedRows = startRun.executeUpdate() + check(affectedRows != 0) + } + + /** + * Finish a run. + */ + fun finishRun(scenario: Long, run: Int, hasFailed: Boolean) { + finishRun.setString(1, if (hasFailed) "fail" else "ok") + finishRun.setInt(2, run) + finishRun.setLong(3, scenario) + + val affectedRows = finishRun.executeUpdate() + check(affectedRows != 0) + } + + /** + * Obtain the latest identifier of the specified statement. + */ + private val Statement.latestId: Long + get() { + val rs = generatedKeys + return try { + check(rs.next()) + rs.getLong(1) + } finally { + rs.close() + } + } + + override fun close() { + conn.close() + } +} |