1 /*
2 * Copyright (c) 2001, 2021, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation. Oracle designates this
8 * particular file as subject to the "Classpath" exception as provided
9 * by Oracle in the LICENSE file that accompanied this code.
10 *
11 * This code is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 * version 2 for more details (a copy is included in the LICENSE file that
15 * accompanied this code).
16 *
17 * You should have received a copy of the GNU General Public License version
18 * 2 along with this work; if not, write to the Free Software Foundation,
19 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20 *
21 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22 * or visit www.oracle.com if you need additional information or have any
23 * questions.
24 */
25
26 package sun.nio.ch;
27
28 import java.io.FileDescriptor;
29 import java.io.IOException;
30 import java.io.UncheckedIOException;
31 import java.lang.invoke.MethodHandles;
32 import java.lang.invoke.VarHandle;
33 import java.lang.ref.Cleaner.Cleanable;
34 import java.lang.reflect.Method;
35 import java.net.DatagramSocket;
36 import java.net.Inet4Address;
37 import java.net.Inet6Address;
38 import java.net.InetAddress;
39 import java.net.InetSocketAddress;
40 import java.net.NetworkInterface;
41 import java.net.PortUnreachableException;
42 import java.net.ProtocolFamily;
43 import java.net.SocketAddress;
44 import java.net.SocketException;
45 import java.net.SocketOption;
46 import java.net.SocketTimeoutException;
47 import java.net.StandardProtocolFamily;
48 import java.net.StandardSocketOptions;
49 import java.nio.ByteBuffer;
50 import java.nio.channels.AlreadyBoundException;
51 import java.nio.channels.AlreadyConnectedException;
52 import java.nio.channels.AsynchronousCloseException;
53 import java.nio.channels.ClosedChannelException;
54 import java.nio.channels.DatagramChannel;
55 import java.nio.channels.IllegalBlockingModeException;
56 import java.nio.channels.MembershipKey;
57 import java.nio.channels.NotYetConnectedException;
58 import java.nio.channels.SelectionKey;
59 import java.nio.channels.spi.AbstractSelectableChannel;
60 import java.nio.channels.spi.SelectorProvider;
61 import java.security.AccessController;
62 import java.security.PrivilegedExceptionAction;
63 import java.util.Collections;
64 import java.util.HashMap;
65 import java.util.HashSet;
66 import java.util.Map;
67 import java.util.Objects;
68 import java.util.Set;
69 import java.util.concurrent.locks.ReentrantLock;
70 import java.util.function.Consumer;
71
72 import jdk.internal.ref.CleanerFactory;
73 import sun.net.ResourceManager;
74 import sun.net.ext.ExtendedSocketOptions;
75 import sun.net.util.IPAddressUtil;
76
77 /**
78 * An implementation of DatagramChannels.
79 */
80
81 class DatagramChannelImpl
82 extends DatagramChannel
83 implements SelChImpl
84 {
85 // Used to make native read and write calls
86 private static final NativeDispatcher nd = new DatagramDispatcher();
87
88 // true if interruptible (can be false to emulate legacy DatagramSocket)
89 private final boolean interruptible;
90
91 // The protocol family of the socket
92 private final ProtocolFamily family;
93
94 // Our file descriptor
95 private final FileDescriptor fd;
96 private final int fdVal;
97
98 // Native sockaddrs and cached InetSocketAddress for receive, protected by readLock
99 private NativeSocketAddress sourceSockAddr;
100 private NativeSocketAddress cachedSockAddr;
101 private InetSocketAddress cachedInetSocketAddress;
102
103 // Native sockaddr and cached objects for send, protected by writeLock
104 private final NativeSocketAddress targetSockAddr;
105 private InetSocketAddress previousTarget;
106 private int previousSockAddrLength;
107
108 // Cleaner to close file descriptor and free native socket address
109 private final Cleanable cleaner;
110
111 // Lock held by current reading or connecting thread
112 private final ReentrantLock readLock = new ReentrantLock();
113
114 // Lock held by current writing or connecting thread
115 private final ReentrantLock writeLock = new ReentrantLock();
116
117 // Lock held by any thread that modifies the state fields declared below
118 // DO NOT invoke a blocking I/O operation while holding this lock!
119 private final Object stateLock = new Object();
120
121 // -- The following fields are protected by stateLock
122
123 // State (does not necessarily increase monotonically)
124 private static final int ST_UNCONNECTED = 0;
125 private static final int ST_CONNECTED = 1;
126 private static final int ST_CLOSING = 2;
127 private static final int ST_CLOSED = 3;
128 private int state;
129
130 // IDs of native threads doing reads and writes, for signalling
131 private long readerThread;
132 private long writerThread;
133
134 // Local and remote (connected) address
135 private InetSocketAddress localAddress;
136 private InetSocketAddress remoteAddress;
137
138 // Local address prior to connecting
139 private InetSocketAddress initialLocalAddress;
140
141 // Socket adaptor, created lazily
142 private static final VarHandle SOCKET;
143 static {
144 try {
145 MethodHandles.Lookup l = MethodHandles.lookup();
146 SOCKET = l.findVarHandle(DatagramChannelImpl.class, "socket", DatagramSocket.class);
147 } catch (Exception e) {
148 throw new InternalError(e);
149 }
150 }
151 private volatile DatagramSocket socket;
152
153 // Multicast support
154 private MembershipRegistry registry;
155
156 // set true when socket is bound and SO_REUSEADDRESS is emulated
157 private boolean reuseAddressEmulated;
158
159 // set true/false when socket is already bound and SO_REUSEADDR is emulated
160 private boolean isReuseAddress;
161
162 // -- End of fields protected by stateLock
163
164
165 DatagramChannelImpl(SelectorProvider sp, boolean interruptible) throws IOException {
166 this(sp, (Net.isIPv6Available()
167 ? StandardProtocolFamily.INET6
168 : StandardProtocolFamily.INET),
169 interruptible);
170 }
171
172 DatagramChannelImpl(SelectorProvider sp, ProtocolFamily family, boolean interruptible)
173 throws IOException
174 {
175 super(sp);
176
177 Objects.requireNonNull(family, "'family' is null");
178 if ((family != StandardProtocolFamily.INET) &&
179 (family != StandardProtocolFamily.INET6)) {
180 throw new UnsupportedOperationException("Protocol family not supported");
181 }
182 if (family == StandardProtocolFamily.INET6 && !Net.isIPv6Available()) {
183 throw new UnsupportedOperationException("IPv6 not available");
184 }
185
186 FileDescriptor fd = null;
187 NativeSocketAddress[] sockAddrs = null;
188
189 ResourceManager.beforeUdpCreate();
190 boolean initialized = false;
191 try {
192 this.interruptible = interruptible;
193 this.family = family;
194 this.fd = fd = Net.socket(family, false);
195 this.fdVal = IOUtil.fdVal(fd);
196
197 sockAddrs = NativeSocketAddress.allocate(3);
198 readLock.lock();
199 try {
200 this.sourceSockAddr = sockAddrs[0];
201 this.cachedSockAddr = sockAddrs[1];
202 } finally {
203 readLock.unlock();
204 }
205 this.targetSockAddr = sockAddrs[2];
206
207 initialized = true;
208 } finally {
209 if (!initialized) {
210 if (sockAddrs != null) NativeSocketAddress.freeAll(sockAddrs);
211 if (fd != null) nd.close(fd);
212 ResourceManager.afterUdpClose();
213 }
214 }
215
216 Runnable releaser = releaserFor(fd, sockAddrs);
217 this.cleaner = CleanerFactory.cleaner().register(this, releaser);
218 }
219
220 DatagramChannelImpl(SelectorProvider sp, FileDescriptor fd)
221 throws IOException
222 {
223 super(sp);
224
225 NativeSocketAddress[] sockAddrs = null;
226
227 ResourceManager.beforeUdpCreate();
228 boolean initialized = false;
229 try {
230 this.interruptible = true;
231 this.family = Net.isIPv6Available()
232 ? StandardProtocolFamily.INET6
233 : StandardProtocolFamily.INET;
234 this.fd = fd;
235 this.fdVal = IOUtil.fdVal(fd);
236
237 sockAddrs = NativeSocketAddress.allocate(3);
238 readLock.lock();
239 try {
240 this.sourceSockAddr = sockAddrs[0];
241 this.cachedSockAddr = sockAddrs[1];
242 } finally {
243 readLock.unlock();
244 }
245 this.targetSockAddr = sockAddrs[2];
246
247 initialized = true;
248 } finally {
249 if (!initialized) {
250 if (sockAddrs != null) NativeSocketAddress.freeAll(sockAddrs);
251 nd.close(fd);
252 ResourceManager.afterUdpClose();
253 }
254 }
255
256 Runnable releaser = releaserFor(fd, sockAddrs);
257 this.cleaner = CleanerFactory.cleaner().register(this, releaser);
258
259 synchronized (stateLock) {
260 this.localAddress = Net.localAddress(fd);
261 }
262 }
263
264 // @throws ClosedChannelException if channel is closed
265 private void ensureOpen() throws ClosedChannelException {
266 if (!isOpen())
267 throw new ClosedChannelException();
268 }
269
270 @Override
271 public DatagramSocket socket() {
272 DatagramSocket socket = this.socket;
273 if (socket == null) {
274 socket = DatagramSocketAdaptor.create(this);
275 if (!SOCKET.compareAndSet(this, null, socket)) {
276 socket = this.socket;
277 }
278 }
279 return socket;
280 }
281
282 @Override
283 public SocketAddress getLocalAddress() throws IOException {
284 synchronized (stateLock) {
285 ensureOpen();
286 // Perform security check before returning address
287 return Net.getRevealedLocalAddress(localAddress);
288 }
289 }
290
291 @Override
292 public SocketAddress getRemoteAddress() throws IOException {
293 synchronized (stateLock) {
294 ensureOpen();
295 return remoteAddress;
296 }
297 }
298
299 /**
300 * Returns the protocol family to specify to set/getSocketOption for the
301 * given socket option.
302 */
303 private ProtocolFamily familyFor(SocketOption<?> name) {
304 assert Thread.holdsLock(stateLock);
305
306 // unspecified (most options)
307 if (SocketOptionRegistry.findOption(name, Net.UNSPEC) != null)
308 return Net.UNSPEC;
309
310 // IPv4 socket
311 if (family == StandardProtocolFamily.INET)
312 return StandardProtocolFamily.INET;
313
314 // IPv6 socket that is unbound
315 if (localAddress == null)
316 return StandardProtocolFamily.INET6;
317
318 // IPv6 socket bound to wildcard or IPv6 address
319 InetAddress address = localAddress.getAddress();
320 if (address.isAnyLocalAddress() || (address instanceof Inet6Address))
321 return StandardProtocolFamily.INET6;
322
323 // IPv6 socket bound to IPv4 address
324 if (Net.canUseIPv6OptionsWithIPv4LocalAddress()) {
325 // IPV6_XXX options can be used
326 return StandardProtocolFamily.INET6;
327 } else {
328 // IPV6_XXX options cannot be used
329 return StandardProtocolFamily.INET;
330 }
331 }
332
333 @Override
334 public <T> DatagramChannel setOption(SocketOption<T> name, T value)
335 throws IOException
336 {
337 Objects.requireNonNull(name);
338 if (!supportedOptions().contains(name))
339 throw new UnsupportedOperationException("'" + name + "' not supported");
340 if (!name.type().isInstance(value))
341 throw new IllegalArgumentException("Invalid value '" + value + "'");
342
343 synchronized (stateLock) {
344 ensureOpen();
345
346 ProtocolFamily family = familyFor(name);
347
348 // Some platforms require both IPV6_XXX and IP_XXX socket options to
349 // be set when the channel's socket is IPv6 and it is used to send
350 // IPv4 multicast datagrams. The IP_XXX socket options are set on a
351 // best effort basis.
352 boolean needToSetIPv4Option = (family != Net.UNSPEC)
353 && (this.family == StandardProtocolFamily.INET6)
354 && Net.shouldSetBothIPv4AndIPv6Options();
355
356 // outgoing multicast interface
357 if (name == StandardSocketOptions.IP_MULTICAST_IF) {
358 assert family != Net.UNSPEC;
359 NetworkInterface interf = (NetworkInterface) value;
360 if (family == StandardProtocolFamily.INET6) {
361 int index = interf.getIndex();
362 if (index == -1)
363 throw new IOException("Network interface cannot be identified");
364 Net.setInterface6(fd, index);
365 }
366 if (family == StandardProtocolFamily.INET || needToSetIPv4Option) {
367 // need IPv4 address to identify interface
368 Inet4Address target = Net.anyInet4Address(interf);
369 if (target != null) {
370 try {
371 Net.setInterface4(fd, Net.inet4AsInt(target));
372 } catch (IOException ioe) {
373 if (family == StandardProtocolFamily.INET) throw ioe;
374 }
375 } else if (family == StandardProtocolFamily.INET) {
376 throw new IOException("Network interface not configured for IPv4");
377 }
378 }
379 return this;
380 }
381
382 // SO_REUSEADDR needs special handling as it may be emulated
383 if (name == StandardSocketOptions.SO_REUSEADDR
384 && Net.useExclusiveBind() && localAddress != null) {
385 reuseAddressEmulated = true;
386 this.isReuseAddress = (Boolean)value;
387 }
388
389 // remaining options don't need any special handling
390 Net.setSocketOption(fd, family, name, value);
391 if (needToSetIPv4Option && family != StandardProtocolFamily.INET) {
392 try {
393 Net.setSocketOption(fd, StandardProtocolFamily.INET, name, value);
394 } catch (IOException ignore) { }
395 }
396
397 return this;
398 }
399 }
400
401 @Override
402 @SuppressWarnings("unchecked")
403 public <T> T getOption(SocketOption<T> name)
404 throws IOException
405 {
406 Objects.requireNonNull(name);
407 if (!supportedOptions().contains(name))
408 throw new UnsupportedOperationException("'" + name + "' not supported");
409
410 synchronized (stateLock) {
411 ensureOpen();
412
413 ProtocolFamily family = familyFor(name);
414
415 if (name == StandardSocketOptions.IP_MULTICAST_IF) {
416 if (family == StandardProtocolFamily.INET) {
417 int address = Net.getInterface4(fd);
418 if (address == 0)
419 return null; // default interface
420
421 InetAddress ia = Net.inet4FromInt(address);
422 NetworkInterface ni = NetworkInterface.getByInetAddress(ia);
423 if (ni == null)
424 throw new IOException("Unable to map address to interface");
425 return (T) ni;
426 } else {
427 int index = Net.getInterface6(fd);
428 if (index == 0)
429 return null; // default interface
430
431 NetworkInterface ni = NetworkInterface.getByIndex(index);
432 if (ni == null)
433 throw new IOException("Unable to map index to interface");
434 return (T) ni;
435 }
436 }
437
438 if (name == StandardSocketOptions.SO_REUSEADDR && reuseAddressEmulated) {
439 return (T) Boolean.valueOf(isReuseAddress);
440 }
441
442 // no special handling
443 return (T) Net.getSocketOption(fd, family, name);
444 }
445 }
446
447 private static class DefaultOptionsHolder {
448 static final Set<SocketOption<?>> defaultOptions = defaultOptions();
449
450 private static Set<SocketOption<?>> defaultOptions() {
451 HashSet<SocketOption<?>> set = new HashSet<>();
452 set.add(StandardSocketOptions.SO_SNDBUF);
453 set.add(StandardSocketOptions.SO_RCVBUF);
454 set.add(StandardSocketOptions.SO_REUSEADDR);
455 if (Net.isReusePortAvailable()) {
456 set.add(StandardSocketOptions.SO_REUSEPORT);
457 }
458 set.add(StandardSocketOptions.SO_BROADCAST);
459 set.add(StandardSocketOptions.IP_TOS);
460 set.add(StandardSocketOptions.IP_MULTICAST_IF);
461 set.add(StandardSocketOptions.IP_MULTICAST_TTL);
462 set.add(StandardSocketOptions.IP_MULTICAST_LOOP);
463 set.addAll(ExtendedSocketOptions.datagramSocketOptions());
464 return Collections.unmodifiableSet(set);
465 }
466 }
467
468 @Override
469 public final Set<SocketOption<?>> supportedOptions() {
470 return DefaultOptionsHolder.defaultOptions;
471 }
472
473 /**
474 * Marks the beginning of a read operation that might block.
475 *
476 * @param blocking true if configured blocking
477 * @param mustBeConnected true if the socket must be connected
478 * @return remote address if connected
479 * @throws ClosedChannelException if the channel is closed
480 * @throws NotYetConnectedException if mustBeConnected and not connected
481 * @throws IOException if socket not bound and cannot be bound
482 */
483 private SocketAddress beginRead(boolean blocking, boolean mustBeConnected)
484 throws IOException
485 {
486 if (blocking && interruptible) {
487 // set hook for Thread.interrupt
488 begin();
489 }
490 SocketAddress remote;
491 synchronized (stateLock) {
492 ensureOpen();
493 remote = remoteAddress;
494 if ((remote == null) && mustBeConnected)
495 throw new NotYetConnectedException();
496 if (localAddress == null)
497 bindInternal(null);
498 if (blocking)
499 readerThread = NativeThread.current();
500 }
501 return remote;
502 }
503
504 /**
505 * Marks the end of a read operation that may have blocked.
506 *
507 * @throws AsynchronousCloseException if the channel was closed asynchronously
508 */
509 private void endRead(boolean blocking, boolean completed)
510 throws AsynchronousCloseException
511 {
512 if (blocking) {
513 synchronized (stateLock) {
514 readerThread = 0;
515 if (state == ST_CLOSING) {
516 tryFinishClose();
517 }
518 }
519 if (interruptible) {
520 // remove hook for Thread.interrupt (may throw AsynchronousCloseException)
521 end(completed);
522 } else if (!completed && !isOpen()) {
523 throw new AsynchronousCloseException();
524 }
525 }
526 }
527
528 @Override
529 public SocketAddress receive(ByteBuffer dst) throws IOException {
530 if (dst.isReadOnly())
531 throw new IllegalArgumentException("Read-only buffer");
532 readLock.lock();
533 try {
534 boolean blocking = isBlocking();
535 SocketAddress sender = null;
536 try {
537 SocketAddress remote = beginRead(blocking, false);
538 boolean connected = (remote != null);
539 @SuppressWarnings("removal")
540 SecurityManager sm = System.getSecurityManager();
541 if (connected || (sm == null)) {
542 // connected or no security manager
543 int n = receive(dst, connected);
544 if (blocking) {
545 while (IOStatus.okayToRetry(n) && isOpen()) {
546 park(Net.POLLIN);
547 n = receive(dst, connected);
548 }
549 }
550 if (n >= 0) {
551 // sender address is in socket address buffer
552 sender = sourceSocketAddress();
553 }
554 } else {
555 // security manager and unconnected
556 sender = untrustedReceive(dst);
557 }
558 return sender;
559 } finally {
560 endRead(blocking, (sender != null));
561 }
562 } finally {
563 readLock.unlock();
564 }
565 }
566
567 /**
568 * Receives a datagram into an untrusted buffer. When there is a security
569 * manager set, and the socket is not connected, datagrams have to be received
570 * into a buffer that is not accessible to the user. The datagram is copied
571 * into the user's buffer when the sender address is accepted by the security
572 * manager.
573 */
574 private SocketAddress untrustedReceive(ByteBuffer dst) throws IOException {
575 @SuppressWarnings("removal")
576 SecurityManager sm = System.getSecurityManager();
577 assert readLock.isHeldByCurrentThread()
578 && sm != null && remoteAddress == null;
579
580 ByteBuffer bb = Util.getTemporaryDirectBuffer(dst.remaining());
581 try {
582 boolean blocking = isBlocking();
583 for (;;) {
584 int n = receive(bb, false);
585 if (blocking) {
586 while (IOStatus.okayToRetry(n) && isOpen()) {
587 park(Net.POLLIN);
588 n = receive(bb, false);
589 }
590 }
591 if (n >= 0) {
592 // sender address is in socket address buffer
593 InetSocketAddress isa = sourceSocketAddress();
594 try {
595 sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
596 bb.flip();
597 dst.put(bb);
598 return isa;
599 } catch (SecurityException se) {
600 // ignore datagram
601 bb.clear();
602 }
603 } else {
604 return null;
605 }
606 }
607 } finally {
608 Util.releaseTemporaryDirectBuffer(bb);
609 }
610 }
611
612 /**
613 * Receives a datagram into the given buffer.
614 *
615 * @apiNote This method is for use by the socket adaptor. The buffer is
616 * assumed to be trusted, meaning it is not accessible to user code.
617 *
618 * @throws IllegalBlockingModeException if the channel is non-blocking
619 * @throws SocketTimeoutException if the timeout elapses
620 */
621 SocketAddress blockingReceive(ByteBuffer dst, long nanos) throws IOException {
622 readLock.lock();
623 try {
624 ensureOpen();
625 if (!isBlocking())
626 throw new IllegalBlockingModeException();
627 @SuppressWarnings("removal")
628 SecurityManager sm = System.getSecurityManager();
629 boolean connected = isConnected();
630 SocketAddress sender;
631 do {
632 if (nanos > 0) {
633 sender = trustedBlockingReceive(dst, nanos);
634 } else {
635 sender = trustedBlockingReceive(dst);
636 }
637 // check sender when security manager set and not connected
638 if (sm != null && !connected) {
639 InetSocketAddress isa = (InetSocketAddress) sender;
640 try {
641 sm.checkAccept(isa.getAddress().getHostAddress(), isa.getPort());
642 } catch (SecurityException e) {
643 sender = null;
644 }
645 }
646 } while (sender == null);
647 return sender;
648 } finally {
649 readLock.unlock();
650 }
651 }
652
653 /**
654 * Receives a datagram into given buffer. This method is used to support
655 * the socket adaptor. The buffer is assumed to be trusted.
656 * @throws SocketTimeoutException if the timeout elapses
657 */
658 private SocketAddress trustedBlockingReceive(ByteBuffer dst)
659 throws IOException
660 {
661 assert readLock.isHeldByCurrentThread() && isBlocking();
662 SocketAddress sender = null;
663 try {
664 SocketAddress remote = beginRead(true, false);
665 boolean connected = (remote != null);
666 int n = receive(dst, connected);
667 while (IOStatus.okayToRetry(n) && isOpen()) {
668 park(Net.POLLIN);
669 n = receive(dst, connected);
670 }
671 if (n >= 0) {
672 // sender address is in socket address buffer
673 sender = sourceSocketAddress();
674 }
675 return sender;
676 } finally {
677 endRead(true, (sender != null));
678 }
679 }
680
681 /**
682 * Receives a datagram into given buffer with a timeout. This method is
683 * used to support the socket adaptor. The buffer is assumed to be trusted.
684 * @throws SocketTimeoutException if the timeout elapses
685 */
686 private SocketAddress trustedBlockingReceive(ByteBuffer dst, long nanos)
687 throws IOException
688 {
689 assert readLock.isHeldByCurrentThread() && isBlocking();
690 SocketAddress sender = null;
691 try {
692 SocketAddress remote = beginRead(true, false);
693 boolean connected = (remote != null);
694
695 // change socket to non-blocking
696 lockedConfigureBlocking(false);
697 try {
698 long startNanos = System.nanoTime();
699 int n = receive(dst, connected);
700 while (n == IOStatus.UNAVAILABLE && isOpen()) {
701 long remainingNanos = nanos - (System.nanoTime() - startNanos);
702 if (remainingNanos <= 0) {
703 throw new SocketTimeoutException("Receive timed out");
704 }
705 park(Net.POLLIN, remainingNanos);
706 n = receive(dst, connected);
707 }
708 if (n >= 0) {
709 // sender address is in socket address buffer
710 sender = sourceSocketAddress();
711 }
712 return sender;
713 } finally {
714 // restore socket to blocking mode (if channel is open)
715 tryLockedConfigureBlocking(true);
716 }
717 } finally {
718 endRead(true, (sender != null));
719 }
720 }
721
722 private int receive(ByteBuffer dst, boolean connected) throws IOException {
723 int pos = dst.position();
724 int lim = dst.limit();
725 assert (pos <= lim);
726 int rem = (pos <= lim ? lim - pos : 0);
727 if (dst instanceof DirectBuffer && rem > 0)
728 return receiveIntoNativeBuffer(dst, rem, pos, connected);
729
730 // Substitute a native buffer. If the supplied buffer is empty
731 // we must instead use a nonempty buffer, otherwise the call
732 // will not block waiting for a datagram on some platforms.
733 int newSize = Math.max(rem, 1);
734 ByteBuffer bb = Util.getTemporaryDirectBuffer(newSize);
735 try {
736 int n = receiveIntoNativeBuffer(bb, newSize, 0, connected);
737 bb.flip();
738 if (n > 0 && rem > 0)
739 dst.put(bb);
740 return n;
741 } finally {
742 Util.releaseTemporaryDirectBuffer(bb);
743 }
744 }
745
746 private int receiveIntoNativeBuffer(ByteBuffer bb, int rem, int pos,
747 boolean connected)
748 throws IOException
749 {
750 int n = receive0(fd,
751 ((DirectBuffer)bb).address() + pos, rem,
752 sourceSockAddr.address(),
753 connected);
754 if (n > 0)
755 bb.position(pos + n);
756 return n;
757 }
758
759 /**
760 * Return an InetSocketAddress to represent the source/sender socket address
761 * in sourceSockAddr. Returns the cached InetSocketAddress if the source
762 * address is the same as the cached address.
763 */
764 private InetSocketAddress sourceSocketAddress() throws IOException {
765 assert readLock.isHeldByCurrentThread();
766 if (cachedInetSocketAddress != null && sourceSockAddr.equals(cachedSockAddr)) {
767 return cachedInetSocketAddress;
768 }
769 InetSocketAddress isa = sourceSockAddr.decode();
770 // swap sourceSockAddr and cachedSockAddr
771 NativeSocketAddress tmp = cachedSockAddr;
772 cachedSockAddr = sourceSockAddr;
773 sourceSockAddr = tmp;
774 cachedInetSocketAddress = isa;
775 return isa;
776 }
777
778 @Override
779 public int send(ByteBuffer src, SocketAddress target)
780 throws IOException
781 {
782 Objects.requireNonNull(src);
783 InetSocketAddress isa = Net.checkAddress(target, family);
784
785 writeLock.lock();
786 try {
787 boolean blocking = isBlocking();
788 int n;
789 boolean completed = false;
790 try {
791 SocketAddress remote = beginWrite(blocking, false);
792 if (remote != null) {
793 // connected
794 if (!target.equals(remote)) {
795 throw new AlreadyConnectedException();
796 }
797 n = IOUtil.write(fd, src, -1, nd);
798 if (blocking) {
799 while (IOStatus.okayToRetry(n) && isOpen()) {
800 park(Net.POLLOUT);
801 n = IOUtil.write(fd, src, -1, nd);
802 }
803 }
804 completed = (n > 0);
805 } else {
806 // not connected
807 @SuppressWarnings("removal")
808 SecurityManager sm = System.getSecurityManager();
809 InetAddress ia = isa.getAddress();
810 if (sm != null) {
811 if (ia.isMulticastAddress()) {
812 sm.checkMulticast(ia);
813 } else {
814 sm.checkConnect(ia.getHostAddress(), isa.getPort());
815 }
816 }
817 if (ia.isLinkLocalAddress())
818 isa = IPAddressUtil.toScopedAddress(isa);
819 if (isa.getPort() == 0)
820 throw new SocketException("Can't send to port 0");
821 n = send(fd, src, isa);
822 if (blocking) {
823 while (IOStatus.okayToRetry(n) && isOpen()) {
824 park(Net.POLLOUT);
825 n = send(fd, src, isa);
826 }
827 }
828 completed = (n >= 0);
829 }
830 } finally {
831 endWrite(blocking, completed);
832 }
833 assert n >= 0 || n == IOStatus.UNAVAILABLE;
834 return IOStatus.normalize(n);
835 } finally {
836 writeLock.unlock();
837 }
838 }
839
840 /**
841 * Sends a datagram from the bytes in given buffer.
842 *
843 * @apiNote This method is for use by the socket adaptor.
844 *
845 * @throws IllegalBlockingModeException if the channel is non-blocking
846 */
847 void blockingSend(ByteBuffer src, SocketAddress target) throws IOException {
848 writeLock.lock();
849 try {
850 ensureOpen();
851 if (!isBlocking())
852 throw new IllegalBlockingModeException();
853 send(src, target);
854 } finally {
855 writeLock.unlock();
856 }
857 }
858
859 private int send(FileDescriptor fd, ByteBuffer src, InetSocketAddress target)
860 throws IOException
861 {
862 if (src instanceof DirectBuffer)
863 return sendFromNativeBuffer(fd, src, target);
864
865 // Substitute a native buffer
866 int pos = src.position();
867 int lim = src.limit();
868 assert (pos <= lim);
869 int rem = (pos <= lim ? lim - pos : 0);
870
871 ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
872 try {
873 bb.put(src);
874 bb.flip();
875 // Do not update src until we see how many bytes were written
876 src.position(pos);
877
878 int n = sendFromNativeBuffer(fd, bb, target);
879 if (n > 0) {
880 // now update src
881 src.position(pos + n);
882 }
883 return n;
884 } finally {
885 Util.releaseTemporaryDirectBuffer(bb);
886 }
887 }
888
889 private int sendFromNativeBuffer(FileDescriptor fd, ByteBuffer bb,
890 InetSocketAddress target)
891 throws IOException
892 {
893 int pos = bb.position();
894 int lim = bb.limit();
895 assert (pos <= lim);
896 int rem = (pos <= lim ? lim - pos : 0);
897
898 int written;
899 try {
900 int addressLen = targetSocketAddress(target);
901 written = send0(fd, ((DirectBuffer)bb).address() + pos, rem,
902 targetSockAddr.address(), addressLen);
903 } catch (PortUnreachableException pue) {
904 if (isConnected())
905 throw pue;
906 written = rem;
907 }
908 if (written > 0)
909 bb.position(pos + written);
910 return written;
911 }
912
913 /**
914 * Encodes the given InetSocketAddress into targetSockAddr, returning the
915 * length of the sockaddr structure (sizeof struct sockaddr or sockaddr6).
916 */
917 private int targetSocketAddress(InetSocketAddress isa) {
918 assert writeLock.isHeldByCurrentThread();
919 // Nothing to do if target address is already in the buffer. Use
920 // identity rather than equals as Inet6Address.equals ignores scope_id.
921 if (isa == previousTarget)
922 return previousSockAddrLength;
923 previousTarget = null;
924 int len = targetSockAddr.encode(family, isa);
925 previousTarget = isa;
926 previousSockAddrLength = len;
927 return len;
928 }
929
930 @Override
931 public int read(ByteBuffer buf) throws IOException {
932 Objects.requireNonNull(buf);
933
934 readLock.lock();
935 try {
936 boolean blocking = isBlocking();
937 int n = 0;
938 try {
939 beginRead(blocking, true);
940 n = IOUtil.read(fd, buf, -1, nd);
941 if (blocking) {
942 while (IOStatus.okayToRetry(n) && isOpen()) {
943 park(Net.POLLIN);
944 n = IOUtil.read(fd, buf, -1, nd);
945 }
946 }
947 } finally {
948 endRead(blocking, n > 0);
949 assert IOStatus.check(n);
950 }
951 return IOStatus.normalize(n);
952 } finally {
953 readLock.unlock();
954 }
955 }
956
957 @Override
958 public long read(ByteBuffer[] dsts, int offset, int length)
959 throws IOException
960 {
961 Objects.checkFromIndexSize(offset, length, dsts.length);
962
963 readLock.lock();
964 try {
965 boolean blocking = isBlocking();
966 long n = 0;
967 try {
968 beginRead(blocking, true);
969 n = IOUtil.read(fd, dsts, offset, length, nd);
970 if (blocking) {
971 while (IOStatus.okayToRetry(n) && isOpen()) {
972 park(Net.POLLIN);
973 n = IOUtil.read(fd, dsts, offset, length, nd);
974 }
975 }
976 } finally {
977 endRead(blocking, n > 0);
978 assert IOStatus.check(n);
979 }
980 return IOStatus.normalize(n);
981 } finally {
982 readLock.unlock();
983 }
984 }
985
986 /**
987 * Marks the beginning of a write operation that might block.
988 * @param blocking true if configured blocking
989 * @param mustBeConnected true if the socket must be connected
990 * @return remote address if connected
991 * @throws ClosedChannelException if the channel is closed
992 * @throws NotYetConnectedException if mustBeConnected and not connected
993 * @throws IOException if socket not bound and cannot be bound
994 */
995 private SocketAddress beginWrite(boolean blocking, boolean mustBeConnected)
996 throws IOException
997 {
998 if (blocking && interruptible) {
999 // set hook for Thread.interrupt
1000 begin();
1001 }
1002 SocketAddress remote;
1003 synchronized (stateLock) {
1004 ensureOpen();
1005 remote = remoteAddress;
1006 if ((remote == null) && mustBeConnected)
1007 throw new NotYetConnectedException();
1008 if (localAddress == null)
1009 bindInternal(null);
1010 if (blocking)
1011 writerThread = NativeThread.current();
1012 }
1013 return remote;
1014 }
1015
1016 /**
1017 * Marks the end of a write operation that may have blocked.
1018 *
1019 * @throws AsynchronousCloseException if the channel was closed asynchronously
1020 */
1021 private void endWrite(boolean blocking, boolean completed)
1022 throws AsynchronousCloseException
1023 {
1024 if (blocking) {
1025 synchronized (stateLock) {
1026 writerThread = 0;
1027 if (state == ST_CLOSING) {
1028 tryFinishClose();
1029 }
1030 }
1031
1032 if (interruptible) {
1033 // remove hook for Thread.interrupt (may throw AsynchronousCloseException)
1034 end(completed);
1035 } else if (!completed && !isOpen()) {
1036 throw new AsynchronousCloseException();
1037 }
1038 }
1039 }
1040
1041 @Override
1042 public int write(ByteBuffer buf) throws IOException {
1043 Objects.requireNonNull(buf);
1044
1045 writeLock.lock();
1046 try {
1047 boolean blocking = isBlocking();
1048 int n = 0;
1049 try {
1050 beginWrite(blocking, true);
1051 n = IOUtil.write(fd, buf, -1, nd);
1052 if (blocking) {
1053 while (IOStatus.okayToRetry(n) && isOpen()) {
1054 park(Net.POLLOUT);
1055 n = IOUtil.write(fd, buf, -1, nd);
1056 }
1057 }
1058 } finally {
1059 endWrite(blocking, n > 0);
1060 assert IOStatus.check(n);
1061 }
1062 return IOStatus.normalize(n);
1063 } finally {
1064 writeLock.unlock();
1065 }
1066 }
1067
1068 @Override
1069 public long write(ByteBuffer[] srcs, int offset, int length)
1070 throws IOException
1071 {
1072 Objects.checkFromIndexSize(offset, length, srcs.length);
1073
1074 writeLock.lock();
1075 try {
1076 boolean blocking = isBlocking();
1077 long n = 0;
1078 try {
1079 beginWrite(blocking, true);
1080 n = IOUtil.write(fd, srcs, offset, length, nd);
1081 if (blocking) {
1082 while (IOStatus.okayToRetry(n) && isOpen()) {
1083 park(Net.POLLOUT);
1084 n = IOUtil.write(fd, srcs, offset, length, nd);
1085 }
1086 }
1087 } finally {
1088 endWrite(blocking, n > 0);
1089 assert IOStatus.check(n);
1090 }
1091 return IOStatus.normalize(n);
1092 } finally {
1093 writeLock.unlock();
1094 }
1095 }
1096
1097 @Override
1098 protected void implConfigureBlocking(boolean block) throws IOException {
1099 readLock.lock();
1100 try {
1101 writeLock.lock();
1102 try {
1103 lockedConfigureBlocking(block);
1104 } finally {
1105 writeLock.unlock();
1106 }
1107 } finally {
1108 readLock.unlock();
1109 }
1110 }
1111
1112 /**
1113 * Adjusts the blocking mode. readLock or writeLock must already be held.
1114 */
1115 private void lockedConfigureBlocking(boolean block) throws IOException {
1116 assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
1117 synchronized (stateLock) {
1118 ensureOpen();
1119 IOUtil.configureBlocking(fd, block);
1120 }
1121 }
1122
1123 /**
1124 * Adjusts the blocking mode if the channel is open. readLock or writeLock
1125 * must already be held.
1126 *
1127 * @return {@code true} if the blocking mode was adjusted, {@code false} if
1128 * the blocking mode was not adjusted because the channel is closed
1129 */
1130 private boolean tryLockedConfigureBlocking(boolean block) throws IOException {
1131 assert readLock.isHeldByCurrentThread() || writeLock.isHeldByCurrentThread();
1132 synchronized (stateLock) {
1133 if (isOpen()) {
1134 IOUtil.configureBlocking(fd, block);
1135 return true;
1136 } else {
1137 return false;
1138 }
1139 }
1140 }
1141
1142 InetSocketAddress localAddress() {
1143 synchronized (stateLock) {
1144 return localAddress;
1145 }
1146 }
1147
1148 InetSocketAddress remoteAddress() {
1149 synchronized (stateLock) {
1150 return remoteAddress;
1151 }
1152 }
1153
1154 @Override
1155 public DatagramChannel bind(SocketAddress local) throws IOException {
1156 readLock.lock();
1157 try {
1158 writeLock.lock();
1159 try {
1160 synchronized (stateLock) {
1161 ensureOpen();
1162 if (localAddress != null)
1163 throw new AlreadyBoundException();
1164 bindInternal(local);
1165 }
1166 } finally {
1167 writeLock.unlock();
1168 }
1169 } finally {
1170 readLock.unlock();
1171 }
1172 return this;
1173 }
1174
1175 private void bindInternal(SocketAddress local) throws IOException {
1176 assert Thread.holdsLock(stateLock )&& (localAddress == null);
1177
1178 InetSocketAddress isa;
1179 if (local == null) {
1180 // only Inet4Address allowed with IPv4 socket
1181 if (family == StandardProtocolFamily.INET) {
1182 isa = new InetSocketAddress(InetAddress.getByName("0.0.0.0"), 0);
1183 } else {
1184 isa = new InetSocketAddress(0);
1185 }
1186 } else {
1187 isa = Net.checkAddress(local, family);
1188 }
1189 @SuppressWarnings("removal")
1190 SecurityManager sm = System.getSecurityManager();
1191 if (sm != null)
1192 sm.checkListen(isa.getPort());
1193
1194 Net.bind(family, fd, isa.getAddress(), isa.getPort());
1195 localAddress = Net.localAddress(fd);
1196 }
1197
1198 @Override
1199 public boolean isConnected() {
1200 synchronized (stateLock) {
1201 return (state == ST_CONNECTED);
1202 }
1203 }
1204
1205 @Override
1206 public DatagramChannel connect(SocketAddress sa) throws IOException {
1207 return connect(sa, true);
1208 }
1209
1210 /**
1211 * Connects the channel's socket.
1212 *
1213 * @param sa the remote address to which this channel is to be connected
1214 * @param check true to check if the channel is already connected.
1215 */
1216 DatagramChannel connect(SocketAddress sa, boolean check) throws IOException {
1217 InetSocketAddress isa = Net.checkAddress(sa, family);
1218 @SuppressWarnings("removal")
1219 SecurityManager sm = System.getSecurityManager();
1220 if (sm != null) {
1221 InetAddress ia = isa.getAddress();
1222 if (ia.isMulticastAddress()) {
1223 sm.checkMulticast(ia);
1224 } else {
1225 sm.checkConnect(ia.getHostAddress(), isa.getPort());
1226 sm.checkAccept(ia.getHostAddress(), isa.getPort());
1227 }
1228 }
1229
1230 readLock.lock();
1231 try {
1232 writeLock.lock();
1233 try {
1234 synchronized (stateLock) {
1235 ensureOpen();
1236 if (check && state == ST_CONNECTED)
1237 throw new AlreadyConnectedException();
1238 if (isa.getPort() == 0)
1239 throw new SocketException("Can't connect to port 0");
1240
1241 // ensure that the socket is bound
1242 if (localAddress == null) {
1243 bindInternal(null);
1244 }
1245
1246 // capture local address before connect
1247 initialLocalAddress = localAddress;
1248
1249 int n = Net.connect(family,
1250 fd,
1251 isa.getAddress(),
1252 isa.getPort());
1253 if (n <= 0)
1254 throw new Error(); // Can't happen
1255
1256 // connected
1257 remoteAddress = isa;
1258 state = ST_CONNECTED;
1259
1260 // refresh local address
1261 localAddress = Net.localAddress(fd);
1262
1263 // flush any packets already received.
1264 boolean blocking = isBlocking();
1265 if (blocking) {
1266 IOUtil.configureBlocking(fd, false);
1267 }
1268 try {
1269 ByteBuffer buf = ByteBuffer.allocate(100);
1270 while (receive(buf, false) >= 0) {
1271 buf.clear();
1272 }
1273 } finally {
1274 if (blocking) {
1275 IOUtil.configureBlocking(fd, true);
1276 }
1277 }
1278 }
1279 } finally {
1280 writeLock.unlock();
1281 }
1282 } finally {
1283 readLock.unlock();
1284 }
1285 return this;
1286 }
1287
1288 @Override
1289 public DatagramChannel disconnect() throws IOException {
1290 readLock.lock();
1291 try {
1292 writeLock.lock();
1293 try {
1294 synchronized (stateLock) {
1295 if (!isOpen() || (state != ST_CONNECTED))
1296 return this;
1297
1298 // disconnect socket
1299 boolean isIPv6 = (family == StandardProtocolFamily.INET6);
1300 disconnect0(fd, isIPv6);
1301
1302 // no longer connected
1303 remoteAddress = null;
1304 state = ST_UNCONNECTED;
1305
1306 // refresh localAddress, should be same as it was prior to connect
1307 localAddress = Net.localAddress(fd);
1308 try {
1309 if (!localAddress.equals(initialLocalAddress)) {
1310 // Workaround connect(2) issues on Linux and macOS
1311 repairSocket(initialLocalAddress);
1312 assert (localAddress != null)
1313 && localAddress.equals(Net.localAddress(fd))
1314 && localAddress.equals(initialLocalAddress);
1315 }
1316 } finally {
1317 initialLocalAddress = null;
1318 }
1319 }
1320 } finally {
1321 writeLock.unlock();
1322 }
1323 } finally {
1324 readLock.unlock();
1325 }
1326 return this;
1327 }
1328
1329 /**
1330 * "Repair" the channel's socket after a disconnect that didn't restore the
1331 * local address.
1332 *
1333 * On Linux, connect(2) dissolves the association but changes the local port
1334 * to 0 when it was initially bound to an ephemeral port. The workaround here
1335 * is to rebind to the original port.
1336 *
1337 * On macOS, connect(2) dissolves the association but rebinds the socket to
1338 * the wildcard address when it was initially bound to a specific address.
1339 * The workaround here is to re-create the socket.
1340 */
1341 private void repairSocket(InetSocketAddress target)
1342 throws IOException
1343 {
1344 assert Thread.holdsLock(stateLock);
1345
1346 // Linux: try to bind the socket to the original address/port
1347 if (localAddress.getPort() == 0) {
1348 assert localAddress.getAddress().equals(target.getAddress());
1349 Net.bind(family, fd, target.getAddress(), target.getPort());
1350 localAddress = Net.localAddress(fd);
1351 return;
1352 }
1353
1354 // capture the value of all existing socket options
1355 Map<SocketOption<?>, Object> map = new HashMap<>();
1356 for (SocketOption<?> option : supportedOptions()) {
1357 Object value = getOption(option);
1358 if (value != null) {
1359 map.put(option, value);
1360 }
1361 }
1362
1363 // macOS: re-create the socket.
1364 FileDescriptor newfd = Net.socket(family, false);
1365 try {
1366 // copy the socket options that are protocol family agnostic
1367 for (Map.Entry<SocketOption<?>, Object> e : map.entrySet()) {
1368 SocketOption<?> option = e.getKey();
1369 if (SocketOptionRegistry.findOption(option, Net.UNSPEC) != null) {
1370 Object value = e.getValue();
1371 try {
1372 Net.setSocketOption(newfd, Net.UNSPEC, option, value);
1373 } catch (IOException ignore) { }
1374 }
1375 }
1376
1377 // copy the blocking mode
1378 if (!isBlocking()) {
1379 IOUtil.configureBlocking(newfd, false);
1380 }
1381
1382 // dup this channel's socket to the new socket. If this succeeds then
1383 // fd will reference the new socket. If it fails then it will still
1384 // reference the old socket.
1385 nd.dup(newfd, fd);
1386 } finally {
1387 // release the file descriptor
1388 nd.close(newfd);
1389 }
1390
1391 // bind to the original local address
1392 try {
1393 Net.bind(family, fd, target.getAddress(), target.getPort());
1394 } catch (IOException ioe) {
1395 // bind failed, socket is left unbound
1396 localAddress = null;
1397 throw ioe;
1398 }
1399
1400 // restore local address
1401 localAddress = Net.localAddress(fd);
1402
1403 // restore all socket options (including those set in first pass)
1404 for (Map.Entry<SocketOption<?>, Object> e : map.entrySet()) {
1405 @SuppressWarnings("unchecked")
1406 SocketOption<Object> option = (SocketOption<Object>) e.getKey();
1407 Object value = e.getValue();
1408 try {
1409 setOption(option, value);
1410 } catch (IOException ignore) { }
1411 }
1412
1413 // restore multicast group membership
1414 MembershipRegistry registry = this.registry;
1415 if (registry != null) {
1416 registry.forEach(k -> {
1417 if (k instanceof MembershipKeyImpl.Type6) {
1418 MembershipKeyImpl.Type6 key6 = (MembershipKeyImpl.Type6) k;
1419 Net.join6(fd, key6.groupAddress(), key6.index(), key6.source());
1420 } else {
1421 MembershipKeyImpl.Type4 key4 = (MembershipKeyImpl.Type4) k;
1422 Net.join4(fd, key4.groupAddress(), key4.interfaceAddress(), key4.source());
1423 }
1424 });
1425 }
1426
1427 // reset registration in all Selectors that this channel is registered with
1428 AbstractSelectableChannels.forEach(this, SelectionKeyImpl::reset);
1429 }
1430
1431 /**
1432 * Defines static methods to access AbstractSelectableChannel non-public members.
1433 */
1434 @SuppressWarnings("removal")
1435 private static class AbstractSelectableChannels {
1436 private static final Method FOREACH;
1437 static {
1438 try {
1439 PrivilegedExceptionAction<Method> pae = () -> {
1440 Method m = AbstractSelectableChannel.class.getDeclaredMethod("forEach", Consumer.class);
1441 m.setAccessible(true);
1442 return m;
1443 };
1444 FOREACH = AccessController.doPrivileged(pae);
1445 } catch (Exception e) {
1446 throw new InternalError(e);
1447 }
1448 }
1449 static void forEach(AbstractSelectableChannel ch, Consumer<SelectionKeyImpl> action) {
1450 try {
1451 FOREACH.invoke(ch, action);
1452 } catch (Exception e) {
1453 throw new InternalError(e);
1454 }
1455 }
1456 }
1457
1458 /**
1459 * Joins channel's socket to the given group/interface and
1460 * optional source address.
1461 */
1462 private MembershipKey innerJoin(InetAddress group,
1463 NetworkInterface interf,
1464 InetAddress source)
1465 throws IOException
1466 {
1467 if (!group.isMulticastAddress())
1468 throw new IllegalArgumentException("Group not a multicast address");
1469
1470 // check multicast address is compatible with this socket
1471 if (group instanceof Inet4Address) {
1472 if (family == StandardProtocolFamily.INET6 && !Net.canIPv6SocketJoinIPv4Group())
1473 throw new IllegalArgumentException("IPv6 socket cannot join IPv4 multicast group");
1474 } else if (group instanceof Inet6Address) {
1475 if (family != StandardProtocolFamily.INET6)
1476 throw new IllegalArgumentException("Only IPv6 sockets can join IPv6 multicast group");
1477 } else {
1478 throw new IllegalArgumentException("Address type not supported");
1479 }
1480
1481 // check source address
1482 if (source != null) {
1483 if (source.isAnyLocalAddress())
1484 throw new IllegalArgumentException("Source address is a wildcard address");
1485 if (source.isMulticastAddress())
1486 throw new IllegalArgumentException("Source address is multicast address");
1487 if (source.getClass() != group.getClass())
1488 throw new IllegalArgumentException("Source address is different type to group");
1489 }
1490
1491 @SuppressWarnings("removal")
1492 SecurityManager sm = System.getSecurityManager();
1493 if (sm != null)
1494 sm.checkMulticast(group);
1495
1496 synchronized (stateLock) {
1497 ensureOpen();
1498
1499 // check the registry to see if we are already a member of the group
1500 if (registry == null) {
1501 registry = new MembershipRegistry();
1502 } else {
1503 // return existing membership key
1504 MembershipKey key = registry.checkMembership(group, interf, source);
1505 if (key != null)
1506 return key;
1507 }
1508
1509 MembershipKeyImpl key;
1510 if ((family == StandardProtocolFamily.INET6) &&
1511 ((group instanceof Inet6Address) || Net.canJoin6WithIPv4Group()))
1512 {
1513 int index = interf.getIndex();
1514 if (index == -1)
1515 throw new IOException("Network interface cannot be identified");
1516
1517 // need multicast and source address as byte arrays
1518 byte[] groupAddress = Net.inet6AsByteArray(group);
1519 byte[] sourceAddress = (source == null) ? null :
1520 Net.inet6AsByteArray(source);
1521
1522 // join the group
1523 int n = Net.join6(fd, groupAddress, index, sourceAddress);
1524 if (n == IOStatus.UNAVAILABLE)
1525 throw new UnsupportedOperationException();
1526
1527 key = new MembershipKeyImpl.Type6(this, group, interf, source,
1528 groupAddress, index, sourceAddress);
1529
1530 } else {
1531 // need IPv4 address to identify interface
1532 Inet4Address target = Net.anyInet4Address(interf);
1533 if (target == null)
1534 throw new IOException("Network interface not configured for IPv4");
1535
1536 int groupAddress = Net.inet4AsInt(group);
1537 int targetAddress = Net.inet4AsInt(target);
1538 int sourceAddress = (source == null) ? 0 : Net.inet4AsInt(source);
1539
1540 // join the group
1541 int n = Net.join4(fd, groupAddress, targetAddress, sourceAddress);
1542 if (n == IOStatus.UNAVAILABLE)
1543 throw new UnsupportedOperationException();
1544
1545 key = new MembershipKeyImpl.Type4(this, group, interf, source,
1546 groupAddress, targetAddress, sourceAddress);
1547 }
1548
1549 registry.add(key);
1550 return key;
1551 }
1552 }
1553
1554 @Override
1555 public MembershipKey join(InetAddress group,
1556 NetworkInterface interf)
1557 throws IOException
1558 {
1559 return innerJoin(group, interf, null);
1560 }
1561
1562 @Override
1563 public MembershipKey join(InetAddress group,
1564 NetworkInterface interf,
1565 InetAddress source)
1566 throws IOException
1567 {
1568 Objects.requireNonNull(source);
1569 return innerJoin(group, interf, source);
1570 }
1571
1572 // package-private
1573 void drop(MembershipKeyImpl key) {
1574 assert key.channel() == this;
1575
1576 synchronized (stateLock) {
1577 if (!key.isValid())
1578 return;
1579
1580 try {
1581 if (key instanceof MembershipKeyImpl.Type6) {
1582 MembershipKeyImpl.Type6 key6 =
1583 (MembershipKeyImpl.Type6)key;
1584 Net.drop6(fd, key6.groupAddress(), key6.index(), key6.source());
1585 } else {
1586 MembershipKeyImpl.Type4 key4 = (MembershipKeyImpl.Type4)key;
1587 Net.drop4(fd, key4.groupAddress(), key4.interfaceAddress(),
1588 key4.source());
1589 }
1590 } catch (IOException ioe) {
1591 // should not happen
1592 throw new AssertionError(ioe);
1593 }
1594
1595 key.invalidate();
1596 registry.remove(key);
1597 }
1598 }
1599
1600 /**
1601 * Finds an existing membership of a multicast group. Returns null if this
1602 * channel's socket is not a member of the group.
1603 *
1604 * @apiNote This method is for use by the socket adaptor
1605 */
1606 MembershipKey findMembership(InetAddress group, NetworkInterface interf) {
1607 synchronized (stateLock) {
1608 if (registry != null) {
1609 return registry.checkMembership(group, interf, null);
1610 } else {
1611 return null;
1612 }
1613 }
1614 }
1615
1616 /**
1617 * Block datagrams from the given source.
1618 */
1619 void block(MembershipKeyImpl key, InetAddress source)
1620 throws IOException
1621 {
1622 assert key.channel() == this;
1623 assert key.sourceAddress() == null;
1624
1625 synchronized (stateLock) {
1626 if (!key.isValid())
1627 throw new IllegalStateException("key is no longer valid");
1628 if (source.isAnyLocalAddress())
1629 throw new IllegalArgumentException("Source address is a wildcard address");
1630 if (source.isMulticastAddress())
1631 throw new IllegalArgumentException("Source address is multicast address");
1632 if (source.getClass() != key.group().getClass())
1633 throw new IllegalArgumentException("Source address is different type to group");
1634
1635 int n;
1636 if (key instanceof MembershipKeyImpl.Type6) {
1637 MembershipKeyImpl.Type6 key6 =
1638 (MembershipKeyImpl.Type6)key;
1639 n = Net.block6(fd, key6.groupAddress(), key6.index(),
1640 Net.inet6AsByteArray(source));
1641 } else {
1642 MembershipKeyImpl.Type4 key4 =
1643 (MembershipKeyImpl.Type4)key;
1644 n = Net.block4(fd, key4.groupAddress(), key4.interfaceAddress(),
1645 Net.inet4AsInt(source));
1646 }
1647 if (n == IOStatus.UNAVAILABLE) {
1648 // ancient kernel
1649 throw new UnsupportedOperationException();
1650 }
1651 }
1652 }
1653
1654 /**
1655 * Unblock the given source.
1656 */
1657 void unblock(MembershipKeyImpl key, InetAddress source) {
1658 assert key.channel() == this;
1659 assert key.sourceAddress() == null;
1660
1661 synchronized (stateLock) {
1662 if (!key.isValid())
1663 throw new IllegalStateException("key is no longer valid");
1664
1665 try {
1666 if (key instanceof MembershipKeyImpl.Type6) {
1667 MembershipKeyImpl.Type6 key6 =
1668 (MembershipKeyImpl.Type6)key;
1669 Net.unblock6(fd, key6.groupAddress(), key6.index(),
1670 Net.inet6AsByteArray(source));
1671 } else {
1672 MembershipKeyImpl.Type4 key4 =
1673 (MembershipKeyImpl.Type4)key;
1674 Net.unblock4(fd, key4.groupAddress(), key4.interfaceAddress(),
1675 Net.inet4AsInt(source));
1676 }
1677 } catch (IOException ioe) {
1678 // should not happen
1679 throw new AssertionError(ioe);
1680 }
1681 }
1682 }
1683
1684 /**
1685 * Closes the socket if there are no I/O operations in progress and the
1686 * channel is not registered with a Selector.
1687 */
1688 private boolean tryClose() throws IOException {
1689 assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
1690 if ((readerThread == 0) && (writerThread == 0) && !isRegistered()) {
1691 state = ST_CLOSED;
1692 try {
1693 // close socket
1694 cleaner.clean();
1695 } catch (UncheckedIOException ioe) {
1696 throw ioe.getCause();
1697 }
1698 return true;
1699 } else {
1700 return false;
1701 }
1702 }
1703
1704 /**
1705 * Invokes tryClose to attempt to close the socket.
1706 *
1707 * This method is used for deferred closing by I/O and Selector operations.
1708 */
1709 private void tryFinishClose() {
1710 try {
1711 tryClose();
1712 } catch (IOException ignore) { }
1713 }
1714
1715 /**
1716 * Closes this channel when configured in blocking mode.
1717 *
1718 * If there is an I/O operation in progress then the socket is pre-closed
1719 * and the I/O threads signalled, in which case the final close is deferred
1720 * until all I/O operations complete.
1721 */
1722 private void implCloseBlockingMode() throws IOException {
1723 synchronized (stateLock) {
1724 assert state < ST_CLOSING;
1725 state = ST_CLOSING;
1726
1727 // if member of any multicast groups then invalidate the keys
1728 if (registry != null)
1729 registry.invalidateAll();
1730
1731 if (!tryClose()) {
1732 long reader = readerThread;
1733 long writer = writerThread;
1734 if (reader != 0 || writer != 0) {
1735 nd.preClose(fd);
1736 if (reader != 0)
1737 NativeThread.signal(reader);
1738 if (writer != 0)
1739 NativeThread.signal(writer);
1740 }
1741 }
1742 }
1743 }
1744
1745 /**
1746 * Closes this channel when configured in non-blocking mode.
1747 *
1748 * If the channel is registered with a Selector then the close is deferred
1749 * until the channel is flushed from all Selectors.
1750 */
1751 private void implCloseNonBlockingMode() throws IOException {
1752 synchronized (stateLock) {
1753 assert state < ST_CLOSING;
1754 state = ST_CLOSING;
1755
1756 // if member of any multicast groups then invalidate the keys
1757 if (registry != null)
1758 registry.invalidateAll();
1759 }
1760
1761 // wait for any read/write operations to complete before trying to close
1762 readLock.lock();
1763 readLock.unlock();
1764 writeLock.lock();
1765 writeLock.unlock();
1766 synchronized (stateLock) {
1767 if (state == ST_CLOSING) {
1768 tryClose();
1769 }
1770 }
1771 }
1772
1773 /**
1774 * Invoked by implCloseChannel to close the channel.
1775 */
1776 @Override
1777 protected void implCloseSelectableChannel() throws IOException {
1778 assert !isOpen();
1779 if (isBlocking()) {
1780 implCloseBlockingMode();
1781 } else {
1782 implCloseNonBlockingMode();
1783 }
1784 }
1785
1786 @Override
1787 public void kill() {
1788 synchronized (stateLock) {
1789 if (state == ST_CLOSING) {
1790 tryFinishClose();
1791 }
1792 }
1793 }
1794
1795 /**
1796 * Translates native poll revent set into a ready operation set
1797 */
1798 public boolean translateReadyOps(int ops, int initialOps, SelectionKeyImpl ski) {
1799 int intOps = ski.nioInterestOps();
1800 int oldOps = ski.nioReadyOps();
1801 int newOps = initialOps;
1802
1803 if ((ops & Net.POLLNVAL) != 0) {
1804 // This should only happen if this channel is pre-closed while a
1805 // selection operation is in progress
1806 // ## Throw an error if this channel has not been pre-closed
1807 return false;
1808 }
1809
1810 if ((ops & (Net.POLLERR | Net.POLLHUP)) != 0) {
1811 newOps = intOps;
1812 ski.nioReadyOps(newOps);
1813 return (newOps & ~oldOps) != 0;
1814 }
1815
1816 if (((ops & Net.POLLIN) != 0) &&
1817 ((intOps & SelectionKey.OP_READ) != 0))
1818 newOps |= SelectionKey.OP_READ;
1819
1820 if (((ops & Net.POLLOUT) != 0) &&
1821 ((intOps & SelectionKey.OP_WRITE) != 0))
1822 newOps |= SelectionKey.OP_WRITE;
1823
1824 ski.nioReadyOps(newOps);
1825 return (newOps & ~oldOps) != 0;
1826 }
1827
1828 public boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl ski) {
1829 return translateReadyOps(ops, ski.nioReadyOps(), ski);
1830 }
1831
1832 public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl ski) {
1833 return translateReadyOps(ops, 0, ski);
1834 }
1835
1836 /**
1837 * Translates an interest operation set into a native poll event set
1838 */
1839 public int translateInterestOps(int ops) {
1840 int newOps = 0;
1841 if ((ops & SelectionKey.OP_READ) != 0)
1842 newOps |= Net.POLLIN;
1843 if ((ops & SelectionKey.OP_WRITE) != 0)
1844 newOps |= Net.POLLOUT;
1845 if ((ops & SelectionKey.OP_CONNECT) != 0)
1846 newOps |= Net.POLLIN;
1847 return newOps;
1848 }
1849
1850 public FileDescriptor getFD() {
1851 return fd;
1852 }
1853
1854 public int getFDVal() {
1855 return fdVal;
1856 }
1857
1858 /**
1859 * Returns an action to release the given file descriptor and socket addresses.
1860 */
1861 private static Runnable releaserFor(FileDescriptor fd, NativeSocketAddress... sockAddrs) {
1862 return () -> {
1863 try {
1864 nd.close(fd);
1865 } catch (IOException ioe) {
1866 throw new UncheckedIOException(ioe);
1867 } finally {
1868 // decrement socket count and release memory
1869 ResourceManager.afterUdpClose();
1870 NativeSocketAddress.freeAll(sockAddrs);
1871 }
1872 };
1873 }
1874
1875 // -- Native methods --
1876
1877 private static native void disconnect0(FileDescriptor fd, boolean isIPv6)
1878 throws IOException;
1879
1880 private static native int receive0(FileDescriptor fd, long address, int len,
1881 long senderAddress, boolean connected)
1882 throws IOException;
1883
1884 private static native int send0(FileDescriptor fd, long address, int len,
1885 long targetAddress, int targetAddressLen)
1886 throws IOException;
1887
1888 static {
1889 IOUtil.load();
1890 }
1891 }