Use counts instead of set.

Author: balhoff

src/main/scala/org/renci/relationgraph/Main.scala

@@ -66,20 +66,20 @@ object Main extends ZCaseApp[Config] {
           classesStream = allClasses(ontology)
           classesTasks = classesStream.map(c => ZIO.effectTotal(processSuperclasses(c, whelk, config)))
           queue <- Queue.unbounded[Restriction]
-          activeRestrictions <- SubscriptionRef.make(Set.empty[Restriction])
-          _ <- traverse(properties, classes, whelk, config.mode, queue, activeRestrictions.ref)
-          restrictionsStream = Stream.fromQueue(queue).map(r => processRestrictionAndExtendQueue(r, classes, whelk, config.mode, queue, activeRestrictions.ref))
+          activeRestrictions <- Ref.make(0)
+          _ <- traverse(properties, classes, whelk, config.mode, queue, activeRestrictions)
+          restrictionsStream = Stream.fromQueue(queue).map(r => processRestrictionAndExtendQueue(r, classes, whelk, config.mode, queue, activeRestrictions))
           allTasks = classesTasks ++ restrictionsStream
           processed = allTasks.mapMParUnordered(JRuntime.getRuntime.availableProcessors)(identity)
-          watchFiber <- activeRestrictions.changes.dropUntil(_.isEmpty).take(1).foreach(_ => queue.shutdown).fork
+          //watchFiber <- activeRestrictions.changes.dropUntil(_.isEmpty).take(1).foreach(_ => queue.shutdown).fork
           _ <- processed.foreach {
             case TriplesGroup(nonredundant, redundant) =>
               ZIO.effect {
                 nonredundant.foreach(nonredundantRDFWriter.triple)
                 redundant.foreach(redundantRDFWriter.triple)
               }
           }
-          _ <- watchFiber.join
+          // _ <- watchFiber.join
           stop <- ZIO.effectTotal(System.currentTimeMillis())
           _ <- ZIO.effectTotal(scribe.info(s"Computed relations in ${(stop - start) / 1000.0}s"))
         } yield ()
@@ -103,16 +103,22 @@ object Main extends ZCaseApp[Config] {
 
   def allClasses(ont: OWLOntology): ZStream[Any, Nothing, OWLClass] = Stream.fromIterable(ont.getClassesInSignature(Imports.INCLUDED).asScala.to(Set) - OWLThing - OWLNothing)
 
-  def traverse(properties: Hierarchy, classes: Hierarchy, reasoner: ReasonerState, mode: Config.OutputMode, queue: Queue[Restriction], activeRestrictions: RefM[Set[Restriction]]): UIO[Unit] =
+  def traverse(properties: Hierarchy, classes: Hierarchy, reasoner: ReasonerState, mode: Config.OutputMode, queue: Queue[Restriction], activeRestrictions: Ref[Int]): UIO[Unit] =
     ZIO.foreach_(properties.subclasses.getOrElse(Top, Set.empty)) { subprop =>
       traverseProperty(subprop, properties, classes, reasoner, mode, queue, activeRestrictions)
     }
 
-  def traverseProperty(property: AtomicConcept, properties: Hierarchy, classes: Hierarchy, whelk: ReasonerState, mode: Config.OutputMode, queue: Queue[Restriction], activeRestrictions: RefM[Set[Restriction]]): UIO[Unit] = {
+  def traverseProperty(property: AtomicConcept, properties: Hierarchy, classes: Hierarchy, whelk: ReasonerState, mode: Config.OutputMode, queue: Queue[Restriction], activeRestrictions: Ref[Int]): UIO[Unit] = {
     val restrictions = if (hasSubClasses(property, whelk)) {
       (classes.subclasses.getOrElse(Top, Set.empty) - Bottom).map(filler => Restriction(Role(property.id), filler))
     } else Set.empty
-    activeRestrictions.update(current => ZIO.succeed(current ++ restrictions)) *> queue.offerAll(restrictions).unit
+    for {
+      _ <- activeRestrictions.update(current => current + restrictions.size)
+      _ <- queue.offerAll(restrictions).unit
+      active <- activeRestrictions.get
+      _ <- queue.shutdown.when(active < 1)
+    } yield ()
+
   }
 
   def hasSubClasses(property: AtomicConcept, whelk: ReasonerState): Boolean = {
@@ -123,14 +129,16 @@ object Main extends ZCaseApp[Config] {
     (updatedWhelk.closureSubsBySuperclass.getOrElse(queryConcept, Set.empty) - Bottom).nonEmpty
   }
 
-  def processRestrictionAndExtendQueue(restriction: Restriction, classes: Hierarchy, whelk: ReasonerState, mode: Config.OutputMode, queue: Queue[Restriction], activeRestrictions: RefM[Set[Restriction]]): UIO[TriplesGroup] = {
+  def processRestrictionAndExtendQueue(restriction: Restriction, classes: Hierarchy, whelk: ReasonerState, mode: Config.OutputMode, queue: Queue[Restriction], activeRestrictions: Ref[Int]): UIO[TriplesGroup] = {
     for {
       triples <- ZIO.effectTotal(processRestriction(restriction, whelk, mode))
       directFillerSubclassesRestrictions = if (triples.redundant.nonEmpty)
         (classes.subclasses.getOrElse(restriction.filler, Set.empty) - Bottom).map(c => Restriction(restriction.property, c))
       else Set.empty
-      _ <- activeRestrictions.update(current => ZIO.succeed(current - restriction ++ directFillerSubclassesRestrictions))
+      _ <- activeRestrictions.update(current => current - 1 + directFillerSubclassesRestrictions.size)
       _ <- queue.offerAll(directFillerSubclassesRestrictions)
+      active <- activeRestrictions.get
+      _ <- queue.shutdown.when(active < 1)
     } yield triples
   }