diff --git a/src/Pure/Admin/build_log.scala b/src/Pure/Admin/build_log.scala --- a/src/Pure/Admin/build_log.scala +++ b/src/Pure/Admin/build_log.scala @@ -1,1113 +1,1120 @@ /* Title: Pure/Admin/build_log.scala Author: Makarius Management of build log files and database storage. */ package isabelle import java.io.{File => JFile} import java.time.format.{DateTimeFormatter, DateTimeParseException} import java.util.Locale import scala.collection.immutable.SortedMap import scala.collection.mutable import scala.util.matching.Regex object Build_Log { /** content **/ /* properties */ object Prop { val build_tags = SQL.Column.string("build_tags") // lines val build_args = SQL.Column.string("build_args") // lines val build_group_id = SQL.Column.string("build_group_id") val build_id = SQL.Column.string("build_id") val build_engine = SQL.Column.string("build_engine") val build_host = SQL.Column.string("build_host") val build_start = SQL.Column.date("build_start") val build_end = SQL.Column.date("build_end") val isabelle_version = SQL.Column.string("isabelle_version") val afp_version = SQL.Column.string("afp_version") val all_props: List[SQL.Column] = List(build_tags, build_args, build_group_id, build_id, build_engine, build_host, build_start, build_end, isabelle_version, afp_version) } /* settings */ object Settings { val ISABELLE_BUILD_OPTIONS = SQL.Column.string("ISABELLE_BUILD_OPTIONS") val ML_PLATFORM = SQL.Column.string("ML_PLATFORM") val ML_HOME = SQL.Column.string("ML_HOME") val ML_SYSTEM = SQL.Column.string("ML_SYSTEM") val ML_OPTIONS = SQL.Column.string("ML_OPTIONS") val ml_settings = List(ML_PLATFORM, ML_HOME, ML_SYSTEM, ML_OPTIONS) val all_settings = ISABELLE_BUILD_OPTIONS :: ml_settings type Entry = (String, String) type T = List[Entry] object Entry { def unapply(s: String): Option[Entry] = for { (a, b) <- Properties.Eq.unapply(s) } yield (a, Library.perhaps_unquote(b)) def getenv(a: String): String = Properties.Eq(a, quote(Isabelle_System.getenv(a))) } def show(): String = cat_lines( List(Entry.getenv("ISABELLE_TOOL_JAVA_OPTIONS"), Entry.getenv(ISABELLE_BUILD_OPTIONS.name), "") ::: ml_settings.map(c => Entry.getenv(c.name))) } /* file names */ def log_date(date: Date): String = String.format(Locale.ROOT, "%s.%05d", DateTimeFormatter.ofPattern("yyyy-MM-dd").format(date.rep), java.lang.Long.valueOf((date.time - date.midnight.time).ms / 1000)) def log_subdir(date: Date): Path = Path.explode("log") + Path.explode(date.rep.getYear.toString) def log_filename(engine: String, date: Date, more: List[String] = Nil): Path = Path.explode((engine :: log_date(date) :: more).mkString("", "_", ".log")) /** log file **/ def print_date(date: Date): String = Log_File.Date_Format(date) object Log_File { /* log file */ def plain_name(name: String): String = { List(".log", ".log.gz", ".log.xz", ".gz", ".xz").find(name.endsWith) match { case Some(s) => Library.try_unsuffix(s, name).get case None => name } } def apply(name: String, lines: List[String]): Log_File = new Log_File(plain_name(name), lines.map(Library.trim_line)) def apply(name: String, text: String): Log_File = new Log_File(plain_name(name), Library.trim_split_lines(text)) def apply(file: JFile): Log_File = { val name = file.getName val text = if (File.is_gz(name)) File.read_gzip(file) else if (File.is_xz(name)) File.read_xz(file) else File.read(file) apply(name, text) } def apply(path: Path): Log_File = apply(path.file) /* log file collections */ def is_log(file: JFile, prefixes: List[String] = List(Build_History.log_prefix, Identify.log_prefix, Identify.log_prefix2, Isatest.log_prefix, AFP_Test.log_prefix), suffixes: List[String] = List(".log", ".log.gz", ".log.xz") ): Boolean = { val name = file.getName prefixes.exists(name.startsWith) && suffixes.exists(name.endsWith) && name != "isatest.log" && name != "afp-test.log" && name != "main.log" } /* date format */ val Date_Format = { val fmts = Date.Formatter.variants( List("EEE MMM d HH:mm:ss O yyyy", "EEE MMM d HH:mm:ss VV yyyy"), List(Locale.ENGLISH, Locale.GERMAN)) ::: List( DateTimeFormatter.RFC_1123_DATE_TIME, Date.Formatter.pattern("EEE MMM d HH:mm:ss yyyy").withZone(Date.timezone_berlin)) def tune_timezone(s: String): String = s match { case "CET" | "MET" => "GMT+1" case "CEST" | "MEST" => "GMT+2" case "EST" => "Europe/Berlin" case _ => s } def tune_weekday(s: String): String = s match { case "Die" => "Di" case "Mit" => "Mi" case "Don" => "Do" case "Fre" => "Fr" case "Sam" => "Sa" case "Son" => "So" case _ => s } def tune(s: String): String = Word.implode( Word.explode(s) match { case a :: "M\uFFFDr" :: bs => tune_weekday(a) :: "Mär" :: bs.map(tune_timezone) case a :: bs => tune_weekday(a) :: bs.map(tune_timezone) case Nil => Nil } ) Date.Format.make(fmts, tune) } } class Log_File private(val name: String, val lines: List[String]) { log_file => override def toString: String = name def text: String = cat_lines(lines) def err(msg: String): Nothing = error("Error in log file " + quote(name) + ": " + msg) /* date format */ object Strict_Date { def unapply(s: String): Some[Date] = try { Some(Log_File.Date_Format.parse(s)) } catch { case exn: DateTimeParseException => log_file.err(exn.getMessage) } } /* inlined text */ def filter(Marker: Protocol_Message.Marker): List[String] = for (Marker(text) <- lines) yield text def find(Marker: Protocol_Message.Marker): Option[String] = lines.collectFirst({ case Marker(text) => text }) def find_match(regexes: List[Regex]): Option[String] = regexes match { case Nil => None case regex :: rest => lines.iterator.map(regex.unapplySeq(_)).find(res => res.isDefined && res.get.length == 1). map(res => res.get.head) orElse find_match(rest) } /* settings */ def get_setting(name: String): Option[Settings.Entry] = lines.collectFirst({ case Settings.Entry(a, b) if a == name => a -> b }) def get_all_settings: Settings.T = for { c <- Settings.all_settings; entry <- get_setting(c.name) } yield entry /* properties (YXML) */ val cache: XML.Cache = XML.Cache.make() def parse_props(text: String): Properties.T = try { cache.props(XML.Decode.properties(YXML.parse_body(text))) } catch { case _: XML.Error => log_file.err("malformed properties") } def filter_props(marker: Protocol_Message.Marker): List[Properties.T] = for (text <- filter(marker) if YXML.detect(text)) yield parse_props(text) def find_props(marker: Protocol_Message.Marker): Option[Properties.T] = for (text <- find(marker) if YXML.detect(text)) yield parse_props(text) /* parse various formats */ def parse_meta_info(): Meta_Info = Build_Log.parse_meta_info(log_file) def parse_build_info(ml_statistics: Boolean = false): Build_Info = Build_Log.parse_build_info(log_file, ml_statistics) def parse_session_info( command_timings: Boolean = false, theory_timings: Boolean = false, ml_statistics: Boolean = false, task_statistics: Boolean = false): Session_Info = Build_Log.parse_session_info( log_file, command_timings, theory_timings, ml_statistics, task_statistics) } /** digested meta info: produced by Admin/build_other in log.xz file **/ object Meta_Info { val empty: Meta_Info = Meta_Info(Nil, Nil) } sealed case class Meta_Info(props: Properties.T, settings: Settings.T) { def is_empty: Boolean = props.isEmpty && settings.isEmpty def get(c: SQL.Column): Option[String] = Properties.get(props, c.name) orElse Properties.get(settings, c.name) def get_date(c: SQL.Column): Option[Date] = get(c).map(Log_File.Date_Format.parse) } object Identify { val log_prefix = "isabelle_identify_" val log_prefix2 = "plain_identify_" def engine(log_file: Log_File): String = if (log_file.name.startsWith(log_prefix2)) "plain_identify" else "identify" def content(date: Date, isabelle_version: Option[String], afp_version: Option[String]): String = terminate_lines( List("isabelle_identify: " + Build_Log.print_date(date), "") ::: isabelle_version.map("Isabelle version: " + _).toList ::: afp_version.map("AFP version: " + _).toList) val Start = new Regex("""^isabelle_identify: (.+)$""") val No_End = new Regex("""$.""") val Isabelle_Version = List(new Regex("""^Isabelle version: (\S+)$""")) val AFP_Version = List(new Regex("""^AFP version: (\S+)$""")) } object Isatest { val log_prefix = "isatest-makeall-" val engine = "isatest" val Start = new Regex("""^------------------- starting test --- (.+) --- (.+)$""") val End = new Regex("""^------------------- test (?:successful|FAILED) --- (.+) --- .*$""") val Isabelle_Version = List(new Regex("""^Isabelle version: (\S+)$""")) } object AFP_Test { val log_prefix = "afp-test-devel-" val engine = "afp-test" val Start = new Regex("""^Start test(?: for .+)? at ([^,]+), (.*)$""") val Start_Old = new Regex("""^Start test(?: for .+)? at ([^,]+)$""") val End = new Regex("""^End test on (.+), .+, elapsed time:.*$""") val Isabelle_Version = List(new Regex("""^Isabelle version: .* -- hg id (\S+)$""")) val AFP_Version = List(new Regex("""^AFP version: .* -- hg id (\S+)$""")) val Bad_Init = new Regex("""^cp:.*: Disc quota exceeded$""") } object Jenkins { val log_prefix = "jenkins_" val engine = "jenkins" val Host = new Regex("""^Building remotely on (\S+) \((\S+)\).*$""") val Start = new Regex("""^(?:Started by an SCM change|Started from command line by admin|).*$""") val Start_Date = new Regex("""^Build started at (.+)$""") val No_End = new Regex("""$.""") val Isabelle_Version = List(new Regex("""^(?:Build for Isabelle id|Isabelle id) (\w+).*$"""), new Regex("""^ISABELLE_CI_REPO_ID="(\w+)".*$"""), new Regex("""^(\w{12}) tip.*$""")) val AFP_Version = List(new Regex("""^(?:Build for AFP id|AFP id) (\w+).*$"""), new Regex("""^ISABELLE_CI_AFP_ID="(\w+)".*$""")) val CONFIGURATION = "=== CONFIGURATION ===" val BUILD = "=== BUILD ===" } private def parse_meta_info(log_file: Log_File): Meta_Info = { def parse(engine: String, host: String, start: Date, End: Regex, Isabelle_Version: List[Regex], AFP_Version: List[Regex] ): Meta_Info = { val build_id = { val prefix = proper_string(host) orElse proper_string(engine) getOrElse "build" prefix + ":" + start.time.ms } val build_engine = if (engine == "") Nil else List(Prop.build_engine.name -> engine) val build_host = if (host == "") Nil else List(Prop.build_host.name -> host) val start_date = List(Prop.build_start.name -> print_date(start)) val end_date = log_file.lines.last match { case End(log_file.Strict_Date(end_date)) => List(Prop.build_end.name -> print_date(end_date)) case _ => Nil } val isabelle_version = log_file.find_match(Isabelle_Version).map(Prop.isabelle_version.name -> _) val afp_version = log_file.find_match(AFP_Version).map(Prop.afp_version.name -> _) Meta_Info((Prop.build_id.name -> build_id) :: build_engine ::: build_host ::: start_date ::: end_date ::: isabelle_version.toList ::: afp_version.toList, log_file.get_all_settings) } log_file.lines match { case line :: _ if Protocol.Meta_Info_Marker.test_yxml(line) => Meta_Info(log_file.find_props(Protocol.Meta_Info_Marker).get, log_file.get_all_settings) case Identify.Start(log_file.Strict_Date(start)) :: _ => parse(Identify.engine(log_file), "", start, Identify.No_End, Identify.Isabelle_Version, Identify.AFP_Version) case Isatest.Start(log_file.Strict_Date(start), host) :: _ => parse(Isatest.engine, host, start, Isatest.End, Isatest.Isabelle_Version, Nil) case AFP_Test.Start(log_file.Strict_Date(start), host) :: _ => parse(AFP_Test.engine, host, start, AFP_Test.End, AFP_Test.Isabelle_Version, AFP_Test.AFP_Version) case AFP_Test.Start_Old(log_file.Strict_Date(start)) :: _ => parse(AFP_Test.engine, "", start, AFP_Test.End, AFP_Test.Isabelle_Version, AFP_Test.AFP_Version) case line :: _ if line.startsWith("\u0000") => Meta_Info.empty case List(Isatest.End(_)) => Meta_Info.empty case _ :: AFP_Test.Bad_Init() :: _ => Meta_Info.empty case Nil => Meta_Info.empty case _ => log_file.err("cannot detect log file format") } } /** build info: toplevel output of isabelle build or Admin/build_other **/ val SESSION_NAME = "session_name" object Session_Status extends Enumeration { val existing, finished, failed, cancelled = Value } sealed case class Session_Entry( chapter: String = "", groups: List[String] = Nil, threads: Option[Int] = None, timing: Timing = Timing.zero, ml_timing: Timing = Timing.zero, sources: Option[String] = None, heap_size: Option[Space] = None, status: Option[Session_Status.Value] = None, errors: List[String] = Nil, theory_timings: Map[String, Timing] = Map.empty, ml_statistics: List[Properties.T] = Nil ) { def proper_groups: Option[String] = if (groups.isEmpty) None else Some(cat_lines(groups)) def finished: Boolean = status == Some(Session_Status.finished) def failed: Boolean = status == Some(Session_Status.failed) } object Build_Info { val sessions_dummy: Map[String, Session_Entry] = Map("" -> Session_Entry(theory_timings = Map("" -> Timing.zero))) } sealed case class Build_Info(sessions: Map[String, Session_Entry]) { def finished_sessions: List[String] = for ((a, b) <- sessions.toList if b.finished) yield a def failed_sessions: List[String] = for ((a, b) <- sessions.toList if b.failed) yield a } private def parse_build_info(log_file: Log_File, parse_ml_statistics: Boolean): Build_Info = { object Chapter_Name { def unapply(s: String): Some[(String, String)] = space_explode('/', s) match { case List(chapter, name) => Some((chapter, name)) case _ => Some(("", s)) } } val Session_No_Groups = new Regex("""^Session (\S+)$""") val Session_Groups = new Regex("""^Session (\S+) \((.*)\)$""") val Session_Finished1 = new Regex("""^Finished (\S+) \((\d+):(\d+):(\d+) elapsed time, (\d+):(\d+):(\d+) cpu time.*$""") val Session_Finished2 = new Regex("""^Finished ([^\s/]+) \((\d+):(\d+):(\d+) elapsed time.*$""") val Session_Timing = new Regex("""^Timing (\S+) \((\d+) threads, (\d+\.\d+)s elapsed time, (\d+\.\d+)s cpu time, (\d+\.\d+)s GC time.*$""") val Session_Started = new Regex("""^(?:Running|Building) (\S+) \.\.\.$""") val Sources = new Regex("""^Sources (\S+) (\S{""" + SHA1.digest_length + """})$""") val Heap = new Regex("""^Heap (\S+) \((\d+) bytes\)$""") object Theory_Timing { def unapply(line: String): Option[(String, (String, Timing))] = Protocol.Theory_Timing_Marker.unapply(line.replace('~', '-')).map(log_file.parse_props) match { case Some((SESSION_NAME, session) :: props) => for (theory <- Markup.Name.unapply(props)) yield (session, theory -> Markup.Timing_Properties.get(props)) case _ => None } } var chapter = Map.empty[String, String] var groups = Map.empty[String, List[String]] var threads = Map.empty[String, Int] var timing = Map.empty[String, Timing] var ml_timing = Map.empty[String, Timing] var started = Set.empty[String] var sources = Map.empty[String, String] var heap_sizes = Map.empty[String, Space] var theory_timings = Map.empty[String, Map[String, Timing]] var ml_statistics = Map.empty[String, List[Properties.T]] var errors = Map.empty[String, List[String]] def all_sessions: Set[String] = chapter.keySet ++ groups.keySet ++ threads.keySet ++ timing.keySet ++ ml_timing.keySet ++ started ++ sources.keySet ++ heap_sizes.keySet ++ theory_timings.keySet ++ ml_statistics.keySet for (line <- log_file.lines) { line match { case Session_No_Groups(Chapter_Name(chapt, name)) => chapter += (name -> chapt) groups += (name -> Nil) case Session_Groups(Chapter_Name(chapt, name), grps) => chapter += (name -> chapt) groups += (name -> Word.explode(grps)) case Session_Started(name) => started += name case Session_Finished1(name, Value.Int(e1), Value.Int(e2), Value.Int(e3), Value.Int(c1), Value.Int(c2), Value.Int(c3)) => val elapsed = Time.hms(e1, e2, e3) val cpu = Time.hms(c1, c2, c3) timing += (name -> Timing(elapsed, cpu, Time.zero)) case Session_Finished2(name, Value.Int(e1), Value.Int(e2), Value.Int(e3)) => val elapsed = Time.hms(e1, e2, e3) timing += (name -> Timing(elapsed, Time.zero, Time.zero)) case Session_Timing(name, Value.Int(t), Value.Double(e), Value.Double(c), Value.Double(g)) => val elapsed = Time.seconds(e) val cpu = Time.seconds(c) val gc = Time.seconds(g) ml_timing += (name -> Timing(elapsed, cpu, gc)) threads += (name -> t) case Sources(name, s) => sources += (name -> s) case Heap(name, Value.Long(size)) => heap_sizes += (name -> Space.bytes(size)) case _ if Protocol.Theory_Timing_Marker.test_yxml(line) => line match { case Theory_Timing(name, theory_timing) => theory_timings += (name -> (theory_timings.getOrElse(name, Map.empty) + theory_timing)) case _ => log_file.err("malformed theory_timing " + quote(line)) } case _ if parse_ml_statistics && Protocol.ML_Statistics_Marker.test_yxml(line) => Protocol.ML_Statistics_Marker.unapply(line).map(log_file.parse_props) match { case Some((SESSION_NAME, name) :: props) => ml_statistics += (name -> (props :: ml_statistics.getOrElse(name, Nil))) case _ => log_file.err("malformed ML_statistics " + quote(line)) } case _ if Protocol.Error_Message_Marker.test_yxml(line) => Protocol.Error_Message_Marker.unapply(line).map(log_file.parse_props) match { case Some(List((SESSION_NAME, name), (Markup.CONTENT, msg))) => errors += (name -> (msg :: errors.getOrElse(name, Nil))) case _ => log_file.err("malformed error message " + quote(line)) } case _ => } } val sessions = Map( (for (name <- all_sessions.toList) yield { val status = if (timing.isDefinedAt(name) || ml_timing.isDefinedAt(name)) Session_Status.finished else if (started(name)) Session_Status.failed else Session_Status.existing val entry = Session_Entry( chapter = chapter.getOrElse(name, ""), groups = groups.getOrElse(name, Nil), threads = threads.get(name), timing = timing.getOrElse(name, Timing.zero), ml_timing = ml_timing.getOrElse(name, Timing.zero), sources = sources.get(name), heap_size = heap_sizes.get(name), status = Some(status), errors = errors.getOrElse(name, Nil).reverse, theory_timings = theory_timings.getOrElse(name, Map.empty), ml_statistics = ml_statistics.getOrElse(name, Nil).reverse) (name -> entry) }):_*) Build_Info(sessions) } /** session info: produced by isabelle build as session database **/ sealed case class Session_Info( session_timing: Properties.T, command_timings: List[Properties.T], theory_timings: List[Properties.T], ml_statistics: List[Properties.T], task_statistics: List[Properties.T], errors: List[String] ) { def error(s: String): Session_Info = copy(errors = errors ::: List(s)) } private def parse_session_info( log_file: Log_File, command_timings: Boolean, theory_timings: Boolean, ml_statistics: Boolean, task_statistics: Boolean ): Session_Info = { Session_Info( session_timing = log_file.find_props(Protocol.Session_Timing_Marker) getOrElse Nil, command_timings = if (command_timings) log_file.filter_props(Protocol.Command_Timing_Marker) else Nil, theory_timings = if (theory_timings) log_file.filter_props(Protocol.Theory_Timing_Marker) else Nil, ml_statistics = if (ml_statistics) log_file.filter_props(Protocol.ML_Statistics_Marker) else Nil, task_statistics = if (task_statistics) log_file.filter_props(Protocol.Task_Statistics_Marker) else Nil, errors = log_file.filter(Protocol.Error_Message_Marker)) } def compress_errors( errors: List[String], cache: Compress.Cache = Compress.Cache.none ): Option[Bytes] = if (errors.isEmpty) None else { Some(Bytes(YXML.string_of_body(XML.Encode.list(XML.Encode.string)(errors))). compress(cache = cache)) } def uncompress_errors(bytes: Bytes, cache: XML.Cache = XML.Cache.make()): List[String] = if (bytes.is_empty) Nil else { XML.Decode.list(YXML.string_of_body)( YXML.parse_body(bytes.uncompress(cache = cache.compress).text, cache = cache)) } /** persistent store **/ /* SQL data model */ object Data { def build_log_table(name: String, columns: List[SQL.Column], body: String = ""): SQL.Table = SQL.Table("isabelle_build_log_" + name, columns, body) /* main content */ val log_name = SQL.Column.string("log_name").make_primary_key val session_name = SQL.Column.string("session_name").make_primary_key val theory_name = SQL.Column.string("theory_name").make_primary_key val chapter = SQL.Column.string("chapter") val groups = SQL.Column.string("groups") val threads = SQL.Column.int("threads") val timing_elapsed = SQL.Column.long("timing_elapsed") val timing_cpu = SQL.Column.long("timing_cpu") val timing_gc = SQL.Column.long("timing_gc") val timing_factor = SQL.Column.double("timing_factor") val ml_timing_elapsed = SQL.Column.long("ml_timing_elapsed") val ml_timing_cpu = SQL.Column.long("ml_timing_cpu") val ml_timing_gc = SQL.Column.long("ml_timing_gc") val ml_timing_factor = SQL.Column.double("ml_timing_factor") val theory_timing_elapsed = SQL.Column.long("theory_timing_elapsed") val theory_timing_cpu = SQL.Column.long("theory_timing_cpu") val theory_timing_gc = SQL.Column.long("theory_timing_gc") val heap_size = SQL.Column.long("heap_size") val status = SQL.Column.string("status") val errors = SQL.Column.bytes("errors") val sources = SQL.Column.string("sources") val ml_statistics = SQL.Column.bytes("ml_statistics") val known = SQL.Column.bool("known") val meta_info_table = build_log_table("meta_info", log_name :: Prop.all_props ::: Settings.all_settings) val sessions_table = build_log_table("sessions", List(log_name, session_name, chapter, groups, threads, timing_elapsed, timing_cpu, timing_gc, timing_factor, ml_timing_elapsed, ml_timing_cpu, ml_timing_gc, ml_timing_factor, heap_size, status, errors, sources)) val theories_table = build_log_table("theories", List(log_name, session_name, theory_name, theory_timing_elapsed, theory_timing_cpu, theory_timing_gc)) val ml_statistics_table = build_log_table("ml_statistics", List(log_name, session_name, ml_statistics)) /* AFP versions */ val isabelle_afp_versions_table: SQL.Table = { val version1 = Prop.isabelle_version val version2 = Prop.afp_version build_log_table("isabelle_afp_versions", List(version1.make_primary_key, version2), SQL.select(List(version1, version2), distinct = true) + meta_info_table + SQL.where(SQL.and(version1.defined, version2.defined))) } /* earliest pull date for repository version (PostgreSQL queries) */ def pull_date(afp: Boolean = false): SQL.Column = if (afp) SQL.Column.date("afp_pull_date") else SQL.Column.date("pull_date") def pull_date_table(afp: Boolean = false): SQL.Table = { val (name, versions) = if (afp) ("afp_pull_date", List(Prop.isabelle_version, Prop.afp_version)) else ("pull_date", List(Prop.isabelle_version)) build_log_table(name, versions.map(_.make_primary_key) ::: List(pull_date(afp)), "SELECT " + versions.mkString(", ") + ", min(" + Prop.build_start + ") AS " + pull_date(afp) + " FROM " + meta_info_table + " WHERE " + SQL.AND((versions ::: List(Prop.build_start)).map(_.defined)) + " GROUP BY " + versions.mkString(", ")) } /* recent entries */ def recent_time(days: Int): PostgreSQL.Source = "now() - INTERVAL '" + days.max(0) + " days'" def recent_pull_date_table( days: Int, rev: String = "", afp_rev: Option[String] = None ): SQL.Table = { val afp = afp_rev.isDefined val rev2 = afp_rev.getOrElse("") val table = pull_date_table(afp) val eq_rev = if_proper(rev, Prop.isabelle_version(table).equal(rev)) val eq_rev2 = if_proper(rev2, Prop.afp_version(table).equal(rev2)) SQL.Table("recent_pull_date", table.columns, table.select(table.columns, sql = SQL.where( SQL.or(pull_date(afp)(table).ident + " > " + recent_time(days), SQL.and(eq_rev, eq_rev2))))) } def select_recent_log_names(days: Int): PostgreSQL.Source = { val table1 = meta_info_table val table2 = recent_pull_date_table(days) table1.select(List(log_name), distinct = true, sql = SQL.join_inner + table2.query_named + " ON " + Prop.isabelle_version(table1) + " = " + Prop.isabelle_version(table2)) } def select_recent_versions( days: Int, rev: String = "", afp_rev: Option[String] = None, sql: PostgreSQL.Source = "" ): PostgreSQL.Source = { val afp = afp_rev.isDefined val version = Prop.isabelle_version val table1 = recent_pull_date_table(days, rev = rev, afp_rev = afp_rev) val table2 = meta_info_table val aux_table = SQL.Table("aux", table2.columns, table2.select(sql = sql)) val columns = table1.columns.map(c => c(table1)) ::: List(known.copy(expr = log_name(aux_table).defined)) SQL.select(columns, distinct = true) + table1.query_named + SQL.join_outer + aux_table.query_named + " ON " + version(table1) + " = " + version(aux_table) + SQL.order_by(List(pull_date(afp)(table1)), descending = true) } /* universal view on main data */ val universal_table: SQL.Table = { val afp_pull_date = pull_date(afp = true) val version1 = Prop.isabelle_version val version2 = Prop.afp_version val table1 = meta_info_table val table2 = pull_date_table(afp = true) val table3 = pull_date_table() val a_columns = log_name :: afp_pull_date :: table1.columns.tail val a_table = SQL.Table("a", a_columns, SQL.select(List(log_name, afp_pull_date) ::: table1.columns.tail.map(_.apply(table1))) + table1 + SQL.join_outer + table2 + " ON " + SQL.and( version1(table1).ident + " = " + version1(table2).ident, version2(table1).ident + " = " + version2(table2).ident)) val b_columns = log_name :: pull_date() :: a_columns.tail val b_table = SQL.Table("b", b_columns, SQL.select( List(log_name(a_table), pull_date()(table3)) ::: a_columns.tail.map(_.apply(a_table))) + a_table.query_named + SQL.join_outer + table3 + " ON " + version1(a_table) + " = " + version1(table3)) val c_columns = b_columns ::: sessions_table.columns.tail val c_table = SQL.Table("c", c_columns, SQL.select(log_name(b_table) :: c_columns.tail) + b_table.query_named + SQL.join_inner + sessions_table + " ON " + log_name(b_table) + " = " + log_name(sessions_table)) SQL.Table("isabelle_build_log", c_columns ::: List(ml_statistics), { SQL.select(c_columns.map(_.apply(c_table)) ::: List(ml_statistics)) + c_table.query_named + SQL.join_outer + ml_statistics_table + " ON " + SQL.and( log_name(c_table).ident + " = " + log_name(ml_statistics_table).ident, session_name(c_table).ident + " = " + session_name(ml_statistics_table).ident) }) } } /* database access */ def store(options: Options, cache: XML.Cache = XML.Cache.make()): Store = new Store(options, cache) class Store private[Build_Log](options: Options, val cache: XML.Cache) { def open_database( user: String = options.string("build_log_database_user"), password: String = options.string("build_log_database_password"), database: String = options.string("build_log_database_name"), host: String = options.string("build_log_database_host"), port: Int = options.int("build_log_database_port"), ssh_host: String = options.string("build_log_ssh_host"), ssh_user: String = options.string("build_log_ssh_user"), ssh_port: Int = options.int("build_log_ssh_port") ): PostgreSQL.Database = { PostgreSQL.open_database( user = user, password = password, database = database, host = host, port = port, ssh = if (ssh_host == "") None else Some(SSH.open_session(options, host = ssh_host, user = ssh_user, port = ssh_port)), ssh_close = true) } def update_database( db: PostgreSQL.Database, dirs: List[Path], ml_statistics: Boolean = false): Unit = { val log_files = dirs.flatMap(dir => File.find_files(dir.file, pred = Log_File.is_log(_), follow_links = true)) write_info(db, log_files, ml_statistics = ml_statistics) db.create_view(Data.pull_date_table()) db.create_view(Data.pull_date_table(afp = true)) db.create_view(Data.universal_table) } def snapshot_database( db: PostgreSQL.Database, sqlite_database: Path, days: Int = 100, ml_statistics: Boolean = false ): Unit = { Isabelle_System.make_directory(sqlite_database.dir) sqlite_database.file.delete using(SQLite.open_database(sqlite_database)) { db2 => db.transaction { db2.transaction { // main content db2.create_table(Data.meta_info_table) db2.create_table(Data.sessions_table) db2.create_table(Data.theories_table) db2.create_table(Data.ml_statistics_table) val recent_log_names = db.using_statement(Data.select_recent_log_names(days))(stmt => stmt.execute_query().iterator(_.string(Data.log_name)).toList) for (log_name <- recent_log_names) { read_meta_info(db, log_name).foreach(meta_info => update_meta_info(db2, log_name, meta_info)) update_sessions(db2, log_name, read_build_info(db, log_name)) if (ml_statistics) { update_ml_statistics(db2, log_name, read_build_info(db, log_name, ml_statistics = true)) } } // pull_date for (afp <- List(false, true)) { val afp_rev = if (afp) Some("") else None val table = Data.pull_date_table(afp) db2.create_table(table) db2.using_statement(table.insert()) { stmt2 => db.using_statement( Data.recent_pull_date_table(days, afp_rev = afp_rev).query) { stmt => val res = stmt.execute_query() while (res.next()) { for ((c, i) <- table.columns.zipWithIndex) { stmt2.string(i + 1) = res.get_string(c) } stmt2.execute() } } } } // full view db2.create_view(Data.universal_table) } } db2.rebuild() } } def domain(db: SQL.Database, table: SQL.Table, column: SQL.Column): Set[String] = db.using_statement(table.select(List(column), distinct = true))(stmt => stmt.execute_query().iterator(_.string(column)).toSet) def update_meta_info(db: SQL.Database, log_name: String, meta_info: Meta_Info): Unit = { val table = Data.meta_info_table db.using_statement(db.insert_permissive(table)) { stmt => stmt.string(1) = log_name for ((c, i) <- table.columns.tail.zipWithIndex) { if (c.T == SQL.Type.Date) stmt.date(i + 2) = meta_info.get_date(c) else stmt.string(i + 2) = meta_info.get(c) } stmt.execute() } } def update_sessions(db: SQL.Database, log_name: String, build_info: Build_Info): Unit = { val table = Data.sessions_table db.using_statement(db.insert_permissive(table)) { stmt => val sessions = if (build_info.sessions.isEmpty) Build_Info.sessions_dummy else build_info.sessions for ((session_name, session) <- sessions) { stmt.string(1) = log_name stmt.string(2) = session_name stmt.string(3) = proper_string(session.chapter) stmt.string(4) = session.proper_groups stmt.int(5) = session.threads stmt.long(6) = session.timing.elapsed.proper_ms stmt.long(7) = session.timing.cpu.proper_ms stmt.long(8) = session.timing.gc.proper_ms stmt.double(9) = session.timing.factor stmt.long(10) = session.ml_timing.elapsed.proper_ms stmt.long(11) = session.ml_timing.cpu.proper_ms stmt.long(12) = session.ml_timing.gc.proper_ms stmt.double(13) = session.ml_timing.factor stmt.long(14) = session.heap_size.map(_.bytes) stmt.string(15) = session.status.map(_.toString) stmt.bytes(16) = compress_errors(session.errors, cache = cache.compress) stmt.string(17) = session.sources stmt.execute() } } } def update_theories(db: SQL.Database, log_name: String, build_info: Build_Info): Unit = { val table = Data.theories_table db.using_statement(db.insert_permissive(table)) { stmt => val sessions = if (build_info.sessions.forall({ case (_, session) => session.theory_timings.isEmpty })) Build_Info.sessions_dummy else build_info.sessions for { (session_name, session) <- sessions (theory_name, timing) <- session.theory_timings } { stmt.string(1) = log_name stmt.string(2) = session_name stmt.string(3) = theory_name stmt.long(4) = timing.elapsed.ms stmt.long(5) = timing.cpu.ms stmt.long(6) = timing.gc.ms stmt.execute() } } } def update_ml_statistics(db: SQL.Database, log_name: String, build_info: Build_Info): Unit = { val table = Data.ml_statistics_table db.using_statement(db.insert_permissive(table)) { stmt => val ml_stats: List[(String, Option[Bytes])] = Par_List.map[(String, Session_Entry), (String, Option[Bytes])]( { case (a, b) => (a, Properties.compress(b.ml_statistics, cache = cache.compress).proper) }, build_info.sessions.iterator.filter(p => p._2.ml_statistics.nonEmpty).toList) val entries = if (ml_stats.nonEmpty) ml_stats else List("" -> None) for ((session_name, ml_statistics) <- entries) { stmt.string(1) = log_name stmt.string(2) = session_name stmt.bytes(3) = ml_statistics stmt.execute() } } } def write_info(db: SQL.Database, files: List[JFile], ml_statistics: Boolean = false): Unit = { abstract class Table_Status(table: SQL.Table) { db.create_table(table) private var known: Set[String] = domain(db, table, Data.log_name) def required(file: JFile): Boolean = !known(Log_File.plain_name(file.getName)) def update_db(db: SQL.Database, log_file: Log_File): Unit def update(log_file: Log_File): Unit = { if (!known(log_file.name)) { update_db(db, log_file) known += log_file.name } } } val status = List( new Table_Status(Data.meta_info_table) { override def update_db(db: SQL.Database, log_file: Log_File): Unit = update_meta_info(db, log_file.name, log_file.parse_meta_info()) }, new Table_Status(Data.sessions_table) { override def update_db(db: SQL.Database, log_file: Log_File): Unit = update_sessions(db, log_file.name, log_file.parse_build_info()) }, new Table_Status(Data.theories_table) { override def update_db(db: SQL.Database, log_file: Log_File): Unit = update_theories(db, log_file.name, log_file.parse_build_info()) }, new Table_Status(Data.ml_statistics_table) { override def update_db(db: SQL.Database, log_file: Log_File): Unit = if (ml_statistics) { update_ml_statistics(db, log_file.name, log_file.parse_build_info(ml_statistics = true)) } }) for (file_group <- files.filter(file => status.exists(_.required(file))). grouped(options.int("build_log_transaction_size") max 1)) { val log_files = Par_List.map[JFile, Log_File](Log_File.apply, file_group) db.transaction { log_files.foreach(log_file => status.foreach(_.update(log_file))) } } } def read_meta_info(db: SQL.Database, log_name: String): Option[Meta_Info] = { val table = Data.meta_info_table val columns = table.columns.tail db.using_statement(table.select(columns, sql = Data.log_name.where_equal(log_name))) { stmt => val res = stmt.execute_query() if (!res.next()) None else { val results = columns.map(c => c.name -> (if (c.T == SQL.Type.Date) res.get_date(c).map(Log_File.Date_Format(_)) else res.get_string(c))) val n = Prop.all_props.length val props = for ((x, Some(y)) <- results.take(n)) yield (x, y) val settings = for ((x, Some(y)) <- results.drop(n)) yield (x, y) Some(Meta_Info(props, settings)) } } } def read_build_info( db: SQL.Database, log_name: String, session_names: List[String] = Nil, ml_statistics: Boolean = false ): Build_Info = { val table1 = Data.sessions_table val table2 = Data.ml_statistics_table val columns1 = table1.columns.tail.map(_.apply(table1)) val (columns, from) = if (ml_statistics) { val columns = columns1 ::: List(Data.ml_statistics(table2)) val join = table1.ident + SQL.join_outer + table2.ident + " ON " + SQL.and( Data.log_name(table1).ident + " = " + Data.log_name(table2).ident, Data.session_name(table1).ident + " = " + Data.session_name(table2).ident) (columns, SQL.enclose(join)) } else (columns1, table1.ident) val where = SQL.where( SQL.and( Data.log_name(table1).equal(log_name), Data.session_name(table1).ident + " <> ''", if_proper(session_names, Data.session_name(table1).member(session_names)))) val sessions = db.using_statement(SQL.select(columns, sql = from + where)) { stmt => stmt.execute_query().iterator({ res => val session_name = res.string(Data.session_name) val session_entry = Session_Entry( chapter = res.string(Data.chapter), groups = split_lines(res.string(Data.groups)), threads = res.get_int(Data.threads), - timing = res.timing(Data.timing_elapsed, Data.timing_cpu, Data.timing_gc), + timing = + res.timing( + Data.timing_elapsed, + Data.timing_cpu, + Data.timing_gc), ml_timing = - res.timing(Data.ml_timing_elapsed, Data.ml_timing_cpu, Data.ml_timing_gc), + res.timing( + Data.ml_timing_elapsed, + Data.ml_timing_cpu, + Data.ml_timing_gc), sources = res.get_string(Data.sources), heap_size = res.get_long(Data.heap_size).map(Space.bytes), status = res.get_string(Data.status).map(Session_Status.withName), errors = uncompress_errors(res.bytes(Data.errors), cache = cache), ml_statistics = if (ml_statistics) { Properties.uncompress(res.bytes(Data.ml_statistics), cache = cache) } else Nil) session_name -> session_entry }).toMap } Build_Info(sessions) } } } diff --git a/src/Pure/Tools/build_job.scala b/src/Pure/Tools/build_job.scala --- a/src/Pure/Tools/build_job.scala +++ b/src/Pure/Tools/build_job.scala @@ -1,751 +1,754 @@ /* Title: Pure/Tools/build_job.scala Author: Makarius Build job running prover process, with rudimentary PIDE session. */ package isabelle import scala.collection.mutable import scala.util.matching.Regex trait Build_Job { def job_name: String def node_info: Host.Node_Info def cancel(): Unit = () def is_finished: Boolean = false def join: (Process_Result, SHA1.Shasum) = (Process_Result.undefined, SHA1.no_shasum) def make_abstract: Build_Job.Abstract = Build_Job.Abstract(job_name, node_info) } object Build_Job { sealed case class Result(node_info: Host.Node_Info, process_result: Process_Result) { def ok: Boolean = process_result.ok } sealed case class Abstract( override val job_name: String, override val node_info: Host.Node_Info ) extends Build_Job { override def make_abstract: Abstract = this } /* build session */ def is_session_name(job_name: String): Boolean = !Long_Name.is_qualified(job_name) def start_session( build_context: Build_Process.Context, session_background: Sessions.Background, input_shasum: SHA1.Shasum, node_info: Host.Node_Info ): Session_Job = new Session_Job(build_context, session_background, input_shasum, node_info) object Session_Context { def load( + uuid: String, name: String, deps: List[String], ancestors: List[String], sources_shasum: SHA1.Shasum, timeout: Time, store: Sessions.Store, progress: Progress = new Progress ): Session_Context = { def default: Session_Context = - new Session_Context(name, deps, ancestors, sources_shasum, timeout, Time.zero, Bytes.empty) + Session_Context( + name, deps, ancestors, sources_shasum, timeout, Time.zero, Bytes.empty, uuid) store.try_open_database(name) match { case None => default case Some(db) => def ignore_error(msg: String) = { progress.echo_warning( "Ignoring bad database " + db + " for session " + quote(name) + if_proper(msg, ":\n" + msg)) default } try { val command_timings = store.read_command_timings(db, name) val elapsed = store.read_session_timing(db, name) match { case Markup.Elapsed(s) => Time.seconds(s) case _ => Time.zero } new Session_Context( - name, deps, ancestors, sources_shasum, timeout, elapsed, command_timings) + name, deps, ancestors, sources_shasum, timeout, elapsed, command_timings, uuid) } catch { case ERROR(msg) => ignore_error(msg) case exn: java.lang.Error => ignore_error(Exn.message(exn)) case _: XML.Error => ignore_error("XML.Error") } finally { db.close() } } } } - final class Session_Context( - val name: String, - val deps: List[String], - val ancestors: List[String], - val sources_shasum: SHA1.Shasum, - val timeout: Time, - val old_time: Time, - val old_command_timings_blob: Bytes + sealed case class Session_Context( + name: String, + deps: List[String], + ancestors: List[String], + sources_shasum: SHA1.Shasum, + timeout: Time, + old_time: Time, + old_command_timings_blob: Bytes, + uuid: String ) { override def toString: String = name } class Session_Job private[Build_Job]( build_context: Build_Process.Context, session_background: Sessions.Background, input_shasum: SHA1.Shasum, override val node_info: Host.Node_Info ) extends Build_Job { private val store = build_context.store private val progress = build_context.progress private val verbose = build_context.verbose def session_name: String = session_background.session_name def job_name: String = session_name - val info: Sessions.Info = session_background.sessions_structure(session_name) - val options: Options = Host.process_policy_options(info.options, node_info.numa_node) + private val info: Sessions.Info = session_background.sessions_structure(session_name) + private val options: Options = Host.process_policy_options(info.options, node_info.numa_node) - val session_sources: Sessions.Sources = + private val session_sources = Sessions.Sources.load(session_background.base, cache = store.cache.compress) - val store_heap = build_context.store_heap(session_name) + private val store_heap = build_context.store_heap(session_name) private val future_result: Future[(Process_Result, SHA1.Shasum)] = Future.thread("build", uninterruptible = true) { val env = Isabelle_System.settings( List("ISABELLE_ML_DEBUGGER" -> options.bool("ML_debugger").toString)) val session_heaps = session_background.info.parent match { case None => Nil case Some(logic) => ML_Process.session_heaps(store, session_background, logic = logic) } val use_prelude = if (session_heaps.isEmpty) Thy_Header.ml_roots.map(_._1) else Nil val eval_store = if (store_heap) { (if (info.theories.nonEmpty) List("ML_Heap.share_common_data ()") else Nil) ::: List("ML_Heap.save_child " + ML_Syntax.print_string_bytes(File.platform_path(store.output_heap(session_name)))) } else Nil def session_blobs(node_name: Document.Node.Name): List[(Command.Blob, Document.Blobs.Item)] = session_background.base.theory_load_commands.get(node_name.theory) match { case None => Nil case Some(spans) => val syntax = session_background.base.theory_syntax(node_name) val master_dir = Path.explode(node_name.master_dir) for (span <- spans; file <- span.loaded_files(syntax).files) yield { val src_path = Path.explode(file) val blob_name = Document.Node.Name(File.symbolic_path(master_dir + src_path)) val bytes = session_sources(blob_name.node).bytes val text = bytes.text val chunk = Symbol.Text_Chunk(text) Command.Blob(blob_name, src_path, Some((SHA1.digest(bytes), chunk))) -> Document.Blobs.Item(bytes, text, chunk, changed = false) } } /* session */ val resources = new Resources(session_background, log = build_context.log, command_timings = build_context.old_command_timings(session_name)) val session = new Session(options, resources) { override val cache: Term.Cache = store.cache override def build_blobs_info(node_name: Document.Node.Name): Command.Blobs_Info = Command.Blobs_Info.make(session_blobs(node_name)) override def build_blobs(node_name: Document.Node.Name): Document.Blobs = Document.Blobs.make(session_blobs(node_name)) } object Build_Session_Errors { private val promise: Promise[List[String]] = Future.promise def result: Exn.Result[List[String]] = promise.join_result def cancel(): Unit = promise.cancel() def apply(errs: List[String]): Unit = { try { promise.fulfill(errs) } catch { case _: IllegalStateException => } } } val export_consumer = Export.consumer(store.open_database(session_name, output = true), store.cache, progress = progress) val stdout = new StringBuilder(1000) val stderr = new StringBuilder(1000) val command_timings = new mutable.ListBuffer[Properties.T] val theory_timings = new mutable.ListBuffer[Properties.T] val session_timings = new mutable.ListBuffer[Properties.T] val runtime_statistics = new mutable.ListBuffer[Properties.T] val task_statistics = new mutable.ListBuffer[Properties.T] def fun( name: String, acc: mutable.ListBuffer[Properties.T], unapply: Properties.T => Option[Properties.T] ): (String, Session.Protocol_Function) = { name -> ((msg: Prover.Protocol_Output) => unapply(msg.properties) match { case Some(props) => acc += props; true case _ => false }) } session.init_protocol_handler(new Session.Protocol_Handler { override def exit(): Unit = Build_Session_Errors.cancel() private def build_session_finished(msg: Prover.Protocol_Output): Boolean = { val (rc, errors) = try { val (rc, errs) = { import XML.Decode._ pair(int, list(x => x))(Symbol.decode_yxml(msg.text)) } val errors = for (err <- errs) yield { val prt = Protocol_Message.expose_no_reports(err) Pretty.string_of(prt, metric = Symbol.Metric) } (rc, errors) } catch { case ERROR(err) => (Process_Result.RC.failure, List(err)) } session.protocol_command("Prover.stop", rc.toString) Build_Session_Errors(errors) true } private def loading_theory(msg: Prover.Protocol_Output): Boolean = msg.properties match { case Markup.Loading_Theory(Markup.Name(name)) => progress.theory(Progress.Theory(name, session = session_name)) false case _ => false } private def export_(msg: Prover.Protocol_Output): Boolean = msg.properties match { case Protocol.Export(args) => export_consumer.make_entry(session_name, args, msg.chunk) true case _ => false } override val functions: Session.Protocol_Functions = List( Markup.Build_Session_Finished.name -> build_session_finished, Markup.Loading_Theory.name -> loading_theory, Markup.EXPORT -> export_, fun(Markup.Theory_Timing.name, theory_timings, Markup.Theory_Timing.unapply), fun(Markup.Session_Timing.name, session_timings, Markup.Session_Timing.unapply), fun(Markup.Task_Statistics.name, task_statistics, Markup.Task_Statistics.unapply)) }) session.command_timings += Session.Consumer("command_timings") { case Session.Command_Timing(props) => for { elapsed <- Markup.Elapsed.unapply(props) elapsed_time = Time.seconds(elapsed) if elapsed_time.is_relevant && elapsed_time >= options.seconds("command_timing_threshold") } command_timings += props.filter(Markup.command_timing_property) } session.runtime_statistics += Session.Consumer("ML_statistics") { case Session.Runtime_Statistics(props) => runtime_statistics += props } session.finished_theories += Session.Consumer[Document.Snapshot]("finished_theories") { case snapshot => if (!progress.stopped) { def export_(name: String, xml: XML.Body, compress: Boolean = true): Unit = { if (!progress.stopped) { val theory_name = snapshot.node_name.theory val args = Protocol.Export.Args(theory_name = theory_name, name = name, compress = compress) val body = Bytes(Symbol.encode(YXML.string_of_body(xml))) export_consumer.make_entry(session_name, args, body) } } def export_text(name: String, text: String, compress: Boolean = true): Unit = export_(name, List(XML.Text(text)), compress = compress) for (command <- snapshot.snippet_command) { export_text(Export.DOCUMENT_ID, command.id.toString, compress = false) } export_text(Export.FILES, cat_lines(snapshot.node_files.map(name => File.symbolic_path(name.path))), compress = false) for ((blob_name, i) <- snapshot.node_files.tail.zipWithIndex) { val xml = snapshot.switch(blob_name).xml_markup() export_(Export.MARKUP + (i + 1), xml) } export_(Export.MARKUP, snapshot.xml_markup()) export_(Export.MESSAGES, snapshot.messages.map(_._1)) } } session.all_messages += Session.Consumer[Any]("build_session_output") { case msg: Prover.Output => val message = msg.message if (msg.is_system) resources.log(Protocol.message_text(message)) if (msg.is_stdout) { stdout ++= Symbol.encode(XML.content(message)) } else if (msg.is_stderr) { stderr ++= Symbol.encode(XML.content(message)) } else if (msg.is_exit) { val err = "Prover terminated" + (msg.properties match { case Markup.Process_Result(result) => ": " + result.print_rc case _ => "" }) Build_Session_Errors(List(err)) } case _ => } build_context.session_setup(session_name, session) val eval_main = Command_Line.ML_tool("Isabelle_Process.init_build ()" :: eval_store) /* process */ val process = Isabelle_Process.start(options, session, session_background, session_heaps, use_prelude = use_prelude, eval_main = eval_main, cwd = info.dir.file, env = env) val timeout_request: Option[Event_Timer.Request] = if (info.timeout_ignored) None else Some(Event_Timer.request(Time.now() + info.timeout) { process.terminate() }) val build_errors = Isabelle_Thread.interrupt_handler(_ => process.terminate()) { Exn.capture { process.await_startup() } match { case Exn.Res(_) => val resources_yxml = resources.init_session_yxml val encode_options: XML.Encode.T[Options] = options => session.prover_options(options).encode val args_yxml = YXML.string_of_body( { import XML.Encode._ pair(string, list(pair(encode_options, list(pair(string, properties)))))( (session_name, info.theories)) }) session.protocol_command("build_session", resources_yxml, args_yxml) Build_Session_Errors.result case Exn.Exn(exn) => Exn.Res(List(Exn.message(exn))) } } val result0 = Isabelle_Thread.interrupt_handler(_ => process.terminate()) { process.await_shutdown() } val was_timeout = timeout_request match { case None => false case Some(request) => !request.cancel() } session.stop() val export_errors = export_consumer.shutdown(close = true).map(Output.error_message_text) val (document_output, document_errors) = try { if (build_errors.isInstanceOf[Exn.Res[_]] && result0.ok && info.documents.nonEmpty) { using(Export.open_database_context(store)) { database_context => val documents = using(database_context.open_session(session_background)) { session_context => Document_Build.build_documents( Document_Build.context(session_context, progress = progress), output_sources = info.document_output, output_pdf = info.document_output) } using(database_context.open_database(session_name, output = true))(session_database => documents.foreach(_.write(session_database.db, session_name))) (documents.flatMap(_.log_lines), Nil) } } else (Nil, Nil) } catch { case exn: Document_Build.Build_Error => (exn.log_lines, exn.log_errors) case Exn.Interrupt.ERROR(msg) => (Nil, List(msg)) } /* process result */ val result1 = { val theory_timing = theory_timings.iterator.flatMap( { case props @ Markup.Name(name) => Some(name -> props) case _ => None }).toMap val used_theory_timings = for { (name, _) <- session_background.base.used_theories } yield theory_timing.getOrElse(name.theory, Markup.Name(name.theory)) val more_output = Library.trim_line(stdout.toString) :: command_timings.toList.map(Protocol.Command_Timing_Marker.apply) ::: used_theory_timings.map(Protocol.Theory_Timing_Marker.apply) ::: session_timings.toList.map(Protocol.Session_Timing_Marker.apply) ::: runtime_statistics.toList.map(Protocol.ML_Statistics_Marker.apply) ::: task_statistics.toList.map(Protocol.Task_Statistics_Marker.apply) ::: document_output result0.output(more_output) .error(Library.trim_line(stderr.toString)) .errors_rc(export_errors ::: document_errors) } val result2 = build_errors match { case Exn.Res(build_errs) => val errs = build_errs ::: document_errors if (errs.nonEmpty) { result1.error_rc.output( errs.flatMap(s => split_lines(Output.error_message_text(s))) ::: errs.map(Protocol.Error_Message_Marker.apply)) } else if (progress.stopped && result1.ok) result1.copy(rc = Process_Result.RC.interrupt) else result1 case Exn.Exn(Exn.Interrupt()) => if (result1.ok) result1.copy(rc = Process_Result.RC.interrupt) else result1 case Exn.Exn(exn) => throw exn } val process_result = if (result2.ok) result2 else if (was_timeout) result2.error(Output.error_message_text("Timeout")).timeout_rc else if (result2.interrupted) result2.error(Output.error_message_text("Interrupt")) else result2 /* output heap */ val output_shasum = if (process_result.ok && store_heap && store.output_heap(session_name).is_file) { SHA1.shasum(ML_Heap.write_digest(store.output_heap(session_name)), session_name) } else SHA1.no_shasum val log_lines = process_result.out_lines.filterNot(Protocol_Message.Marker.test) val build_log = Build_Log.Log_File(session_name, process_result.out_lines). parse_session_info( command_timings = true, theory_timings = true, ml_statistics = true, task_statistics = true) // write log file if (process_result.ok) { File.write_gzip(store.output_log_gz(session_name), terminate_lines(log_lines)) } else File.write(store.output_log(session_name), terminate_lines(log_lines)) // write database using(store.open_database(session_name, output = true))(db => store.write_session_info(db, session_name, session_sources, build_log = if (process_result.timeout) build_log.error("Timeout") else build_log, build = Sessions.Build_Info( sources = build_context.sources_shasum(session_name), input_heaps = input_shasum, output_heap = output_shasum, process_result.rc, build_context.uuid))) // messages process_result.err_lines.foreach(progress.echo) if (process_result.ok) { if (verbose) { val props = build_log.session_timing val threads = Markup.Session_Timing.Threads.unapply(props) getOrElse 1 val timing = Markup.Timing_Properties.get(props) progress.echo( "Timing " + session_name + " (" + threads + " threads, " + timing.message_factor + ")") } progress.echo( "Finished " + session_name + " (" + process_result.timing.message_resources + ")") } else { progress.echo( session_name + " FAILED (see also \"isabelle log -H Error " + session_name + "\")") if (!process_result.interrupted) { val tail = info.options.int("process_output_tail") val suffix = if (tail == 0) log_lines else log_lines.drop(log_lines.length - tail max 0) val prefix = if (log_lines.length == suffix.length) Nil else List("...") progress.echo(Library.trim_line(cat_lines(prefix ::: suffix))) } } (process_result.copy(out_lines = log_lines), output_shasum) } override def cancel(): Unit = future_result.cancel() override def is_finished: Boolean = future_result.is_finished override def join: (Process_Result, SHA1.Shasum) = future_result.join } /* theory markup/messages from session database */ def read_theory( theory_context: Export.Theory_Context, unicode_symbols: Boolean = false ): Option[Document.Snapshot] = { def decode_bytes(bytes: Bytes): String = Symbol.output(unicode_symbols, UTF8.decode_permissive(bytes)) def read(name: String): Export.Entry = theory_context(name, permissive = true) def read_xml(name: String): XML.Body = YXML.parse_body(decode_bytes(read(name).bytes), cache = theory_context.cache) def read_source_file(name: String): Sessions.Source_File = theory_context.session_context.source_file(name) for { id <- theory_context.document_id() (thy_file, blobs_files) <- theory_context.files(permissive = true) } yield { val master_dir = Path.explode(Url.strip_base_name(thy_file).getOrElse( error("Cannot determine theory master directory: " + quote(thy_file)))) val blobs = blobs_files.map { name => val path = Path.explode(name) val src_path = File.relative_path(master_dir, path).getOrElse(path) val file = read_source_file(name) val bytes = file.bytes val text = decode_bytes(bytes) val chunk = Symbol.Text_Chunk(text) Command.Blob(Document.Node.Name(name), src_path, Some((file.digest, chunk))) -> Document.Blobs.Item(bytes, text, chunk, changed = false) } val thy_source = decode_bytes(read_source_file(thy_file).bytes) val thy_xml = read_xml(Export.MARKUP) val blobs_xml = for (i <- (1 to blobs.length).toList) yield read_xml(Export.MARKUP + i) val markups_index = Command.Markup_Index.make(blobs.map(_._1)) val markups = Command.Markups.make( for ((index, xml) <- markups_index.zip(thy_xml :: blobs_xml)) yield index -> Markup_Tree.from_XML(xml)) val results = Command.Results.make( for (elem @ XML.Elem(Markup(_, Markup.Serial(i)), _) <- read_xml(Export.MESSAGES)) yield i -> elem) val command = Command.unparsed(thy_source, theory = true, id = id, node_name = Document.Node.Name(thy_file, theory = theory_context.theory), blobs_info = Command.Blobs_Info.make(blobs), markups = markups, results = results) val doc_blobs = Document.Blobs.make(blobs) Document.State.init.snippet(command, doc_blobs) } } /* print messages */ def print_log( options: Options, sessions: List[String], theories: List[String] = Nil, message_head: List[Regex] = Nil, message_body: List[Regex] = Nil, verbose: Boolean = false, progress: Progress = new Progress, margin: Double = Pretty.default_margin, breakgain: Double = Pretty.default_breakgain, metric: Pretty.Metric = Symbol.Metric, unicode_symbols: Boolean = false ): Unit = { val store = Sessions.store(options) val session = new Session(options, Resources.bootstrap) def check(filter: List[Regex], make_string: => String): Boolean = filter.isEmpty || { val s = Output.clean_yxml(make_string) filter.forall(r => r.findFirstIn(Output.clean_yxml(s)).nonEmpty) } def print(session_name: String): Unit = { using(Export.open_session_context0(store, session_name)) { session_context => val result = for { db <- session_context.session_db() theories = store.read_theories(db, session_name) errors = store.read_errors(db, session_name) info <- store.read_build(db, session_name) } yield (theories, errors, info.return_code) result match { case None => store.error_database(session_name) case Some((used_theories, errors, rc)) => theories.filterNot(used_theories.toSet) match { case Nil => case bad => error("Unknown theories " + commas_quote(bad)) } val print_theories = if (theories.isEmpty) used_theories else used_theories.filter(theories.toSet) for (thy <- print_theories) { val thy_heading = "\nTheory " + quote(thy) + " (in " + session_name + ")" + ":" read_theory(session_context.theory(thy), unicode_symbols = unicode_symbols) match { case None => progress.echo(thy_heading + " MISSING") case Some(snapshot) => val rendering = new Rendering(snapshot, options, session) val messages = rendering.text_messages(Text.Range.full) .filter(message => verbose || Protocol.is_exported(message.info)) if (messages.nonEmpty) { val line_document = Line.Document(snapshot.node.source) val buffer = new mutable.ListBuffer[String] for (Text.Info(range, elem) <- messages) { val line = line_document.position(range.start).line1 val pos = Position.Line_File(line, snapshot.node_name.node) def message_text: String = Protocol.message_text(elem, heading = true, pos = pos, margin = margin, breakgain = breakgain, metric = metric) val ok = check(message_head, Protocol.message_heading(elem, pos)) && check(message_body, XML.content(Pretty.unformatted(List(elem)))) if (ok) buffer += message_text } if (buffer.nonEmpty) { progress.echo(thy_heading) buffer.foreach(progress.echo) } } } } if (errors.nonEmpty) { val msg = Symbol.output(unicode_symbols, cat_lines(errors)) progress.echo("\nBuild errors:\n" + Output.error_message_text(msg)) } if (rc != Process_Result.RC.ok) { progress.echo("\n" + Process_Result.RC.print_long(rc)) } } } } val errors = new mutable.ListBuffer[String] for (session_name <- sessions) { Exn.interruptible_capture(print(session_name)) match { case Exn.Res(_) => case Exn.Exn(exn) => errors += Exn.message(exn) } } if (errors.nonEmpty) error(cat_lines(errors.toList)) } /* Isabelle tool wrapper */ val isabelle_tool = Isabelle_Tool("log", "print messages from build database", Scala_Project.here, { args => /* arguments */ var message_head = List.empty[Regex] var message_body = List.empty[Regex] var unicode_symbols = false var theories: List[String] = Nil var margin = Pretty.default_margin var options = Options.init() var verbose = false val getopts = Getopts(""" Usage: isabelle log [OPTIONS] [SESSIONS ...] Options are: -H REGEX filter messages by matching against head -M REGEX filter messages by matching against body -T NAME restrict to given theories (multiple options possible) -U output Unicode symbols -m MARGIN margin for pretty printing (default: """ + margin + """) -o OPTION override Isabelle system OPTION (via NAME=VAL or NAME) -v print all messages, including information etc. Print messages from the build database of the given sessions, without any checks against current sources nor session structure: results from old sessions or failed builds can be printed as well. Multiple options -H and -M are conjunctive: all given patterns need to match. Patterns match any substring, but ^ or $ may be used to match the start or end explicitly. """, "H:" -> (arg => message_head = message_head ::: List(arg.r)), "M:" -> (arg => message_body = message_body ::: List(arg.r)), "T:" -> (arg => theories = theories ::: List(arg)), "U" -> (_ => unicode_symbols = true), "m:" -> (arg => margin = Value.Double.parse(arg)), "o:" -> (arg => options = options + arg), "v" -> (_ => verbose = true)) val sessions = getopts(args) val progress = new Console_Progress() if (sessions.isEmpty) progress.echo_warning("No sessions to print") else { print_log(options, sessions, theories = theories, message_head = message_head, message_body = message_body, verbose = verbose, margin = margin, progress = progress, unicode_symbols = unicode_symbols) } }) } diff --git a/src/Pure/Tools/build_process.scala b/src/Pure/Tools/build_process.scala --- a/src/Pure/Tools/build_process.scala +++ b/src/Pure/Tools/build_process.scala @@ -1,637 +1,717 @@ /* Title: Pure/Tools/build_process.scala Author: Makarius Build process for sessions, with build database, optional heap, and optional presentation. */ package isabelle import scala.math.Ordering import scala.annotation.tailrec object Build_Process { /** static context **/ object Context { def apply( store: Sessions.Store, build_deps: Sessions.Deps, progress: Progress = new Progress, hostname: String = Isabelle_System.hostname(), numa_shuffling: Boolean = false, build_heap: Boolean = false, max_jobs: Int = 1, fresh_build: Boolean = false, no_build: Boolean = false, verbose: Boolean = false, session_setup: (String, Session) => Unit = (_, _) => (), uuid: String = UUID.random().toString ): Context = { val sessions_structure = build_deps.sessions_structure val build_graph = sessions_structure.build_graph val sessions = Map.from( for ((name, (info, _)) <- build_graph.iterator) yield { val deps = info.parent.toList val ancestors = sessions_structure.build_requirements(deps) val sources_shasum = build_deps.sources_shasum(name) val session_context = Build_Job.Session_Context.load( - name, deps, ancestors, sources_shasum, info.timeout, store, progress = progress) + uuid, name, deps, ancestors, sources_shasum, info.timeout, store, + progress = progress) name -> session_context }) val sessions_time = { val maximals = build_graph.maximals.toSet def descendants_time(name: String): Double = { if (maximals.contains(name)) sessions(name).old_time.seconds else { val descendants = build_graph.all_succs(List(name)).toSet val g = build_graph.restrict(descendants) (0.0 :: g.maximals.flatMap { desc => val ps = g.all_preds(List(desc)) if (ps.exists(p => !sessions.isDefinedAt(p))) None else Some(ps.map(p => sessions(p).old_time.seconds).sum) }).max } } Map.from( for (name <- sessions.keysIterator) yield name -> descendants_time(name)).withDefaultValue(0.0) } val ordering = new Ordering[String] { def compare(name1: String, name2: String): Int = sessions_time(name2) compare sessions_time(name1) match { case 0 => sessions(name2).timeout compare sessions(name1).timeout match { case 0 => name1 compare name2 case ord => ord } case ord => ord } } val numa_nodes = Host.numa_nodes(enabled = numa_shuffling) new Context(store, build_deps, sessions, ordering, progress, hostname, numa_nodes, build_heap = build_heap, max_jobs = max_jobs, fresh_build = fresh_build, no_build = no_build, verbose = verbose, session_setup, uuid = uuid) } } + // static context of one particular instance, identified by uuid final class Context private( val store: Sessions.Store, val build_deps: Sessions.Deps, - val sessions: Map[String, Build_Job.Session_Context], + val sessions: State.Sessions, val ordering: Ordering[String], val progress: Progress, val hostname: String, val numa_nodes: List[Int], val build_heap: Boolean, val max_jobs: Int, val fresh_build: Boolean, val no_build: Boolean, val verbose: Boolean, val session_setup: (String, Session) => Unit, val uuid: String ) { def build_options: Options = store.options val log: Logger = build_options.string("system_log") match { case "" => No_Logger case "-" => Logger.make(progress) case log_file => Logger.make(Some(Path.explode(log_file))) } def sessions_structure: Sessions.Structure = build_deps.sessions_structure def sources_shasum(name: String): SHA1.Shasum = sessions(name).sources_shasum def old_command_timings(name: String): List[Properties.T] = sessions.get(name) match { case Some(session_context) => Properties.uncompress(session_context.old_command_timings_blob, cache = store.cache) case None => Nil } def store_heap(name: String): Boolean = build_heap || Sessions.is_pure(name) || sessions.valuesIterator.exists(_.ancestors.contains(name)) } /** dynamic state **/ case class Entry(name: String, deps: List[String], info: JSON.Object.T = JSON.Object.empty) { def is_ready: Boolean = deps.isEmpty def resolve(dep: String): Entry = if (deps.contains(dep)) copy(deps = deps.filterNot(_ == dep)) else this } case class Result( process_result: Process_Result, output_shasum: SHA1.Shasum, node_info: Host.Node_Info, current: Boolean ) { def ok: Boolean = process_result.ok } + object State { + type Sessions = Map[String, Build_Job.Session_Context] + type Pending = List[Entry] + type Running = Map[String, Build_Job] + type Results = Map[String, Result] + } + + // dynamic state of various instances, distinguished by uuid sealed case class State( serial: Long = 0, numa_index: Int = 0, - pending: List[Entry] = Nil, - running: Map[String, Build_Job] = Map.empty, - results: Map[String, Build_Process.Result] = Map.empty + sessions: State.Sessions = Map.empty, // static build targets + pending: State.Pending = Nil, // dynamic build "queue" + running: State.Running = Map.empty, // presently running jobs + results: State.Results = Map.empty // finished results ) { def numa_next(numa_nodes: List[Int]): (Option[Int], State) = if (numa_nodes.isEmpty) (None, this) else { val available = numa_nodes.zipWithIndex val used = Set.from(for (job <- running.valuesIterator; i <- job.node_info.numa_node) yield i) val candidates = available.drop(numa_index) ::: available.take(numa_index) val (n, i) = candidates.find({ case (n, i) => i == numa_index && !used(n) }) orElse candidates.find({ case (n, _) => !used(n) }) getOrElse candidates.head (Some(n), copy(numa_index = (i + 1) % available.length)) } def finished: Boolean = pending.isEmpty def remove_pending(name: String): State = copy(pending = pending.flatMap( entry => if (entry.name == name) None else Some(entry.resolve(name)))) def is_running(name: String): Boolean = running.isDefinedAt(name) def stop_running(): Unit = running.valuesIterator.foreach(_.cancel()) def finished_running(): List[Build_Job] = List.from(running.valuesIterator.filter(_.is_finished)) def add_running(name: String, job: Build_Job): State = copy(running = running + (name -> job)) def remove_running(name: String): State = copy(running = running - name) def make_result( name: String, process_result: Process_Result, output_shasum: SHA1.Shasum, node_info: Host.Node_Info = Host.Node_Info.none, current: Boolean = false ): State = { val entry = name -> Build_Process.Result(process_result, output_shasum, node_info, current) copy(results = results + entry) } } /** SQL data model **/ object Data { val database = Path.explode("$ISABELLE_HOME_USER/build.db") def make_table(name: String, columns: List[SQL.Column], body: String = ""): SQL.Table = SQL.Table("isabelle_build" + if_proper(name, "_" + name), columns, body = body) object Generic { val uuid = SQL.Column.string("uuid") val name = SQL.Column.string("name") def sql_equal(uuid: String = "", name: String = ""): SQL.Source = SQL.and( if_proper(uuid, Generic.uuid.equal(uuid)), if_proper(name, Generic.name.equal(name))) def sql_member(uuid: String = "", names: Iterable[String] = Nil): SQL.Source = SQL.and( if_proper(uuid, Generic.uuid.equal(uuid)), if_proper(names, Generic.name.member(names))) } object Base { val uuid = Generic.uuid.make_primary_key val ml_platform = SQL.Column.string("ml_platform") val options = SQL.Column.string("options") val table = make_table("", List(uuid, ml_platform, options)) } object Serial { val serial = SQL.Column.long("serial") val table = make_table("serial", List(serial)) } + object Sessions { + val name = Generic.name.make_primary_key + val deps = SQL.Column.string("deps") + val ancestors = SQL.Column.string("ancestors") + val sources = SQL.Column.string("sources") + val timeout = SQL.Column.long("timeout") + val old_time = SQL.Column.long("old_time") + val old_command_timings = SQL.Column.bytes("old_command_timings") + val uuid = Generic.uuid + + val table = make_table("sessions", + List(name, deps, ancestors, sources, timeout, old_time, old_command_timings, uuid)) + } + object Pending { val name = Generic.name.make_primary_key val deps = SQL.Column.string("deps") val info = SQL.Column.string("info") val table = make_table("pending", List(name, deps, info)) } object Running { val name = Generic.name.make_primary_key val hostname = SQL.Column.string("hostname") val numa_node = SQL.Column.int("numa_node") val table = make_table("running", List(name, hostname, numa_node)) } object Results { val name = Generic.name.make_primary_key val hostname = SQL.Column.string("hostname") val numa_node = SQL.Column.string("numa_node") val rc = SQL.Column.int("rc") val out = SQL.Column.string("out") val err = SQL.Column.string("err") val timing_elapsed = SQL.Column.long("timing_elapsed") val timing_cpu = SQL.Column.long("timing_cpu") val timing_gc = SQL.Column.long("timing_gc") val table = make_table("results", List(name, hostname, numa_node, rc, out, err, timing_elapsed, timing_cpu, timing_gc)) } def get_serial(db: SQL.Database): Long = db.using_statement(Serial.table.select())(stmt => stmt.execute_query().iterator(_.long(Serial.serial)).nextOption.getOrElse(0L)) def set_serial(db: SQL.Database, serial: Long): Unit = if (get_serial(db) != serial) { db.using_statement(Serial.table.delete())(_.execute()) db.using_statement(Serial.table.insert()) { stmt => stmt.long(1) = serial stmt.execute() } } + def read_sessions_domain(db: SQL.Database): Set[String] = + db.using_statement(Sessions.table.select(List(Sessions.name)))(stmt => + Set.from(stmt.execute_query().iterator(_.string(Sessions.name)))) + + def read_sessions(db: SQL.Database, names: Iterable[String] = Nil): State.Sessions = + db.using_statement( + Sessions.table.select(sql = if_proper(names, Sessions.name.where_member(names))) + ) { stmt => + Map.from(stmt.execute_query().iterator { res => + val name = res.string(Sessions.name) + val deps = split_lines(res.string(Sessions.deps)) + val ancestors = split_lines(res.string(Sessions.ancestors)) + val sources_shasum = SHA1.fake_shasum(res.string(Sessions.sources)) + val timeout = Time.ms(res.long(Sessions.timeout)) + val old_time = Time.ms(res.long(Sessions.old_time)) + val old_command_timings_blob = res.bytes(Sessions.old_command_timings) + val uuid = res.string(Sessions.uuid) + name -> Build_Job.Session_Context(name, deps, ancestors, sources_shasum, + timeout, old_time, old_command_timings_blob, uuid) + }) + } + + def update_sessions(db:SQL.Database, sessions: State.Sessions): Boolean = { + val old_sessions = read_sessions_domain(db) + val insert = sessions.iterator.filterNot(p => old_sessions.contains(p._1)).toList + + for ((name, session) <- insert) { + db.using_statement(Sessions.table.insert()) { stmt => + stmt.string(1) = name + stmt.string(2) = cat_lines(session.deps) + stmt.string(3) = cat_lines(session.ancestors) + stmt.string(4) = session.sources_shasum.toString + stmt.long(5) = session.timeout.ms + stmt.long(6) = session.old_time.ms + stmt.bytes(7) = session.old_command_timings_blob + stmt.string(8) = session.uuid + stmt.execute() + } + } + + insert.nonEmpty + } + def read_pending(db: SQL.Database): List[Entry] = db.using_statement(Pending.table.select(sql = SQL.order_by(List(Pending.name)))) { stmt => List.from( stmt.execute_query().iterator { res => val name = res.string(Pending.name) val deps = res.string(Pending.deps) val info = res.string(Pending.info) Entry(name, split_lines(deps), info = JSON.Object.parse(info)) }) } - def update_pending(db: SQL.Database, pending: List[Entry]): Boolean = { + def update_pending(db: SQL.Database, pending: State.Pending): Boolean = { val old_pending = read_pending(db) val (delete, insert) = Library.symmetric_difference(old_pending, pending) if (delete.nonEmpty) { db.using_statement( Pending.table.delete( sql = SQL.where(Generic.sql_member(names = delete.map(_.name)))))(_.execute()) } for (entry <- insert) { db.using_statement(Pending.table.insert()) { stmt => stmt.string(1) = entry.name stmt.string(2) = cat_lines(entry.deps) stmt.string(3) = JSON.Format(entry.info) stmt.execute() } } delete.nonEmpty || insert.nonEmpty } def read_running(db: SQL.Database): List[Build_Job.Abstract] = db.using_statement(Running.table.select(sql = SQL.order_by(List(Running.name)))) { stmt => List.from( stmt.execute_query().iterator { res => val name = res.string(Running.name) val hostname = res.string(Running.hostname) val numa_node = res.get_int(Running.numa_node) Build_Job.Abstract(name, Host.Node_Info(hostname, numa_node)) }) } - def update_running(db: SQL.Database, running: Map[String, Build_Job]): Boolean = { + def update_running(db: SQL.Database, running: State.Running): Boolean = { val old_running = read_running(db) val abs_running = running.valuesIterator.map(_.make_abstract).toList val (delete, insert) = Library.symmetric_difference(old_running, abs_running) if (delete.nonEmpty) { db.using_statement( Running.table.delete( sql = SQL.where(Generic.sql_member(names = delete.map(_.job_name)))))(_.execute()) } for (job <- insert) { db.using_statement(Running.table.insert()) { stmt => stmt.string(1) = job.job_name stmt.string(2) = job.node_info.hostname stmt.int(3) = job.node_info.numa_node stmt.execute() } } delete.nonEmpty || insert.nonEmpty } + def read_results_domain(db: SQL.Database): Set[String] = + db.using_statement(Results.table.select(List(Results.name)))(stmt => + Set.from(stmt.execute_query().iterator(_.string(Results.name)))) + def read_results(db: SQL.Database, names: List[String] = Nil): Map[String, Build_Job.Result] = db.using_statement( - Results.table.select(sql = if_proper(names, Results.name.where_member(names)))) { stmt => - Map.from( - stmt.execute_query().iterator { res => + Results.table.select(sql = if_proper(names, Results.name.where_member(names))) + ) { stmt => + Map.from(stmt.execute_query().iterator { res => val name = res.string(Results.name) val hostname = res.string(Results.hostname) val numa_node = res.get_int(Results.numa_node) val rc = res.int(Results.rc) val out = res.string(Results.out) val err = res.string(Results.err) - val timing_elapsed = res.long(Results.timing_elapsed) - val timing_cpu = res.long(Results.timing_cpu) - val timing_gc = res.long(Results.timing_gc) + val timing = + res.timing( + Results.timing_elapsed, + Results.timing_cpu, + Results.timing_gc) val node_info = Host.Node_Info(hostname, numa_node) val process_result = Process_Result(rc, out_lines = split_lines(out), err_lines = split_lines(err), - timing = Timing(Time.ms(timing_elapsed), Time.ms(timing_cpu), Time.ms(timing_gc))) + timing = timing) name -> Build_Job.Result(node_info, process_result) }) - } + } - def read_results_name(db: SQL.Database): Set[String] = - db.using_statement(Results.table.select(List(Results.name)))(stmt => - Set.from(stmt.execute_query().iterator(_.string(Results.name)))) - - def update_results(db: SQL.Database, results: Map[String, Build_Process.Result]): Boolean = { - val old_results = read_results_name(db) + def update_results(db: SQL.Database, results: State.Results): Boolean = { + val old_results = read_results_domain(db) val insert = results.iterator.filterNot(p => old_results.contains(p._1)).toList for ((name, result) <- insert) { val node_info = result.node_info val process_result = result.process_result db.using_statement(Results.table.insert()) { stmt => stmt.string(1) = name stmt.string(2) = node_info.hostname stmt.int(3) = node_info.numa_node stmt.int(4) = process_result.rc stmt.string(5) = cat_lines(process_result.out_lines) stmt.string(6) = cat_lines(process_result.err_lines) stmt.long(7) = process_result.timing.elapsed.ms stmt.long(8) = process_result.timing.cpu.ms stmt.long(9) = process_result.timing.gc.ms stmt.execute() } } insert.nonEmpty } def init_database(db: SQL.Database, build_context: Build_Process.Context): Unit = { val tables = List( Base.table, Serial.table, + Sessions.table, Pending.table, Running.table, Results.table, Host.Data.Node_Info.table) for (table <- tables) db.create_table(table) val old_pending = Data.read_pending(db) if (old_pending.nonEmpty) { error("Cannot init build process, because of unfinished " + commas_quote(old_pending.map(_.name))) } for (table <- tables) db.using_statement(table.delete())(_.execute()) db.using_statement(Base.table.insert()) { stmt => stmt.string(1) = build_context.uuid stmt.string(2) = Isabelle_System.getenv("ML_PLATFORM") stmt.string(3) = build_context.store.options.make_prefs(Options.init(prefs = "")) stmt.execute() } } def update_database( db: SQL.Database, uuid: String, hostname: String, state: State ): State = { val changed = List( + update_sessions(db, state.sessions), update_pending(db, state.pending), update_running(db, state.running), update_results(db, state.results), Host.Data.update_numa_index(db, hostname, state.numa_index)) val serial0 = get_serial(db) val serial = if (changed.exists(identity)) serial0 + 1 else serial0 set_serial(db, serial) state.copy(serial = serial) } } } /** main process **/ class Build_Process(protected val build_context: Build_Process.Context) extends AutoCloseable { /* context */ protected val store: Sessions.Store = build_context.store protected val build_options: Options = store.options protected val build_deps: Sessions.Deps = build_context.build_deps protected val progress: Progress = build_context.progress protected val verbose: Boolean = build_context.verbose /* global state: internal var vs. external database */ private var _state: Build_Process.State = init_state(Build_Process.State()) private val _database: Option[SQL.Database] = if (!build_options.bool("build_database_test")) None else if (store.database_server) Some(store.open_database_server()) else { val db = SQLite.open_database(Build_Process.Data.database) try { Isabelle_System.chmod("600", Build_Process.Data.database) } catch { case exn: Throwable => db.close(); throw exn } Some(db) } def close(): Unit = synchronized { _database.foreach(_.close()) } private def setup_database(): Unit = synchronized { for (db <- _database) { db.transaction { Build_Process.Data.init_database(db, build_context) } db.rebuild() } } protected def synchronized_database[A](body: => A): A = synchronized { _database match { case None => body case Some(db) => db.transaction { body } } } private def sync_database(): Unit = synchronized_database { for (db <- _database) { _state = Build_Process.Data.update_database( db, build_context.uuid, build_context.hostname, _state) } } /* policy operations */ protected def init_state(state: Build_Process.State): Build_Process.State = { + val sessions1 = + build_context.sessions.foldLeft(state.sessions) { case (map, (name, session)) => + if (state.sessions.isDefinedAt(name)) map + else map + (name -> session) + } + val old_pending = state.pending.iterator.map(_.name).toSet val new_pending = List.from( for { (name, session_context) <- build_context.sessions.iterator if !old_pending(name) } yield Build_Process.Entry(name, session_context.deps)) - state.copy(pending = new_pending ::: state.pending) + val pending1 = new_pending ::: state.pending + + state.copy(sessions = sessions1, pending = pending1) } protected def next_job(state: Build_Process.State): Option[String] = if (state.running.size < (build_context.max_jobs max 1)) { state.pending.filter(entry => entry.is_ready && !state.is_running(entry.name)) .sortBy(_.name)(build_context.ordering) .headOption.map(_.name) } else None protected def start_session(state: Build_Process.State, session_name: String): Build_Process.State = { val ancestor_results = for (a <- build_context.sessions(session_name).ancestors) yield state.results(a) val input_shasum = if (ancestor_results.isEmpty) { SHA1.shasum_meta_info(SHA1.digest(Path.explode("$POLYML_EXE"))) } else SHA1.flat_shasum(ancestor_results.map(_.output_shasum)) val store_heap = build_context.store_heap(session_name) val (current, output_shasum) = store.check_output(session_name, sources_shasum = build_context.sources_shasum(session_name), input_shasum = input_shasum, fresh_build = build_context.fresh_build, store_heap = store_heap) val all_current = current && ancestor_results.forall(_.current) if (all_current) { state .remove_pending(session_name) .make_result(session_name, Process_Result.ok, output_shasum, current = true) } else if (build_context.no_build) { progress.echo_if(verbose, "Skipping " + session_name + " ...") state. remove_pending(session_name). make_result(session_name, Process_Result.error, output_shasum) } else if (!ancestor_results.forall(_.ok) || progress.stopped) { progress.echo(session_name + " CANCELLED") state .remove_pending(session_name) .make_result(session_name, Process_Result.undefined, output_shasum) } else { progress.echo((if (store_heap) "Building " else "Running ") + session_name + " ...") store.init_output(session_name) val (numa_node, state1) = state.numa_next(build_context.numa_nodes) val node_info = Host.Node_Info(build_context.hostname, numa_node) val job = Build_Job.start_session( build_context, build_deps.background(session_name), input_shasum, node_info) state1.add_running(session_name, job) } } /* run */ def run(): Map[String, Process_Result] = { def finished(): Boolean = synchronized_database { _state.finished } def sleep(): Unit = Isabelle_Thread.interrupt_handler(_ => progress.stop()) { build_options.seconds("editor_input_delay").sleep() } def start(): Boolean = synchronized_database { next_job(_state) match { case Some(name) => if (Build_Job.is_session_name(name)) { _state = start_session(_state, name) true } else error("Unsupported build job name " + quote(name)) case None => false } } if (finished()) { progress.echo_warning("Nothing to build") Map.empty[String, Process_Result] } else { setup_database() while (!finished()) { if (progress.stopped) synchronized_database { _state.stop_running() } for (job <- synchronized_database { _state.finished_running() }) { val job_name = job.job_name val (process_result, output_shasum) = job.join synchronized_database { _state = _state. remove_pending(job_name). remove_running(job_name). make_result(job_name, process_result, output_shasum, node_info = job.node_info) } } if (!start()) { sync_database() sleep() } } synchronized_database { for ((name, result) <- _state.results) yield name -> result.process_result } } } }