-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Adjusts handling of asynchronous execution of extraction pipelines.
Signed-off-by: Ralph Gasser <[email protected]>
- Loading branch information
1 parent
9b07f60
commit 563f042
Showing
8 changed files
with
143 additions
and
128 deletions.
There are no files selected for viewing
141 changes: 78 additions & 63 deletions
141
...core/src/main/kotlin/org/vitrivr/engine/core/config/pipeline/execution/ExecutionServer.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,102 +1,117 @@ | ||
| package org.vitrivr.engine.core.config.pipeline.execution | ||
|
|
||
| import io.github.oshai.kotlinlogging.KLogger | ||
| import io.github.oshai.kotlinlogging.KotlinLogging | ||
| import kotlinx.coroutines.* | ||
| import kotlinx.coroutines.flow.cancellable | ||
| import kotlinx.coroutines.flow.collect | ||
| import kotlinx.coroutines.flow.takeWhile | ||
| import org.vitrivr.engine.core.config.pipeline.Pipeline | ||
| import org.vitrivr.engine.core.config.pipeline.PipelineBuilder | ||
| import org.vitrivr.engine.core.model.metamodel.Schema | ||
| import org.vitrivr.engine.core.operators.Operator | ||
| import org.vitrivr.engine.core.operators.ingest.AbstractSegmenter | ||
| import org.vitrivr.engine.core.operators.ingest.Extractor | ||
| import java.util.UUID | ||
| import java.util.concurrent.BlockingQueue | ||
| import java.util.* | ||
| import java.util.concurrent.ConcurrentHashMap | ||
| import java.util.concurrent.ExecutorService | ||
| import java.util.concurrent.Executors | ||
| import java.util.concurrent.LinkedBlockingDeque | ||
|
|
||
|
|
||
| private val logger: KLogger = KotlinLogging.logger {} | ||
|
|
||
| /** | ||
| * A | ||
| * An execution environment for data ingest and retrieval. | ||
| * | ||
| * @author Ralph Gasser | ||
| * @version 1.0.0 | ||
| */ | ||
| class ExecutionServer private constructor(schema: Schema){ | ||
|
|
||
| companion object { | ||
|
|
||
| @Volatile private var instances: MutableMap<Schema, ExecutionServer> = mutableMapOf() | ||
|
|
||
| fun getInstance(schema: Schema) = | ||
| instances[schema] ?: synchronized(this) { // synchronized to avoid concurrency problem | ||
| instances[schema] ?: ExecutionServer(schema).also { instances[schema] = it } | ||
| } | ||
| } | ||
|
|
||
| class ExecutionServer { | ||
|
|
||
| /** The [ExecutorService] used to execution [] */ | ||
| private val executor: ExecutorService = Executors.newCachedThreadPool() | ||
|
|
||
| /** The [CoroutineDispatcher] used for execution. */ | ||
| private val dispatcher: CoroutineDispatcher = this.executor.asCoroutineDispatcher() | ||
|
|
||
| var indexJobQueue: BlockingQueue<Pair<Pipeline,UUID>> = LinkedBlockingDeque() | ||
|
|
||
| init { | ||
| this.run() | ||
| } | ||
| /** A [ConcurrentHashMap] of all ongoing [Job]s. */ | ||
| private val jobs = ConcurrentHashMap<UUID, Job>() | ||
|
|
||
| fun isPending(uuid: UUID): Int { | ||
| return this.indexJobQueue.indexOf(this.indexJobQueue.find { it.second == uuid }) | ||
| } | ||
| /** A [ConcurrentHashMap] of all ongoing [Job]s. */ | ||
| private val jobHistory = Collections.synchronizedList(ArrayList<Triple<UUID, ExecutionStatus, Long>>(100)) | ||
|
|
||
| fun enqueueIndexJob(pipeline: Pipeline): UUID { | ||
| val uuid = UUID.randomUUID() | ||
| return this.enqueueIndexJob(pipeline, uuid) | ||
| } | ||
| /** | ||
| * Checks the status of the [Job] with the provided [UUID]. | ||
| * | ||
| * @param uuid [UUID] of the [Job] to check. | ||
| * @return [ExecutionStatus] of [Job] | ||
| */ | ||
| fun status(uuid: UUID): ExecutionStatus { | ||
| /* Check if job is still running. */ | ||
| val jobs = this.jobs[uuid] | ||
| if (jobs != null) { | ||
| return ExecutionStatus.RUNNING | ||
| } | ||
|
|
||
| /* Check list for job. */ | ||
| for (job in this.jobHistory) { | ||
| if (job.first == uuid) { | ||
| return job.second | ||
| } | ||
| } | ||
|
|
||
| fun enqueueIndexJob(pipeline: Pipeline, uuid: UUID): UUID{ | ||
| this.indexJobQueue.add(Pair(pipeline, uuid)) | ||
| return uuid; | ||
| /* Otherwise, job is unknown. */ | ||
| return ExecutionStatus.UNKNOWN | ||
| } | ||
|
|
||
| /** | ||
| * Executes an extraction job using a [List] of [Extractor]s. | ||
| * Cancels the [Job] with the provided [UUID]. | ||
| * | ||
| * @param extractors The [List] of [Extractor]s to execute. | ||
| * @param uuid [UUID] of the [Job] to check. | ||
| * @return True, if job is running, false otherwise. | ||
| */ | ||
| private fun extract(pipeline: Pipeline) = runBlocking { | ||
| val scope = CoroutineScope(this@ExecutionServer.dispatcher) | ||
| val jobs = pipeline.getLeaves().map { e -> scope.launch { e.toFlow(scope).takeWhile { it != AbstractSegmenter.TerminalRetrievable }.collect() } } | ||
| jobs.forEach { it.join() } | ||
| fun cancel(uuid: UUID): Boolean { | ||
| val job = this.jobs[uuid] | ||
| return if (job != null) { | ||
| job.cancel() | ||
| true | ||
| } else { | ||
| false | ||
| } | ||
| } | ||
|
|
||
|
|
||
| private fun run() { | ||
| Thread { | ||
| val running = true | ||
| while (running) { | ||
| val pipeline = indexJobQueue.take() | ||
| try { | ||
| this.extract(pipeline.first) | ||
| logger.debug { "Extraction with pipeline '${pipeline.second}' finished." } | ||
| } catch (e: Exception) { | ||
| logger.error { "Error while executing extraction job: ${e.message}" } | ||
| /** | ||
| * Executes an extraction [Pipeline] in a blocking fashion, i.e., the call will block until the [Pipeline] has been executed. | ||
| * | ||
| * @param pipeline The [Pipeline] to execute. | ||
| * @return [UUID] identifying the job. | ||
| */ | ||
| fun extractAsync(pipeline: Pipeline): UUID { | ||
| val jobId = UUID.randomUUID() | ||
| val scope = CoroutineScope(this@ExecutionServer.dispatcher) + CoroutineName("index-job-$jobId") | ||
| val job = scope.launch { | ||
| try { | ||
| val jobs = pipeline.getLeaves().map { e -> this.launch { e.toFlow(scope).cancellable().takeWhile { it != AbstractSegmenter.TerminalRetrievable }.collect() } } | ||
| jobs.forEach { it.join() } | ||
| this@ExecutionServer.jobHistory.add(Triple(jobId, ExecutionStatus.COMPLETED, System.currentTimeMillis())) | ||
| } catch (e: Throwable) { | ||
| this@ExecutionServer.jobHistory.add(Triple(jobId, ExecutionStatus.FAILED, System.currentTimeMillis())) | ||
| } finally { | ||
| this@ExecutionServer.jobs.remove(jobId) | ||
| if (this@ExecutionServer.jobHistory.size > 100) { | ||
| this@ExecutionServer.jobHistory.removeFirst() | ||
| } | ||
| // wait | ||
| Thread.sleep(10000) | ||
| } | ||
| }.start() | ||
|
|
||
| } | ||
| this.jobs[jobId] = job | ||
| return jobId | ||
| } | ||
|
|
||
| /** | ||
| * Shuts down the [ExecutorService] used by this [ExecutionServer] | ||
| * Executes an extraction [Pipeline] in a blocking fashion, i.e., the call will block until the [Pipeline] has been executed. | ||
| * | ||
| * This is mainly for testing purposes! | ||
| * | ||
| * @param pipeline The [Pipeline] to execute. | ||
| */ | ||
| fun shutdown() { | ||
| this.executor.shutdown() | ||
| fun extract(pipeline: Pipeline) { | ||
| val jobId = UUID.randomUUID() | ||
| val scope = CoroutineScope(this@ExecutionServer.dispatcher) + CoroutineName("index-job-$jobId") | ||
| runBlocking { | ||
| val jobs = pipeline.getLeaves().map { e -> scope.launch { e.toFlow(this).takeWhile { it != AbstractSegmenter.TerminalRetrievable }.collect() } } | ||
| jobs.forEach { it.join() } | ||
| } | ||
| } | ||
| } |
14 changes: 14 additions & 0 deletions
14
...core/src/main/kotlin/org/vitrivr/engine/core/config/pipeline/execution/ExecutionStatus.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| package org.vitrivr.engine.core.config.pipeline.execution | ||
|
|
||
| /** | ||
| * A status enumeration for [ExecutionServer] jobs. | ||
| * | ||
| * @author Ralph Gasser | ||
| * @version 1.0.0 | ||
| */ | ||
| enum class ExecutionStatus { | ||
| UNKNOWN, | ||
| RUNNING, | ||
| FAILED, | ||
| COMPLETED | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.