load theories in stages, to reduce ML heap requirements;

--- a/src/Pure/PIDE/document_status.scala	Wed Sep 04 21:42:11 2019 +0200
+++ b/src/Pure/PIDE/document_status.scala	Wed Sep 04 22:00:38 2019 +0200
@@ -244,6 +244,12 @@
         node_status <- get(name)
       } yield (name, node_status)).toList
 
+    def consolidated(name: Document.Node.Name): Boolean =
+      rep.get(name) match {
+        case Some(st) => st.consolidated
+        case None => false
+      }
+
     def quasi_consolidated(name: Document.Node.Name): Boolean =
       rep.get(name) match {
         case Some(st) => st.quasi_consolidated

--- a/src/Pure/PIDE/headless.scala	Wed Sep 04 21:42:11 2019 +0200
+++ b/src/Pure/PIDE/headless.scala	Wed Sep 04 22:00:38 2019 +0200
@@ -44,6 +44,62 @@
       (nodes.iterator ++ nodes_committed.iterator).forall({ case (_, st) => st.ok })
   }
 
+  private object Checkpoints_State
+  {
+    object Status extends Enumeration
+    {
+      val INIT, LOADED, LOADED_DESCENDANTS = Value
+    }
+
+    def init(nodes: List[Document.Node.Name]): Checkpoints_State =
+      Checkpoints_State(Status.INIT, nodes)
+
+    val last: Checkpoints_State =
+      Checkpoints_State(Status.LOADED_DESCENDANTS, Nil)
+  }
+
+  private sealed case class Checkpoints_State(
+    status: Checkpoints_State.Status.Value,
+    nodes: List[Document.Node.Name])
+  {
+    def next(
+      dep_graph: Document.Theory_Graph[Unit],
+      finished: Document.Node.Name => Boolean): (List[Document.Node.Name], Checkpoints_State) =
+    {
+      import Checkpoints_State.Status
+
+      def descendants: List[Document.Node.Name] =
+        nodes match {
+          case Nil => dep_graph.topological_order
+          case current :: rest =>
+            val dep_graph1 =
+              if (rest.isEmpty) dep_graph
+              else {
+                val exclude = dep_graph.all_succs(rest).toSet
+                dep_graph.restrict(name => !exclude(name))
+              }
+            dep_graph1.all_succs(List(current))
+        }
+
+      def requirements: List[Document.Node.Name] =
+        dep_graph.all_preds(nodes.headOption.toList).reverse
+
+      val (load_theories, st1) =
+        (status, nodes) match {
+          case (Status.INIT, Nil) =>
+            (descendants, Checkpoints_State.last)
+          case (Status.INIT, current :: _) =>
+            (requirements, copy(status = Status.LOADED))
+          case (Status.LOADED, current :: rest) if finished(current) =>
+            (descendants, copy(status = Status.LOADED_DESCENDANTS))
+          case (Status.LOADED_DESCENDANTS, _ :: rest) if descendants.forall(finished) =>
+            Checkpoints_State.init(rest).next(dep_graph, finished)
+          case _ => (Nil, this)
+        }
+      (load_theories.filterNot(finished), st1)
+    }
+  }
+
   class Session private[Headless](
     session_name: String,
     _session_options: => Options,
@@ -52,6 +108,10 @@
     session =>
 
 
+    private def loaded_theory(name: Document.Node.Name): Boolean =
+      resources.session_base.loaded_theory(name.theory)
+
+
     /* options */
 
     def default_check_delay: Time = session_options.seconds("headless_check_delay")
@@ -81,15 +141,21 @@
     /* theories */
 
     private sealed case class Use_Theories_State(
+      dependencies: resources.Dependencies[Unit],
+      checkpoints_state: Checkpoints_State,
+      watchdog_timeout: Time,
+      commit: Option[(Document.Snapshot, Document_Status.Node_Status) => Unit],
       last_update: Time = Time.now(),
       nodes_status: Document_Status.Nodes_Status = Document_Status.Nodes_Status.empty,
       already_committed: Map[Document.Node.Name, Document_Status.Node_Status] = Map.empty,
       result: Option[Exn.Result[Use_Theories_Result]] = None)
     {
+      def dep_graph: Document.Theory_Graph[Unit] = dependencies.theory_graph
+
       def update(new_nodes_status: Document_Status.Nodes_Status): Use_Theories_State =
         copy(last_update = Time.now(), nodes_status = new_nodes_status)
 
-      def watchdog(watchdog_timeout: Time): Boolean =
+      def watchdog: Boolean =
         watchdog_timeout > Time.zero && Time.now() - last_update > watchdog_timeout
 
       def finished_result: Boolean = result.isDefined
@@ -100,55 +166,63 @@
       def cancel_result: Use_Theories_State =
         if (finished_result) this else copy(result = Some(Exn.Exn(Exn.Interrupt())))
 
-      def check_result(
-          state: Document.State,
-          version: Document.Version,
-          dep_theories: List[Document.Node.Name],
-          beyond_limit: Boolean,
-          watchdog_timeout: Time,
-          commit: Option[(Document.Snapshot, Document_Status.Node_Status) => Unit])
-        : Use_Theories_State =
+      def clean: Set[Document.Node.Name] =
+        already_committed.keySet -- checkpoints_state.nodes
+
+      def check(state: Document.State, version: Document.Version, beyond_limit: Boolean)
+        : (List[Document.Node.Name], Use_Theories_State) =
       {
         val already_committed1 =
-          if (commit.isDefined) {
-            (already_committed /: dep_theories)({ case (committed, name) =>
-              def parents_committed: Boolean =
-                version.nodes(name).header.imports.forall(parent =>
-                  resources.session_base.loaded_theory(parent) || committed.isDefinedAt(parent))
-              if (!committed.isDefinedAt(name) && parents_committed &&
-                  state.node_consolidated(version, name))
-              {
-                val snapshot = stable_snapshot(state, version, name)
-                val status = Document_Status.Node_Status.make(state, version, name)
-                commit.get.apply(snapshot, status)
-                committed + (name -> status)
-              }
-              else committed
-            })
+          commit match {
+            case None => already_committed
+            case Some(commit_fn) =>
+              (already_committed /: dep_graph.topological_order)(
+                { case (committed, name) =>
+                    def parents_committed: Boolean =
+                      version.nodes(name).header.imports.forall(parent =>
+                        loaded_theory(parent) || committed.isDefinedAt(parent))
+                    if (!committed.isDefinedAt(name) && parents_committed &&
+                        state.node_consolidated(version, name))
+                    {
+                      val snapshot = stable_snapshot(state, version, name)
+                      val status = Document_Status.Node_Status.make(state, version, name)
+                      commit_fn(snapshot, status)
+                      committed + (name -> status)
+                    }
+                    else committed
+                })
           }
-          else already_committed
 
         val result1 =
           if (!finished_result &&
-            (beyond_limit || watchdog(watchdog_timeout) ||
-              dep_theories.forall(name =>
+            (beyond_limit || watchdog ||
+              dep_graph.keys_iterator.forall(name =>
                 already_committed1.isDefinedAt(name) ||
                 state.node_consolidated(version, name) ||
                 nodes_status.quasi_consolidated(name))))
           {
             val nodes =
-              for (name <- dep_theories)
-              yield { (name -> Document_Status.Node_Status.make(state, version, name)) }
+              (for (name <- dep_graph.keys_iterator)
+                yield { (name -> Document_Status.Node_Status.make(state, version, name)) }).toList
             val nodes_committed =
-              for {
-                name <- dep_theories
+              (for {
+                name <- dep_graph.keys_iterator
                 status <- already_committed1.get(name)
-              } yield (name -> status)
+              } yield (name -> status)).toList
             Some(Exn.Res(new Use_Theories_Result(state, version, nodes, nodes_committed)))
           }
           else result
 
-        copy(already_committed = already_committed1, result = result1)
+        val (load_theories, checkpoints_state1) =
+          checkpoints_state.next(dep_graph,
+            name =>
+              loaded_theory(name) ||
+              already_committed1.isDefinedAt(name) ||
+              nodes_status.consolidated(name))
+
+        (load_theories,
+          copy(already_committed = already_committed1, result = result1,
+            checkpoints_state = checkpoints_state1))
       }
     }
 
@@ -163,6 +237,7 @@
       nodes_status_delay: Time = default_nodes_status_delay,
       id: UUID.T = UUID.random(),
       share_common_data: Boolean = false,
+      checkpoints: Set[Document.Node.Name] = Set.empty,
       // commit: must not block, must not fail
       commit: Option[(Document.Snapshot, Document_Status.Node_Status) => Unit] = None,
       commit_cleanup_delay: Time = default_commit_cleanup_delay,
@@ -176,20 +251,29 @@
         resources.dependencies(import_names, progress = progress).check_errors
       }
       val dep_theories = dependencies.theories
+      val dep_theories_set = dep_theories.toSet
       val dep_files =
         dependencies.loaded_files(false).flatMap(_._2).
           map(path => Document.Node.Name(resources.append("", path)))
 
-      val use_theories_state = Synchronized(Use_Theories_State())
+      val use_theories_state =
+      {
+        val checkpoints_state =
+          Checkpoints_State.init(
+            if (checkpoints.isEmpty) Nil
+            else dependencies.theory_graph.topological_order.filter(checkpoints(_)))
+        Synchronized(Use_Theories_State(dependencies, checkpoints_state, watchdog_timeout, commit))
+      }
 
-      def check_result_state(beyond_limit: Boolean = false)
+      def check_state(beyond_limit: Boolean = false)
       {
         val state = session.current_state()
-        state.stable_tip_version match {
-          case Some(version) =>
-            use_theories_state.change(
-              _.check_result(state, version, dep_theories, beyond_limit, watchdog_timeout, commit))
-          case None =>
+        for (version <- state.stable_tip_version) {
+          val load_theories = use_theories_state.change_result(_.check(state, version, beyond_limit))
+          if (load_theories.nonEmpty) {
+            resources.load_theories(
+              session, id, load_theories, dep_files, unicode_symbols, share_common_data, progress)
+          }
         }
       }
 
@@ -201,7 +285,7 @@
             if (progress.stopped) use_theories_state.change(_.cancel_result)
             else {
               check_count += 1
-              check_result_state(check_limit > 0 && check_count > check_limit)
+              check_state(check_limit > 0 && check_count > check_limit)
             }
           }
       }
@@ -215,12 +299,9 @@
 
         val delay_commit_clean =
           Standard_Thread.delay_first(commit_cleanup_delay max Time.zero) {
-            val clean = use_theories_state.value.already_committed.keySet
-            resources.clean_theories(session, id, clean)
+            resources.clean_theories(session, id, use_theories_state.value.clean)
           }
 
-        val dep_theories_set = dep_theories.toSet
-
         Session.Consumer[Session.Commands_Changed](getClass.getName) {
           case changed =>
             if (changed.nodes.exists(dep_theories_set)) {
@@ -256,7 +337,7 @@
 
               theory_progress.foreach(progress.theory(_))
 
-              check_result_state()
+              check_state()
 
               if (commit.isDefined && commit_cleanup_delay > Time.zero) {
                 if (use_theories_state.value.finished_result)
@@ -269,8 +350,7 @@
 
       try {
         session.commands_changed += consumer
-        resources.load_theories(
-          session, id, dep_theories, dep_files, unicode_symbols, share_common_data, progress)
+        check_state()
         use_theories_state.guarded_access(_.join_result)
         check_progress.cancel
       }
@@ -398,7 +478,7 @@
 
       /* theories */
 
-      lazy val theory_graph: Graph[Document.Node.Name, Unit] =
+      lazy val theory_graph: Document.Theory_Graph[Unit] =
         Document.theory_graph(
           for ((name, theory) <- theories.toList)
           yield ((name, ()), theory.node_header.imports.filter(theories.isDefinedAt(_))))

--- a/src/Pure/Tools/dump.scala	Wed Sep 04 21:42:11 2019 +0200
+++ b/src/Pure/Tools/dump.scala	Wed Sep 04 22:00:38 2019 +0200
@@ -206,24 +206,16 @@
       }
 
 
-      // run
+      // synchronous body
 
       try {
-        val dump_checkpoints = deps.dump_checkpoints
-        def commit(snapshot: Document.Snapshot, status: Document_Status.Node_Status)
-        {
-          if (dump_checkpoints.contains(snapshot.node_name)) {
-            session.protocol_command("ML_Heap.share_common_data")
-          }
-          Consumer.apply(snapshot, status)
-        }
-
         val use_theories_result =
           session.use_theories(used_theories.map(_.theory),
             unicode_symbols = unicode_symbols,
             share_common_data = true,
             progress = progress,
-            commit = Some(commit _))
+            checkpoints = deps.dump_checkpoints,
+            commit = Some(Consumer.apply _))
 
         val bad_theories = Consumer.shutdown()
         val bad_msgs =