src/Pure/System/isabelle_process.scala
author wenzelm
Mon Jul 15 10:25:35 2013 +0200 (2013-07-15 ago)
changeset 52655 3b2b1ef13979
parent 52582 31467a4b1466
child 52799 6a4498b048b7
permissions -rw-r--r--
more careful termination of removed execs, leaving running execs undisturbed;
wenzelm@43283
     1
/*  Title:      Pure/System/isabelle_process.scala
wenzelm@27949
     2
    Author:     Makarius
wenzelm@27963
     3
    Options:    :folding=explicit:collapseFolds=1:
wenzelm@27949
     4
wenzelm@27949
     5
Isabelle process management -- always reactive due to multi-threaded I/O.
wenzelm@27949
     6
*/
wenzelm@27949
     7
wenzelm@27949
     8
package isabelle
wenzelm@27949
     9
wenzelm@43520
    10
import java.lang.System
wenzelm@27955
    11
import java.util.concurrent.LinkedBlockingQueue
wenzelm@28045
    12
import java.io.{BufferedReader, BufferedWriter, InputStreamReader, OutputStreamWriter,
wenzelm@38270
    13
  InputStream, OutputStream, BufferedOutputStream, IOException}
wenzelm@27949
    14
wenzelm@32474
    15
import scala.actors.Actor
wenzelm@32474
    16
import Actor._
wenzelm@27949
    17
wenzelm@27973
    18
wenzelm@32474
    19
object Isabelle_Process
wenzelm@32474
    20
{
wenzelm@46772
    21
  /* messages */
wenzelm@27949
    22
wenzelm@44733
    23
  sealed abstract class Message
wenzelm@43721
    24
wenzelm@43721
    25
  class Input(name: String, args: List[String]) extends Message
wenzelm@43721
    26
  {
wenzelm@43721
    27
    override def toString: String =
wenzelm@50201
    28
      XML.Elem(Markup(Markup.PROVER_COMMAND, List((Markup.NAME, name))),
wenzelm@43721
    29
        args.map(s =>
wenzelm@50201
    30
          List(XML.Text("\n"), XML.elem(Markup.PROVER_ARG, YXML.parse_body(s)))).flatten).toString
wenzelm@43721
    31
  }
wenzelm@43721
    32
wenzelm@46772
    33
  class Output(val message: XML.Elem) extends Message
wenzelm@34100
    34
  {
wenzelm@43747
    35
    def kind: String = message.markup.name
wenzelm@43780
    36
    def properties: Properties.T = message.markup.properties
wenzelm@43747
    37
    def body: XML.Body = message.body
wenzelm@37689
    38
wenzelm@50201
    39
    def is_init = kind == Markup.INIT
wenzelm@50201
    40
    def is_exit = kind == Markup.EXIT
wenzelm@50201
    41
    def is_stdout = kind == Markup.STDOUT
wenzelm@50201
    42
    def is_stderr = kind == Markup.STDERR
wenzelm@50201
    43
    def is_system = kind == Markup.SYSTEM
wenzelm@50201
    44
    def is_status = kind == Markup.STATUS
wenzelm@50201
    45
    def is_report = kind == Markup.REPORT
wenzelm@50201
    46
    def is_protocol = kind == Markup.PROTOCOL
wenzelm@46121
    47
    def is_syslog = is_init || is_exit || is_system || is_stderr
wenzelm@34100
    48
wenzelm@34100
    49
    override def toString: String =
wenzelm@34100
    50
    {
wenzelm@29522
    51
      val res =
wenzelm@38236
    52
        if (is_status || is_report) message.body.map(_.toString).mkString
wenzelm@46774
    53
        else if (is_protocol) "..."
wenzelm@37689
    54
        else Pretty.string_of(message.body)
wenzelm@38230
    55
      if (properties.isEmpty)
wenzelm@29572
    56
        kind.toString + " [[" + res + "]]"
wenzelm@29572
    57
      else
wenzelm@29572
    58
        kind.toString + " " +
wenzelm@38230
    59
          (for ((x, y) <- properties) yield x + "=" + y).mkString("{", ",", "}") + " [[" + res + "]]"
wenzelm@27949
    60
    }
wenzelm@27973
    61
  }
wenzelm@27973
    62
}
wenzelm@27973
    63
wenzelm@27973
    64
