[ruby-dev:27468] Re: 1.8.4 relrase plan?

山本です。

>| とりあえずpreview1出してみませんか、そろそろ。
>
>native win32な関係者の皆さんにお伺いしたいんですけど、1.9から
>rb_w32_select(とそれに付随するもの)をbackportしてもいいでしょ
>うか?
>やるならpreview1の出る前しかないと思うので。

うーん、まだ時期尚早なんじゃないかという気がします。現在の HEAD の
ものをそのまま持ってきて何度も testrb drb をしてると、固まることが
あります。私のいじった最後のリビジョンだと固まらないようですが、
それすら自信がありません。

どうも do_select の中の catch_interrupt があると高確率で固まるよう
ですが、そもそも RUBY_CRITICAL がシングルスレッドの ruby でなぜ必要なのか
とか、catch_interrupt が何をしているのかとかまったくわかってないので・・・

とりあえず、HEAD の select をバックポートするパッチをつけておきます。

* gets でもスレッドが切り替わる

* パイプでもスレッドが切り替わる。ただし、バッファサイズ以上書き込んだ場合
  には今まで同様ブロックしてしまう。

* ソケットとコンソールを混ぜて使ったりした場合、パフォーマンスが向上している

  極端な例では

    require "net/http"

    Thread.new {
      loop do
	STDOUT.write(".") # busy on console (this is worst case though)
      end
    }

    # socket operation took long time. (sometimes timed out)
    Net::HTTP.start("www.ruby-lang.org") do |http|
	http.get("/cgi-bin/cvsweb.cgi/ruby/array.c?rev=1.179")
    end

Index: win32.c
===================================================================
RCS file: /src/ruby/win32/win32.c,v
retrieving revision 1.103.2.33
diff -u -w -b -p -r1.103.2.33 win32.c
--- win32.c	14 Oct 2005 13:09:11 -0000	1.103.2.33
+++ win32.c	20 Oct 2005 03:53:48 -0000
@@ -1866,33 +1866,152 @@ rb_w32_fdisset(int fd, fd_set *set)
 static int NtSocketsInitialized = 0;
 
 static int
