8282726: java/net/vthread/BlockingSocketOps.java timeout/hang intermittently on Windows

Reviewed-by: djelinski

src/java.base/aix/classes/sun/nio/ch/PollsetPoller.java

@@ -66,7 +66,7 @@ class PollsetPoller extends Poller {
     }
 
     @Override
-    void implDeregister(int fd) {
+    void implDeregister(int fd, boolean polled) {
         int ret = Pollset.pollsetCtl(setid, Pollset.PS_DELETE, fd, 0);
         assert ret == 0;
     }

src/java.base/linux/classes/sun/nio/ch/EPollPoller.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017, 2022, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2023, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -62,8 +62,11 @@ class EPollPoller extends Poller {
     }
 
     @Override
-    void implDeregister(int fdVal) {
+    void implDeregister(int fdVal, boolean polled) {
-        EPoll.ctl(epfd, EPOLL_CTL_DEL, fdVal, 0);
+        // event is disabled if already polled
+        if (!polled) {
+            EPoll.ctl(epfd, EPOLL_CTL_DEL, fdVal, 0);
+        }
     }
 
     @Override

src/java.base/macosx/classes/sun/nio/ch/KQueuePoller.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2017, 2022, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2017, 2023, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -56,8 +56,11 @@ class KQueuePoller extends Poller {
     }
 
     @Override
-    void implDeregister(int fdVal) {
+    void implDeregister(int fdVal, boolean polled) {
-        KQueue.register(kqfd, fdVal, filter, EV_DELETE);
+        // event was deleted if already polled
+        if (!polled) {
+            KQueue.register(kqfd, fdVal, filter, EV_DELETE);
+        }
     }
 
     @Override

src/java.base/share/classes/sun/nio/ch/Poller.java

@@ -94,14 +94,16 @@ abstract class Poller {
     }
 
     /**
-     * Register the file descriptor.
+     * Register the file descriptor. The registration is "one shot", meaning it should
+     * be polled at most once.
      */
     abstract void implRegister(int fdVal) throws IOException;
 
     /**
      * Deregister the file descriptor.
+     * @param polled true if the file descriptor has already been polled
      */
-    abstract void implDeregister(int fdVal);
+    abstract void implDeregister(int fdVal, boolean polled);
 
     /**
      * Poll for events. The {@link #polled(int)} method is invoked for each
@@ -182,23 +184,23 @@ abstract class Poller {
     }
 
     /**
-     * Registers the file descriptor.
+     * Registers the file descriptor to be polled at most once when the file descriptor
+     * is ready for I/O.
      */
     private void register(int fdVal) throws IOException {
-        Thread previous = map.putIfAbsent(fdVal, Thread.currentThread());
+        Thread previous = map.put(fdVal, Thread.currentThread());
         assert previous == null;
         implRegister(fdVal);
     }
 
     /**
-     * Deregister the file descriptor, a no-op if already polled.
+     * Deregister the file descriptor so that the file descriptor is not polled.
      */
     private void deregister(int fdVal) {
         Thread previous = map.remove(fdVal);
-        assert previous == null || previous == Thread.currentThread();
+        boolean polled = (previous == null);
-        if (previous != null) {
+        assert polled || previous == Thread.currentThread();
-            implDeregister(fdVal);
+        implDeregister(fdVal, polled);
-        }
     }
 
     /**

src/java.base/windows/classes/sun/nio/ch/PipeImpl.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2002, 2022, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2002, 2023, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -129,14 +129,7 @@ class PipeImpl
                         }
 
                         // Establish connection (assume connection is eagerly accepted)
-                        if (sa instanceof InetSocketAddress
+                        sc1 = SocketChannel.open(sa);
-                                && Thread.currentThread().isVirtual()) {
-                            // workaround "lost event" issue on older releases of Windows
-                            sc1 = SocketChannel.open();
-                            sc1.socket().connect(sa, 10_000);
-                        } else {
-                            sc1 = SocketChannel.open(sa);
-                        }
                         RANDOM_NUMBER_GENERATOR.nextBytes(secret.array());
                         do {
                             sc1.write(secret);

src/java.base/windows/classes/sun/nio/ch/WEPollPoller.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020, 2022, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2020, 2023, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
@@ -46,16 +46,13 @@ class WEPollPoller extends Poller {
 
     @Override
     void implRegister(int fdVal) throws IOException {
-        // re-arm
+        int err = WEPoll.ctl(handle, EPOLL_CTL_ADD, fdVal, (event | EPOLLONESHOT));
-        int err = WEPoll.ctl(handle, EPOLL_CTL_MOD, fdVal, (event | EPOLLONESHOT));
-        if (err == ENOENT)
-            err = WEPoll.ctl(handle, EPOLL_CTL_ADD, fdVal, (event | EPOLLONESHOT));
         if (err != 0)
             throw new IOException("epoll_ctl failed: " + err);
     }
 
     @Override
-    void implDeregister(int fdVal) {
+    void implDeregister(int fdVal, boolean polled) {
         WEPoll.ctl(handle, EPOLL_CTL_DEL, fdVal, 0);
     }
 

test/jdk/java/net/vthread/BlockingSocketOps.java

@@ -685,7 +685,7 @@ class BlockingSocketOps {
                 Socket s1 = new Socket();
                 Socket s2;
                 try {
-                    s1.connect(listener.getLocalSocketAddress(), 10_000);
+                    s1.connect(listener.getLocalSocketAddress());
                     s2 = listener.accept();
                 } catch (IOException ioe) {
                     s1.close();

test/jdk/java/nio/channels/vthread/BlockingChannelOps.java

@@ -808,7 +808,7 @@ class BlockingChannelOps {
                 SocketChannel sc1 = SocketChannel.open();
                 SocketChannel sc2 = null;
                 try {
-                    sc1.socket().connect(listener.getLocalAddress(), 10_000);
+                    sc1.socket().connect(listener.getLocalAddress());
                     sc2 = listener.accept();
                 } catch (IOException ioe) {
                     sc1.close();