src/Pure/System/isabelle_process.scala
author wenzelm
Sat Sep 18 21:33:56 2010 +0200 (2010-09-18 ago)
changeset 39524 59ebce09ce6e
parent 39523 d8971680b0fc
child 39525 72e949a0425b
permissions -rw-r--r--
recovered basic session stop/restart;
     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 case class Input_Text(text: String)
   151   private case class Input_Chunks(chunks: List[Array[Byte]])
   152   private case object Close
   153 
   154 
   155   /* raw stdin */
   156 
   157   private def stdin_actor(name: String, stream: OutputStream): Actor =
   158     Simple_Thread.actor(name) {
   159       val writer = new BufferedWriter(new OutputStreamWriter(stream, Standard_System.charset))
   160       var finished = false
   161       while (!finished) {
   162         try {
   163           //{{{
   164           receive {
   165             case Input_Text(text) =>
   166               writer.write(text)
   167               writer.flush
   168             case Close =>
   169               writer.close
   170               finished = true
   171             case bad => System.err.println(name + ": ignoring bad message " + bad)
   172           }
   173           //}}}
   174         }
   175         catch {
   176           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   177         }
   178       }
   179       put_result(Markup.SYSTEM, name + " terminated")
   180     }
   181 
   182 
   183   /* raw stdout */
   184 
   185   private def stdout_actor(name: String, stream: InputStream): Actor =
   186     Simple_Thread.actor(name) {
   187       val reader = new BufferedReader(new InputStreamReader(stream, Standard_System.charset))
   188       var result = new StringBuilder(100)
   189 
   190       var finished = false
   191       while (!finished) {
   192         try {
   193           //{{{
   194           var c = -1
   195           var done = false
   196           while (!done && (result.length == 0 || reader.ready)) {
   197             c = reader.read
   198             if (c >= 0) result.append(c.asInstanceOf[Char])
   199             else done = true
   200           }
   201           if (result.length > 0) {
   202             put_result(Markup.STDOUT, result.toString)
   203             result.length = 0
   204           }
   205           else {
   206             reader.close
   207             finished = true
   208             close()
   209           }
   210           //}}}
   211         }
   212         catch {
   213           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   214         }
   215       }
   216       put_result(Markup.SYSTEM, name + " terminated")
   217     }
   218 
   219 
   220   /* command input */
   221 
   222   private def input_actor(name: String, fifo: String): Actor =
   223     Simple_Thread.actor(name) {
   224       val stream = new BufferedOutputStream(system.fifo_output_stream(fifo))  // FIXME potentially blocking forever
   225       var finished = false
   226       while (!finished) {
   227         try {
   228           //{{{
   229           receive {
   230             case Input_Chunks(chunks) =>
   231               stream.write(Standard_System.string_bytes(
   232                   chunks.map(_.length).mkString("", ",", "\n")));
   233               chunks.foreach(stream.write(_));
   234               stream.flush
   235             case Close =>
   236               stream.close
   237               finished = true
   238             case bad => System.err.println(name + ": ignoring bad message " + bad)
   239           }
   240           //}}}
   241         }
   242         catch {
   243           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   244         }
   245       }
   246       put_result(Markup.SYSTEM, name + " terminated")
   247     }
   248 
   249 
   250   /* message output */
   251 
   252   private class Protocol_Error(msg: String) extends Exception(msg)
   253 
   254   private def message_actor(name: String, fifo: String): Actor =
   255     Simple_Thread.actor(name) {
   256       val stream = system.fifo_input_stream(fifo)  // FIXME potentially blocking forever
   257       val default_buffer = new Array[Byte](65536)
   258       var c = -1
   259 
   260       def read_chunk(): XML.Body =
   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           val header = read_chunk()
   293           val body = read_chunk()
   294           header match {
   295             case List(XML.Elem(Markup(name, props), Nil))
   296                 if name.size == 1 && Kind.markup.isDefinedAt(name(0)) =>
   297               put_result(Kind.markup(name(0)), props, body)
   298             case _ => throw new Protocol_Error("bad header: " + header.toString)
   299           }
   300         }
   301         catch {
   302           case e: IOException =>
   303             put_result(Markup.SYSTEM, "Cannot read message:\n" + e.getMessage)
   304           case e: Protocol_Error =>
   305             put_result(Markup.SYSTEM, "Malformed message:\n" + e.getMessage)
   306         }
   307       } while (c != -1)
   308       stream.close
   309       close()
   310 
   311       put_result(Markup.SYSTEM, name + " terminated")
   312     }
   313 
   314 
   315 
   316   /** init **/
   317 
   318   /* exec process */
   319 
   320   private val in_fifo = system.mk_fifo()
   321   private val out_fifo = system.mk_fifo()
   322   private def rm_fifos() { system.rm_fifo(in_fifo); system.rm_fifo(out_fifo) }
   323 
   324   try {
   325     val cmdline =
   326       List(system.getenv_strict("ISABELLE_PROCESS"), "-W", in_fifo + ":" + out_fifo) ++ args
   327     proc = Some(system.execute(true, cmdline: _*))
   328   }
   329   catch { case e: IOException => rm_fifos(); throw(e) }
   330 
   331 
   332   /* I/O actors */
   333 
   334   private val command_input = input_actor("command_input", in_fifo)
   335   message_actor("message_output", out_fifo)
   336 
   337   private val standard_input = stdin_actor("standard_input", proc.get.getOutputStream)
   338   stdout_actor("standard_output", proc.get.getInputStream)
   339 
   340 
   341   /* exit process */
   342 
   343   Simple_Thread.actor("process_exit") {
   344     proc match {
   345       case None =>
   346       case Some(p) =>
   347         val rc = p.waitFor()
   348         Thread.sleep(300)  // FIXME property!?
   349         put_result(Markup.SYSTEM, "Isabelle process terminated")
   350         put_result(Markup.EXIT, rc.toString)
   351     }
   352     rm_fifos()
   353   }
   354 
   355 
   356 
   357   /** main methods **/
   358 
   359   def input_raw(text: String): Unit = standard_input ! Input_Text(text)
   360 
   361   def input_bytes(name: String, args: Array[Byte]*): Unit =
   362     command_input ! Input_Chunks(Standard_System.string_bytes(name) :: args.toList)
   363 
   364   def input(name: String, args: String*): Unit =
   365     input_bytes(name, args.map(Standard_System.string_bytes): _*)
   366 
   367   def close(): Unit = { standard_input ! Close; command_input ! Close }
   368 }