TheAnachronism/docspell
1
0
Fork 0
mirror of https://github.com/TheAnachronism/docspell.git synced 2025-06-06 15:15:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

Properly initialize thread pools

Browse Source
This commit is contained in:
Eike Kettner 2020-03-17 22:37:12 +01:00
parent 00ca6b5697
commit f0449dd2ce
3 changed files with 40 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
modules/common/src/main/scala/docspell/common/ThreadFactories.scala
View File

@ -2,6 +2,8 @@ package docspell.common
import java.util.concurrent.atomic.AtomicLong import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.{Executors, ThreadFactory} import java.util.concurrent.{Executors, ThreadFactory}
import cats.effect._
import scala.concurrent._
object ThreadFactories { object ThreadFactories {
@ -17,4 +19,16 @@ object ThreadFactories {
} }
} }
def executorResource[F[_]: Sync](
c: => ExecutionContextExecutorService
): Resource[F, ExecutionContextExecutorService] =
Resource.make(Sync[F].delay(c))(ec => Sync[F].delay(ec.shutdown))
def cached[F[_]: Sync](tf: ThreadFactory): Resource[F, ExecutionContextExecutorService] =
executorResource(
ExecutionContext.fromExecutorService(Executors.newCachedThreadPool(tf))
)
def fixed[F[_]: Sync](n: Int, tf: ThreadFactory): Resource[F, ExecutionContextExecutorService] =
executorResource(ExecutionContext.fromExecutorService(Executors.newFixedThreadPool(n, tf)))
} }

23
modules/joex/src/main/scala/docspell/joex/Main.scala
View File

@ -3,9 +3,8 @@ package docspell.joex
import cats.effect.{Blocker, ExitCode, IO, IOApp} import cats.effect.{Blocker, ExitCode, IO, IOApp}
import cats.implicits._ import cats.implicits._
import scala.concurrent.ExecutionContext
import java.util.concurrent.Executors
import java.nio.file.{Files, Paths} import java.nio.file.{Files, Paths}
import scala.concurrent.ExecutionContext
import docspell.common.{Banner, ThreadFactories} import docspell.common.{Banner, ThreadFactories}
import org.log4s._ import org.log4s._
@ -13,13 +12,8 @@ import org.log4s._
object Main extends IOApp { object Main extends IOApp {
private[this] val logger = getLogger private[this] val logger = getLogger
val blockingEC: ExecutionContext = ExecutionContext.fromExecutor( val blockingEC = ThreadFactories.cached[IO](ThreadFactories.ofName("docspell-joex-blocking"))
Executors.newCachedThreadPool(ThreadFactories.ofName("docspell-joex-blocking")) val connectEC = ThreadFactories.fixed[IO](5, ThreadFactories.ofName("docspell-joex-dbconnect"))
)
val blocker = Blocker.liftExecutionContext(blockingEC)
val connectEC: ExecutionContext = ExecutionContext.fromExecutorService(
Executors.newFixedThreadPool(5, ThreadFactories.ofName("docspell-joex-dbconnect"))
)
def run(args: List[String]) = { def run(args: List[String]) = {
args match { args match {
@ -52,6 +46,15 @@ object Main extends IOApp {
cfg.baseUrl cfg.baseUrl
) )
logger.info(s"\n${banner.render("***>")}") logger.info(s"\n${banner.render("***>")}")
JoexServer.stream[IO](cfg, connectEC, blockingEC, blocker).compile.drain.as(ExitCode.Success) val pools = for {
cec <- connectEC
bec <- blockingEC
blocker = Blocker.liftExecutorService(bec)
} yield Pools(cec, bec, blocker)
pools.use(p =>
JoexServer.stream[IO](cfg, p.connectEC, p.clientEC, p.blocker).compile.drain.as(ExitCode.Success)
)
} }
case class Pools(connectEC: ExecutionContext, clientEC: ExecutionContext, blocker: Blocker)
} }

23
modules/restserver/src/main/scala/docspell/restserver/Main.scala
View File

@ -4,7 +4,6 @@ import cats.effect._
import cats.implicits._ import cats.implicits._
import scala.concurrent.ExecutionContext import scala.concurrent.ExecutionContext
import java.util.concurrent.Executors
import java.nio.file.{Files, Paths} import java.nio.file.{Files, Paths}
import docspell.common.{Banner, ThreadFactories} import docspell.common.{Banner, ThreadFactories}
@ -13,14 +12,8 @@ import org.log4s._
object Main extends IOApp { object Main extends IOApp {
private[this] val logger = getLogger private[this] val logger = getLogger
val blockingEc: ExecutionContext = ExecutionContext.fromExecutor( val blockingEC = ThreadFactories.cached[IO](ThreadFactories.ofName("docspell-restserver-blocking"))
Executors.newCachedThreadPool(ThreadFactories.ofName("docspell-restserver-blocking")) val connectEC = ThreadFactories.fixed[IO](5, ThreadFactories.ofName("docspell-dbconnect"))
)
val blocker = Blocker.liftExecutionContext(blockingEc)
val connectEC: ExecutionContext = ExecutionContext.fromExecutorService(
Executors.newFixedThreadPool(5, ThreadFactories.ofName("docspell-dbconnect"))
)
def run(args: List[String]) = { def run(args: List[String]) = {
args match { args match {
@ -52,7 +45,17 @@ object Main extends IOApp {
cfg.appId, cfg.appId,
cfg.baseUrl cfg.baseUrl
) )
val pools = for {
cec <- connectEC
bec <- blockingEC
blocker = Blocker.liftExecutorService(bec)
} yield Pools(cec, bec, blocker)
logger.info(s"\n${banner.render("***>")}") logger.info(s"\n${banner.render("***>")}")
RestServer.stream[IO](cfg, connectEC, blockingEc, blocker).compile.drain.as(ExitCode.Success) pools.use(p =>
RestServer.stream[IO](cfg, p.connectEC, p.clientEC, p.blocker).compile.drain.as(ExitCode.Success)
)
} }
case class Pools(connectEC: ExecutionContext, clientEC: ExecutionContext, blocker: Blocker)
} }