Fix thread blocking issue, confirmed with integration tests

hms-dbmi · Dec 18, 2023 · 83640e7 · 83640e7
1 parent 1a505a6
commit 83640e7
Show file tree

Hide file tree

Showing 5 changed files with 18 additions and 3 deletions.
diff --git a/...g/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/processing/GenomicProcessorConfig.java b/...g/src/main/java/edu/harvard/hms/dbmi/avillach/hpds/processing/GenomicProcessorConfig.java
@@ -38,6 +38,16 @@ public GenomicProcessor localDistributedGenomicProcessor() {
         return new GenomicProcessorParentImpl(processorNodes);
     }
 
+    @Bean(name = "integrationTestGenomicProcessor")
+    @ConditionalOnProperty(prefix = "hpds.genomicProcessor", name = "impl", havingValue = "integrationTest")
+    public GenomicProcessor integrationTestGenomicProcessor() {
+        // todo: parameterize these
+        return new GenomicProcessorParentImpl(List.of(
+                new GenomicProcessorNodeImpl("/Users/ryan/dev/pic-sure-hpds-test/data/orchestration/1040.22/all/"),
+                new GenomicProcessorNodeImpl("/Users/ryan/dev/pic-sure-hpds-test/data/orchestration/1040.20/all/")
+        ));
+    }
+
     @Bean(name = "remoteGenomicProcessor")
     @ConditionalOnProperty(prefix = "hpds.genomicProcessor", name = "impl", havingValue = "remote")
     public GenomicProcessor remoteGenomicProcessor() {

diff --git a/...src/main/java/edu/harvard/hms/dbmi/avillach/hpds/processing/GenomicProcessorNodeImpl.java b/...src/main/java/edu/harvard/hms/dbmi/avillach/hpds/processing/GenomicProcessorNodeImpl.java
@@ -78,7 +78,7 @@ public GenomicProcessorNodeImpl(String genomicDataDirectory) {
 
     @Override
     public Mono<BigInteger> getPatientMask(DistributableQuery distributableQuery) {
-        return Mono.fromCallable(() -> runGetPatientMask(distributableQuery));
+        return Mono.fromCallable(() -> runGetPatientMask(distributableQuery)).subscribeOn(Schedulers.boundedElastic());
     }
     public BigInteger runGetPatientMask(DistributableQuery distributableQuery) {
 //		log.debug("filterdIDSets START size: " + filteredIdSets.size());

diff --git a/...c/main/java/edu/harvard/hms/dbmi/avillach/hpds/processing/GenomicProcessorParentImpl.java b/...c/main/java/edu/harvard/hms/dbmi/avillach/hpds/processing/GenomicProcessorParentImpl.java
@@ -48,7 +48,6 @@ public GenomicProcessorParentImpl(List<GenomicProcessor> nodes) {
     @Override
     public Mono<BigInteger> getPatientMask(DistributableQuery distributableQuery) {
         Mono<BigInteger> result = Flux.just(nodes.toArray(GenomicProcessor[]::new))
-                .publishOn(Schedulers.boundedElastic())
                 .flatMap(node -> node.getPatientMask(distributableQuery))
                 .reduce(BigInteger::or);
         return result;

diff --git a/service/src/main/resources/application-integration-test.properties b/service/src/main/resources/application-integration-test.properties
@@ -0,0 +1,5 @@
+SMALL_JOB_LIMIT = 100
+SMALL_TASK_THREADS = 1
+LARGE_TASK_THREADS = 1
+
+hpds.genomicProcessor.impl=integrationTest
diff --git a/...ava/edu/harvard/hms/dbmi/avillach/hpds/service/util/AbstractProcessorIntegrationTest.java b/...ava/edu/harvard/hms/dbmi/avillach/hpds/service/util/AbstractProcessorIntegrationTest.java
@@ -13,7 +13,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
 import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.ActiveProfiles;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
 import java.util.*;
@@ -24,6 +24,7 @@
 @ExtendWith(SpringExtension.class)
 @EnableAutoConfiguration
 @SpringBootTest(classes = HpdsApplication.class)
+@ActiveProfiles("integration-test")
 public class AbstractProcessorIntegrationTest {
 
     @Autowired