-extract_file_fd(fd_set *set, fd_set *fileset)
+extract_fd(fd_set *dst, fd_set *src, int (*func)(SOCKET))
 {
-    int idx;
+    int s = 0;
+    if (!src || !dst) return 0;
 
-    fileset->fd_count = 0;
-    if (!set)
-	return 0;
-    for (idx = 0; idx < set->fd_count; idx++) {
-	SOCKET fd = set->fd_array[idx];
+    while (s < src->fd_count) {
+        SOCKET fd = src->fd_array[s];
 
-	if (!is_socket(fd)) {
-	    int i;
+	if (!func || (*func)(fd)) { /* move it to dst */
+	    int d;
 
-	    for (i = 0; i < fileset->fd_count; i++) {
-		if (fileset->fd_array[i] == fd) {
-		    break;
+	    for (d = 0; d < dst->fd_count; d++) {
+		if (dst->fd_array[d] == fd) break;
+	    }
+	    if (d == dst->fd_count && dst->fd_count < FD_SETSIZE) {
+		dst->fd_array[dst->fd_count++] = fd;
+	    }
+	    memmove(
+		&src->fd_array[s],
+		&src->fd_array[s+1], 
+		sizeof(src->fd_array[0]) * (--src->fd_count - s));
+	}
+	else s++;
+    }
+
+    return dst->fd_count;
+}
+
+static int
+is_not_socket(SOCKET sock)
+{
+    return !is_socket(sock);
+}
+
+static int
+is_pipe(SOCKET sock) /* DONT call this for SOCKET! it clains it is PIPE. */
+{
+    int ret;
+
+    RUBY_CRITICAL(
+	ret = (GetFileType((HANDLE)sock) == FILE_TYPE_PIPE)
+    );
+
+    return ret;
+}
+
+static int
+is_readable_pipe(SOCKET sock) /* call this for pipe only */
+{
+    int ret;
+    DWORD n = 0;
+
+    RUBY_CRITICAL(
+	if (PeekNamedPipe((HANDLE)sock, NULL, 0, NULL, &n, NULL)) {
+	    ret = (n > 0);
+	}
+	else {
+	    ret = (GetLastError() == ERROR_BROKEN_PIPE); /* pipe was closed */
+	}
+    );
+
+    return ret;
+}
+
+static int
+is_console(SOCKET sock) /* DONT call this for SOCKET! */
+{
+    int ret;
+    DWORD n = 0;
+    INPUT_RECORD ir;
+
+    RUBY_CRITICAL(
+	ret = (PeekConsoleInput((HANDLE)sock, &ir, 1, &n))
+    );
+
+    return ret;
+}
+
+static int
+is_readable_console(SOCKET sock) /* call this for console only */
+{
+    int ret = 0;
+    DWORD n = 0;
+    INPUT_RECORD ir;
+
+    RUBY_CRITICAL(
+	if (PeekConsoleInput((HANDLE)sock, &ir, 1, &n) && n > 0) {
+	    if (ir.EventType == KEY_EVENT && ir.Event.KeyEvent.bKeyDown &&
+		ir.Event.KeyEvent.uChar.AsciiChar) {
+		ret = 1;
+	    }
+	    else {
+		ReadConsoleInput((HANDLE)sock, &ir, 1, &n);
+	    }
+	}
+    );
+
+    return ret;
+}
+
+static void catch_interrupt(void);
+static long 
+do_select(int nfds, fd_set *rd, fd_set *wr, fd_set *ex,
+            struct timeval *timeout)
+{
+    long r = 0;
+
+    if (nfds == 0) {
+	if (timeout)
+	    rb_w32_sleep(timeout->tv_sec * 1000 + timeout->tv_usec / 1000);
+	else
+	    rb_w32_sleep(INFINITE);
 		}
+    else {
+#if !USE_INTERRUPT_WINSOCK
+	int trap_immediate = rb_trap_immediate;
+#endif	/* !USE_INTERRUPT_WINSOCK */
+	RUBY_CRITICAL(
+	    r = select(nfds, rd, wr, ex, timeout);
+	    if (r == SOCKET_ERROR) {
+		errno = map_errno(WSAGetLastError());
+		r = -1;
 	    }
-	    if (i == fileset->fd_count) {
-		if (fileset->fd_count < FD_SETSIZE) {
-		    fileset->fd_array[i] = fd;
-		    fileset->fd_count++;
+	);
+#if !USE_INTERRUPT_WINSOCK
+	rb_trap_immediate = trap_immediate;
+	catch_interrupt();
+#endif	/* !USE_INTERRUPT_WINSOCK */
 		}
+
+    return r;
 	    }
+
+static int 
+subst(struct timeval *rest, const struct timeval *wait)
+{
+    while (rest->tv_usec < wait->tv_usec) {
+	if (rest->tv_sec <= wait->tv_sec) {
+	    return 0;
 	}
+	rest->tv_sec -= 1;
+	rest->tv_usec += 1000 * 1000;
     }
-    return fileset->fd_count;
+    rest->tv_sec -= wait->tv_sec;
+    rest->tv_usec -= wait->tv_usec;
+    return 1;
 }
 
 long 
@@ -1900,35 +2019,47 @@ rb_w32_select (int nfds, fd_set *rd, fd_
 	       struct timeval *timeout)
 {
     long r;
-    fd_set file_rd;
-    fd_set file_wr;
-#ifdef USE_INTERRUPT_WINSOCK
+    fd_set pipe_rd;
+    fd_set cons_rd;
+    fd_set else_rd;
+    fd_set else_wr;
+#if USE_INTERRUPT_WINSOCK
     fd_set trap;
+    int ret;
 #endif /* USE_INTERRUPT_WINSOCK */
-    int file_nfds;
+    int nonsock = 0;
 
+    if (nfds < 0 || (timeout && (timeout->tv_sec < 0 || timeout->tv_usec < 0))) {
+	errno = EINVAL;
+	return -1;
+    }
     if (!NtSocketsInitialized) {
 	StartSockets();
     }
+
+    // assume else_{rd,wr} (other than socket, pipe reader, console reader)
+    // are always readable/writable. but this implementation still has
+    // problem. if pipe's buffer is full, writing to pipe will block
+    // until some data is read from pipe. but ruby is single threaded system,
+    // so whole system will be blocked forever.
+
+    else_rd.fd_count = 0;
+    nonsock += extract_fd(&else_rd, rd, is_not_socket);
+
+    pipe_rd.fd_count = 0;
+    extract_fd(&pipe_rd, &else_rd, is_pipe); // should not call is_pipe for socket
+
+    cons_rd.fd_count = 0;
+    extract_fd(&cons_rd, &else_rd, is_console); // ditto
+
+    else_wr.fd_count = 0;
+    nonsock += extract_fd(&else_wr, wr, is_not_socket);
+
     r = 0;
     if (rd && rd->fd_count > r) r = rd->fd_count;
     if (wr && wr->fd_count > r) r = wr->fd_count;
     if (ex && ex->fd_count > r) r = ex->fd_count;
     if (nfds > r) nfds = r;
-    if (nfds == 0 && timeout) {
-	Sleep(timeout->tv_sec * 1000 + timeout->tv_usec / 1000);
-	return 0;
-    }
-    file_nfds = extract_file_fd(rd, &file_rd);
-    file_nfds += extract_file_fd(wr, &file_wr);
-    if (file_nfds)
-    {
-	// assume normal files are always readable/writable
-	// fake read/write fd_set and return value
-	if (rd) *rd = file_rd;
-	if (wr) *wr = file_wr;
-	return file_nfds;
-    }
 
 #if USE_INTERRUPT_WINSOCK
     if (ex)
@@ -1941,12 +2072,54 @@ rb_w32_select (int nfds, fd_set *rd, fd_
     ex = &trap;
 #endif /* USE_INTERRUPT_WINSOCK */
 
-    RUBY_CRITICAL({
-	r = select(nfds, rd, wr, ex, timeout);
-	if (r == SOCKET_ERROR) {
-	    errno = map_errno(WSAGetLastError());
+    if (nonsock) {
+	struct timeval rest;
+	struct timeval wait;
+	struct timeval zero;
+	if (timeout) rest = *timeout;
+	wait.tv_sec = 0; wait.tv_usec = 10 * 1000; // 10ms
+	zero.tv_sec = 0; zero.tv_usec = 0;         //  0ms
+	do {
+	    // modifying {else,pipe,cons}_rd is safe because
+	    // if they are modified, function returns immediately.
+	    extract_fd(&else_rd, &pipe_rd, is_readable_pipe);
+	    extract_fd(&else_rd, &cons_rd, is_readable_console);
+
+	    if (else_rd.fd_count || else_wr.fd_count) {
+		r = do_select(nfds, rd, wr, ex, &zero); // polling
+		if (r < 0) break; // XXX: should I ignore error and return signaled handles?
+		r += extract_fd(rd, &else_rd, NULL); // move all
+		r += extract_fd(wr, &else_wr, NULL); // move all
+		break;
 	}
-    });
+	    else {
+		fd_set orig_rd;
+		fd_set orig_wr;
+		fd_set orig_ex;
+		if (rd) orig_rd = *rd;
+		if (wr) orig_wr = *wr;
+		if (ex) orig_ex = *ex;
+		r = do_select(nfds, rd, wr, ex, &wait);
+		if (r != 0) break; // signaled or error
+		if (rd) *rd = orig_rd;
+		if (wr) *wr = orig_wr;
+		if (ex) *ex = orig_ex;
+	    }
+	} while (!timeout || subst(&rest, &wait));
+    }
+    else
+	r = do_select(nfds, rd, wr, ex, timeout);
+
+#if USE_INTERRUPT_WINSOCK
+    RUBY_CRITICAL(ret = __WSAFDIsSet((SOCKET)interrupted_event, ex));
+    if (ret) {
+	// In this case, we must restore all FDs. But this is only a test
+	// code, so we think about that later.
+	errno = EINTR;
+	r = -1;
+    }
+#endif /* USE_INTERRUPT_WINSOCK */
+
     return r;
 }