Fix MCS deadlock: poll() timeouts, future cancellation, SMSD 6.10.1

Root cause: CompletionService.take() blocks forever when MCS workers
hang inside the SMSD engine. Replaced with poll(timeout) at three levels:

- GraphMatcher: 15s per-pair poll + 60s overall matcher budget
- CallableAtomMappingTool: 120s per-algorithm poll
- Phase 1 inline .call() moved to executor with timeout

Orphaned futures are now cancelled (interrupt=true) after timeout to
free the shared static MAPPING_EXECUTOR pool. GraphMatcher finally
block uses shutdownNow() to interrupt stuck local MCS threads.

Also: SMSD upgraded to 6.10.1; energy filter NPE guard added in
SmsdReactionMappingEngine.applyDefaultFilters().

Author: asad

ALGORITHM.md

@@ -171,7 +171,7 @@ For each reactant-product pair *(R_i, P_j)*, compute a Maximum Common Subgraph (
         φ_{ij} := {(a_k, a_k) : k = 1…|A(R_i)|}   (direct 1:1 mapping)
         skip MCS
 
-Canonical SMILES are generated by `MolGraph.toCanonicalSmiles()` (SMSD 6.9.1), which encodes tetrahedral chirality (`@`/`@@`) and E/Z geometry (`/`/`\`). This is essential: using a stereo-unaware generator would incorrectly short-circuit enantiomers (e.g. (R)-lactic acid ≡ (S)-lactic acid) to a spurious identity mapping.
+Canonical SMILES are generated by `MolGraph.toCanonicalSmiles()` (SMSD 6.10.1), which encodes tetrahedral chirality (`@`/`@@`) and E/Z geometry (`/`/`\`). This is essential: using a stereo-unaware generator would incorrectly short-circuit enantiomers (e.g. (R)-lactic acid ≡ (S)-lactic acid) to a spurious identity mapping.
 
 **Stage 2 — Size ratio filter:**
 
@@ -393,7 +393,7 @@ RINGS resolves the majority of reactions via the funnel at a 2-4x computational
 
 | Component | Version | Role |
 |-----------|---------|------|
-| SMSD | 6.9.1 | MCS engine: VF2++ subgraph isomorphism, circular/path fingerprints, MolGraph canonical SMILES (stereo-aware) |
+| SMSD | 6.10.1 | MCS engine: VF2++ subgraph isomorphism, circular/path fingerprints, MolGraph canonical SMILES (stereo-aware) |
 | CDK | 2.12 | Molecule I/O, atom typing, aromaticity perception, ring finding |
 | Java | 21+ | Platform |
 

changes.log

@@ -284,7 +284,14 @@ c) clean up
 -----------------------
 Changes (2026-04-03) — v3.9.0
 -----------------------
-a) Identity shortcut stereo + multiplicity fix: isIdentityReaction() now uses
+a) SMSD upgraded to 6.10.1 with energy filter NPE guard.
+b) Deadlock fix: GraphMatcher and CallableAtomMappingTool now use
+   CompletionService.poll(timeout) instead of take() (which blocks forever),
+   with per-pair budget (15s), overall matcher budget (60s), and algorithm
+   worker timeout (120s). Orphaned futures are cancelled on timeout to free
+   the shared executor pool. Phase 1 inline call moved to executor with
+   timeout to prevent main-thread deadlocks.
+c) Identity shortcut stereo + multiplicity fix: isIdentityReaction() now uses
    SmiFlavor.Canonical|Stereo (E/Z and R/S are distinguished) and a sorted
    List instead of a TreeSet (stoichiometric multiplicity is preserved).
    Previously, F/C=C/F>>F/C=C\F was incorrectly classified as a transporter

pom.xml

@@ -178,7 +178,7 @@
         <dependency>
             <groupId>com.bioinceptionlabs</groupId>
             <artifactId>smsd</artifactId>
-            <version>6.9.1</version>
+            <version>6.10.1</version>
         </dependency>
 
         <!-- https://mvnrepository.com/artifact/commons-cli/commons-cli -->

src/main/java/com/bioinceptionlabs/reactionblast/mapping/CallableAtomMappingTool.java

@@ -59,6 +59,8 @@ public class CallableAtomMappingTool implements Serializable {
     private final static ILoggingTool LOGGER
             = createLoggingTool(CallableAtomMappingTool.class);
     private static final long serialVersionUID = 0x29e2adb1716b13eL;
+    /** Hard timeout per algorithm worker. Covers all MCS pairs + matrix selection. */
+    private static final long ALGORITHM_TIMEOUT_MS = 120_000L; // 2 minutes
     private static final int MAPPING_PARALLELISM
             = Math.max(2, Math.min(3, Runtime.getRuntime().availableProcessors()));
     private static final ExecutorService MAPPING_EXECUTOR
@@ -146,10 +148,13 @@ private void generateAtomAtomMapping(
         boolean hasRings = hasRingSystems(standardizedReaction);
         IMappingAlgorithm firstPass = (checkComplex && hasRings) ? RINGS : MIN;
         if (totalMolecules <= 5) {
+            java.util.concurrent.Future<Reactor> phase1Future = null;
             try {
-                Reactor firstPassResult = new MappingThread(
+                phase1Future = MAPPING_EXECUTOR.submit(new MappingThread(
                         "IMappingAlgorithm." + firstPass.name(),
-                        standardizedReaction, firstPass, removeHydrogen).call();
+                        standardizedReaction, firstPass, removeHydrogen));
+                Reactor firstPassResult = phase1Future.get(
+                        ALGORITHM_TIMEOUT_MS, java.util.concurrent.TimeUnit.MILLISECONDS);
                 putSolution(firstPass, firstPassResult);
 
                 if (isMappingAcceptable(firstPassResult)) {
@@ -158,6 +163,11 @@ private void generateAtomAtomMapping(
                     return;
                 }
                 LOGGER.debug(firstPass + " mapping insufficient — running remaining algorithms");
+            } catch (java.util.concurrent.TimeoutException e) {
+                LOGGER.warn(firstPass + " phase timed out after " + ALGORITHM_TIMEOUT_MS + "ms");
+                if (phase1Future != null) {
+                    phase1Future.cancel(true);
+                }
             } catch (InterruptedException | ExecutionException e) {
                 LOGGER.debug(firstPass + " phase failed: " + e.getMessage());
                 LOGGER.error(e);
@@ -184,24 +194,31 @@ private void generateAtomAtomMapping(
             remaining = new IMappingAlgorithm[]{MIN, MAX, MIXTURE, RINGS};
         }
 
+        java.util.List<java.util.concurrent.Future<Reactor>> submittedFutures = new java.util.ArrayList<>();
         try {
             CompletionService<Reactor> cs = new ExecutorCompletionService<>(MAPPING_EXECUTOR);
             int jobCounter = 0;
             for (IMappingAlgorithm algo : remaining) {
                 LOGGER.debug("Submitting " + algo.description());
-                cs.submit(new MappingThread("IMappingAlgorithm." + algo.name(),
-                        standardizedReaction, algo, removeHydrogen));
+                submittedFutures.add(cs.submit(new MappingThread("IMappingAlgorithm." + algo.name(),
+                        standardizedReaction, algo, removeHydrogen)));
                 jobCounter++;
             }
+            int collected = 0;
             for (int i = 0; i < jobCounter; i++) {
                 try {
-                    Reactor chosen = cs.take().get();
+                    java.util.concurrent.Future<Reactor> future =
+                            cs.poll(ALGORITHM_TIMEOUT_MS, java.util.concurrent.TimeUnit.MILLISECONDS);
+                    if (future == null) {
+                        LOGGER.warn("Algorithm poll timed out after " + ALGORITHM_TIMEOUT_MS
+                                + "ms — " + (jobCounter - collected) + " remaining algorithms skipped");
+                        break;
+                    }
+                    collected++;
+                    Reactor chosen = future.get(); // already complete
                     putSolution(chosen.getAlgorithm(), chosen);
                 } catch (ExecutionException e) {
-                    // One algorithm worker failed — log and continue collecting the rest.
-                    // The lower MCS layer already handles per-pair failures; this catch
-                    // prevents a single bad algorithm from silently dropping all remaining
-                    // successful results.
+                    collected++;
                     LOGGER.debug("Algorithm worker failed: " + e.getCause());
                     LOGGER.error(e);
                 } catch (InterruptedException e) {
@@ -210,6 +227,13 @@ private void generateAtomAtomMapping(
                     break;
                 }
             }
+            // Cancel any orphaned workers still running in the shared pool
+            // so they don't starve future mapping requests
+            for (java.util.concurrent.Future<Reactor> f : submittedFutures) {
+                if (!f.isDone()) {
+                    f.cancel(true);
+                }
+            }
             LOGGER.debug("======DONE CallableAtomMappingTool=======");
         } catch (Exception e) {
             LOGGER.debug("ERROR: in AtomMappingTool: " + e.getMessage());

src/main/java/com/bioinceptionlabs/reactionblast/mapping/GraphMatcher.java

@@ -78,6 +78,12 @@ public class GraphMatcher extends Debugger {
     private static final int SINGLE_SUBGRAPH_MATCH = 1;
     private static final long SUBGRAPH_TIMEOUT_MS = 5_000L;
     private static final long MCS_TIMEOUT_MS = 10_000L;
+    /** Hard timeout per poll() call waiting for the next completed MCS pair. */
+    private static final long MCS_POLL_TIMEOUT_MS = 15_000L;
+    /** Overall wall-clock budget for the entire matcher() call. */
+    private static final long MATCHER_BUDGET_MS = 60_000L;
+    /** Hard timeout for the executor shutdown after collection. */
+    private static final long MATCHER_SHUTDOWN_TIMEOUT_MS = 2_000L;
 
     /**
      * @author Syed Asad Rahman <asad.rahman@bioinceptionlabs.com>
@@ -449,22 +455,45 @@ public static Collection<MCSSolution> matcher(Holder mh) throws Exception {
 
             LOGGER.debug("submited " + taskCounter + " jobs");
             Collection<MCSSolution> threadedUniqueMCSSolutions = new ArrayList<>();
+            int collected = 0;
+            long matcherDeadline = currentTimeMillis() + MATCHER_BUDGET_MS;
             for (int count = 0; count < taskCounter; count++) {
                 try {
-                    MCSSolution isomorphism = callablesQueue.take().get();
-                    threadedUniqueMCSSolutions.add(isomorphism);
+                    long remaining = matcherDeadline - currentTimeMillis();
+                    if (remaining <= 0) {
+                        LOGGER.warn("Matcher budget (" + MATCHER_BUDGET_MS
+                                + "ms) exhausted — " + (taskCounter - collected)
+                                + " remaining pairs skipped");
+                        break;
+                    }
+                    long pollMs = Math.min(remaining, MCS_POLL_TIMEOUT_MS);
+                    java.util.concurrent.Future<MCSSolution> future =
+                            callablesQueue.poll(pollMs, TimeUnit.MILLISECONDS);
+                    if (future == null) {
+                        LOGGER.warn("MCS poll timed out after " + pollMs
+                                + "ms — " + (taskCounter - collected)
+                                + " remaining pairs will be skipped");
+                        break;
+                    }
+                    MCSSolution isomorphism = future.get(); // already complete
+                    if (isomorphism != null) {
+                        threadedUniqueMCSSolutions.add(isomorphism);
+                    }
+                    collected++;
                 } catch (ExecutionException ex) {
+                    collected++;
                     Throwable cause = ex.getCause() != null ? ex.getCause() : ex;
                     LOGGER.error(SEVERE, "MCS worker failed", cause);
                 }
             }
             // Add directly-constructed identity mappings (bypassed MCSThread)
             threadedUniqueMCSSolutions.addAll(directMCSSolutions);
-            // This will make the executor accept no new threads
-            // and finish all existing threads in the queue
+            // Shut down the local executor; interrupt any stuck threads
             executor.shutdown();
-            // Wait until all threads are finished
-            executor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
+            if (!executor.awaitTermination(MATCHER_SHUTDOWN_TIMEOUT_MS, TimeUnit.MILLISECONDS)) {
+                LOGGER.warn("MCS executor did not shut down cleanly — forcing shutdown");
+                executor.shutdownNow();
+            }
 
             LOGGER.debug("==Gathering MCS solution from the Thread==");
             long replayedMappings = 0;
@@ -502,7 +531,7 @@ public static Collection<MCSSolution> matcher(Holder mh) throws Exception {
             LOGGER.error(SEVERE, null, ex);
         } finally {
             if (executor != null) {
-                executor.shutdown();
+                executor.shutdownNow();
             }
         }
         return unmodifiableCollection(mcsSolutions);

src/main/java/com/bioinceptionlabs/reactionblast/mapping/SmsdReactionMappingEngine.java

@@ -105,7 +105,17 @@ public BaseMapping findSubstructure(IQueryAtomContainer query,
     @Override
     public void applyDefaultFilters(BaseMapping mapping) {
         if (mapping != null) {
-            mapping.setChemFilters(true, true, true);
+            try {
+                mapping.setChemFilters(true, true, true);
+            } catch (NullPointerException ex) {
+                // SMSD 6.10.1 energy filter can NPE on certain molecule pairs
+                // where bond-energy lookup returns null. Fall back to stereo+fragment only.
+                try {
+                    mapping.setChemFilters(true, true, false);
+                } catch (Exception fallback) {
+                    // last resort — no filters at all
+                }
+            }
         }
     }
 }