src/Pure/System/isabelle_process.scala
author wenzelm
Sun Aug 08 19:36:31 2010 +0200 (2010-08-08 ago)
changeset 38236 d8c7be27e01d
parent 38231 968844caaff9
child 38253 3d4e521014f7
permissions -rw-r--r--
explicitly distinguish Output.status (essential feedback) vs. Output.report (useful markup);
     1 /*  Title:      Pure/System/isabelle_process.ML
     2     Author:     Makarius
     3     Options:    :folding=explicit:collapseFolds=1:
     4 
     5 Isabelle process management -- always reactive due to multi-threaded I/O.
     6 */
     7 
     8 package isabelle
     9 
    10 import java.util.concurrent.LinkedBlockingQueue
    11 import java.io.{BufferedReader, BufferedWriter, InputStreamReader, OutputStreamWriter,
    12   InputStream, OutputStream, IOException}
    13 
    14 import scala.actors.Actor
    15 import Actor._
    16 
    17 
    18 object Isabelle_Process
    19 {
    20   /* results */
    21 
    22   object Kind {
    23     // message markup
    24     val markup = Map(
    25       ('A' : Int) -> Markup.INIT,
    26       ('B' : Int) -> Markup.STATUS,
    27       ('C' : Int) -> Markup.REPORT,
    28       ('D' : Int) -> Markup.WRITELN,
    29       ('E' : Int) -> Markup.TRACING,
    30       ('F' : Int) -> Markup.WARNING,
    31       ('G' : Int) -> Markup.ERROR,
    32       ('H' : Int) -> Markup.DEBUG)
    33     def is_raw(kind: String) =
    34       kind == Markup.STDOUT
    35     def is_control(kind: String) =
    36       kind == Markup.SYSTEM ||
    37       kind == Markup.SIGNAL ||
    38       kind == Markup.EXIT
    39     def is_system(kind: String) =
    40       kind == Markup.SYSTEM ||
    41       kind == Markup.STDIN ||
    42       kind == Markup.SIGNAL ||
    43       kind == Markup.EXIT ||
    44       kind == Markup.STATUS
    45   }
    46 
    47   class Result(val message: XML.Elem)
    48   {
    49     def kind = message.markup.name
    50     def properties = message.markup.properties
    51     def body = message.body
    52 
    53     def is_raw = Kind.is_raw(kind)
    54     def is_control = Kind.is_control(kind)
    55     def is_system = Kind.is_system(kind)
    56     def is_status = kind == Markup.STATUS
    57     def is_report = kind == Markup.REPORT
    58     def is_ready = is_status && body == List(XML.Elem(Markup.Ready, Nil))
    59 
    60     override def toString: String =
    61     {
    62       val res =
    63         if (is_status || is_report) message.body.map(_.toString).mkString
    64         else Pretty.string_of(message.body)
    65       if (properties.isEmpty)
    66         kind.toString + " [[" + res + "]]"
    67       else
    68         kind.toString + " " +
    69           (for ((x, y) <- properties) yield x + "=" + y).mkString("{", ",", "}") + " [[" + res + "]]"
    70     }
    71 
    72     def cache(c: XML.Cache): Result = new Result(c.cache_tree(message).asInstanceOf[XML.Elem])
    73   }
    74 }
    75 
    76 
    77 class Isabelle_Process(system: Isabelle_System, receiver: Actor, args: String*)
    78 {
    79   import Isabelle_Process._
    80 
    81 
    82   /* demo constructor */
    83 
    84   def this(args: String*) =
    85     this(new Isabelle_System,
    86       actor { loop { react { case res => Console.println(res) } } }, args: _*)
    87 
    88 
    89   /* process information */
    90 
    91   @volatile private var proc: Process = null
    92   @volatile private var closing = false
    93   @volatile private var pid: String = null
    94 
    95 
    96   /* results */
    97 
    98   private def put_result(kind: String, props: List[(String, String)], body: List[XML.Tree])
    99   {
   100     if (kind == Markup.INIT) {
   101       for ((Markup.PID, p) <- props) pid = p
   102     }
   103     receiver ! new Result(XML.Elem(Markup(kind, props), body))
   104   }
   105 
   106   private def put_result(kind: String, text: String)
   107   {
   108     put_result(kind, Nil, List(XML.Text(system.symbols.decode(text))))
   109   }
   110 
   111 
   112   /* signals */
   113 
   114   def interrupt() = synchronized {
   115     if (proc == null) error("Cannot interrupt Isabelle: no process")
   116     if (pid == null) put_result(Markup.SYSTEM, "Cannot interrupt: unknown pid")
   117     else {
   118       try {
   119         if (system.execute(true, "kill", "-INT", pid).waitFor == 0)
   120           put_result(Markup.SIGNAL, "INT")
   121         else
   122           put_result(Markup.SYSTEM, "Cannot interrupt: kill command failed")
   123       }
   124       catch { case e: IOException => error("Cannot interrupt Isabelle: " + e.getMessage) }
   125     }
   126   }
   127 
   128   def kill() = synchronized {
   129     if (proc == 0) error("Cannot kill Isabelle: no process")
   130     else {
   131       try_close()
   132       Thread.sleep(500)  // FIXME property!?
   133       put_result(Markup.SIGNAL, "KILL")
   134       proc.destroy
   135       proc = null
   136       pid = null
   137     }
   138   }
   139 
   140 
   141   /* output being piped into the process */
   142 
   143   private val output = new LinkedBlockingQueue[String]
   144 
   145   private def output_raw(text: String) = synchronized {
   146     if (proc == null) error("Cannot output to Isabelle: no process")
   147     if (closing) error("Cannot output to Isabelle: already closing")
   148     output.put(text)
   149   }
   150 
   151   def output_sync(text: String) =
   152     output_raw(" \\<^sync>\n; " + text + " \\<^sync>;\n")
   153 
   154 
   155   def command(text: String) =
   156     output_sync("Isabelle.command " + Isabelle_Syntax.encode_string(text))
   157 
   158   def command(props: List[(String, String)], text: String) =
   159     output_sync("Isabelle.command " + Isabelle_Syntax.encode_properties(props) + " " +
   160       Isabelle_Syntax.encode_string(text))
   161 
   162   def ML_val(text: String) =
   163     output_sync("ML_val " + Isabelle_Syntax.encode_string(text))
   164 
   165   def ML_command(text: String) =
   166     output_sync("ML_command " + Isabelle_Syntax.encode_string(text))
   167 
   168   def close() = synchronized {    // FIXME watchdog/timeout
   169     output_raw("\u0000")
   170     closing = true
   171   }
   172 
   173   def try_close() = synchronized {
   174     if (proc != null && !closing) {
   175       try { close() }
   176       catch { case _: RuntimeException => }
   177     }
   178   }
   179 
   180 
   181   /* stdin */
   182 
   183   private class Stdin_Thread(out_stream: OutputStream) extends Thread("isabelle: stdin") {
   184     override def run() = {
   185       val writer = new BufferedWriter(new OutputStreamWriter(out_stream, Standard_System.charset))
   186       var finished = false
   187       while (!finished) {
   188         try {
   189           //{{{
   190           val s = output.take
   191           if (s == "\u0000") {
   192             writer.close
   193             finished = true
   194           }
   195           else {
   196             put_result(Markup.STDIN, s)
   197             writer.write(s)
   198             writer.flush
   199           }
   200           //}}}
   201         }
   202         catch {
   203           case e: IOException => put_result(Markup.SYSTEM, "Stdin thread: " + e.getMessage)
   204         }
   205       }
   206       put_result(Markup.SYSTEM, "Stdin thread terminated")
   207     }
   208   }
   209 
   210 
   211   /* stdout */
   212 
   213   private class Stdout_Thread(in_stream: InputStream) extends Thread("isabelle: stdout") {
   214     override def run() = {
   215       val reader = new BufferedReader(new InputStreamReader(in_stream, Standard_System.charset))
   216       var result = new StringBuilder(100)
   217 
   218       var finished = false
   219       while (!finished) {
   220         try {
   221           //{{{
   222           var c = -1
   223           var done = false
   224           while (!done && (result.length == 0 || reader.ready)) {
   225             c = reader.read
   226             if (c >= 0) result.append(c.asInstanceOf[Char])
   227             else done = true
   228           }
   229           if (result.length > 0) {
   230             put_result(Markup.STDOUT, result.toString)
   231             result.length = 0
   232           }
   233           else {
   234             reader.close
   235             finished = true
   236             try_close()
   237           }
   238           //}}}
   239         }
   240         catch {
   241           case e: IOException => put_result(Markup.SYSTEM, "Stdout thread: " + e.getMessage)
   242         }
   243       }
   244       put_result(Markup.SYSTEM, "Stdout thread terminated")
   245     }
   246   }
   247 
   248 
   249   /* messages */
   250 
   251   private class Message_Thread(fifo: String) extends Thread("isabelle: messages")
   252   {
   253     private class Protocol_Error(msg: String) extends Exception(msg)
   254     override def run()
   255     {
   256       val stream = system.fifo_stream(fifo)
   257       val default_buffer = new Array[Byte](65536)
   258       var c = -1
   259 
   260       def read_chunk(): List[XML.Tree] =
   261       {
   262         //{{{
   263         // chunk size
   264         var n = 0
   265         c = stream.read
   266         while (48 <= c && c <= 57) {
   267           n = 10 * n + (c - 48)
   268           c = stream.read
   269         }
   270         if (c != 10) throw new Protocol_Error("bad message chunk header")
   271 
   272         // chunk content
   273         val buf =
   274           if (n <= default_buffer.size) default_buffer
   275           else new Array[Byte](n)
   276 
   277         var i = 0
   278         var m = 0
   279         do {
   280           m = stream.read(buf, i, n - i)
   281           i += m
   282         } while (m > 0 && n > i)
   283 
   284         if (i != n) throw new Protocol_Error("bad message chunk content")
   285 
   286         YXML.parse_body_failsafe(YXML.decode_chars(system.symbols.decode, buf, 0, n))
   287         //}}}
   288       }
   289 
   290       do {
   291         try {
   292           //{{{
   293           c = stream.read
   294           var non_sync = 0
   295           while (c >= 0 && c != 2) {
   296             non_sync += 1
   297             c = stream.read
   298           }
   299           if (non_sync > 0)
   300             throw new Protocol_Error("lost synchronization -- skipping " + non_sync + " bytes")
   301           if (c == 2) {
   302             val header = read_chunk()
   303             val body = read_chunk()
   304             header match {
   305               case List(XML.Elem(Markup(name, props), Nil))
   306                   if name.size == 1 && Kind.markup.isDefinedAt(name(0)) =>
   307                 put_result(Kind.markup(name(0)), props, body)
   308               case _ => throw new Protocol_Error("bad header: " + header.toString)
   309             }
   310           }
   311           //}}}
   312         }
   313         catch {
   314           case e: IOException =>
   315             put_result(Markup.SYSTEM, "Cannot read message:\n" + e.getMessage)
   316           case e: Protocol_Error =>
   317             put_result(Markup.SYSTEM, "Malformed message:\n" + e.getMessage)
   318         }
   319       } while (c != -1)
   320       stream.close
   321       try_close()
   322 
   323       put_result(Markup.SYSTEM, "Message thread terminated")
   324     }
   325   }
   326 
   327 
   328 
   329   /** main **/
   330 
   331   {
   332     /* messages */
   333 
   334     val message_fifo = system.mk_fifo()
   335     def rm_fifo() = system.rm_fifo(message_fifo)
   336 
   337     val message_thread = new Message_Thread(message_fifo)
   338     message_thread.start
   339 
   340 
   341     /* exec process */
   342 
   343     try {
   344       val cmdline = List(system.getenv_strict("ISABELLE_PROCESS"), "-W", message_fifo) ++ args
   345       proc = system.execute(true, cmdline: _*)
   346     }
   347     catch {
   348       case e: IOException =>
   349         rm_fifo()
   350         error("Failed to execute Isabelle process: " + e.getMessage)
   351     }
   352 
   353 
   354     /* stdin/stdout */
   355 
   356     new Stdin_Thread(proc.getOutputStream).start
   357     new Stdout_Thread(proc.getInputStream).start
   358 
   359 
   360     /* exit */
   361 
   362     new Thread("isabelle: exit") {
   363       override def run() = {
   364         val rc = proc.waitFor()
   365         Thread.sleep(300)  // FIXME property!?
   366         put_result(Markup.SYSTEM, "Exit thread terminated")
   367         put_result(Markup.EXIT, rc.toString)
   368         rm_fifo()
   369       }
   370     }.start
   371   }
   372 }