src/Pure/System/isabelle_process.scala
author wenzelm
Sat Sep 18 17:14:47 2010 +0200 (2010-09-18 ago)
changeset 39519 b376f53bcc18
parent 39513 fce2202892c4
child 39523 d8971680b0fc
permissions -rw-r--r--
slightly more robust Isabelle_Process startup -- NB: openening fifo streams synchronizes with other end, which may fail to reach that point;
     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, BufferedOutputStream, 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     def is_raw(kind: String) =
    33       kind == Markup.STDOUT
    34     def is_control(kind: String) =
    35       kind == Markup.SYSTEM ||
    36       kind == Markup.SIGNAL ||
    37       kind == Markup.EXIT
    38     def is_system(kind: String) =
    39       kind == Markup.SYSTEM ||
    40       kind == Markup.INPUT ||
    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 }
    73 
    74 
    75 class Isabelle_Process(system: Isabelle_System, receiver: Actor, args: String*)
    76 {
    77   import Isabelle_Process._
    78 
    79 
    80   /* demo constructor */
    81 
    82   def this(args: String*) =
    83     this(new Isabelle_System,
    84       actor { loop { react { case res => Console.println(res) } } }, args: _*)
    85 
    86 
    87   /* process information */
    88 
    89   @volatile private var proc: Option[Process] = None
    90   @volatile private var pid: Option[String] = None
    91 
    92 
    93   /* results */
    94 
    95   private val xml_cache = new XML.Cache(131071)
    96 
    97   private def put_result(kind: String, props: List[(String, String)], body: XML.Body)
    98   {
    99     if (pid.isEmpty && kind == Markup.INIT)
   100       pid = props.find(_._1 == Markup.PID).map(_._1)
   101 
   102     val msg = XML.Elem(Markup(kind, props), Isar_Document.clean_message(body))
   103     xml_cache.cache_tree(msg)((message: XML.Tree) =>
   104       receiver ! new Result(message.asInstanceOf[XML.Elem]))
   105   }
   106 
   107   private def put_result(kind: String, text: String)
   108   {
   109     put_result(kind, Nil, List(XML.Text(system.symbols.decode(text))))
   110   }
   111 
   112 
   113   /* signals */
   114 
   115   def interrupt()
   116   {
   117     if (proc.isEmpty) put_result(Markup.SYSTEM, "Cannot interrupt Isabelle: no process")
   118     else
   119       pid match {
   120         case None => put_result(Markup.SYSTEM, "Cannot interrupt Isabelle: unknowd pid")
   121         case Some(i) =>
   122           try {
   123             if (system.execute(true, "kill", "-INT", i).waitFor == 0)
   124               put_result(Markup.SIGNAL, "INT")
   125             else
   126               put_result(Markup.SYSTEM, "Cannot interrupt Isabelle: kill command failed")
   127           }
   128           catch { case e: IOException => error("Cannot interrupt Isabelle: " + e.getMessage) }
   129       }
   130   }
   131 
   132   def kill()
   133   {
   134     proc match {
   135       case None => put_result(Markup.SYSTEM, "Cannot kill Isabelle: no process")
   136       case Some(p) =>
   137         close()
   138         Thread.sleep(500)  // FIXME !?
   139         put_result(Markup.SIGNAL, "KILL")
   140         p.destroy
   141         proc = None
   142         pid = None
   143     }
   144   }
   145 
   146 
   147 
   148   /** stream actors **/
   149 
   150   private val in_fifo = system.mk_fifo()
   151   private val out_fifo = system.mk_fifo()
   152   private def rm_fifos() = { system.rm_fifo(in_fifo); system.rm_fifo(out_fifo) }
   153 
   154   private case class Input_Text(text: String)
   155   private case class Input_Chunks(chunks: List[Array[Byte]])
   156   private case object Close
   157 
   158 
   159   /* raw stdin */
   160 
   161   private def stdin_actor(name: String, stream: OutputStream): Actor =
   162     Simple_Thread.actor(name) {
   163       val writer = new BufferedWriter(new OutputStreamWriter(stream, Standard_System.charset))
   164       var finished = false
   165       while (!finished) {
   166         try {
   167           //{{{
   168           receive {
   169             case Input_Text(text) =>
   170               // FIXME echo input?!
   171               writer.write(text)
   172               writer.flush
   173             case Close =>
   174               writer.close
   175               finished = true
   176             case bad => System.err.println(name + ": ignoring bad message " + bad)
   177           }
   178           //}}}
   179         }
   180         catch {
   181           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   182         }
   183       }
   184       put_result(Markup.SYSTEM, name + " terminated")
   185     }
   186 
   187 
   188   /* raw stdout */
   189 
   190   private def stdout_actor(name: String, stream: InputStream): Actor =
   191     Simple_Thread.actor(name) {
   192       val reader = new BufferedReader(new InputStreamReader(stream, Standard_System.charset))
   193       var result = new StringBuilder(100)
   194 
   195       var finished = false
   196       while (!finished) {
   197         try {
   198           //{{{
   199           var c = -1
   200           var done = false
   201           while (!done && (result.length == 0 || reader.ready)) {
   202             c = reader.read
   203             if (c >= 0) result.append(c.asInstanceOf[Char])
   204             else done = true
   205           }
   206           if (result.length > 0) {
   207             put_result(Markup.STDOUT, result.toString)
   208             result.length = 0
   209           }
   210           else {
   211             reader.close
   212             finished = true
   213             close()
   214           }
   215           //}}}
   216         }
   217         catch {
   218           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   219         }
   220       }
   221       put_result(Markup.SYSTEM, name + " terminated")
   222     }
   223 
   224 
   225   /* command input */
   226 
   227   private def input_actor(name: String): Actor =
   228     Simple_Thread.actor(name) {
   229       val stream = new BufferedOutputStream(system.fifo_output_stream(in_fifo))  // FIXME potentially blocking forever
   230       var finished = false
   231       while (!finished) {
   232         try {
   233           //{{{
   234           receive {
   235             case Input_Chunks(chunks) =>
   236               stream.write(Standard_System.string_bytes(
   237                   chunks.map(_.length).mkString("", ",", "\n")));
   238               chunks.foreach(stream.write(_));
   239               stream.flush
   240             case Close =>
   241               stream.close
   242               finished = true
   243             case bad => System.err.println(name + ": ignoring bad message " + bad)
   244           }
   245           //}}}
   246         }
   247         catch {
   248           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   249         }
   250       }
   251       put_result(Markup.SYSTEM, name + " terminated")
   252     }
   253 
   254 
   255   /* message output */
   256 
   257   private class Protocol_Error(msg: String) extends Exception(msg)
   258 
   259   private def message_actor(name: String): Actor =
   260     Simple_Thread.actor(name) {
   261       val stream = system.fifo_input_stream(out_fifo)  // FIXME potentially blocking forever
   262       val default_buffer = new Array[Byte](65536)
   263       var c = -1
   264 
   265       def read_chunk(): XML.Body =
   266       {
   267         //{{{
   268         // chunk size
   269         var n = 0
   270         c = stream.read
   271         while (48 <= c && c <= 57) {
   272           n = 10 * n + (c - 48)
   273           c = stream.read
   274         }
   275         if (c != 10) throw new Protocol_Error("bad message chunk header")
   276 
   277         // chunk content
   278         val buf =
   279           if (n <= default_buffer.size) default_buffer
   280           else new Array[Byte](n)
   281 
   282         var i = 0
   283         var m = 0
   284         do {
   285           m = stream.read(buf, i, n - i)
   286           i += m
   287         } while (m > 0 && n > i)
   288 
   289         if (i != n) throw new Protocol_Error("bad message chunk content")
   290 
   291         YXML.parse_body_failsafe(YXML.decode_chars(system.symbols.decode, buf, 0, n))
   292         //}}}
   293       }
   294 
   295       do {
   296         try {
   297           val header = read_chunk()
   298           val body = read_chunk()
   299           header match {
   300             case List(XML.Elem(Markup(name, props), Nil))
   301                 if name.size == 1 && Kind.markup.isDefinedAt(name(0)) =>
   302               put_result(Kind.markup(name(0)), props, body)
   303             case _ => throw new Protocol_Error("bad header: " + header.toString)
   304           }
   305         }
   306         catch {
   307           case e: IOException =>
   308             put_result(Markup.SYSTEM, "Cannot read message:\n" + e.getMessage)
   309           case e: Protocol_Error =>
   310             put_result(Markup.SYSTEM, "Malformed message:\n" + e.getMessage)
   311         }
   312       } while (c != -1)
   313       stream.close
   314       close()
   315 
   316       put_result(Markup.SYSTEM, name + " terminated")
   317     }
   318 
   319 
   320 
   321   /** init **/
   322 
   323   /* exec process */
   324 
   325   try {
   326     val cmdline =
   327       List(system.getenv_strict("ISABELLE_PROCESS"), "-W", in_fifo + ":" + out_fifo) ++ args
   328     proc = Some(system.execute(true, cmdline: _*))
   329   }
   330   catch {
   331     case e: IOException =>
   332       rm_fifos()
   333       error("Failed to execute Isabelle process: " + e.getMessage)
   334   }
   335 
   336 
   337   /* I/O actors */
   338 
   339   private val command_input = input_actor("command_input")
   340   message_actor("message_output")
   341 
   342   private val standard_input = stdin_actor("standard_input", proc.get.getOutputStream)
   343   stdout_actor("standard_output", proc.get.getInputStream)
   344 
   345 
   346   /* exit process */
   347 
   348   Simple_Thread.actor("process_exit") {
   349     proc match {
   350       case None =>
   351       case Some(p) =>
   352         val rc = p.waitFor()
   353         Thread.sleep(300)  // FIXME property!?
   354         put_result(Markup.SYSTEM, "process_exit terminated")
   355         put_result(Markup.EXIT, rc.toString)
   356     }
   357     rm_fifos()
   358   }
   359 
   360 
   361 
   362   /** main methods **/
   363 
   364   def input_raw(text: String): Unit = standard_input ! Input_Text(text)
   365 
   366   def input_bytes(name: String, args: Array[Byte]*): Unit =
   367     command_input ! Input_Chunks(Standard_System.string_bytes(name) :: args.toList)
   368 
   369   def input(name: String, args: String*): Unit =
   370     input_bytes(name, args.map(Standard_System.string_bytes): _*)
   371 
   372   def close(): Unit = command_input ! Close
   373 }