Allow Async w/ Subprocess for POSIX Systems (#2953)

* add native_fflush libc passthru

* add `pipe2` which combines `pipe` and `fcntl` - see docs

* POSIX subprocess: incorporate async reader capability and and fflush for writer

* use the real-name O_NONBLOCK since it's available

* typo

* ok... remove `pipe2` and do this with an existing method......

* Update releasenotes.md

---------

Co-authored-by: Christoffer Lerno <christoffer@aegik.com>

lib/std/io/os/file_libc.c3

@@ -76,6 +76,11 @@ fn usz? native_fread(CFile file, char[] buffer) @inline
 	return libc::fread(buffer.ptr, 1, buffer.len, file);
 }
 
+fn void? native_fflush(CFile file) @inline @maydiscard
+{
+	if (libc::fflush(file) != 0) return io::GENERAL_ERROR~;
+}
+
 macro fault file_open_errno() @local
 {
 	switch (libc::errno())

lib/std/io/os/file_nolibc.c3

@@ -8,6 +8,7 @@ alias FseekFn = fn void?(void*, long, SeekOrigin);
 alias FtellFn = fn long?(void*);
 alias FwriteFn = fn usz?(void*, char[] buffer);
 alias FreadFn = fn usz?(void*, char[] buffer);
+alias FflushFn = fn void?(void*);
 alias RemoveFn = fn void?(String);
 alias FputcFn = fn void?(int, void*);
 alias SetModifiedTimeFn = fn void?(String, libc::Time_t);
@@ -19,6 +20,7 @@ FseekFn native_fseek_fn @weak @if(!$defined(native_fseek_fn));
 FtellFn native_ftell_fn @weak @if(!$defined(native_ftell_fn));
 FwriteFn native_fwrite_fn @weak @if(!$defined(native_fwrite_fn));
 FreadFn native_fread_fn @weak @if(!$defined(native_fread_fn));
+FflushFn native_fflush_fn @weak @if(!$defined(native_fflush_fn));
 RemoveFn native_remove_fn @weak @if(!$defined(native_remove_fn));
 FputcFn native_fputc_fn @weak @if(!$defined(native_fputc_fn));
 SetModifiedTimeFn native_set_modified_time_fn @weak @if(!$defined(native_set_modified_time_fn));
@@ -78,6 +80,12 @@ fn usz? native_fread(CFile file, char[] buffer) @inline
 	return io::UNSUPPORTED_OPERATION~;
 }
 
+fn void? native_fflush(CFile file) @inline @maydiscard
+{
+	if (native_fflush_fn) return native_fflush_fn(file);
+	return io::UNSUPPORTED_OPERATION~;
+}
+
 fn void? native_fputc(CInt c, CFile stream) @inline
 {
 	if (native_fputc_fn) return native_fputc_fn(c, stream);

lib/std/os/subprocess.c3

@@ -1,5 +1,5 @@
 module std::os::process @if(env::WIN32 || env::POSIX);
-import std::io, libc, std::os;
+import std::io, libc, std::os, std::net::os;
 
 // This code is based on https://github.com/sheredom/subprocess.h
 
@@ -321,10 +321,24 @@ fn SubProcess? create(String[] command_line, SubProcessOptions options = {}, Str
 		if (posix::pipe(&stdoutfd)) return FAILED_TO_OPEN_STDOUT~;
 		if (!options.combined_stdout_stderr && posix::pipe(&stderrfd)) return FAILED_TO_OPEN_STDERR~;
 
+		if (options.read_async)
+		{
+			((NativeSocket)stdoutfd[0]).set_non_blocking(true)!;
+			((NativeSocket)stdoutfd[1]).set_non_blocking(true)!;
+			if (!options.combined_stdout_stderr)
+			{
+				((NativeSocket)stderrfd[0]).set_non_blocking(true)!;
+				((NativeSocket)stderrfd[1]).set_non_blocking(true)!;
+			}
+		}
+
 		if (posix::spawn_file_actions_addclose(&actions, stdinfd[1])) return FAILED_TO_OPEN_STDIN~;
 		if (posix::spawn_file_actions_adddup2(&actions, stdinfd[0], libc::STDIN_FD)) return FAILED_TO_OPEN_STDIN~;
+		if (posix::spawn_file_actions_addclose(&actions, stdinfd[0])) return FAILED_TO_OPEN_STDIN~;
+
 		if (posix::spawn_file_actions_addclose(&actions, stdoutfd[0])) return FAILED_TO_OPEN_STDOUT~;
 		if (posix::spawn_file_actions_adddup2(&actions, stdoutfd[1], libc::STDOUT_FD)) return FAILED_TO_OPEN_STDOUT~;
+		if (posix::spawn_file_actions_addclose(&actions, stdoutfd[1])) return FAILED_TO_OPEN_STDERR~;
 
 		if (options.combined_stdout_stderr)
 		{
@@ -334,13 +348,14 @@ fn SubProcess? create(String[] command_line, SubProcessOptions options = {}, Str
 		{
 			if (posix::spawn_file_actions_addclose(&actions, stderrfd[0])) return FAILED_TO_OPEN_STDERR~;
 			if (posix::spawn_file_actions_adddup2(&actions, stderrfd[1], libc::STDERR_FD)) return FAILED_TO_OPEN_STDERR~;
+			if (posix::spawn_file_actions_addclose(&actions, stderrfd[1])) return FAILED_TO_OPEN_STDERR~;
 		}
 	}
 
 	Pid_t child;
 	@stack_mem(2048; Allocator mem)
 	{
-    	ZString* command_line_copy = copy_command_line(mem, command_line);
+		ZString* command_line_copy = copy_command_line(mem, command_line);
 		ZString* used_environment = options.inherit_environment ? posix::environ : copy_env(mem, environment);
 		if (options.search_user_path)
 		{
@@ -485,10 +500,15 @@ fn usz? read_from_file_win32(CFile file, Win32_HANDLE event_handle, char* buffer
 	}
 	return bytes_read;
 }
+
 fn usz? read_from_file_posix(CFile file, char* buffer, usz size) @if(env::POSIX) @local
 {
 	isz bytes_read = libc::read(libc::fileno(file), buffer, size);
-	if (bytes_read < 0) return READ_FAILED~;
+	if (bytes_read < 0)
+	{
+		if (libc::errno() == errno::EAGAIN) return 0;   // nothing to read but O_NONBLOCK (async) is used
+		return READ_FAILED~;
+	}
 	return bytes_read;
 }
 
@@ -537,5 +557,6 @@ fn bool? SubProcess.is_running(&self)
 fn usz? SubProcess.write_to_stdin(&self, char[] buffer)
 {
 	if (!self.stdin_file) return FAILED_TO_OPEN_STDIN~;
+	defer try (void)os::native_fflush(self.stdin_file);
 	return os::native_fwrite(self.stdin_file, buffer);
 }

releasenotes.md

@@ -36,6 +36,7 @@
 - Stream functions now use long/ulong rather than isz/usz for seek/available. 
 - `instream.seek` is replaced by `set_cursor` and `cursor`.
 - `instream.available`, `cursor` etc are long/ulong rather than isz/usz to be correct on 32-bit.
+- Enable asynchronous, non-blocking reads of subprocess STDOUT/STDERR pipes on POSIX systems.
 
 ### Fixes
 - Add error message if directory with output file name already exists