* win32/win32.c (collect_file_fd): rename from extract_file_fd.

* win32/win32.c (extract_pipe_fd, peek_pipe): new functions.

* win32/win32.c (rb_w32_select): check pipes by polling them.


git-svn-id: svn+ssh://ci.ruby-lang.org/ruby/trunk@9159 b2dd03c8-39d4-4d8f-98ff-823fe69b080e

ChangeLog

@@ -1,3 +1,11 @@
+Wed Sep 14 23:28:28 2005  NAKAMURA Usaku  <usa@ruby-lang.org>
+
+	* win32/win32.c (collect_file_fd): rename from extract_file_fd.
+
+	* win32/win32.c (extract_pipe_fd, peek_pipe): new functions.
+
+	* win32/win32.c (rb_w32_select): check pipes by polling them.
+
 Wed Sep 14 22:40:26 2005  Nobuyoshi Nakada  <nobu@ruby-lang.org>
 
 	* dir.c (ruby_glob): glob function not using ruby exception system.

io.c

@@ -4850,11 +4850,7 @@ rb_io_s_pipe(VALUE klass)
     int pipes[2], state;
     VALUE r, w, args[3];
 
-#ifdef _WIN32
-    if (_pipe(pipes, 1024, O_BINARY) == -1)
-#else
     if (pipe(pipes) == -1)
-#endif
 	rb_sys_fail(0);
 
     args[0] = klass;

win32/win32.c

@@ -1910,7 +1910,7 @@ rb_w32_fdisset(int fd, fd_set *set)
 static int NtSocketsInitialized = 0;
 
 static int
-extract_file_fd(fd_set *set, fd_set *fileset)
+collect_file_fd(fd_set *set, fd_set *fileset)
 {
     int idx;
 
@@ -1939,6 +1939,76 @@ extract_file_fd(fd_set *set, fd_set *fileset)
     return fileset->fd_count;
 }
 
+static int
+extract_pipe_fd(fd_set *set, fd_set *fileset)
+{
+    int idx;
+
+    fileset->fd_count = 0;
+    if (!set)
+	return 0;
+    for (idx = 0; idx < set->fd_count; idx++) {
+	DWORD type;
+	int fd = set->fd_array[idx];
+	RUBY_CRITICAL(type = GetFileType((HANDLE)fd));
+
+	if (type == FILE_TYPE_PIPE) {
+	    int i;
+
+	    for (i = 0; i < fileset->fd_count; i++) {
+		if (fileset->fd_array[i] == fd) {
+		    break;
+		}
+	    }
+	    if (i == fileset->fd_count) {
+		if (fileset->fd_count < FD_SETSIZE) {
+		    fileset->fd_array[i] = fd;
+		    fileset->fd_count++;
+		}
+	    }
+
+	    for (i = idx; i < set->fd_count - 1; i++) {
+		set->fd_array[i] = set->fd_array[i + 1];
+		i++;
+	    }
+	    set->fd_count--;
+	    idx--;
+	}
+    }
+    return fileset->fd_count;
+}
+
+static int
+peek_pipe(fd_set *set, fd_set *fileset)
+{
+    int idx;
+
+    fileset->fd_count = 0;
+    for (idx = 0; idx < set->fd_count; idx++) {
+	DWORD n;
+	int fd = set->fd_array[idx];
+	if (PeekNamedPipe((HANDLE)fd, NULL, 0, NULL, &n, NULL) && n > 0) {
+	    int i;
+
+	    for (i = 0; i < fileset->fd_count; i++) {
+		if (fileset->fd_array[i] == fd) {
+		    break;
+		}
+	    }
+	    if (i == fileset->fd_count) {
+		if (fileset->fd_count < FD_SETSIZE) {
+		    fileset->fd_array[i] = fd;
+		    fileset->fd_count++;
+		}
+	    }
+	}
+    }
+
+    return fileset->fd_count;
+}
+
+#define PIPE_PEEK_INTERVAL (10 * 1000)	/* usec */
+
 long 
 rb_w32_select(int nfds, fd_set *rd, fd_set *wr, fd_set *ex,
 	      struct timeval *timeout)
