Ledger API Server: Named threadpools (#9588)

Some cached threadpools weren't given names, meaning at runtime there are a bunch of pool-x-thread-y threads. This makes it hard to understand which threads are being used for what. The following pool names were introduced: append-only indexer: input-mapping-pool, batching-pool ProgramResource: program-resource-pool kvutils PackageCommitter: package-preload-executor CHANGELOG_BEGIN CHANGELOG_END
digital-asset · May 6, 2021 · 5128206 · 5128206
1 parent 26a53d8
commit 5128206
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 3 deletions.
diff --git a/...r/participant-integration-api/src/main/scala/platform/indexer/parallel/AsyncSupport.scala b/...r/participant-integration-api/src/main/scala/platform/indexer/parallel/AsyncSupport.scala
@@ -8,6 +8,7 @@ import java.util.concurrent.Executors
 import com.codahale.metrics.{InstrumentedExecutorService, MetricRegistry}
 import com.daml.ledger.resources.ResourceOwner
 import com.daml.metrics.MetricName
+import com.google.common.util.concurrent.ThreadFactoryBuilder
 
 import scala.concurrent.{ExecutionContext, Future}
 
@@ -20,10 +21,14 @@ object AsyncSupport {
 
   def asyncPool(
       size: Int,
+      namePrefix: String,
       withMetric: Option[(MetricName, MetricRegistry)] = None,
   ): ResourceOwner[Executor] =
     ResourceOwner.forCloseable { () =>
-      val executor = Executors.newFixedThreadPool(size)
+      val executor = Executors.newFixedThreadPool(
+        size,
+        new ThreadFactoryBuilder().setNameFormat(s"$namePrefix-%d").build,
+      )
       val workerE = withMetric match {
         case Some((metricName, metricRegistry)) =>
           new InstrumentedExecutorService(

diff --git a/...ant-integration-api/src/main/scala/platform/indexer/parallel/ParallelIndexerFactory.scala b/...ant-integration-api/src/main/scala/platform/indexer/parallel/ParallelIndexerFactory.scala
@@ -48,10 +48,12 @@ object ParallelIndexerFactory {
     for {
       inputMapperExecutor <- asyncPool(
         inputMappingParallelism,
+        "input-mapping-pool",
         Some(metrics.daml.parallelIndexer.inputMapping.executor -> metrics.registry),
       )
       batcherExecutor <- asyncPool(
         batchingParallelism,
+        "batching-pool",
         Some(metrics.daml.parallelIndexer.batching.executor -> metrics.registry),
       )
       dbDispatcher <- DbDispatcher

diff --git a/...src/main/scala/com/daml/ledger/participant/state/kvutils/committer/PackageCommitter.scala b/...src/main/scala/com/daml/ledger/participant/state/kvutils/committer/PackageCommitter.scala
@@ -340,7 +340,7 @@ final private[kvutils] class PackageCommitter(
 
   private lazy val preloadExecutor =
     Executors.newSingleThreadExecutor { (runnable: Runnable) =>
-      val t = new Thread(runnable)
+      val t = new Thread(runnable, "package-preload-executor")
       t.setDaemon(true)
       t
     }

diff --git a/libs-scala/resources/src/main/scala/com/digitalasset/resources/ProgramResource.scala b/libs-scala/resources/src/main/scala/com/digitalasset/resources/ProgramResource.scala
@@ -3,6 +3,7 @@
 
 package com.daml.resources
 
+import java.util.concurrent.atomic.AtomicLong
 import java.util.concurrent.{Executors, TimeUnit}
 
 import com.daml.logging.ContextualizedLogger
@@ -20,7 +21,12 @@ final class ProgramResource[Context: HasExecutionContext, T](
 ) {
   private val logger = ContextualizedLogger.get(getClass)
 
-  private val executorService = Executors.newCachedThreadPool()
+  private val executorService = {
+    val counter = new AtomicLong(0L)
+    Executors.newCachedThreadPool((runnable: Runnable) =>
+      new Thread(runnable, s"program-resource-pool-${counter.incrementAndGet()}")
+    )
+  }
 
   def run(newContext: ExecutionContext => Context): Unit = {
     newLoggingContext { implicit loggingContext =>
@@ -72,4 +78,5 @@ object ProgramResource {
   trait SuppressedStartupException {
     self: Exception =>
   }
+
 }