Blocked OS threads can prevent a program from shutting down

[yorickpeterse]

Please describe the bug

Consider the following program (per the Discord):

import std.byte_array (ByteArray)
import std.fmt (fmt)
import std.process
import std.stdio (Stdin, Stdout)
import std.sync
import std.time

type enum Out {
  case Timeout
  case Read(Int)
}

type async SleepProcess {
  let @timeout: time.Duration
  let @chan: sync.Channel[Out]

  fn async run {
    process.sleep(@timeout)
    @chan.send(recover Out.Timeout)
  }
}

type async Cb {
  let @inner: fn -> Int
  let @chan: sync.Channel[Out]

  fn async mut run {
    let read = @inner.call

    @chan.send(recover Out.Read(read))
  }
}

type async Main {
  fn async main {
    let chan = sync.Channel.new
    let sleep = SleepProcess(
      timeout: recover time.Duration.from_secs(1),
      chan: recover chan.clone,
    )

    let cb = Cb(
      inner: recover {
        fn {
          let arr = ByteArray.new
          let stdin = Stdin.new
          let stdout = Stdout.new
          let read = stdin.read(arr, 10).or_panic('failed to read from stdin!')

          stdout.print('read ${read} bytes')
          stdout.print(fmt(arr))
          stdout.print(arr.to_string)
          read
        }
      },
      chan: recover chan.clone,
    )

    cb.run
    sleep.run

    let val = match chan.receive {
      case Timeout -> 'got a timeout!'
      case Read(i) -> 'read ${i} bytes'
    }

    let stdout = Stdout.new

    stdout.print(val)
    stdout.flush()
  }
}

When running this the program will hang after printing "got a timeout!". This is because the process calling Stdin.read blocks the OS thread. This in turn results in us getting stuck when joining the process threads here:

inko/rt/src/scheduler/process.rs

Lines 1013 to 1057 in 73e5f0f

	let _ = scope(move |s| {
	s.builder()
	.name("proc monitor".to_string())
	.spawn(move |_| Monitor::new(&self.pool).run())
	.unwrap();
	s.builder()
	.name("epoch".to_string())
	.spawn(move |_| {
	epoch_loop(state);
	})
	.unwrap();
	for id in 1..self.primary {
	let poll_id = id % pollers;
	s.builder()
	.name(format!("proc {}", id))
	.spawn(move |_| {
	pin_thread_to_core(id % cores);
	Thread::new(id, poll_id, self.pool.clone()).run(state)
	})
	.expect("failed to start a process thread");
	}
	for id in 0..self.backup {
	let poll_id = id % pollers;
	s.builder()
	.name(format!("backup {}", id))
	.spawn(move |_| {
	pin_thread_to_core(id % cores);
	Thread::backup(poll_id, self.pool.clone()).run(state)
	})
	.expect("failed to start a backup thread");
	}
	self.pool.schedule(process);
	// The current thread is used for running the main process. This
	// makes it possible for this process to interface with libraries
	// that require the same thread to be used for all operations (e.g.
	// most GUI libraries).
	Thread::new(0, 0, self.pool.clone()).run_main(state);
	});

For the worker threads we shouldn't need to use scoped, which means we can fix this by just spawning the threads the usual way and forgetting about them when shutting down. This does require that instead of threads capturing a &State, they instead each get a unique RcState.

Operating system

Fedora

Inko version

main

[yorickpeterse]

Come to think of it, I think we don't need to use scoped here at all.