Put that in your pipe and smoke it

extra/io/unix/backend/backend.factor

@@ -1,10 +1,11 @@
 ! Copyright (C) 2004, 2008 Slava Pestov.
 ! See http://factorcode.org/license.txt for BSD license.
-USING: alien generic assocs kernel kernel.private math
+USING: alien alien.c-types generic assocs kernel kernel.private
-io.ports sequences strings structs sbufs threads unix
+math io.ports sequences strings structs sbufs threads unix
 vectors io.buffers io.backend io.encodings math.parser
 continuations system libc qualified namespaces io.timeouts
-io.encodings.utf8 destructors accessors summary combinators ;
+io.encodings.utf8 destructors accessors summary combinators
+locals ;
 QUALIFIED: io
 IN: io.unix.backend
 
@@ -12,17 +13,19 @@ GENERIC: handle-fd ( handle -- fd )
 
 TUPLE: fd fd disposed ;
 
+: init-fd ( fd -- fd )
+    [
+        |dispose
+        dup fd>> F_SETFL O_NONBLOCK fcntl io-error
+        dup fd>> F_SETFD FD_CLOEXEC fcntl io-error
+    ] with-destructors ;
+
 : <fd> ( n -- fd )
     #! We drop the error code rather than calling io-error,
     #! since on OS X 10.3, this operation fails from init-io
     #! when running the Factor.app (presumably because fd 0 and
     #! 1 are closed).
-    fd new
+    f fd boa ;
-        swap
-        [ F_SETFL O_NONBLOCK fcntl drop ]
-        [ F_SETFD FD_CLOEXEC fcntl drop ]
-        [ >>fd ]
-        tri ;
 
 M: fd dispose
     dup disposed>> [ drop ] [
@@ -146,8 +149,55 @@ M: unix (wait-to-write) ( port -- )
 M: unix io-multiplex ( ms/f -- )
     mx get-global wait-for-events ;
 
+! On Unix, you're not supposed to set stdin to non-blocking
+! because the fd might be shared with another process (either
+! parent or child). So what we do is have the VM start a thread
+! which pumps data from the real stdin to a pipe. We set the
+! pipe to non-blocking, and read from it instead of the real
+! stdin. Very crufty, but it will suffice until we get native
+! threading support at the language level.
+TUPLE: stdin control size data ;
+
+M: stdin dispose
+    [
+        [ control>> &dispose drop ]
+        [ size>> &dispose drop ]
+        [ data>> &dispose drop ]
+        tri
+    ] with-destructors ;
+
+: wait-for-stdin ( stdin -- n )
+    [ control>> CHAR: X over io:stream-write1 io:stream-flush ]
+    [ size>> "uint" heap-size swap io:stream-read *uint ]
+    bi ;
+
+:: refill-stdin ( buffer stdin size -- )
+    stdin data>> handle-fd buffer buffer-end size read
+    dup 0 < [
+        drop
+        err_no EINTR = [ buffer stdin size refill-stdin ] [ (io-error) ] if
+    ] [
+        size = [ "Error reading stdin pipe" throw ] unless
+        size buffer n>buffer
+    ] if ;
+
+M: stdin refill
+    [ buffer>> ] [ dup wait-for-stdin ] bi* refill-stdin f ;
+
+: control-write-fd ( -- fd ) "control_write" f dlsym *uint ;
+
+: size-read-fd ( -- fd ) "size_read" f dlsym *uint ;
+
+: data-read-fd ( -- fd ) "stdin_read" f dlsym *uint ;
+
+: <stdin> ( -- stdin )
+    control-write-fd <fd> <output-port>
+    size-read-fd <fd> init-fd <input-port>
+    data-read-fd <fd>
+    stdin boa ;
+
 M: unix (init-stdio) ( -- )
-    0 <fd> <input-port>
+    <stdin> <input-port>
     1 <fd> <output-port>
     2 <fd> <output-port> ;
 

extra/io/unix/files/files.factor

@@ -19,7 +19,7 @@ M: unix cd ( path -- ) [ chdir ] unix-system-call drop ;
 : open-read ( path -- fd ) O_RDONLY file-mode open-file ;
 
 M: unix (file-reader) ( path -- stream )
-    open-read <fd> <input-port> ;
+    open-read <fd> init-fd <input-port> ;
 
 : write-flags { O_WRONLY O_CREAT O_TRUNC } flags ; inline
 
@@ -27,7 +27,7 @@ M: unix (file-reader) ( path -- stream )
     write-flags file-mode open-file ;
 
 M: unix (file-writer) ( path -- stream )
-    open-write <fd> <output-port> ;
+    open-write <fd> init-fd <output-port> ;
 
 : append-flags { O_WRONLY O_APPEND O_CREAT } flags ; inline
 
@@ -38,7 +38,7 @@ M: unix (file-writer) ( path -- stream )
     ] with-destructors ;
 
 M: unix (file-appender) ( path -- stream )
-    open-append <fd> <output-port> ;
+    open-append <fd> init-fd <output-port> ;
 
 : touch-mode ( -- n )
     { O_WRONLY O_APPEND O_CREAT O_EXCL } flags ; foldable

