8231259: (dc) DatagramChannel::disconnect re-binds socket to the wildcard address (macOS)

Reviewed-by: dfuchs, chegar

src/java.base/share/classes/java/nio/channels/spi/AbstractSelectableChannel.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2000, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2000, 2019, 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
@@ -34,6 +34,8 @@ import java.nio.channels.IllegalSelectorException;
 import java.nio.channels.SelectableChannel;
 import java.nio.channels.SelectionKey;
 import java.nio.channels.Selector;
+import java.util.Arrays;
+import java.util.function.Consumer;
 
 
 /**
@@ -171,6 +173,20 @@ public abstract class AbstractSelectableChannel
         }
     }
 
+    /**
+     * Invokes an action for each key.
+     *
+     * This method is invoked by DatagramChannelImpl::disconnect.
+     */
+    private void forEach(Consumer<SelectionKey> action) {
+        synchronized (keyLock) {
+            SelectionKey[] keys = this.keys;
+            if (keys != null) {
+                Arrays.stream(keys).filter(k -> k != null).forEach(action::accept);
+            }
+        }
+    }
+
     /**
      * Registers this channel with the given selector, returning a selection key.
      *

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

@@ -31,6 +31,7 @@ import java.io.UncheckedIOException;
 import java.lang.invoke.MethodHandles;
 import java.lang.invoke.VarHandle;
 import java.lang.ref.Cleaner.Cleanable;
+import java.lang.reflect.Method;
 import java.net.DatagramSocket;
 import java.net.Inet4Address;
 import java.net.Inet6Address;
@@ -54,12 +55,18 @@ import java.nio.channels.IllegalBlockingModeException;
 import java.nio.channels.MembershipKey;
 import java.nio.channels.NotYetConnectedException;
 import java.nio.channels.SelectionKey;
+import java.nio.channels.spi.AbstractSelectableChannel;
 import java.nio.channels.spi.SelectorProvider;
+import java.security.AccessController;
+import java.security.PrivilegedExceptionAction;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Consumer;
 
 import jdk.internal.ref.CleanerFactory;
 import sun.net.ResourceManager;
@@ -113,10 +120,13 @@ class DatagramChannelImpl
     private long readerThread;
     private long writerThread;
 
-    // Binding and remote address (when connected)
+    // Local and remote (connected) address
     private InetSocketAddress localAddress;
     private InetSocketAddress remoteAddress;
 
+    // Local address prior to connecting
+    private InetSocketAddress initialLocalAddress;
+
     // Socket adaptor, created lazily
     private static final VarHandle SOCKET;
     static {
@@ -1103,6 +1113,9 @@ class DatagramChannelImpl
                         bindInternal(null);
                     }
 
+                    // capture local address before connect
+                    initialLocalAddress = localAddress;
+
                     int n = Net.connect(family,
                                         fd,
                                         isa.getAddress(),
@@ -1160,21 +1173,19 @@ class DatagramChannelImpl
                     remoteAddress = null;
                     state = ST_UNCONNECTED;
 
-                    // check whether rebind is needed
-                    InetSocketAddress isa = Net.localAddress(fd);
-                    if (isa.getPort() == 0) {
-                        // On Linux, if bound to ephemeral port,
-                        // disconnect does not preserve that port.
-                        // In this case, try to rebind to the previous port.
-                        int port = localAddress.getPort();
-                        localAddress = isa; // in case Net.bind fails
-                        Net.bind(family, fd, isa.getAddress(), port);
-                        isa = Net.localAddress(fd); // refresh address
-                        assert isa.getPort() == port;
+                    // refresh localAddress, should be same as it was prior to connect
+                    localAddress = Net.localAddress(fd);
+                    try {
+                        if (!localAddress.equals(initialLocalAddress)) {
+                            // Workaround connect(2) issues on Linux and macOS
+                            repairSocket(initialLocalAddress);
+                            assert (localAddress != null)
+                                    && localAddress.equals(Net.localAddress(fd))
+                                    && localAddress.equals(initialLocalAddress);
+                        }
+                    } finally {
+                        initialLocalAddress = null;
                     }
-
-                    // refresh localAddress
-                    localAddress = isa;
                 }
             } finally {
                 writeLock.unlock();
@@ -1185,6 +1196,134 @@ class DatagramChannelImpl
         return this;
     }
 
+    /**
+     * "Repair" the channel's socket after a disconnect that didn't restore the
+     * local address.
+     *
+     * On Linux, connect(2) dissolves the association but changes the local port
+     * to 0 when it was initially bound to an ephemeral port. The workaround here
+     * is to rebind to the original port.
+     *
+     * On macOS, connect(2) dissolves the association but rebinds the socket to
+     * the wildcard address when it was initially bound to a specific address.
+     * The workaround here is to re-create the socket.
+     */
+    private void repairSocket(InetSocketAddress target)
+        throws IOException
+    {
+        assert Thread.holdsLock(stateLock);
+
+        // Linux: try to bind the socket to the original address/port
+        if (localAddress.getPort() == 0) {
+            assert localAddress.getAddress().equals(target.getAddress());
+            Net.bind(family, fd, target.getAddress(), target.getPort());
+            localAddress = Net.localAddress(fd);
+            return;
+        }
+
+        // capture the value of all existing socket options
+        Map<SocketOption<?>, Object> map = new HashMap<>();
+        for (SocketOption<?> option : supportedOptions()) {
+            Object value = getOption(option);
+            if (value != null) {
+                map.put(option, value);
+            }
+        }
+
+        // macOS: re-create the socket.
+        FileDescriptor newfd = Net.socket(family, false);
+        try {
+            // copy the socket options that are protocol family agnostic
+            for (Map.Entry<SocketOption<?>, Object> e : map.entrySet()) {
+                SocketOption<?> option = e.getKey();
+                if (SocketOptionRegistry.findOption(option, Net.UNSPEC) != null) {
+                    Object value = e.getValue();
+                    try {
+                        Net.setSocketOption(newfd, Net.UNSPEC, option, value);
+                    } catch (IOException ignore) { }
+                }
+            }
+
+            // copy the blocking mode
+            if (!isBlocking()) {
+                IOUtil.configureBlocking(newfd, false);
+            }
+
+            // dup this channel's socket to the new socket. If this succeeds then
+            // fd will reference the new socket. If it fails then it will still
+            // reference the old socket.
+            nd.dup(newfd, fd);
+        } finally {
+            // release the file descriptor
+            nd.close(newfd);
+        }
+
+        // bind to the original local address
+        try {
+            Net.bind(family, fd, target.getAddress(), target.getPort());
+        } catch (IOException ioe) {
+            // bind failed, socket is left unbound
+            localAddress = null;
+            throw ioe;
+        }
+
+        // restore local address
+        localAddress = Net.localAddress(fd);
+
+        // restore all socket options (including those set in first pass)
+        for (Map.Entry<SocketOption<?>, Object> e : map.entrySet()) {
+            @SuppressWarnings("unchecked")
+            SocketOption<Object> option = (SocketOption<Object>) e.getKey();
+            Object value = e.getValue();
+            try {
+                setOption(option, value);
+            } catch (IOException ignore) { }
+        }
+
+        // restore multicast group membership
+        MembershipRegistry registry = this.registry;
+        if (registry != null) {
+            registry.forEach(k -> {
+                if (k instanceof MembershipKeyImpl.Type6) {
+                    MembershipKeyImpl.Type6 key6 = (MembershipKeyImpl.Type6) k;
+                    Net.join6(fd, key6.groupAddress(), key6.index(), key6.source());
+                } else {
+                    MembershipKeyImpl.Type4 key4 = (MembershipKeyImpl.Type4) k;
+                    Net.join4(fd, key4.groupAddress(), key4.interfaceAddress(), key4.source());
+                }
+            });
+        }
+
+        // reset registration in all Selectors that this channel is registered with
+        AbstractSelectableChannels.forEach(this, SelectionKeyImpl::reset);
+    }
+
+    /**
+     * Defines static methods to access AbstractSelectableChannel non-public members.
+     */
+    private static class AbstractSelectableChannels {
+        private static final Method FOREACH;
+        static {
+            try {
+                PrivilegedExceptionAction<Method> pae = () -> {
+                    Method m = AbstractSelectableChannel.class.getDeclaredMethod("forEach", Consumer.class);
+                    m.setAccessible(true);
+                    return m;
+                };
+                FOREACH = AccessController.doPrivileged(pae);
+            } catch (Exception e) {
+                throw new InternalError(e);
+            }
+        }
+        static void forEach(AbstractSelectableChannel ch, Consumer<SelectionKeyImpl> action) {
+            try {
+                FOREACH.invoke(ch, action);
+            } catch (Exception e) {
+                throw new InternalError(e);
+            }
+        }
+    }
+
     /**
      * Joins channel's socket to the given group/interface and
      * optional source address.

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2008, 2009, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2008, 2019, 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
@@ -25,10 +25,14 @@
 
 package sun.nio.ch;
 
-import java.nio.channels.*;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
-import java.util.*;
+import java.nio.channels.MembershipKey;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
 
 /**
  * Simple registry of membership keys for a MulticastChannel.
@@ -38,8 +42,8 @@ import java.util.*;
 
 class MembershipRegistry {
 
-    // map multicast group to keys
-    private Map<InetAddress,List<MembershipKeyImpl>> groups = null;
+    // map multicast group to list of keys
+    private Map<InetAddress, List<MembershipKeyImpl>> groups;
 
     MembershipRegistry() {
     }
@@ -116,16 +120,29 @@ class MembershipRegistry {
         }
     }
 
+    @FunctionalInterface
+    interface ThrowingConsumer<T, X extends Throwable> {
+        void accept(T action) throws X;
+    }
+
     /**
-     * Invalidate all keys in the registry
+     * Invoke an action for each key in the registry
      */