@@ -1946,10 +2016,14 @@ rb_w32_select(int nfds, fd_set *rd, fd_set *wr, fd_set *ex,
     long r;
     fd_set file_rd;
     fd_set file_wr;
+    fd_set pipe_rd;
 #ifdef USE_INTERRUPT_WINSOCK
     fd_set trap;
 #endif /* USE_INTERRUPT_WINSOCK */
     int file_nfds;
+    int pipe_nfds;
+    struct timeval remainder;
+    struct timeval wait;
 
     if (!NtSocketsInitialized) {
 	StartSockets();
@@ -1963,11 +2037,13 @@ rb_w32_select(int nfds, fd_set *rd, fd_set *wr, fd_set *ex,
 	Sleep(timeout->tv_sec * 1000 + timeout->tv_usec / 1000);
 	return 0;
     }
-    file_nfds = extract_file_fd(rd, &file_rd);
+    pipe_nfds = extract_pipe_fd(rd, &pipe_rd);
-    file_nfds += extract_file_fd(wr, &file_wr);
+    nfds -= pipe_nfds;
-    if (file_nfds)
+    file_nfds = collect_file_fd(rd, &file_rd);
-    {
+    file_nfds += collect_file_fd(wr, &file_wr);
-	// assume normal files are always readable/writable
+    if (file_nfds) {
+	// assume normal files are always readable/writable,
+	// and pipes are always writable.
 	// fake read/write fd_set and return value
 	if (rd) *rd = file_rd;
 	if (wr) *wr = file_wr;
@@ -1985,12 +2061,52 @@ rb_w32_select(int nfds, fd_set *rd, fd_set *wr, fd_set *ex,
     ex = &trap;
 #endif /* USE_INTERRUPT_WINSOCK */
 
-    RUBY_CRITICAL({
+    if (timeout)
-	r = select(nfds, rd, wr, ex, timeout);
+	remainder = *timeout;
+    wait.tv_sec = 0;
+    RUBY_CRITICAL(do{
+	if (pipe_nfds) {
+	    r = peek_pipe(&pipe_rd, &file_rd);
+	    if (r > 0) {
+		if (rd) *rd = file_rd;
+		if (wr) wr->fd_count = 0;
+		break;
+	    }
+	}
+	if (timeout && remainder.tv_sec == 0 &&
+	    remainder.tv_usec < PIPE_PEEK_INTERVAL) {
+	    wait.tv_usec = remainder.tv_usec;
+	    remainder.tv_usec = 0;
+	}
+	else if (timeout && remainder.tv_usec < PIPE_PEEK_INTERVAL) {
+	    remainder.tv_sec--;
+	    remainder.tv_usec += 1000 * 1000 * 1000L- PIPE_PEEK_INTERVAL;
+	    wait.tv_usec = PIPE_PEEK_INTERVAL;
+	}
+	else {
+	    if (timeout)
+		remainder.tv_usec -= PIPE_PEEK_INTERVAL;
+	    wait.tv_usec = PIPE_PEEK_INTERVAL;
+	}
+	if (nfds == 0) {
+	    Sleep(wait.tv_sec * 1000 + wait.tv_usec / 1000);
+	}
+	else {
+	    r = select(nfds, rd, wr, ex, &wait);
 	    if (r == SOCKET_ERROR) {
 		errno = map_errno(WSAGetLastError());
+		r = -1;
+		break;
 	    }
-    });
+	    else if (r > 0) {
+		break;
+	    }
+	    else if (timeout &&
+		     remainder.tv_sec == 0 && remainder.tv_usec == 0) {
+		break;
+	    }
+	}
+    } while (0));
     return r;
 }
 

win32/win32.h

@@ -93,6 +93,7 @@ extern "C++" {
 #define strcasecmp(s1, s2)	stricmp(s1, s2)
 #define strncasecmp(s1, s2, n)	strnicmp(s1, s2, n)
 
+#define pipe(p)			_pipe(p, 2048L, O_BINARY)
 #define close(h)		rb_w32_close(h)
 #define fclose(f)		rb_w32_fclose(f)
 #define getpid()		rb_w32_getpid()