src/Pure/System/isabelle_process.scala
author wenzelm
Mon Aug 23 16:53:22 2010 +0200 (2010-08-23)
changeset 38639 f642faca303e
parent 38636 b7647ca7de5a
child 39439 1c294d150ded
permissions -rw-r--r--
main session actor as independent thread, to avoid starvation via regular worker pool;
tuned;
     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       ('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.INPUT ||
    42       kind == Markup.STDIN ||
    43       kind == Markup.SIGNAL ||
    44       kind == Markup.EXIT ||
    45       kind == Markup.STATUS
    46   }
    47 
    48   class Result(val message: XML.Elem)
    49   {
    50     def kind = message.markup.name
    51     def properties = message.markup.properties
    52     def body = message.body
    53 
    54     def is_raw = Kind.is_raw(kind)
    55     def is_control = Kind.is_control(kind)
    56     def is_system = Kind.is_system(kind)
    57     def is_status = kind == Markup.STATUS
    58     def is_report = kind == Markup.REPORT
    59     def is_ready = is_status && body == List(XML.Elem(Markup.Ready, Nil))
    60 
    61     override def toString: String =
    62     {
    63       val res =
    64         if (is_status || is_report) message.body.map(_.toString).mkString
    65         else Pretty.string_of(message.body)
    66       if (properties.isEmpty)
    67         kind.toString + " [[" + res + "]]"
    68       else
    69         kind.toString + " " +
    70           (for ((x, y) <- properties) yield x + "=" + y).mkString("{", ",", "}") + " [[" + res + "]]"
    71     }
    72   }
    73 }
    74 
    75 
    76 class Isabelle_Process(system: Isabelle_System, receiver: Actor, args: String*)
    77 {
    78   import Isabelle_Process._
    79 
    80 
    81   /* demo constructor */
    82 
    83   def this(args: String*) =
    84     this(new Isabelle_System,
    85       actor { loop { react { case res => Console.println(res) } } }, args: _*)
    86 
    87 
    88   /* process information */
    89 
    90   @volatile private var proc: Option[Process] = None
    91   @volatile private var pid: Option[String] = None
    92 
    93 
    94   /* results */
    95 
    96   private val xml_cache = new XML.Cache(131071)
    97 
    98   private def put_result(kind: String, props: List[(String, String)], body: XML.Body)
    99   {
   100     if (pid.isEmpty && kind == Markup.INIT)
   101       pid = props.find(_._1 == Markup.PID).map(_._1)
   102 
   103     xml_cache.cache_tree(XML.Elem(Markup(kind, props), body))((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   case class Input_Text(text: String)
   151   case class Input_Chunks(chunks: List[Array[Byte]])
   152   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               // FIXME echo input?!
   167               writer.write(text)
   168               writer.flush
   169             case Close =>
   170               writer.close
   171               finished = true
   172             case bad => System.err.println(name + ": ignoring bad message " + bad)
   173           }
   174           //}}}
   175         }
   176         catch {
   177           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   178         }
   179       }
   180       put_result(Markup.SYSTEM, name + " terminated")
   181     }
   182 
   183 
   184   /* raw stdout */
   185 
   186   private def stdout_actor(name: String, stream: InputStream): Actor =
   187     Simple_Thread.actor(name) {
   188       val reader = new BufferedReader(new InputStreamReader(stream, Standard_System.charset))
   189       var result = new StringBuilder(100)
   190 
   191       var finished = false
   192       while (!finished) {
   193         try {
   194           //{{{
   195           var c = -1
   196           var done = false
   197           while (!done && (result.length == 0 || reader.ready)) {
   198             c = reader.read
   199             if (c >= 0) result.append(c.asInstanceOf[Char])
   200             else done = true
   201           }
   202           if (result.length > 0) {
   203             put_result(Markup.STDOUT, result.toString)
   204             result.length = 0
   205           }
   206           else {
   207             reader.close
   208             finished = true
   209             close()
   210           }
   211           //}}}
   212         }
   213         catch {
   214           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   215         }
   216       }
   217       put_result(Markup.SYSTEM, name + " terminated")
   218     }
   219 
   220 
   221   /* command input */
   222 
   223   private def input_actor(name: String, raw_stream: OutputStream): Actor =
   224     Simple_Thread.actor(name) {
   225       val stream = new BufferedOutputStream(raw_stream)
   226       var finished = false
   227       while (!finished) {
   228         try {
   229           //{{{
   230           receive {
   231             case Input_Chunks(chunks) =>
   232               stream.write(Standard_System.string_bytes(
   233                   chunks.map(_.length).mkString("", ",", "\n")));
   234               chunks.foreach(stream.write(_));
   235               stream.flush
   236             case Close =>
   237               stream.close
   238               finished = true
   239             case bad => System.err.println(name + ": ignoring bad message " + bad)
   240           }
   241           //}}}
   242         }
   243         catch {
   244           case e: IOException => put_result(Markup.SYSTEM, name + ": " + e.getMessage)
   245         }
   246       }
   247       put_result(Markup.SYSTEM, name + " terminated")
   248     }
   249 
   250 
   251   /* message output */
   252 
   253   private class Protocol_Error(msg: String) extends Exception(msg)
   254 
   255   private def message_actor(name: String, stream: InputStream): Actor =
   256     Simple_Thread.actor(name) {
   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 {
   330     case e: IOException =>
   331       rm_fifos()
   332       error("Failed to execute Isabelle process: " + e.getMessage)
   333   }
   334 
   335 
   336   /* I/O actors */
   337 
   338   private val standard_input = stdin_actor("standard_input", proc.get.getOutputStream)
   339   stdout_actor("standard_output", proc.get.getInputStream)
   340 
   341   private val command_input = input_actor("command_input", system.fifo_output_stream(in_fifo))
   342   message_actor("message_output", system.fifo_input_stream(out_fifo))
   343 
   344 
   345   /* exit process */
   346 
   347   Simple_Thread.actor("process_exit") {
   348     proc match {
   349       case None =>
   350       case Some(p) =>
   351         val rc = p.waitFor()
   352         Thread.sleep(300)  // FIXME property!?
   353         put_result(Markup.SYSTEM, "process_exit terminated")
   354         put_result(Markup.EXIT, rc.toString)
   355     }
   356     rm_fifos()
   357   }
   358 
   359 
   360 
   361   /** main methods **/
   362 
   363   def input_raw(text: String): Unit = standard_input ! Input_Text(text)
   364 
   365   def input_bytes(name: String, args: Array[Byte]*): Unit =
   366     command_input ! Input_Chunks(Standard_System.string_bytes(name) :: args.toList)
   367 
   368   def input(name: String, args: String*): Unit =
   369     input_bytes(name, args.map(Standard_System.string_bytes): _*)
   370 
   371   def close(): Unit = command_input ! Close
   372 }