extra/io/unix/launcher/launcher.factor

@@ -30,17 +30,14 @@ USE: unix
         } at set-priority
     ] when* ;
 
+: reset-fd ( fd -- )
+    [ F_SETFL 0 fcntl io-error ] [ F_SETFD 0 fcntl io-error ] bi ;
+
 : redirect-fd ( oldfd fd -- )
     2dup = [ 2drop ] [ dup2 io-error ] if ;
 
-: reset-fd ( fd -- )
-    #! We drop the error code because on *BSD, fcntl of
-    #! /dev/null fails.
-    [ F_SETFL 0 fcntl drop ]
-    [ F_SETFD 0 fcntl drop ] bi ;
-
 : redirect-inherit ( obj mode fd -- )
-    2nip reset-fd ;
+    3drop ;
 
 : redirect-file ( obj mode fd -- )
     >r >r normalize-path r> file-mode

extra/io/unix/linux/monitors/monitors.factor

@@ -23,7 +23,7 @@ TUPLE: linux-monitor < monitor wd inotify watches disposed ;
 : wd>monitor ( wd -- monitor ) watches get at ;
 
 : <inotify> ( -- port/f )
-    inotify_init dup 0 < [ drop f ] [ <fd> <input-port> ] if ;
+    inotify_init dup 0 < [ drop f ] [ <fd> init-fd <input-port> ] if ;
 
 : inotify-fd ( -- fd ) inotify get handle>> handle-fd ;
 

extra/io/unix/pipes/pipes.factor

@@ -8,4 +8,4 @@ QUALIFIED: io.pipes
 M: unix io.pipes:(pipe) ( -- pair )
     2 "int" <c-array>
     dup pipe io-error
-    2 c-int-array> first2 [ <fd> ] bi@ io.pipes:pipe boa ;
+    2 c-int-array> first2 [ <fd> init-fd ] bi@ io.pipes:pipe boa ;

extra/io/unix/sockets/sockets.factor

@@ -13,7 +13,7 @@ EXCLUDE: io.sockets => accept ;
 IN: io.unix.sockets
 
 : socket-fd ( domain type -- fd )
-    0 socket dup io-error <fd> |dispose ;
+    0 socket dup io-error <fd> init-fd |dispose ;
 
 : set-socket-option ( fd level opt -- )
     >r >r handle-fd r> r> 1 <int> "int" heap-size setsockopt io-error ;
