Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Issue 259 Add executor framework for running viewports in multiple pa…
…rallel threads (#338) * #259 Added executor framework for running multiple viewport calculations on separate thread pool. * #259 left default thread count as one in SimulMain
- Loading branch information
1 parent
9ade0d1
commit bba484a
Showing
12 changed files
with
398 additions
and
73 deletions.
There are no files selected for viewing
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
114 changes: 114 additions & 0 deletions
114
...box/src/main/scala/org/finos/toolbox/thread/LifeCycleRunOncePerThreadExecutorRunner.scala
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,114 @@ | ||
| package org.finos.toolbox.thread | ||
|
|
||
| import com.typesafe.scalalogging.StrictLogging | ||
| import org.finos.toolbox.lifecycle.{LifecycleContainer, LifecycleEnabled} | ||
| import org.finos.toolbox.logging.LogAtFrequency | ||
| import org.finos.toolbox.thread.executor.ResubmitExecutor | ||
| import org.finos.toolbox.time.Clock | ||
|
|
||
| import java.util.concurrent.atomic.AtomicBoolean | ||
| import java.util.concurrent.{Callable, ConcurrentSkipListSet, FutureTask, LinkedBlockingQueue, TimeUnit} | ||
| import scala.jdk.CollectionConverters.CollectionHasAsScala | ||
|
|
||
| trait WorkItem[T] extends Comparable[WorkItem[T]]{ | ||
| def doWork(): T | ||
| def compareTo(o: WorkItem[T]): Int = { | ||
| if(o.hashCode() == this.hashCode()){ | ||
| 0 | ||
| }else if( o.hashCode() < this.hashCode()){ | ||
| 1 | ||
| }else if(o.hashCode() > this.hashCode()){ | ||
| -1 | ||
| }else{ | ||
| 0 | ||
| } | ||
| } | ||
| } | ||
|
|
||
| abstract class LifeCycleRunOncePerThreadExecutorRunner[T](val name: String, val countOfThreads: Int, val generateWorkFunc: () => List[WorkItem[T]]) (implicit lifecycle: LifecycleContainer, clock: Clock) extends LifeCycleRunner(name, () => ()) with StrictLogging{ | ||
|
|
||
| lifecycle(this) | ||
|
|
||
| private var retryExecutor: Option[ResubmitExecutor[T]] = None | ||
| private final val workQueue = new LinkedBlockingQueue[Runnable]() | ||
| private val selfRef = this; | ||
| private final val setOfWork = new ConcurrentSkipListSet[WorkItem[T]]() | ||
|
|
||
| override def doStart(): Unit = { | ||
| logger.info("Starting up viewport runner...") | ||
| retryExecutor = Some(new ResubmitExecutor[T](countOfThreads, countOfThreads, 1000, TimeUnit.SECONDS, workQueue){ | ||
| override def newCallable(r: FutureTask[T], t: Throwable): Callable[T] = { | ||
| selfRef.newCallable(r) | ||
| } | ||
| override def shouldResubmit(r: FutureTask[T], t: Throwable): Boolean = { | ||
| setOfWork.contains(newWorkItem(r)) | ||
| } | ||
| override def newWorkItem(r: FutureTask[T]): WorkItem[T] = selfRef.newWorkItem(r) | ||
| }) | ||
| runInBackground() | ||
| } | ||
|
|
||
| override protected def getRunnable() = { | ||
| () => { | ||
|
|
||
| while (true) { | ||
| val start = clock.now() | ||
|
|
||
| val workList = generateWorkFunc() | ||
| val addedWork = workList.filter(!setOfWork.contains(_)) | ||
| val removedWork = CollectionHasAsScala(setOfWork).asScala.filter(item => !workList.contains(item)) | ||
|
|
||
| removedWork.foreach( item => { | ||
| setOfWork.remove(item) | ||
| logger.info("Removed work item from viewport threadpool:" + item) | ||
| }) | ||
|
|
||
| addedWork.foreach(item => { | ||
| //println("Adding" + item.hashCode()) | ||
| setOfWork.add(item) | ||
| }) | ||
|
|
||
| retryExecutor match { | ||
| case Some(executor) => { | ||
| addedWork.foreach(work => { | ||
| executor.submit(new Callable[T] { | ||
| override def call(): T = { | ||
| logger.info("Adding work to vp threadpool.." + work) | ||
| work.doWork() | ||
| } | ||
| }) | ||
| }) | ||
| } | ||
| case None => | ||
| } | ||
|
|
||
| val end = clock.now() | ||
|
|
||
| doMinCycleTime(start, end) | ||
| if (Thread.interrupted()) { | ||
| //shouldContinue.set(false) | ||
| logger.debug(s"[$name] interrupted or run once, going to exit") | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| def newCallable(r: FutureTask[T]): Callable[T] | ||
| def newWorkItem(r: FutureTask[T]): WorkItem[T] | ||
|
|
||
| override def doStop(): Unit = { | ||
| retryExecutor match { | ||
| case Some(executor) => { | ||
| executor.shutdown() | ||
| } | ||
| case None => //all good | ||
| } | ||
| logger.info(s"[$name] is exiting....") | ||
| } | ||
|
|
||
| override def doInitialize(): Unit = {} | ||
|
|
||
| override def doDestroy(): Unit = {} | ||
|
|
||
| override val lifecycleId: String = this.getClass.getName + "#" + this.hashCode() | ||
| } |
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
56 changes: 56 additions & 0 deletions
56
toolbox/src/main/scala/org/finos/toolbox/thread/executor/ResubmitExecutor.scala
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,56 @@ | ||
| package org.finos.toolbox.thread.executor | ||
|
|
||
| import com.typesafe.scalalogging.StrictLogging | ||
| import org.finos.toolbox.logging.LogAtFrequency | ||
| import org.finos.toolbox.thread.WorkItem | ||
| import org.finos.toolbox.time.Clock | ||
|
|
||
| import java.util | ||
| import java.util.concurrent.atomic.AtomicBoolean | ||
| import java.util.concurrent.{BlockingQueue, Callable, FutureTask, RunnableFuture, ThreadPoolExecutor} | ||
| import scala.concurrent.duration.TimeUnit | ||
|
|
||
| /** | ||
| * This is a java executor implementation, which will automatically resubmit the existing job into the queue | ||
| * when its complete. | ||
| * | ||
| */ | ||
| abstract class ResubmitExecutor[T](corePoolSize: Int, maxPoolSize: Int, keepAliveTime: Long, timeUnit: TimeUnit, | ||
| workQueue: BlockingQueue[Runnable])(implicit clock: Clock) extends ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, timeUnit, workQueue) with StrictLogging { | ||
|
|
||
| private val logEvery = new LogAtFrequency(5_000) | ||
| private val shuttingDown = new AtomicBoolean(false) | ||
|
|
||
| override def shutdown(): Unit = { | ||
| shuttingDown.set(true) | ||
| super.shutdown() | ||
| } | ||
|
|
||
| override def shutdownNow(): util.List[Runnable] = { | ||
| shuttingDown.set(true) | ||
| super.shutdownNow() | ||
| } | ||
|
|
||
| override def afterExecute(r: Runnable, t: Throwable): Unit = { | ||
| super.afterExecute(r, t) | ||
|
|
||
| if(!shuttingDown.get()){ | ||
| val futureTask = r.asInstanceOf[FutureTask[T]] | ||
| if(shouldResubmit(futureTask, t)){ | ||
| retry(futureTask, t) | ||
| if(logEvery.shouldLog()){ | ||
| logger.info("Finished runnable" + futureTask.get() + " resubmitting...") | ||
| } | ||
| } | ||
|
|
||
| } | ||
| } | ||
|
|
||
| def retry(runnable: FutureTask[T], t: Throwable): Unit ={ | ||
| this.submit(newCallable(runnable, t)) | ||
| } | ||
|
|
||
| def newCallable(r: FutureTask[T], t: Throwable): Callable[T] | ||
| def shouldResubmit(r: FutureTask[T], t: Throwable): Boolean | ||
| def newWorkItem(r: FutureTask[T]): WorkItem[T] | ||
| } |
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.