Add proper support for initializing ontologies from stream

arteymix · arteymix · commit d8e8f5c87d4e · 2023-04-06T10:20:31.000-07:00
Deprecate loadTermsInNameSpace and use the improved initialize() method
from a stream.

Improve documentations for the ontology service interface.

Add more checks for the current thread interruption in initialize().
diff --git a/src/ubic/basecode/ontology/jena/AbstractOntologyService.java b/src/ubic/basecode/ontology/jena/AbstractOntologyService.java
@@ -88,9 +88,15 @@ public abstract class AbstractOntologyService implements OntologyService {
 
     private boolean isInitialized = false;
 
-
-    @Override
     public void initialize( boolean forceLoad, boolean forceIndexing ) {
+        initialize( null, forceLoad, forceIndexing );
+    }
+
+    public void initialize( InputStream stream, boolean forceIndexing ) {
+        initialize( stream, true, forceIndexing );
+    }
+
+    private void initialize( InputStream stream, boolean forceLoad, boolean forceIndexing ) {
         if ( !forceLoad && isInitialized ) {
             log.warn( "{} is already loaded, and force=false, not restarting", this );
             return;
@@ -118,20 +124,21 @@ public void initialize( boolean forceLoad, boolean forceIndexing ) {
         log.info( "Loading ontology: {}...", this );
         StopWatch loadTime = StopWatch.createStarted();
 
-        // use temporary variables so we can minimize the critical region for replacing the service's state
-        Map<String, OntologyTerm> terms = new HashMap<>();
-        Map<String, OntologyIndividual> individuals = new HashMap<>();
+        // use temporary variables, so that we can minimize the critical region for replacing the service's state
         OntModel model;
         SearchIndex index;
 
-        if ( Thread.currentThread().isInterrupted() ) {
-            log.warn( "The current thread is interrupted, initialization of {} will be stop.", this );
+        // loading the model from disk or URL is lengthy
+        if ( checkIfInterrupted() )
             return;
-        }
 
-        model = loadModel(); // can take a while.
+        model = stream != null ? loadModelFromStream( stream ) : loadModel(); // can take a while.
         assert model != null;
 
+        // retrieving restrictions is lengthy
+        if ( checkIfInterrupted() )
+            return;
+
         // compute additional restrictions
         Set<Restriction> additionalRestrictions = model.listRestrictions()
                 .filterKeep( new RestrictionWithOnPropertyFilter( additionalProperties ) )
@@ -147,12 +154,15 @@ public void initialize( boolean forceLoad, boolean forceIndexing ) {
          */
         boolean force = forceReindexing || changed || !indexExists;
 
+        // indexing is lengthy, don't bother if we're interrupted
+        if ( checkIfInterrupted() )
+            return;
+
         index = OntologyIndexer.indexOntology( getCacheName(), model, force );
 
-        if ( Thread.currentThread().isInterrupted() ) {
-            log.warn( "The current thread is interrupted, initialization of {} will be stop.", this );
+        // if interrupted, we don't need to replace the model and clear the *old* cache
+        if ( checkIfInterrupted() )
             return;
-        }
 
         Lock lock = rwLock.writeLock();
         try {
@@ -161,18 +171,25 @@ public void initialize( boolean forceLoad, boolean forceIndexing ) {
             this.additionalRestrictions = additionalRestrictions;
             this.index = index;
             this.isInitialized = true;
+            // now that the terms have been replaced, we can clear old caches
+            OntologyLoader.deleteOldCache( getCacheName() );
         } finally {
             lock.unlock();
         }
 
-        // now that the terms have been replaced, we can clear old caches
-        OntologyLoader.deleteOldCache( getCacheName() );
-
         loadTime.stop();
 
         log.info( "Finished loading {} in {}s", this, String.format( "%.2f", loadTime.getTime() / 1000.0 ) );
     }
 
+    private boolean checkIfInterrupted() {
+        if ( Thread.interrupted() ) {
+            log.warn( "The current thread is interrupted, initialization of {} will be stop.", this );
+            return true;
+        }
+        return false;
+    }
+
     /**
      * Do not do this except before re-indexing.
      */
@@ -496,6 +513,14 @@ public void waitForInitializationThread() throws InterruptedException {
      */
     protected abstract OntModel loadModel();
 
+
+    /**
+     * Load a model from a given input stream.
+     */
+    protected OntModel loadModelFromStream( InputStream is ) {
+        return OntologyLoader.loadMemoryModel( is, this.getOntologyUrl() );
+    }
+
     protected String getCacheName() {
         return getOntologyName();
     }
@@ -559,48 +584,29 @@ private void initSearchByAlternativeId() {
 
     @Override
     public void loadTermsInNameSpace( InputStream is, boolean forceIndex ) {
-        Lock lock = rwLock.writeLock();
-        try {
-            lock.lock();
-            this.isInitialized = false;
-
-            if ( initializationThread != null && initializationThread.isAlive() ) {
-                log.warn( "{} initialization is already running, trying to cancel ...", this );
-                initializationThread.interrupt();
-                // wait for the thread to die.
-                int maxWait = 10;
-                int wait = 0;
-                while ( initializationThread.isAlive() ) {
-                    try {
-                        initializationThread.join( 5000 );
-                        log.warn( "Waiting for auto-initialization to stop so manual initialization can begin ..." );
-                    } catch ( InterruptedException e ) {
-                        Thread.currentThread().interrupt();
-                        log.warn( "Got interrupted while waiting for the initialization thread of {} to finish.", this );
-                        return;
-                    }
-                    ++wait;
-                    if ( wait >= maxWait && !initializationThread.isAlive() ) {
-                        throw new RuntimeException( String.format( "Got tired of waiting for %s's initialization thread.", this ) );
-                    }
+        // wait for the initialization thread to finish
+        if ( initializationThread != null && initializationThread.isAlive() ) {
+            log.warn( "{} initialization is already running, trying to cancel ...", this );
+            initializationThread.interrupt();
+            // wait for the thread to die.
+            int maxWait = 10;
+            int wait = 0;
+            while ( initializationThread.isAlive() ) {
+                try {
+                    initializationThread.join( 5000 );
+                    log.warn( "Waiting for auto-initialization to stop so manual initialization can begin ..." );
+                } catch ( InterruptedException e ) {
+                    Thread.currentThread().interrupt();
+                    log.warn( "Got interrupted while waiting for the initialization thread of {} to finish.", this );
+                    return;
+                }
+                ++wait;
+                if ( wait >= maxWait && !initializationThread.isAlive() ) {
+                    throw new RuntimeException( String.format( "Got tired of waiting for %s's initialization thread.", this ) );
                 }
             }
-
-            this.model = OntologyLoader.loadMemoryModel( is, this.getOntologyUrl() );
-            this.additionalRestrictions = model.listRestrictions()
-                    .filterKeep( new RestrictionWithOnPropertyFilter( additionalProperties ) )
-                    .toSet();
-            this.index = OntologyIndexer.getSubjectIndex( getCacheName() );
-            if ( index == null || forceIndex ) {
-                this.index = OntologyIndexer.indexOntology( getCacheName(), model, true /* force */ );
-            }
-
-            isInitialized = true;
-        } finally {
-            lock.unlock();
         }
-
-        log.info( "Ontology {} is ready!", this );
+        initialize( is, forceIndex );
     }
 
     @Override
diff --git a/src/ubic/basecode/ontology/providers/OntologyService.java b/src/ubic/basecode/ontology/providers/OntologyService.java
@@ -21,6 +21,13 @@ public interface OntologyService {
      */
     void initialize( boolean forceLoad, boolean forceIndexing );
 
+    /**
+     * Initialize this ontology service from a stream.
+     * <p>
+     * Note that when this method of initialization is used, the ontology cache is not created on-disk.
+     */
+    void initialize( InputStream stream, boolean forceIndexing );
+
     /**
      * Looks for any individuals that match the given search string.
      * <p>
@@ -97,21 +104,55 @@ default Collection<OntologyTerm> findTerm( String search ) throws OntologySearch
      */
     OntologyTerm getTerm( String uri );
 
+    /**
+     * Obtain all the individuals for a given term URI.
+     */
     Collection<OntologyIndividual> getTermIndividuals( String uri );
 
+    /**
+     * Obtain all the parents of a given set of terms, excluding obsolete terms.
+     *
+     * @see #getParents(Collection, boolean, boolean, boolean)
+     */
     default Set<OntologyTerm> getParents( Collection<OntologyTerm> terms, boolean direct, boolean includeAdditionalProperties ) {
         return getParents( terms, direct, includeAdditionalProperties, false );
     }
 
+    /**
+     * Obtain all the parents of a given set of terms.
+     *
+     * @param terms                       set of terms whose parents are retrieved
+     * @param direct                      only retain direct parents
+     * @param includeAdditionalProperties also include parents matched via additional properties
+     * @param keepObsoletes               retain obsolete terms
+     * @return a set of parent terms
+     */
     Set<OntologyTerm> getParents( Collection<OntologyTerm> terms, boolean direct, boolean includeAdditionalProperties, boolean keepObsoletes );
 
+    /**
+     * Obtain all the children of a given set of terms, excluding obsolete terms.
+     *
+     * @see #getChildren(Collection, boolean, boolean, boolean)
+     */
     default Set<OntologyTerm> getChildren( Collection<OntologyTerm> terms, boolean direct, boolean includeAdditionalProperties ) {
         return getChildren( terms, direct, includeAdditionalProperties, false );
     }
 
+    /**
+     * Obtain all the children of a given set of terms.
+     *
+     * @param terms                       set of terms whose children are retrieved
+     * @param direct                      only retain direct children
+     * @param includeAdditionalProperties also include children matched via additional properties
+     * @param keepObsoletes               retain obsolete terms
+     * @return a set of child terms
+     */
     Set<OntologyTerm> getChildren( Collection<OntologyTerm> terms, boolean direct, boolean includeAdditionalProperties, boolean keepObsoletes );
 
 
+    /**
+     * Check if this ontology is enabled.
+     */
     boolean isEnabled();
 
     /**
@@ -133,10 +174,19 @@ default Set<OntologyTerm> getChildren( Collection<OntologyTerm> terms, boolean d
      */
     void startInitializationThread( boolean forceLoad, boolean forceIndexing );
 
+    /**
+     * Check if the initialization thread is alive.
+     */
     boolean isInitializationThreadAlive();
 
+    /**
+     * Check if the initialization thread is cancelled.
+     */
     boolean isInitializationThreadCancelled();
 
+    /**
+     * Cancel a running initialization thread.
+     */
     void cancelInitializationThread();
 
     /**
@@ -147,6 +197,7 @@ default Set<OntologyTerm> getChildren( Collection<OntologyTerm> terms, boolean d
     /**
      * Index the ontology for performing full-text searches.
      *
+     * @param force if true, perform indexing even if an index already exists
      * @see #findIndividuals(String)
      * @see #findTerm(String)
      * @see #findResources(String)
@@ -159,6 +210,9 @@ default Set<OntologyTerm> getChildren( Collection<OntologyTerm> terms, boolean d
      *
      * @param is         input stream from which the ontology model is loaded
      * @param forceIndex initialize the index. Otherwise it will only be initialized if it doesn't exist.
+     * @deprecated use {@link #initialize(InputStream, boolean)} instead and possibly {@link #cancelInitializationThread()}
+     * prior to get any running initialization thread out of the way
      */
+    @Deprecated
     void loadTermsInNameSpace( InputStream is, boolean forceIndex );
 }
diff --git a/test/ubic/basecode/ontology/OntologyTermTest.java b/test/ubic/basecode/ontology/OntologyTermTest.java
@@ -52,7 +52,7 @@ protected String getCacheName() {
         };
         try ( InputStream is = new GZIPInputStream( requireNonNull( OntologyTermTest.class.getResourceAsStream( "/data/uberon.owl.gz" ) ) ) ) {
             // FIXME: indexing Uberon is very slow, so we disable it so if the tests are breaking, try force-indexing
-            uberon.loadTermsInNameSpace( is, false );
+            uberon.initialize( is, false );
         }
     }
 
@@ -61,7 +61,7 @@ public void testGetChildren() throws Exception {
         // DOID:4159
         DiseaseOntologyService s = new DiseaseOntologyService();
         InputStream is = new GZIPInputStream( requireNonNull( this.getClass().getResourceAsStream( "/data/doid.short.owl.gz" ) ) );
-        s.loadTermsInNameSpace( is, false );
+        s.initialize( is, false );
 
         OntologyTerm t = s.getTerm( "http://purl.obolibrary.org/obo/DOID_4159" );
 
@@ -105,7 +105,7 @@ public void testGetChildrenHasProperPart() throws Exception {
         NIFSTDOntologyService s = new NIFSTDOntologyService();
         InputStream is = new GZIPInputStream( requireNonNull( this.getClass().getResourceAsStream(
                 "/data/NIF-GrossAnatomy.small.owl.xml.gz" ) ) );
-        s.loadTermsInNameSpace( is, false );
+        s.initialize( is, false );
 
         OntologyTerm t = s.getTerm( "http://ontology.neuinfo.org/NIF/BiomaterialEntities/NIF-GrossAnatomy.owl#birnlex_734" );
         assertNotNull( t );
@@ -138,7 +138,7 @@ public void testGetParents() throws Exception {
         DiseaseOntologyService s = new DiseaseOntologyService();
         InputStream is = new GZIPInputStream( requireNonNull( this.getClass().getResourceAsStream( "/data/doid.short.owl.gz" ) ) );
 
-        s.loadTermsInNameSpace( is, false );
+        s.initialize( is, false );
 
         /*
          * Note that this test uses the 'new style' URIs for DO, but at this writing we actually use purl.org not
@@ -202,7 +202,7 @@ public void testGetParentsHasProperPart() throws Exception {
         InputStream is = new GZIPInputStream( requireNonNull( this.getClass().getResourceAsStream(
                 "/data/NIF-GrossAnatomy.small.owl.xml.gz" ) ) );
         assertNotNull( is );
-        s.loadTermsInNameSpace( is, false );
+        s.initialize( is, false );
 
         // Mammillary princeps fasciculus: part of white matter, hypothalamus, etc.
         OntologyTerm t = s
@@ -245,7 +245,7 @@ public void testGetParentsHasProperPart() throws Exception {
     public void testRejectNonEnglish() throws Exception {
         CellLineOntologyService s = new CellLineOntologyService();
         InputStream is = new GZIPInputStream( requireNonNull( this.getClass().getResourceAsStream( "/data/clo_merged.sample.owl.xml.gz" ) ) );
-        s.loadTermsInNameSpace( is, false );
+        s.initialize( is, false );
 
         OntologyTerm t = s.getTerm( "http://purl.obolibrary.org/obo/CLO_0000292" );
         assertEquals( "immortal larynx-derived cell line cell", t.getLabel() );