@@ -77,7 +77,7 @@ M: object (server) ( addrspec -- handle )
 M: object (accept) ( server addrspec -- fd sockaddr )
     2dup do-accept
     {
-        { [ over 0 >= ] [ >r 2nip <fd> r> ] }
+        { [ over 0 >= ] [ >r 2nip <fd> init-fd r> ] }
         { [ err_no EINTR = ] [ 2drop (accept) ] }
         { [ err_no EAGAIN = ] [
             2drop

vm/debug.c

@@ -325,7 +325,6 @@ void find_code_references(CELL look_for_)
 
 void factorbug(void)
 {
-	reset_stdio();
 	open_console();
 
 	printf("Starting low level debugger...\n");

vm/factor.c

@@ -28,7 +28,13 @@ void default_parameters(F_PARAMETERS *p)
 
 	p->secure_gc = false;
 	p->fep = false;
+
+#ifdef WINDOWS
 	p->console = false;
+#else
+	p->console = true;
+#endif
+
 	p->stack_traces = true;
 }
 
@@ -92,6 +98,9 @@ void init_factor(F_PARAMETERS *p)
 	init_c_io();
 	init_signals();
 
+	if(p->console)
+		open_console();
+
 	stack_chain = NULL;
 	profiling_p = false;
 	performing_gc = false;
@@ -180,9 +189,6 @@ void init_factor_from_args(F_CHAR *image, int argc, F_CHAR **argv, bool embedded
 	userenv[EXECUTABLE_ENV] = tag_object(from_native_string(executable_path));
 	userenv[EMBEDDED_ENV] = (embedded ? T : F);
 
-	if(p.console)
-		open_console();
-
 	if(p.fep)
 		factorbug();
 

vm/mach_signal.c

@@ -166,8 +166,6 @@ void mach_initialize (void)
 {
 	mach_port_t self;
 	exception_mask_t mask;
-	pthread_attr_t attr;
-	pthread_t thread;
 
 	self = mach_task_self ();
 
@@ -186,13 +184,7 @@ void mach_initialize (void)
 	mask = EXC_MASK_BAD_ACCESS | EXC_MASK_ARITHMETIC;
 
 	/* Create the thread listening on the exception port.  */
-	if (pthread_attr_init (&attr) != 0)
+	start_thread(mach_exception_thread);
-		fatal_error("pthread_attr_init() failed",0);
-	if (pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED) != 0)
-		fatal_error("pthread_attr_setdetachstate() failed",0);
-	if (pthread_create (&thread, &attr, mach_exception_thread, NULL) != 0)
-		fatal_error("pthread_create() failed",0);
-	pthread_attr_destroy (&attr);
 
 	/* Replace the exception port info for these exceptions with our own.
 	Note that we replace the exception port for the entire task, not only

vm/os-unix.c

@@ -1,5 +1,19 @@
 #include "master.h"
 
+void start_thread(void *(*start_routine)(void *))
+{
+	pthread_attr_t attr;
+	pthread_t thread;
+
+	if (pthread_attr_init (&attr) != 0)
+		fatal_error("pthread_attr_init() failed",0);
+	if (pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED) != 0)
+		fatal_error("pthread_attr_setdetachstate() failed",0);
+	if (pthread_create (&thread, &attr, start_routine, NULL) != 0)
+		fatal_error("pthread_create() failed",0);
+	pthread_attr_destroy (&attr);
+}
+
 static void *null_dll;
 
 s64 current_millis(void)
@@ -264,10 +278,128 @@ void unix_init_signals(void)
 	sigaction_safe(SIGPIPE,&ignore_sigaction,NULL);
 }
 
-void reset_stdio(void)
+/* On Unix, shared fds such as stdin cannot be set to non-blocking mode
+(http://homepages.tesco.net/J.deBoynePollard/FGA/dont-set-shared-file-descriptors-to-non-blocking-mode.html)
+so we kludge around this by spawning a thread, which waits on a control pipe
+for a signal, upon receiving this signal it reads one block of data from stdin
+and writes it to a data pipe. Upon completion, it writes a 4-byte integer to
+the size pipe, indicating how much data was written to the data pipe.
+
+The read end of the size pipe can be set to non-blocking. */
+__attribute__((visibility("default"))) int stdin_read;
+__attribute__((visibility("default"))) int stdin_write;
+
+__attribute__((visibility("default"))) int control_read;
+__attribute__((visibility("default"))) int control_write;
+
+__attribute__((visibility("default"))) int size_read;
+__attribute__((visibility("default"))) int size_write;
+
+void safe_close(int fd)
 {
-	fcntl(0,F_SETFL,0);
+	if(close(fd) < 0)
-	fcntl(1,F_SETFL,0);
+		fatal_error("error closing fd",errno);
 }
 
-void open_console(void) { }
+bool check_write(int fd, void *data, size_t size)
+{
+	if(write(fd,data,size) == size)
+		return true;
+	else
+	{
+		if(errno == EINTR)
+			return check_write(fd,data,size);
+		else
+			return false;
+	}
+}
+
+void safe_write(int fd, void *data, size_t size)
+{
+	if(!check_write(fd,data,size))
+		fatal_error("error writing fd",errno);
+}
+
+void safe_read(int fd, void *data, size_t size)
+{
+	if(read(fd,data,size) != size)
+		fatal_error("error reading fd",errno);
+}
+
+void *stdin_loop(void *arg)
+{
+	unsigned char buf[4096];
+	bool loop_running = true;
+
+	while(loop_running)
+	{
+		safe_read(control_read,buf,1);
+		if(buf[0] != 'X')
+			fatal_error("stdin_loop: bad data on control fd",buf[0]);
+
+		for(;;)
+		{
+			size_t bytes = read(0,buf,sizeof(buf));
+			if(bytes < 0)
+			{
+				if(errno == EINTR)
+					continue;
+				else
+				{
+					loop_running = false;
+					break;
+				}
+			}
+			else if(bytes >= 0)
+			{
+				safe_write(size_write,&bytes,sizeof(bytes));
+
+				if(write(stdin_write,buf,bytes) != bytes)
+					loop_running = false;
+				break;
+			}
+		}
+	}
+
+
+	safe_close(stdin_write);
+	safe_close(control_write);
+
+	return NULL;
+}
+
+void open_console(void)
+{
+	int filedes[2];
+
+	if(pipe(filedes) < 0)
+		fatal_error("Error opening control pipe",errno);
+
+	control_read = filedes[0];
+	control_write = filedes[1];
+
+	if(pipe(filedes) < 0)
+		fatal_error("Error opening size pipe",errno);
+
+	size_read = filedes[0];
+	size_write = filedes[1];
+
+	if(pipe(filedes) < 0)
+		fatal_error("Error opening stdin pipe",errno);
+
+	stdin_read = filedes[0];
+	stdin_write = filedes[1];
+
+	start_thread(stdin_loop);
+}
+
+DLLEXPORT void wait_for_stdin(void)
+{
+	if(write(control_write,"X",1) != 1)
+	{
+		if(errno == EINTR)
+			wait_for_stdin();
+		else
+			fatal_error("Error writing control fd",errno);
+	}
+}

vm/os-unix.h

@@ -38,5 +38,4 @@ void dump_stack_signal(int signal, siginfo_t* siginfo, void* uap);
 s64 current_millis(void);
 void sleep_millis(CELL msec);
 
-void reset_stdio(void);
 void open_console(void);

vm/os-windows.h

@@ -47,5 +47,3 @@ long getpagesize (void);
 
 s64 current_millis(void);
 
-INLINE void reset_stdio(void) {}
-