os/process.fz

# This file is part of the Fuzion language implementation.
#
# The Fuzion language implementation is free software: you can redistribute it
# and/or modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, version 3 of the License.
#
# The Fuzion language implementation is distributed in the hope that it will be
# useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public
# License for more details.
#
# You should have received a copy of the GNU General Public License along with The
# Fuzion language implementation.  If not, see <https://www.gnu.org/licenses/>.


# -----------------------------------------------------------------------
#
#  Tokiwa Software GmbH, Germany
#
#  Source code of Fuzion standard library feature process
#
# -----------------------------------------------------------------------

# type denoting a started process
#
module:public process(public pid i64, module std_in, std_out, std_err i64) : property.orderable is

  # how many bytes are read at a time
  # POSIX.1 requires PIPE_BUF to be at least 512 bytes
  # NYI: UNDER DEVELOPMENT: move this to effect?
  pipe_buffer_size := 4096


  # predefined read handler for reading from
  # standard out of the process
  #
  std_out_read_handler => read_handler std_out


  # predefined read handler for reading from
  # standard err of the process
  #
  std_err_read_handler => read_handler std_err


  # predefined write handler for writing to
  # standard in of the process
  #
  std_in_write_handler => write_handler std_in


  # read bytes from standard out
  #
  read_handler(desc i64) : io.Read_Handler is
    public redef read(count i32) choice (array u8) io.end_of_file error =>
      arr := array u8 pipe_buffer_size i->0
      res := fzE_pipe_read desc arr.internal_array.data arr.count
      if res = -1
        error "error reading from stdout."
      else if res = 0
        io.end_of_file
      else
        arr
          .slice 0 res
          .as_array


  # write bytes to standard in
  #
  write_handler (desc i64) : io.Write_Handler is
    public redef write (bytes Sequence u8) outcome unit =>
      if fzE_pipe_write desc bytes.as_array.internal_array.data bytes.count = -1
        error "error while writing"
      else
        unit


  # write bytes to stdin of child process
  #
  public write_bytes (bytes Sequence u8) outcome i32
  =>
    write_bytes std_in bytes


  # helper function
  #
  write_bytes (desc i64, bytes Sequence u8) outcome i32
  pre debug: bytes.count > 0
  =>
    bytes
      .chunk pipe_buffer_size
      .reduce_or_error 0 (r, t)->
        arr := t.as_array
        bw := fzE_pipe_write desc arr.internal_array.data arr.count
        if bw = -1
          abort (outcome i32) (error "error while writing. wrote $r bytes already.")
        else
          r+bw


  # install buffered reader for reading from stdout of process
  # and run `f`.
  public with_out(T type, LM type : mutate, f () -> outcome T) outcome T =>
    (io.buffered LM).reader (std_out_read_handler) pipe_buffer_size ! f


  # install buffered reader for reading from stdout of process
  # and run `f`.
  public with_err(T type, LM type : mutate, f () -> outcome T) outcome T =>
    (io.buffered LM).reader (std_err_read_handler) pipe_buffer_size ! f


  public with_in(T type, LM type : mutate, f () -> T) outcome T =>
    (io.buffered LM).writer std_in_write_handler pipe_buffer_size ! f


  # write string to stdin of child process
  #
  public write_string (s String) outcome i32 =>
    write_bytes s.utf8


  # pipe this processes output to new process
  #
  public infix | (process_and_args Sequence String) outcome os.process
  pre
    process_and_args ∀ .is_ascii
    !process_and_args[0].contains_whitespace
  =>
    (os.process.start process_and_args.first.val (process_and_args.drop 1)).bind p->

      # NYI: UNDER DEVELOPMENT: should wire pipe directly to process
      # NYI: UNDER DEVELOPMENT: this works only when pipes buffer is large enough for
      # what is being written/read to/from pipes
      # thread to pipe from one process to other
      _ := concur.threads.spawn ()->
        _ :=
          for p1rb := std_out_read_handler.read pipe_buffer_size
          while
            match p1rb
              s array u8 => (write_bytes p.std_in s).ok
              * => false

        # NYI: UNDER DEVELOPMENT: error handling
        _ := fzE_pipe_close p.std_in
        _ := fzE_pipe_close process.this.std_out
        _ := fzE_pipe_close process.this.std_err

      p


  # close standard input of this process
  #
  public close_in outcome unit =>
    if fzE_pipe_close std_in != 0
      error "closing standard in was not successful, error code: $fzE_last_error"


  # wait for this process
  #
  public wait outcome u32 =>
    # NYI: process should be ref?
    os.processes.wait (os.process pid std_in std_out std_err)


  # send signal to processm
  #
  public send_signal(sig signal) outcome unit=>
    fzE_send_signal pid sig.as_i32 = 0 ? unit : error "error sending signal to process $pid: $fzE_last_error"


  # start a process with name, arguments and environment variables
  #
  public type.start(n String, args Sequence String, env_vars container.Map String String) outcome os.process=>
    os.processes.start n args env_vars


  # start a process with name and arguments
  #
  public type.start(n String, args Sequence String) outcome os.process=>
    start n args (container.ps_map String String).empty


  # start a process with name and no arguments
  #
  public type.start(n String) outcome os.process =>
    start n []


  # equality
  #
  public fixed redef type.equality(a, b process.this) bool => a.pid = b.pid


  # lteq
  #
  public fixed redef type.lteq(a, b process.this) bool =>
    a.pid <= b.pid



  # NYI: UNDER DEVELOPMENT:
  # - check if process is still running
  # - send signal to process
  # - get the exit code from a process?
  # - change nice level
  # - ...