wenzelm@45055
    65
class Isabelle_Process(
wenzelm@45055
    66
    receiver: Isabelle_Process.Message => Unit = Console.println(_),
wenzelm@45055
    67
    args: List[String] = Nil)
wenzelm@29192
    68
{
wenzelm@31797
    69
  import Isabelle_Process._
wenzelm@29194
    70
wenzelm@27973
    71
wenzelm@48705
    72
  /* text representation */
wenzelm@48705
    73
wenzelm@48705
    74
  def encode(s: String): String = Symbol.encode(s)
wenzelm@48705
    75
  def decode(s: String): String = Symbol.decode(s)
wenzelm@48705
    76
wenzelm@48705
    77
  object Encode
wenzelm@48705
    78
  {
wenzelm@48705
    79
    val string: XML.Encode.T[String] = (s => XML.Encode.string(encode(s)))
wenzelm@48705
    80
  }
wenzelm@48705
    81
wenzelm@48705
    82
wenzelm@46772
    83
  /* output */
wenzelm@39573
    84
wenzelm@51664
    85
  val xml_cache = new XML.Cache()
wenzelm@51664
    86
wenzelm@46772
    87
  private def system_output(text: String)
wenzelm@39573
    88
  {
wenzelm@50201
    89
    receiver(new Output(XML.Elem(Markup(Markup.SYSTEM, Nil), List(XML.Text(text)))))
wenzelm@39573
    90
  }
wenzelm@39573
    91
wenzelm@46772
    92
  private def output_message(kind: String, props: Properties.T, body: XML.Body)
wenzelm@39573
    93
  {
wenzelm@50201
    94
    if (kind == Markup.INIT) system_channel.accepted()
wenzelm@50201
    95
    if (kind == Markup.PROTOCOL)
wenzelm@46772
    96
      receiver(new Output(XML.Elem(Markup(kind, props), body)))
wenzelm@43746
    97
    else {
wenzelm@49445
    98
      val main = XML.Elem(Markup(kind, props), Protocol.clean_message(body))
wenzelm@49445
    99
      val reports = Protocol.message_reports(props, body)
wenzelm@51663
   100
      for (msg <- main :: reports) receiver(new Output(xml_cache.elem(msg)))
wenzelm@43746
   101
    }
wenzelm@39573
   102
  }
wenzelm@39573
   103
wenzelm@48016
   104
  private def exit_message(rc: Int)
wenzelm@39573
   105
  {
wenzelm@50201
   106
    output_message(Markup.EXIT, Markup.Return_Code(rc),
wenzelm@48016
   107
      List(XML.Text("Return code: " + rc.toString)))
wenzelm@39573
   108
  }
wenzelm@39573
   109
wenzelm@39573
   110
wenzelm@52581
   111
  /* command input actor */
wenzelm@39528
   112
wenzelm@39528
   113
  private case class Input_Chunks(chunks: List[Array[Byte]])
wenzelm@39528
   114
wenzelm@39528
   115
  private case object Close
wenzelm@39585
   116
  private def close(p: (Thread, Actor))
wenzelm@39585
   117
  {
wenzelm@39585
   118
    if (p != null && p._1.isAlive) {
wenzelm@39585
   119
      p._2 ! Close
wenzelm@39585
   120
      p._1.join
wenzelm@39585
   121
    }
wenzelm@39585
   122
  }
wenzelm@39528
   123
wenzelm@39585
   124
  @volatile private var command_input: (Thread, Actor) = null
wenzelm@39528
   125
wenzelm@39528
   126
wenzelm@39585
   127
  /** process manager **/
wenzelm@39528
   128
wenzelm@45158
   129
  private val system_channel = System_Channel()
wenzelm@39528
   130
wenzelm@39585
   131
  private val process =
wenzelm@39528
   132
    try {
wenzelm@39528
   133
      val cmdline =
wenzelm@45027
   134
        Isabelle_System.getenv_strict("ISABELLE_PROCESS") ::
wenzelm@45055
   135
          (system_channel.isabelle_args ::: args)
wenzelm@48353
   136
      new Isabelle_System.Managed_Process(null, null, false, cmdline: _*)
wenzelm@39528
   137
    }
wenzelm@45027
   138
    catch { case e: IOException => system_channel.accepted(); throw(e) }
wenzelm@39528
   139
wenzelm@48355
   140
  val (_, process_result) =
wenzelm@39587
   141
    Simple_Thread.future("process_result") { process.join }
wenzelm@39587
   142
wenzelm@39585
   143
  private def terminate_process()
wenzelm@39585
   144
  {
wenzelm@39585
   145
    try { process.terminate }
wenzelm@46772
   146
    catch { case e: IOException => system_output("Failed to terminate Isabelle: " + e.getMessage) }
wenzelm@39585
   147
  }
wenzelm@39528
   148
wenzelm@39585
   149
  private val process_manager = Simple_Thread.fork("process_manager")
wenzelm@39572
   150
  {
wenzelm@46548
   151
    val (startup_failed, startup_errors) =
wenzelm@39528
   152
    {
wenzelm@48020
   153
      var finished: Option[Boolean] = None
wenzelm@39528
   154
      val result = new StringBuilder(100)
wenzelm@48020
   155
      while (finished.isEmpty && (process.stderr.ready || !process_result.is_finished)) {
wenzelm@46548
   156
        while (finished.isEmpty && process.stderr.ready) {
wenzelm@48020
   157
          try {
wenzelm@48020
   158
            val c = process.stderr.read
wenzelm@48020
   159
            if (c == 2) finished = Some(true)
wenzelm@48020
   160
            else result += c.toChar
wenzelm@48020
   161
          }
wenzelm@48020
   162
          catch { case _: IOException => finished = Some(false) }
wenzelm@39528
   163
        }
wenzelm@48020
   164
        Thread.sleep(10)
wenzelm@39528
   165
      }
wenzelm@39623
   166
      (finished.isEmpty || !finished.get, result.toString.trim)
wenzelm@39528
   167
    }
wenzelm@46772
   168
    if (startup_errors != "") system_output(startup_errors)
wenzelm@39572
   169
wenzelm@52581
   170
    process.stdin.close
wenzelm@39528
   171
    if (startup_failed) {
wenzelm@48016
   172
      exit_message(127)
wenzelm@39585
   173
      Thread.sleep(300)
wenzelm@39585
   174
      terminate_process()
wenzelm@39587
   175
      process_result.join
wenzelm@39528
   176
    }
wenzelm@39528
   177
    else {
wenzelm@45027
   178
      val (command_stream, message_stream) = system_channel.rendezvous()
wenzelm@39530
   179
wenzelm@46773
   180
      val stdout = physical_output_actor(false)
wenzelm@46773
   181
      val stderr = physical_output_actor(true)
wenzelm@39585
   182
      command_input = input_actor(command_stream)
wenzelm@39572
   183
      val message = message_actor(message_stream)
wenzelm@39528
   184
wenzelm@39587
   185
      val rc = process_result.join
wenzelm@46772
   186
      system_output("process terminated")
wenzelm@52581
   187
      close(command_input)
wenzelm@52581
   188
      for ((thread, _) <- List(stdout, stderr, command_input, message))
wenzelm@45633
   189
        thread.join
wenzelm@46772
   190
      system_output("process_manager terminated")
wenzelm@48016
   191
      exit_message(rc)
wenzelm@39528
   192
    }
wenzelm@45027
   193
    system_channel.accepted()
wenzelm@39528
   194
  }
wenzelm@27973
   195
wenzelm@39572
   196
wenzelm@39585
   197
  /* management methods */
wenzelm@27973
   198
wenzelm@39585
   199
  def join() { process_manager.join() }
wenzelm@39528
   200
wenzelm@39585
   201
  def terminate()
wenzelm@38270
   202
  {
wenzelm@52581
   203
    close(command_input)
wenzelm@46772
   204
    system_output("Terminating Isabelle process")
wenzelm@39585
   205
    terminate_process()
wenzelm@27973
   206
  }
wenzelm@27973
   207
wenzelm@27949
   208
wenzelm@27949
   209
wenzelm@38259
   210
  /** stream actors **/
wenzelm@27949
   211
wenzelm@46773
   212
  /* physical output */
wenzelm@27949
   213
wenzelm@46773
   214
  private def physical_output_actor(err: Boolean): (Thread, Actor) =
wenzelm@39528
   215
  {
wenzelm@45633
   216
    val (name, reader, markup) =
wenzelm@50201
   217
      if (err) ("standard_error", process.stderr, Markup.STDERR)
wenzelm@50201
   218
      else ("standard_output", process.stdout, Markup.STDOUT)
wenzelm@45633
   219
wenzelm@38636
   220
    Simple_Thread.actor(name) {
wenzelm@46769
   221
      try {
wenzelm@46769
   222
        var result = new StringBuilder(100)
wenzelm@46769
   223
        var finished = false
wenzelm@46769
   224
        while (!finished) {
wenzelm@28045
   225
          //{{{
wenzelm@28045
   226
          var c = -1
wenzelm@28045
   227
          var done = false
wenzelm@45633
   228
          while (!done && (result.length == 0 || reader.ready)) {
wenzelm@45633
   229
            c = reader.read
wenzelm@28045
   230
            if (c >= 0) result.append(c.asInstanceOf[Char])
wenzelm@28045
   231
            else done = true
wenzelm@28045
   232
          }
wenzelm@28045
   233
          if (result.length > 0) {
wenzelm@48705
   234
            output_message(markup, Nil, List(XML.Text(decode(result.toString))))
wenzelm@28045
   235
            result.length = 0
wenzelm@27949
   236
          }
wenzelm@27949
   237
          else {
wenzelm@45633
   238
            reader.close
wenzelm@28045
   239
            finished = true
wenzelm@38270
   240
          }
wenzelm@38270
   241
          //}}}
wenzelm@38270
   242
        }
wenzelm@38270
   243
      }
wenzelm@46772
   244
      catch { case e: IOException => system_output(name + ": " + e.getMessage) }
wenzelm@46772
   245
      system_output(name + " terminated")
wenzelm@38270
   246
    }
wenzelm@39528
   247
  }
wenzelm@38270
   248
wenzelm@38270
   249
wenzelm@38270
   250
  /* command input */
wenzelm@38270
   251
wenzelm@39572
   252
  private def input_actor(raw_stream: OutputStream): (Thread, Actor) =
wenzelm@39528
   253
  {
wenzelm@39528
   254
    val name = "command_input"
wenzelm@38636
   255
    Simple_Thread.actor(name) {
wenzelm@46769
   256
      try {
wenzelm@46769
   257
        val stream = new BufferedOutputStream(raw_stream)
wenzelm@46769
   258
        var finished = false
wenzelm@46769
   259
        while (!finished) {
wenzelm@38270
   260
          //{{{
wenzelm@38270
   261
          receive {
wenzelm@38270
   262
            case Input_Chunks(chunks) =>
wenzelm@50203
   263
              stream.write(UTF8.string_bytes(chunks.map(_.length).mkString("", ",", "\n")))
wenzelm@50203
   264
              chunks.foreach(stream.write(_))
wenzelm@38270
   265
              stream.flush
wenzelm@38270
   266
            case Close =>
wenzelm@38270
   267
              stream.close
wenzelm@38270
   268
              finished = true
wenzelm@38270
   269
            case bad => System.err.println(name + ": ignoring bad message " + bad)
wenzelm@27949
   270
          }
wenzelm@28045
   271
          //}}}
wenzelm@27963
   272
        }
wenzelm@27949
   273
      }
wenzelm@46772
   274
      catch { case e: IOException => system_output(name + ": " + e.getMessage) }
wenzelm@46772
   275
      system_output(name + " terminated")
wenzelm@27949
   276
    }
wenzelm@39528
   277
  }
wenzelm@27949
   278
wenzelm@27949
   279
wenzelm@38259
   280
  /* message output */
wenzelm@27949
   281
wenzelm@39572
   282
  private def message_actor(stream: InputStream): (Thread, Actor) =
wenzelm@39526
   283
  {
wenzelm@39526
   284
    class EOF extends Exception
wenzelm@39526
   285
    class Protocol_Error(msg: String) extends Exception(msg)
wenzelm@38259
   286
wenzelm@39528
   287
    val name = "message_output"
wenzelm@38636
   288
    Simple_Thread.actor(name) {
wenzelm@34100
   289
      val default_buffer = new Array[Byte](65536)
wenzelm@34100
   290
      var c = -1
wenzelm@27949
   291
wenzelm@48705
   292
      def read_chunk(do_decode: Boolean): XML.Body =
wenzelm@34100
   293
      {
wenzelm@34100
   294
        //{{{
wenzelm@34100
   295
        // chunk size
wenzelm@34100
   296
        var n = 0
wenzelm@34100
   297
        c = stream.read
wenzelm@39526
   298
        if (c == -1) throw new EOF
wenzelm@34100
   299
        while (48 <= c && c <= 57) {
wenzelm@34100
   300
          n = 10 * n + (c - 48)
wenzelm@34100
   301
          c = stream.read
wenzelm@34100
   302
        }
wenzelm@34100
   303
        if (c != 10) throw new Protocol_Error("bad message chunk header")
wenzelm@34100
   304
wenzelm@34100
   305
        // chunk content
wenzelm@34100
   306
        val buf =
wenzelm@34100
   307
          if (n <= default_buffer.size) default_buffer
wenzelm@34100
   308
          else new Array[Byte](n)
wenzelm@34100
   309
wenzelm@34100
   310
        var i = 0
wenzelm@34100
   311
        var m = 0
wenzelm@34100
   312
        do {
wenzelm@34100
   313
          m = stream.read(buf, i, n - i)
wenzelm@39731
   314
          if (m != -1) i += m
wenzelm@39731
   315
        } while (m != -1 && n > i)
wenzelm@34100
   316
wenzelm@34100
   317
        if (i != n) throw new Protocol_Error("bad message chunk content")
wenzelm@34100
   318
wenzelm@48705
   319
        if (do_decode)
wenzelm@50203
   320
          YXML.parse_body_failsafe(UTF8.decode_chars(decode, buf, 0, n))
wenzelm@50203
   321
        else List(XML.Text(UTF8.decode_chars(s => s, buf, 0, n).toString))
wenzelm@34100
   322
        //}}}
wenzelm@34100
   323
      }
wenzelm@34100
   324
wenzelm@46769
   325
      try {
wenzelm@46769
   326
        do {
wenzelm@46769
   327
          try {
wenzelm@46769
   328
            //{{{
wenzelm@46769
   329
            val header = read_chunk(true)
wenzelm@46769
   330
            header match {
wenzelm@49677
   331
              case List(XML.Elem(Markup(name, props), Nil)) =>
wenzelm@49677
   332
                val kind = name.intern
wenzelm@50201
   333
                val body = read_chunk(kind != Markup.PROTOCOL)
wenzelm@46772
   334
                output_message(kind, props, body)
wenzelm@46769
   335
              case _ =>
wenzelm@46769
   336
                read_chunk(false)
wenzelm@46769
   337
                throw new Protocol_Error("bad header: " + header.toString)
wenzelm@46769
   338
            }
wenzelm@46769
   339
            //}}}
wenzelm@28063
   340
          }
wenzelm@46769
   341
          catch { case _: EOF => }
wenzelm@46769
   342
        } while (c != -1)
wenzelm@46769
   343
      }
wenzelm@46769
   344
      catch {
wenzelm@46772
   345
        case e: IOException => system_output("Cannot read message:\n" + e.getMessage)
wenzelm@46772
   346
        case e: Protocol_Error => system_output("Malformed message:\n" + e.getMessage)
wenzelm@46769
   347
      }
wenzelm@34100
   348
      stream.close
wenzelm@34100
   349
wenzelm@46772
   350
      system_output(name + " terminated")
wenzelm@27949
   351
    }
wenzelm@39526
   352
  }
wenzelm@27949
   353
wenzelm@27949
   354
wenzelm@38259
   355
  /** main methods **/
wenzelm@38259
   356
wenzelm@52582
   357
  def protocol_command_raw(name: String, args: Array[Byte]*): Unit =
wenzelm@50203
   358
    command_input._2 ! Input_Chunks(UTF8.string_bytes(name) :: args.toList)
wenzelm@38372
   359
wenzelm@52582
   360
  def protocol_command(name: String, args: String*)
wenzelm@43721
   361
  {
wenzelm@44732
   362
    receiver(new Input(name, args.toList))
wenzelm@52582
   363
    protocol_command_raw(name, args.map(UTF8.string_bytes): _*)
wenzelm@43721
   364
  }
wenzelm@38259
   365
wenzelm@50117
   366
  def options(opts: Options): Unit =
wenzelm@52582
   367
    protocol_command("Isabelle_Process.options", YXML.string_of_body(opts.encode))
wenzelm@27949
   368
}