Process monitor async write buffering

[SeanTAllen]

Based on work originally done by @lispmeister. I brought it up to date
and fixed problematic test failures.

Add write buffering and asio events for writes to STDIN of the child
We now register an asio event on the STDIN file descriptor.
We write as much as possible. If we can't write everything we store the
ByteSeq in a list together with an offset and set the _stdin_writeable
flag to false. Once the OS signals STDIN is writeable again we try to
write the chunks stored in the list starting with the head chunk.

Closes #945

[SeanTAllen]

Here is the updated version of the test script designed to verify #945 is fixed:

updated test script for latest ponyc:

// original credit to @lispmeister
use "process"
use "files"
use "options"

actor Main
  let _env: Env

  new create(env: Env) =>
    _env = env
    var lines: (I64 | None) = None
    var required_args_are_present: Bool = true
    // parse commandline
    try
      var options = Options(_env.args)
      options
      .add("lines", "l", I64Argument)
      for option in options do
        match option
        | ("lines", let arg: I64) => lines = arg
        else
          env.err.print("stdin-process-test: unknown argument")
          env.err.print("stdin-process-test: usage: [--lines=2000]")
        end
      end

      // check lines value
      if lines is None then
        env.err.print("dagon: Must supply required '--lines' argument")
        required_args_are_present = false
      elseif (lines as I64) < 0 then
          env.err.print("stdin-process-test: lines can't be negative")
          required_args_are_present = false
      end

      if not required_args_are_present then
        env.err.print("stdin-process-test: error parsing arguments. Bailing out!")
        return
      end

      // create a notifier
      let client = ProcessClient(_env)
      let notifier: ProcessNotify iso = consume client
      // define the binary to run
      try
        let path = FilePath(env.root as AmbientAuth, "/usr/bin/wc")?
        // define the arguments; first arg is always the binary name
        let args: Array[String] iso = recover Array[String](2) end
        args.push("wc")
        args.push("-l")
        // define the environment variable for the execution
        let vars: Array[String] iso = recover Array[String](2) end
        vars.push("HOME=/")
        vars.push("PATH=/bin")
        // create a ProcessMonitor and spawn the child process
        let auth = _env.root as AmbientAuth
        let pm: ProcessMonitor = ProcessMonitor(auth, consume notifier, path,
        consume args, consume vars)
        // write to STDIN of the child process
        var i:I64 = 0
        while i < (lines as I64) do
          pm.write("..............................\n")
          i = i + 1
        end
        pm.done_writing() // closing stdin
      else
        env.err.print("stdin-process-test: Could not create FilePath!")
      end
    else
      env.err.print("stdin-process-test: error parsing arguments")
    end


// define a client that implements the ProcessNotify interface
class ProcessClient is ProcessNotify
  let _env: Env

  new iso create(env: Env) =>
    _env = env

  fun ref stdout(process: ProcessMonitor ref, data: Array[U8] iso) =>
    let out = String.from_array(consume data)
    _env.out.print("STDOUT: " + out)

  fun ref stderr(process: ProcessMonitor ref, data: Array[U8] iso) =>
    let err = String.from_array(consume data)
    _env.out.print("STDERR: " + err)

  fun ref failed(process: ProcessMonitor ref, err: ProcessError) =>
    match err
    | ExecveError   => _env.out.print("ProcessError: ExecveError")
    | PipeError     => _env.out.print("ProcessError: PipeError")
    | ForkError     => _env.out.print("ProcessError: ForkError")
    | WaitpidError  => _env.out.print("ProcessError: WaitpidError")
    | WriteError    => _env.out.print("ProcessError: WriteError")
    | KillError     => _env.out.print("ProcessError: KillError")
    | CapError      => _env.out.print("ProcessError: CapError")
    | Unsupported   => _env.out.print("ProcessError: Unsupported")
    else
      _env.out.print("Unknown ProcessError!")
    end

  fun ref dispose(process: ProcessMonitor ref, child_exit_code: I32) =>
    let code: I32 = consume child_exit_code
    _env.out.print("Child exit code: " + code.string())

To trigger error run with command line option like --lines=100000. basically a large value that will trigger a write error.

[SeanTAllen]

Found a flaw in the test logic. Fixing.

[SeanTAllen]

This seems to have some timeout issues that are environmental. I can't reproduce it locally. But I'm going to keep trying to verify.