-    void invalidateAll() {
+     <X extends Throwable>
+     void forEach(ThrowingConsumer<MembershipKeyImpl, X> action) throws X {
         if (groups != null) {
-            for (InetAddress group: groups.keySet()) {
-                for (MembershipKeyImpl key: groups.get(group)) {
-                    key.invalidate();
+            for (List<MembershipKeyImpl> keys : groups.values()) {
+                for (MembershipKeyImpl key : keys) {
+                    action.accept(key);
                 }
             }
         }
     }
+
+    /**
+     * Invalidate all keys in the registry
+     */
+    void invalidateAll() {
+        forEach(MembershipKeyImpl::invalidate);
+    }
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2000, 2019, 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
@@ -25,15 +25,15 @@
 
 package sun.nio.ch;
 
-import java.io.*;
+import java.io.FileDescriptor;
+import java.io.IOException;
 
 /**
  * Allows different platforms to call different native methods
  * for read and write operations.
  */
 
-abstract class NativeDispatcher
-{
+abstract class NativeDispatcher {
 
     abstract int read(FileDescriptor fd, long address, int len)
         throws IOException;
@@ -77,4 +77,13 @@ abstract class NativeDispatcher
         // Do nothing by default; this is only needed on Unix
     }
 
+    /**
+     * Duplicates a file descriptor.
+     * @param fd1 the file descriptor to duplicate
+     * @param fd2 the new file descriptor, the socket or file that it is connected
+     *            to will be closed by this method
+     */
+    void dup(FileDescriptor fd1, FileDescriptor fd2) throws IOException {
+        throw new UnsupportedOperationException();
+    }
 }

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2000, 2018, Oracle and/or its affiliates. All rights reserved.
+ * Copyright (c) 2000, 2019, 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
@@ -58,6 +58,9 @@ public final class SelectionKeyImpl
     // registered events in kernel, used by some Selector implementations
     private int registeredEvents;
 
+    // registered events need to be reset, used by some Selector implementations
+    private volatile boolean reset;
+
     // index of key in pollfd array, used by some Selector implementations
     private int index;
 
@@ -184,6 +187,26 @@ public final class SelectionKeyImpl
         index = i;
     }
 
+    /**
+     * Sets the reset flag, re-queues the key, and wakeups up the Selector
+     */
+    void reset() {
+        reset = true;
+        selector.setEventOps(this);
+        selector.wakeup();
+    }
+
+    /**
+     * Clears the reset flag, returning the previous value of the flag
+     */
+    boolean getAndClearReset() {
+        assert Thread.holdsLock(selector);
+        boolean r = reset;
+        if (r)
+            reset = false;
+        return r;
+    }
+
     @Override
     public String toString() {
         StringBuilder sb = new StringBuilder();