Non-blocking Handle read

Stream: Is there code for X?

[Oliver Dressler]

Is there a non-blocking way to read from a Handle, w timeout or similar?

Currently I use:

let proc ← IO.Process.spawn {
  cmd := "cmd"
  stdin := .piped
  stdout := .piped
  stderr := .piped
}

proc.stdin.putStr "prompt"
proc.stdin.flush
let line ← proc.stdout.getLine  -- Non-blocking would be nice

[Siddhartha Gadgil]

There is now an Async. Perhaps the following code (my attempt to learn Async) includes what you need:

import Std.Internal.Async
import Std.Sync.Mutex
open Std.Internal.IO Async Std

def writeSlow (n: Nat) (s: String := "")(count : Std.Mutex Nat) : Async Nat := do
  IO.sleep (n * 1000).toUInt32
  IO.eprintln s!"Write operation completed after {n} seconds; phase: {s}"
  count.atomically (·.modify (fun n ↦ n + 1))
  let c ← count.atomically (·.get)
  IO.eprintln s!"Total writes completed: {c}"
  return n

partial def loop (n: Nat) (s: String) (count : Std.Mutex Nat) (stdin : IO.FS.Stream) : Async Unit := do
  for m in [1:n] do
    background (prio := Task.Priority.default) (writeSlow m s count)
  IO.eprintln s!"Write operations started in phase {s}. Waiting for input. Type 'q' to quit"
  let inp ← stdin.getLine
  unless inp.startsWith "q" do
    loop n inp count stdin

def main (args: List String) : IO Unit := do
  AsyncTask.block <| ←
    Async.toIO do
      let n := args[0]! |>.toNat!
      let count ←  Std.Mutex.new 0
      let stdin ← IO.getStdin
      loop n "intial" count stdin

[Henrik Böving]

Oliver Dressler said:

Is there a non-blocking way to read from a Handle, w timeout or similar?

Currently I use:

let proc ← IO.Process.spawn {
  cmd := "cmd"
  stdin := .piped
  stdout := .piped
  stderr := .piped
}

proc.stdin.putStr "prompt"
proc.stdin.flush
let line ← proc.stdout.getLine  -- Non-blocking would be nice

Unfortunately the async efforts haven't reached the process API yet.

Siddhartha Gadgil said:

There is now an Async. Perhaps the following code (my attempt to learn Async) includes what you need:

import Std.Internal.Async
import Std.Sync.Mutex
open Std.Internal.IO Async Std

def writeSlow (n: Nat) (s: String := "")(count : Std.Mutex Nat) : Async Nat := do
  IO.sleep (n * 1000).toUInt32
  IO.eprintln s!"Write operation completed after {n} seconds; phase: {s}"
  count.atomically (·.modify (fun n ↦ n + 1))
  let c ← count.atomically (·.get)
  IO.eprintln s!"Total writes completed: {c}"
  return n

partial def loop (n: Nat) (s: String) (count : Std.Mutex Nat) (stdin : IO.FS.Stream) : Async Unit := do
  for m in [1:n] do
    background (prio := Task.Priority.default) (writeSlow m s count)
  IO.eprintln s!"Write operations started in phase {s}. Waiting for input. Type 'q' to quit"
  let inp ← stdin.getLine
  unless inp.startsWith "q" do
    loop n inp count stdin

def main (args: List String) : IO Unit := do
  AsyncTask.block <| ←
    Async.toIO do
      let n := args[0]! |>.toNat!
      let count ←  Std.Mutex.new 0
      let stdin ← IO.getStdin
      loop n "intial" count stdin

Unfortunately that code is not at all async and will just block all over the place, IO.sleep is a blocking sleeping API, the one that would not block would be Std.Internal.IO.Async.sleep and also getLine is not an asynchronous but a blocking one. The Async monad can currently lift blocking functions from IO to do things like print etc. but just because your program type checks inside of the Async monad doesn't mean there is blocking going on.

Furthermore the proper API to read with a timeout would be to use a selectable: https://leanprover-community.github.io/mathlib4_docs/Std/Internal/Async/Select.html#Std.Internal.IO.Async.Selectable.one but again this API currently only works with things like sockets and not yet stdout handles.

[Siddhartha Gadgil]

Thanks @Henrik Böving
What I tested and found was that the process in the background runs and outputs in StdErr while waiting for input, and when I input new processes are launched.

This is the minimal I needed - a task running in the background while waiting for a new request, and if the new request comes then launching on another thread. The background function does this.

Could you suggest how to improve the code? I see one in the above : I should be using Std.Internal.IO.Async.sleep to not block a thread (but my actual code has no `IO.sleep).

[Henrik Böving]

There is no way to do it properly with actually 0 blockage right now. The getLine has no proper alternative

[Siddhartha Gadgil]

Does that mean one thread blocks because of the getLine, but the other stuff can run in the background?

[Henrik Böving]

Yes writeSlow will run on separate threads (assuming that you have more than 1 on your machine)

[Siddhartha Gadgil]

For me losing 1 thread is an acceptable price till the Async API gets more goodies. Acutally the StdIO is also a hack while waiting for the http server (currently a Python server wraps StdIO and StdIn).

[Oliver Dressler]

Thanks for all the info! I also tried async, task and readToEnd but none fit my use case.
I can currently work around this until the async handle read is available. It just forces me to know exactly how many lines to expect.

[Siddhartha Gadgil]

@Henrik Böving am I correct that if I want to poll without blocking any resources, I can use Std.Internal.IO.Async.sleep and map the "result" to a recursion on a loop?

[Henrik Böving]

You will still block on getLine regardless.

[Siddhartha Gadgil]

I mean if I poll with a curl request instead of having getLine.