1 /*
2 * Copyright (c) 2001, 2019, 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.net.DatagramSocket;
31 import java.net.Inet4Address;
32 import java.net.Inet6Address;
33 import java.net.InetAddress;
34 import java.net.InetSocketAddress;
35 import java.net.NetworkInterface;
36 import java.net.PortUnreachableException;
37 import java.net.ProtocolFamily;
38 import java.net.SocketAddress;
39 import java.net.SocketOption;
40 import java.net.StandardProtocolFamily;
41 import java.net.StandardSocketOptions;
42 import java.nio.ByteBuffer;
43 import java.nio.channels.AlreadyBoundException;
44 import java.nio.channels.AlreadyConnectedException;
45 import java.nio.channels.AsynchronousCloseException;
46 import java.nio.channels.ClosedChannelException;
47 import java.nio.channels.DatagramChannel;
48 import java.nio.channels.MembershipKey;
49 import java.nio.channels.NotYetConnectedException;
50 import java.nio.channels.SelectionKey;
51 import java.nio.channels.spi.SelectorProvider;
52 import java.util.Collections;
53 import java.util.HashSet;
54 import java.util.Objects;
55 import java.util.Set;
56 import java.util.concurrent.locks.ReentrantLock;
57
58 import sun.net.ResourceManager;
59 import sun.net.ext.ExtendedSocketOptions;
60 import sun.net.util.IPAddressUtil;
61
62 /**
63 * An implementation of DatagramChannels.
64 */
65
66 class DatagramChannelImpl
67 extends DatagramChannel
68 implements SelChImpl
69 {
70 // Used to make native read and write calls
71 private static NativeDispatcher nd = new DatagramDispatcher();
72
73 // The protocol family of the socket
74 private final ProtocolFamily family;
75
76 // Our file descriptor
77 private final FileDescriptor fd;
78 private final int fdVal;
79
80 // Cached InetAddress and port for unconnected DatagramChannels
81 // used by receive0
82 private InetAddress cachedSenderInetAddress;
83 private int cachedSenderPort;
84
85 // Lock held by current reading or connecting thread
86 private final ReentrantLock readLock = new ReentrantLock();
87
88 // Lock held by current writing or connecting thread
89 private final ReentrantLock writeLock = new ReentrantLock();
90
91 // Lock held by any thread that modifies the state fields declared below
92 // DO NOT invoke a blocking I/O operation while holding this lock!
93 private final Object stateLock = new Object();
94
95 // -- The following fields are protected by stateLock
96
97 // State (does not necessarily increase monotonically)
98 private static final int ST_UNCONNECTED = 0;
99 private static final int ST_CONNECTED = 1;
100 private static final int ST_CLOSING = 2;
101 private static final int ST_CLOSED = 3;
102 private int state;
103
104 // IDs of native threads doing reads and writes, for signalling
105 private long readerThread;
106 private long writerThread;
107
108 // Binding and remote address (when connected)
109 private InetSocketAddress localAddress;
110 private InetSocketAddress remoteAddress;
111
112 // Our socket adaptor, if any
113 private DatagramSocket socket;
114
115 // Multicast support
116 private MembershipRegistry registry;
117
118 // set true when socket is bound and SO_REUSEADDRESS is emulated
119 private boolean reuseAddressEmulated;
120
121 // set true/false when socket is already bound and SO_REUSEADDR is emulated
122 private boolean isReuseAddress;
123
124 // -- End of fields protected by stateLock
125
126 public DatagramChannelImpl(SelectorProvider sp)
127 throws IOException
128 {
129 super(sp);
130 ResourceManager.beforeUdpCreate();
131 try {
132 this.family = Net.isIPv6Available()
133 ? StandardProtocolFamily.INET6
134 : StandardProtocolFamily.INET;
135 this.fd = Net.socket(family, false);
136 this.fdVal = IOUtil.fdVal(fd);
137 } catch (IOException ioe) {
138 ResourceManager.afterUdpClose();
139 throw ioe;
140 }
141 }
142
143 public DatagramChannelImpl(SelectorProvider sp, ProtocolFamily family)
144 throws IOException
145 {
146 super(sp);
147 Objects.requireNonNull(family, "'family' is null");
148 if ((family != StandardProtocolFamily.INET) &&
149 (family != StandardProtocolFamily.INET6)) {
150 throw new UnsupportedOperationException("Protocol family not supported");
151 }
152 if (family == StandardProtocolFamily.INET6) {
153 if (!Net.isIPv6Available()) {
154 throw new UnsupportedOperationException("IPv6 not available");
155 }
156 }
157
158 ResourceManager.beforeUdpCreate();
159 try {
160 this.family = family;
161 this.fd = Net.socket(family, false);
162 this.fdVal = IOUtil.fdVal(fd);
163 } catch (IOException ioe) {
164 ResourceManager.afterUdpClose();
165 throw ioe;
166 }
167 }
168
169 public DatagramChannelImpl(SelectorProvider sp, FileDescriptor fd)
170 throws IOException
171 {
172 super(sp);
173
174 // increment UDP count to match decrement when closing
175 ResourceManager.beforeUdpCreate();
176
177 this.family = Net.isIPv6Available()
178 ? StandardProtocolFamily.INET6
179 : StandardProtocolFamily.INET;
180 this.fd = fd;
181 this.fdVal = IOUtil.fdVal(fd);
182 synchronized (stateLock) {
183 this.localAddress = Net.localAddress(fd);
184 }
185 }
186
187 // @throws ClosedChannelException if channel is closed
188 private void ensureOpen() throws ClosedChannelException {
189 if (!isOpen())
190 throw new ClosedChannelException();
191 }
192
193 @Override
194 public DatagramSocket socket() {
195 synchronized (stateLock) {
196 if (socket == null)
197 socket = DatagramSocketAdaptor.create(this);
198 return socket;
199 }
200 }
201
202 @Override
203 public SocketAddress getLocalAddress() throws IOException {
204 synchronized (stateLock) {
205 ensureOpen();
206 // Perform security check before returning address
207 return Net.getRevealedLocalAddress(localAddress);
208 }
209 }
210
211 @Override
212 public SocketAddress getRemoteAddress() throws IOException {
213 synchronized (stateLock) {
214 ensureOpen();
215 return remoteAddress;
216 }
217 }
218
219 @Override
220 public <T> DatagramChannel setOption(SocketOption<T> name, T value)
221 throws IOException
222 {
223 Objects.requireNonNull(name);
224 if (!supportedOptions().contains(name))
225 throw new UnsupportedOperationException("'" + name + "' not supported");
226 if (!name.type().isInstance(value))
227 throw new IllegalArgumentException("Invalid value '" + value + "'");
228
229 synchronized (stateLock) {
230 ensureOpen();
231
232 if (name == StandardSocketOptions.IP_TOS ||
233 name == StandardSocketOptions.IP_MULTICAST_TTL ||
234 name == StandardSocketOptions.IP_MULTICAST_LOOP)
235 {
236 // options are protocol dependent
237 Net.setSocketOption(fd, family, name, value);
238 return this;
239 }
240
241 if (name == StandardSocketOptions.IP_MULTICAST_IF) {
242 NetworkInterface interf = (NetworkInterface)value;
243 if (family == StandardProtocolFamily.INET6) {
244 int index = interf.getIndex();
245 if (index == -1)
246 throw new IOException("Network interface cannot be identified");
247 Net.setInterface6(fd, index);
248 } else {
249 // need IPv4 address to identify interface
250 Inet4Address target = Net.anyInet4Address(interf);
251 if (target == null)
252 throw new IOException("Network interface not configured for IPv4");
253 int targetAddress = Net.inet4AsInt(target);
254 Net.setInterface4(fd, targetAddress);
255 }
256 return this;
257 }
258 if (name == StandardSocketOptions.SO_REUSEADDR
259 && Net.useExclusiveBind() && localAddress != null) {
260 reuseAddressEmulated = true;
261 this.isReuseAddress = (Boolean)value;
262 }
263
264 // remaining options don't need any special handling
265 Net.setSocketOption(fd, Net.UNSPEC, name, value);
266 return this;
267 }
268 }
269
270 @Override
271 @SuppressWarnings("unchecked")
272 public <T> T getOption(SocketOption<T> name)
273 throws IOException
274 {
275 Objects.requireNonNull(name);
276 if (!supportedOptions().contains(name))
277 throw new UnsupportedOperationException("'" + name + "' not supported");
278
279 synchronized (stateLock) {
280 ensureOpen();
281
282 if (name == StandardSocketOptions.IP_TOS ||
283 name == StandardSocketOptions.IP_MULTICAST_TTL ||
284 name == StandardSocketOptions.IP_MULTICAST_LOOP)
285 {
286 return (T) Net.getSocketOption(fd, family, name);
287 }
288
289 if (name == StandardSocketOptions.IP_MULTICAST_IF) {
290 if (family == StandardProtocolFamily.INET) {
291 int address = Net.getInterface4(fd);
292 if (address == 0)
293 return null; // default interface
294
295 InetAddress ia = Net.inet4FromInt(address);
296 NetworkInterface ni = NetworkInterface.getByInetAddress(ia);
297 if (ni == null)
298 throw new IOException("Unable to map address to interface");
299 return (T) ni;
300 } else {
301 int index = Net.getInterface6(fd);
302 if (index == 0)
303 return null; // default interface
304
305 NetworkInterface ni = NetworkInterface.getByIndex(index);
306 if (ni == null)
307 throw new IOException("Unable to map index to interface");
308 return (T) ni;
309 }
310 }
311
312 if (name == StandardSocketOptions.SO_REUSEADDR && reuseAddressEmulated) {
313 return (T)Boolean.valueOf(isReuseAddress);
314 }
315
316 // no special handling
317 return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
318 }
319 }
320
321 private static class DefaultOptionsHolder {
322 static final Set<SocketOption<?>> defaultOptions = defaultOptions();
323
324 private static Set<SocketOption<?>> defaultOptions() {
325 HashSet<SocketOption<?>> set = new HashSet<>();
326 set.add(StandardSocketOptions.SO_SNDBUF);
327 set.add(StandardSocketOptions.SO_RCVBUF);
328 set.add(StandardSocketOptions.SO_REUSEADDR);
329 if (Net.isReusePortAvailable()) {
330 set.add(StandardSocketOptions.SO_REUSEPORT);
331 }
332 set.add(StandardSocketOptions.SO_BROADCAST);
333 set.add(StandardSocketOptions.IP_TOS);
334 set.add(StandardSocketOptions.IP_MULTICAST_IF);
335 set.add(StandardSocketOptions.IP_MULTICAST_TTL);
336 set.add(StandardSocketOptions.IP_MULTICAST_LOOP);
337 set.addAll(ExtendedSocketOptions.datagramSocketOptions());
338 return Collections.unmodifiableSet(set);
339 }
340 }
341
342 @Override
343 public final Set<SocketOption<?>> supportedOptions() {
344 return DefaultOptionsHolder.defaultOptions;
345 }
346
347 /**
348 * Marks the beginning of a read operation that might block.
349 *
350 * @param blocking true if configured blocking
351 * @param mustBeConnected true if the socket must be connected
352 * @return remote address if connected
353 * @throws ClosedChannelException if the channel is closed
354 * @throws NotYetConnectedException if mustBeConnected and not connected
355 * @throws IOException if socket not bound and cannot be bound
356 */
357 private SocketAddress beginRead(boolean blocking, boolean mustBeConnected)
358 throws IOException
359 {
360 if (blocking) {
361 // set hook for Thread.interrupt
362 begin();
363 }
364 SocketAddress remote;
365 synchronized (stateLock) {
366 ensureOpen();
367 remote = remoteAddress;
368 if ((remote == null) && mustBeConnected)
369 throw new NotYetConnectedException();
370 if (localAddress == null)
371 bindInternal(null);
372 if (blocking)
373 readerThread = NativeThread.current();
374 }
375 return remote;
376 }
377
378 /**
379 * Marks the end of a read operation that may have blocked.
380 *
381 * @throws AsynchronousCloseException if the channel was closed asynchronously
382 */
383 private void endRead(boolean blocking, boolean completed)
384 throws AsynchronousCloseException
385 {
386 if (blocking) {
387 synchronized (stateLock) {
388 readerThread = 0;
389 if (state == ST_CLOSING) {
390 tryFinishClose();
391 }
392 }
393 // remove hook for Thread.interrupt
394 end(completed);
395 }
396 }
397
398 private SocketAddress sender; // Set by receive0 (## ugh)
399
400 @Override
401 public SocketAddress receive(ByteBuffer dst) throws IOException {
402 if (dst.isReadOnly())
403 throw new IllegalArgumentException("Read-only buffer");
404
405 readLock.lock();
406 try {
407 boolean blocking = isBlocking();
408 int n = 0;
409 ByteBuffer bb = null;
410 try {
411 SocketAddress remote = beginRead(blocking, false);
412 boolean connected = (remote != null);
413 SecurityManager sm = System.getSecurityManager();
414 if (connected || (sm == null)) {
415 // connected or no security manager
416 n = receive(fd, dst, connected);
417 if (blocking) {
418 while (IOStatus.okayToRetry(n) && isOpen()) {
419 park(Net.POLLIN);
420 n = receive(fd, dst, connected);
421 }
422 } else if (n == IOStatus.UNAVAILABLE) {
423 return null;
424 }
425 } else {
426 // Cannot receive into user's buffer when running with a
427 // security manager and not connected
428 bb = Util.getTemporaryDirectBuffer(dst.remaining());
429 for (;;) {
430 n = receive(fd, bb, connected);
431 if (blocking) {
432 while (IOStatus.okayToRetry(n) && isOpen()) {
433 park(Net.POLLIN);
434 n = receive(fd, bb, connected);
435 }
436 } else if (n == IOStatus.UNAVAILABLE) {
437 return null;
438 }
439 InetSocketAddress isa = (InetSocketAddress)sender;
440 try {
441 sm.checkAccept(isa.getAddress().getHostAddress(),
442 isa.getPort());
443 } catch (SecurityException se) {
444 // Ignore packet
445 bb.clear();
446 n = 0;
447 continue;
448 }
449 bb.flip();
450 dst.put(bb);
451 break;
452 }
453 }
454 assert sender != null;
455 return sender;
456 } finally {
457 if (bb != null)
458 Util.releaseTemporaryDirectBuffer(bb);
459 endRead(blocking, n > 0);
460 assert IOStatus.check(n);
461 }
462 } finally {
463 readLock.unlock();
464 }
465 }
466
467 private int receive(FileDescriptor fd, ByteBuffer dst, boolean connected)
468 throws IOException
469 {
470 int pos = dst.position();
471 int lim = dst.limit();
472 assert (pos <= lim);
473 int rem = (pos <= lim ? lim - pos : 0);
474 if (dst instanceof DirectBuffer && rem > 0)
475 return receiveIntoNativeBuffer(fd, dst, rem, pos, connected);
476
477 // Substitute a native buffer. If the supplied buffer is empty
478 // we must instead use a nonempty buffer, otherwise the call
479 // will not block waiting for a datagram on some platforms.
480 int newSize = Math.max(rem, 1);
481 ByteBuffer bb = Util.getTemporaryDirectBuffer(newSize);
482 try {
483 int n = receiveIntoNativeBuffer(fd, bb, newSize, 0, connected);
484 bb.flip();
485 if (n > 0 && rem > 0)
486 dst.put(bb);
487 return n;
488 } finally {
489 Util.releaseTemporaryDirectBuffer(bb);
490 }
491 }
492
493 private int receiveIntoNativeBuffer(FileDescriptor fd, ByteBuffer bb,
494 int rem, int pos, boolean connected)
495 throws IOException
496 {
497 int n = receive0(fd, ((DirectBuffer)bb).address() + pos, rem, connected);
498 if (n > 0)
499 bb.position(pos + n);
500 return n;
501 }
502
503 @Override
504 public int send(ByteBuffer src, SocketAddress target)
505 throws IOException
506 {
507 Objects.requireNonNull(src);
508 InetSocketAddress isa = Net.checkAddress(target, family);
509
510 writeLock.lock();
511 try {
512 boolean blocking = isBlocking();
513 int n = 0;
514 try {
515 SocketAddress remote = beginWrite(blocking, false);
516 if (remote != null) {
517 // connected
518 if (!target.equals(remote)) {
519 throw new AlreadyConnectedException();
520 }
521 n = IOUtil.write(fd, src, -1, nd);
522 if (blocking) {
523 while (IOStatus.okayToRetry(n) && isOpen()) {
524 park(Net.POLLOUT);
525 n = IOUtil.write(fd, src, -1, nd);
526 }
527 }
528 } else {
529 // not connected
530 SecurityManager sm = System.getSecurityManager();
531 InetAddress ia = isa.getAddress();
532 if (sm != null) {
533 if (ia.isMulticastAddress()) {
534 sm.checkMulticast(ia);
535 } else {
536 sm.checkConnect(ia.getHostAddress(), isa.getPort());
537 }
538 }
539 if (ia.isLinkLocalAddress())
540 isa = IPAddressUtil.toScopedAddress(isa);
541 n = send(fd, src, isa);
542 if (blocking) {
543 while (IOStatus.okayToRetry(n) && isOpen()) {
544 park(Net.POLLOUT);
545 n = send(fd, src, isa);
546 }
547 }
548 }
549 } finally {
550 endWrite(blocking, n > 0);
551 assert IOStatus.check(n);
552 }
553 return IOStatus.normalize(n);
554 } finally {
555 writeLock.unlock();
556 }
557 }
558
559 private int send(FileDescriptor fd, ByteBuffer src, InetSocketAddress target)
560 throws IOException
561 {
562 if (src instanceof DirectBuffer)
563 return sendFromNativeBuffer(fd, src, target);
564
565 // Substitute a native buffer
566 int pos = src.position();
567 int lim = src.limit();
568 assert (pos <= lim);
569 int rem = (pos <= lim ? lim - pos : 0);
570
571 ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
572 try {
573 bb.put(src);
574 bb.flip();
575 // Do not update src until we see how many bytes were written
576 src.position(pos);
577
578 int n = sendFromNativeBuffer(fd, bb, target);
579 if (n > 0) {
580 // now update src
581 src.position(pos + n);
582 }
583 return n;
584 } finally {
585 Util.releaseTemporaryDirectBuffer(bb);
586 }
587 }
588
589 private int sendFromNativeBuffer(FileDescriptor fd, ByteBuffer bb,
590 InetSocketAddress target)
591 throws IOException
592 {
593 int pos = bb.position();
594 int lim = bb.limit();
595 assert (pos <= lim);
596 int rem = (pos <= lim ? lim - pos : 0);
597
598 boolean preferIPv6 = (family != StandardProtocolFamily.INET);
599 int written;
600 try {
601 written = send0(preferIPv6, fd, ((DirectBuffer)bb).address() + pos,
602 rem, target.getAddress(), target.getPort());
603 } catch (PortUnreachableException pue) {
604 if (isConnected())
605 throw pue;
606 written = rem;
607 }
608 if (written > 0)
609 bb.position(pos + written);
610 return written;
611 }
612
613 @Override
614 public int read(ByteBuffer buf) throws IOException {
615 Objects.requireNonNull(buf);
616
617 readLock.lock();
618 try {
619 boolean blocking = isBlocking();
620 int n = 0;
621 try {
622 beginRead(blocking, true);
623 n = IOUtil.read(fd, buf, -1, nd);
624 if (blocking) {
625 while (IOStatus.okayToRetry(n) && isOpen()) {
626 park(Net.POLLIN);
627 n = IOUtil.read(fd, buf, -1, nd);
628 }
629 }
630 } finally {
631 endRead(blocking, n > 0);
632 assert IOStatus.check(n);
633 }
634 return IOStatus.normalize(n);
635 } finally {
636 readLock.unlock();
637 }
638 }
639
640 @Override
641 public long read(ByteBuffer[] dsts, int offset, int length)
642 throws IOException
643 {
644 Objects.checkFromIndexSize(offset, length, dsts.length);
645
646 readLock.lock();
647 try {
648 boolean blocking = isBlocking();
649 long n = 0;
650 try {
651 beginRead(blocking, true);
652 n = IOUtil.read(fd, dsts, offset, length, nd);
653 if (blocking) {
654 while (IOStatus.okayToRetry(n) && isOpen()) {
655 park(Net.POLLIN);
656 n = IOUtil.read(fd, dsts, offset, length, nd);
657 }
658 }
659 } finally {
660 endRead(blocking, n > 0);
661 assert IOStatus.check(n);
662 }
663 return IOStatus.normalize(n);
664 } finally {
665 readLock.unlock();
666 }
667 }
668
669 /**
670 * Marks the beginning of a write operation that might block.
671 * @param blocking true if configured blocking
672 * @param mustBeConnected true if the socket must be connected
673 * @return remote address if connected
674 * @throws ClosedChannelException if the channel is closed
675 * @throws NotYetConnectedException if mustBeConnected and not connected
676 * @throws IOException if socket not bound and cannot be bound
677 */
678 private SocketAddress beginWrite(boolean blocking, boolean mustBeConnected)
679 throws IOException
680 {
681 if (blocking) {
682 // set hook for Thread.interrupt
683 begin();
684 }
685 SocketAddress remote;
686 synchronized (stateLock) {
687 ensureOpen();
688 remote = remoteAddress;
689 if ((remote == null) && mustBeConnected)
690 throw new NotYetConnectedException();
691 if (localAddress == null)
692 bindInternal(null);
693 if (blocking)
694 writerThread = NativeThread.current();
695 }
696 return remote;
697 }
698
699 /**
700 * Marks the end of a write operation that may have blocked.
701 *
702 * @throws AsynchronousCloseException if the channel was closed asynchronously
703 */
704 private void endWrite(boolean blocking, boolean completed)
705 throws AsynchronousCloseException
706 {
707 if (blocking) {
708 synchronized (stateLock) {
709 writerThread = 0;
710 if (state == ST_CLOSING) {
711 tryFinishClose();
712 }
713 }
714 // remove hook for Thread.interrupt
715 end(completed);
716 }
717 }
718
719 @Override
720 public int write(ByteBuffer buf) throws IOException {
721 Objects.requireNonNull(buf);
722
723 writeLock.lock();
724 try {
725 boolean blocking = isBlocking();
726 int n = 0;
727 try {
728 beginWrite(blocking, true);
729 n = IOUtil.write(fd, buf, -1, nd);
730 if (blocking) {
731 while (IOStatus.okayToRetry(n) && isOpen()) {
732 park(Net.POLLOUT);
733 n = IOUtil.write(fd, buf, -1, nd);
734 }
735 }
736 } finally {
737 endWrite(blocking, n > 0);
738 assert IOStatus.check(n);
739 }
740 return IOStatus.normalize(n);
741 } finally {
742 writeLock.unlock();
743 }
744 }
745
746 @Override
747 public long write(ByteBuffer[] srcs, int offset, int length)
748 throws IOException
749 {
750 Objects.checkFromIndexSize(offset, length, srcs.length);
751
752 writeLock.lock();
753 try {
754 boolean blocking = isBlocking();
755 long n = 0;
756 try {
757 beginWrite(blocking, true);
758 n = IOUtil.write(fd, srcs, offset, length, nd);
759 if (blocking) {
760 while (IOStatus.okayToRetry(n) && isOpen()) {
761 park(Net.POLLOUT);
762 n = IOUtil.write(fd, srcs, offset, length, nd);
763 }
764 }
765 } finally {
766 endWrite(blocking, n > 0);
767 assert IOStatus.check(n);
768 }
769 return IOStatus.normalize(n);
770 } finally {
771 writeLock.unlock();
772 }
773 }
774
775 @Override
776 protected void implConfigureBlocking(boolean block) throws IOException {
777 readLock.lock();
778 try {
779 writeLock.lock();
780 try {
781 synchronized (stateLock) {
782 ensureOpen();
783 IOUtil.configureBlocking(fd, block);
784 }
785 } finally {
786 writeLock.unlock();
787 }
788 } finally {
789 readLock.unlock();
790 }
791 }
792
793 InetSocketAddress localAddress() {
794 synchronized (stateLock) {
795 return localAddress;
796 }
797 }
798
799 InetSocketAddress remoteAddress() {
800 synchronized (stateLock) {
801 return remoteAddress;
802 }
803 }
804
805 @Override
806 public DatagramChannel bind(SocketAddress local) throws IOException {
807 readLock.lock();
808 try {
809 writeLock.lock();
810 try {
811 synchronized (stateLock) {
812 ensureOpen();
813 if (localAddress != null)
814 throw new AlreadyBoundException();
815 bindInternal(local);
816 }
817 } finally {
818 writeLock.unlock();
819 }
820 } finally {
821 readLock.unlock();
822 }
823 return this;
824 }
825
826 private void bindInternal(SocketAddress local) throws IOException {
827 assert Thread.holdsLock(stateLock )&& (localAddress == null);
828
829 InetSocketAddress isa;
830 if (local == null) {
831 // only Inet4Address allowed with IPv4 socket
832 if (family == StandardProtocolFamily.INET) {
833 isa = new InetSocketAddress(InetAddress.getByName("0.0.0.0"), 0);
834 } else {
835 isa = new InetSocketAddress(0);
836 }
837 } else {
838 isa = Net.checkAddress(local, family);
839 }
840 SecurityManager sm = System.getSecurityManager();
841 if (sm != null)
842 sm.checkListen(isa.getPort());
843
844 Net.bind(family, fd, isa.getAddress(), isa.getPort());
845 localAddress = Net.localAddress(fd);
846 }
847
848 @Override
849 public boolean isConnected() {
850 synchronized (stateLock) {
851 return (state == ST_CONNECTED);
852 }
853 }
854
855 @Override
856 public DatagramChannel connect(SocketAddress sa) throws IOException {
857 InetSocketAddress isa = Net.checkAddress(sa, family);
858 SecurityManager sm = System.getSecurityManager();
859 if (sm != null) {
860 InetAddress ia = isa.getAddress();
861 if (ia.isMulticastAddress()) {
862 sm.checkMulticast(ia);
863 } else {
864 sm.checkConnect(ia.getHostAddress(), isa.getPort());
865 sm.checkAccept(ia.getHostAddress(), isa.getPort());
866 }
867 }
868
869 readLock.lock();
870 try {
871 writeLock.lock();
872 try {
873 synchronized (stateLock) {
874 ensureOpen();
875 if (state == ST_CONNECTED)
876 throw new AlreadyConnectedException();
877
878 int n = Net.connect(family,
879 fd,
880 isa.getAddress(),
881 isa.getPort());
882 if (n <= 0)
883 throw new Error(); // Can't happen
884
885 // connected
886 remoteAddress = isa;
887 state = ST_CONNECTED;
888
889 // refresh local address
890 localAddress = Net.localAddress(fd);
891
892 // flush any packets already received.
893 boolean blocking = isBlocking();
894 if (blocking) {
895 IOUtil.configureBlocking(fd, false);
896 }
897 try {
898 ByteBuffer buf = ByteBuffer.allocate(100);
899 while (receive(fd, buf, false) > 0) {
900 buf.clear();
901 }
902 } finally {
903 if (blocking) {
904 IOUtil.configureBlocking(fd, true);
905 }
906 }
907 }
908 } finally {
909 writeLock.unlock();
910 }
911 } finally {
912 readLock.unlock();
913 }
914 return this;
915 }
916
917 @Override
918 public DatagramChannel disconnect() throws IOException {
919 readLock.lock();
920 try {
921 writeLock.lock();
922 try {
923 synchronized (stateLock) {
924 if (!isOpen() || (state != ST_CONNECTED))
925 return this;
926
927 // disconnect socket
928 boolean isIPv6 = (family == StandardProtocolFamily.INET6);
929 disconnect0(fd, isIPv6);
930
931 // no longer connected
932 remoteAddress = null;
933 state = ST_UNCONNECTED;
934
935 // refresh local address
936 localAddress = Net.localAddress(fd);
937 }
938 } finally {
939 writeLock.unlock();
940 }
941 } finally {
942 readLock.unlock();
943 }
944 return this;
945 }
946
947 /**
948 * Joins channel's socket to the given group/interface and
949 * optional source address.
950 */
951 private MembershipKey innerJoin(InetAddress group,
952 NetworkInterface interf,
953 InetAddress source)
954 throws IOException
955 {
956 if (!group.isMulticastAddress())
957 throw new IllegalArgumentException("Group not a multicast address");
958
959 // check multicast address is compatible with this socket
960 if (group instanceof Inet4Address) {
961 if (family == StandardProtocolFamily.INET6 && !Net.canIPv6SocketJoinIPv4Group())
962 throw new IllegalArgumentException("IPv6 socket cannot join IPv4 multicast group");
963 } else if (group instanceof Inet6Address) {
964 if (family != StandardProtocolFamily.INET6)
965 throw new IllegalArgumentException("Only IPv6 sockets can join IPv6 multicast group");
966 } else {
967 throw new IllegalArgumentException("Address type not supported");
968 }
969
970 // check source address
971 if (source != null) {
972 if (source.isAnyLocalAddress())
973 throw new IllegalArgumentException("Source address is a wildcard address");
974 if (source.isMulticastAddress())
975 throw new IllegalArgumentException("Source address is multicast address");
976 if (source.getClass() != group.getClass())
977 throw new IllegalArgumentException("Source address is different type to group");
978 }
979
980 SecurityManager sm = System.getSecurityManager();
981 if (sm != null)
982 sm.checkMulticast(group);
983
984 synchronized (stateLock) {
985 ensureOpen();
986
987 // check the registry to see if we are already a member of the group
988 if (registry == null) {
989 registry = new MembershipRegistry();
990 } else {
991 // return existing membership key
992 MembershipKey key = registry.checkMembership(group, interf, source);
993 if (key != null)
994 return key;
995 }
996
997 MembershipKeyImpl key;
998 if ((family == StandardProtocolFamily.INET6) &&
999 ((group instanceof Inet6Address) || Net.canJoin6WithIPv4Group()))
1000 {
1001 int index = interf.getIndex();
1002 if (index == -1)
1003 throw new IOException("Network interface cannot be identified");
1004
1005 // need multicast and source address as byte arrays
1006 byte[] groupAddress = Net.inet6AsByteArray(group);
1007 byte[] sourceAddress = (source == null) ? null :
1008 Net.inet6AsByteArray(source);
1009
1010 // join the group
1011 int n = Net.join6(fd, groupAddress, index, sourceAddress);
1012 if (n == IOStatus.UNAVAILABLE)
1013 throw new UnsupportedOperationException();
1014
1015 key = new MembershipKeyImpl.Type6(this, group, interf, source,
1016 groupAddress, index, sourceAddress);
1017
1018 } else {
1019 // need IPv4 address to identify interface
1020 Inet4Address target = Net.anyInet4Address(interf);
1021 if (target == null)
1022 throw new IOException("Network interface not configured for IPv4");
1023
1024 int groupAddress = Net.inet4AsInt(group);
1025 int targetAddress = Net.inet4AsInt(target);
1026 int sourceAddress = (source == null) ? 0 : Net.inet4AsInt(source);
1027
1028 // join the group
1029 int n = Net.join4(fd, groupAddress, targetAddress, sourceAddress);
1030 if (n == IOStatus.UNAVAILABLE)
1031 throw new UnsupportedOperationException();
1032
1033 key = new MembershipKeyImpl.Type4(this, group, interf, source,
1034 groupAddress, targetAddress, sourceAddress);
1035 }
1036
1037 registry.add(key);
1038 return key;
1039 }
1040 }
1041
1042 @Override
1043 public MembershipKey join(InetAddress group,
1044 NetworkInterface interf)
1045 throws IOException
1046 {
1047 return innerJoin(group, interf, null);
1048 }
1049
1050 @Override
1051 public MembershipKey join(InetAddress group,
1052 NetworkInterface interf,
1053 InetAddress source)
1054 throws IOException
1055 {
1056 Objects.requireNonNull(source);
1057 return innerJoin(group, interf, source);
1058 }
1059
1060 // package-private
1061 void drop(MembershipKeyImpl key) {
1062 assert key.channel() == this;
1063
1064 synchronized (stateLock) {
1065 if (!key.isValid())
1066 return;
1067
1068 try {
1069 if (key instanceof MembershipKeyImpl.Type6) {
1070 MembershipKeyImpl.Type6 key6 =
1071 (MembershipKeyImpl.Type6)key;
1072 Net.drop6(fd, key6.groupAddress(), key6.index(), key6.source());
1073 } else {
1074 MembershipKeyImpl.Type4 key4 = (MembershipKeyImpl.Type4)key;
1075 Net.drop4(fd, key4.groupAddress(), key4.interfaceAddress(),
1076 key4.source());
1077 }
1078 } catch (IOException ioe) {
1079 // should not happen
1080 throw new AssertionError(ioe);
1081 }
1082
1083 key.invalidate();
1084 registry.remove(key);
1085 }
1086 }
1087
1088 /**
1089 * Block datagrams from given source if a memory to receive all
1090 * datagrams.
1091 */
1092 void block(MembershipKeyImpl key, InetAddress source)
1093 throws IOException
1094 {
1095 assert key.channel() == this;
1096 assert key.sourceAddress() == null;
1097
1098 synchronized (stateLock) {
1099 if (!key.isValid())
1100 throw new IllegalStateException("key is no longer valid");
1101 if (source.isAnyLocalAddress())
1102 throw new IllegalArgumentException("Source address is a wildcard address");
1103 if (source.isMulticastAddress())
1104 throw new IllegalArgumentException("Source address is multicast address");
1105 if (source.getClass() != key.group().getClass())
1106 throw new IllegalArgumentException("Source address is different type to group");
1107
1108 int n;
1109 if (key instanceof MembershipKeyImpl.Type6) {
1110 MembershipKeyImpl.Type6 key6 =
1111 (MembershipKeyImpl.Type6)key;
1112 n = Net.block6(fd, key6.groupAddress(), key6.index(),
1113 Net.inet6AsByteArray(source));
1114 } else {
1115 MembershipKeyImpl.Type4 key4 =
1116 (MembershipKeyImpl.Type4)key;
1117 n = Net.block4(fd, key4.groupAddress(), key4.interfaceAddress(),
1118 Net.inet4AsInt(source));
1119 }
1120 if (n == IOStatus.UNAVAILABLE) {
1121 // ancient kernel
1122 throw new UnsupportedOperationException();
1123 }
1124 }
1125 }
1126
1127 /**
1128 * Unblock given source.
1129 */
1130 void unblock(MembershipKeyImpl key, InetAddress source) {
1131 assert key.channel() == this;
1132 assert key.sourceAddress() == null;
1133
1134 synchronized (stateLock) {
1135 if (!key.isValid())
1136 throw new IllegalStateException("key is no longer valid");
1137
1138 try {
1139 if (key instanceof MembershipKeyImpl.Type6) {
1140 MembershipKeyImpl.Type6 key6 =
1141 (MembershipKeyImpl.Type6)key;
1142 Net.unblock6(fd, key6.groupAddress(), key6.index(),
1143 Net.inet6AsByteArray(source));
1144 } else {
1145 MembershipKeyImpl.Type4 key4 =
1146 (MembershipKeyImpl.Type4)key;
1147 Net.unblock4(fd, key4.groupAddress(), key4.interfaceAddress(),
1148 Net.inet4AsInt(source));
1149 }
1150 } catch (IOException ioe) {
1151 // should not happen
1152 throw new AssertionError(ioe);
1153 }
1154 }
1155 }
1156
1157 /**
1158 * Closes the socket if there are no I/O operations in progress and the
1159 * channel is not registered with a Selector.
1160 */
1161 private boolean tryClose() throws IOException {
1162 assert Thread.holdsLock(stateLock) && state == ST_CLOSING;
1163 if ((readerThread == 0) && (writerThread == 0) && !isRegistered()) {
1164 state = ST_CLOSED;
1165 try {
1166 nd.close(fd);
1167 } finally {
1168 // notify resource manager
1169 ResourceManager.afterUdpClose();
1170 }
1171 return true;
1172 } else {
1173 return false;
1174 }
1175 }
1176
1177 /**
1178 * Invokes tryClose to attempt to close the socket.
1179 *
1180 * This method is used for deferred closing by I/O and Selector operations.
1181 */
1182 private void tryFinishClose() {
1183 try {
1184 tryClose();
1185 } catch (IOException ignore) { }
1186 }
1187
1188 /**
1189 * Closes this channel when configured in blocking mode.
1190 *
1191 * If there is an I/O operation in progress then the socket is pre-closed
1192 * and the I/O threads signalled, in which case the final close is deferred
1193 * until all I/O operations complete.
1194 */
1195 private void implCloseBlockingMode() throws IOException {
1196 synchronized (stateLock) {
1197 assert state < ST_CLOSING;
1198 state = ST_CLOSING;
1199
1200 // if member of any multicast groups then invalidate the keys
1201 if (registry != null)
1202 registry.invalidateAll();
1203
1204 if (!tryClose()) {
1205 long reader = readerThread;
1206 long writer = writerThread;
1207 if (reader != 0 || writer != 0) {
1208 nd.preClose(fd);
1209 if (reader != 0)
1210 NativeThread.signal(reader);
1211 if (writer != 0)
1212 NativeThread.signal(writer);
1213 }
1214 }
1215 }
1216 }
1217
1218 /**
1219 * Closes this channel when configured in non-blocking mode.
1220 *
1221 * If the channel is registered with a Selector then the close is deferred
1222 * until the channel is flushed from all Selectors.
1223 */
1224 private void implCloseNonBlockingMode() throws IOException {
1225 synchronized (stateLock) {
1226 assert state < ST_CLOSING;
1227 state = ST_CLOSING;
1228
1229 // if member of any multicast groups then invalidate the keys
1230 if (registry != null)
1231 registry.invalidateAll();
1232 }
1233
1234 // wait for any read/write operations to complete before trying to close
1235 readLock.lock();
1236 readLock.unlock();
1237 writeLock.lock();
1238 writeLock.unlock();
1239 synchronized (stateLock) {
1240 if (state == ST_CLOSING) {
1241 tryClose();
1242 }
1243 }
1244 }
1245
1246 /**
1247 * Invoked by implCloseChannel to close the channel.
1248 */
1249 @Override
1250 protected void implCloseSelectableChannel() throws IOException {
1251 assert !isOpen();
1252 if (isBlocking()) {
1253 implCloseBlockingMode();
1254 } else {
1255 implCloseNonBlockingMode();
1256 }
1257 }
1258
1259 @Override
1260 public void kill() {
1261 synchronized (stateLock) {
1262 if (state == ST_CLOSING) {
1263 tryFinishClose();
1264 }
1265 }
1266 }
1267
1268 @SuppressWarnings("deprecation")
1269 protected void finalize() throws IOException {
1270 // fd is null if constructor threw exception
1271 if (fd != null)
1272 close();
1273 }
1274
1275 /**
1276 * Translates native poll revent set into a ready operation set
1277 */
1278 public boolean translateReadyOps(int ops, int initialOps, SelectionKeyImpl ski) {
1279 int intOps = ski.nioInterestOps();
1280 int oldOps = ski.nioReadyOps();
1281 int newOps = initialOps;
1282
1283 if ((ops & Net.POLLNVAL) != 0) {
1284 // This should only happen if this channel is pre-closed while a
1285 // selection operation is in progress
1286 // ## Throw an error if this channel has not been pre-closed
1287 return false;
1288 }
1289
1290 if ((ops & (Net.POLLERR | Net.POLLHUP)) != 0) {
1291 newOps = intOps;
1292 ski.nioReadyOps(newOps);
1293 return (newOps & ~oldOps) != 0;
1294 }
1295
1296 if (((ops & Net.POLLIN) != 0) &&
1297 ((intOps & SelectionKey.OP_READ) != 0))
1298 newOps |= SelectionKey.OP_READ;
1299
1300 if (((ops & Net.POLLOUT) != 0) &&
1301 ((intOps & SelectionKey.OP_WRITE) != 0))
1302 newOps |= SelectionKey.OP_WRITE;
1303
1304 ski.nioReadyOps(newOps);
1305 return (newOps & ~oldOps) != 0;
1306 }
1307
1308 public boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl ski) {
1309 return translateReadyOps(ops, ski.nioReadyOps(), ski);
1310 }
1311
1312 public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl ski) {
1313 return translateReadyOps(ops, 0, ski);
1314 }
1315
1316 /**
1317 * Poll this channel's socket for reading up to the given timeout.
1318 * @return {@code true} if the socket is polled
1319 */
1320 boolean pollRead(long timeout) throws IOException {
1321 boolean blocking = isBlocking();
1322 assert Thread.holdsLock(blockingLock()) && blocking;
1323
1324 readLock.lock();
1325 try {
1326 boolean polled = false;
1327 try {
1328 beginRead(blocking, false);
1329 int events = Net.poll(fd, Net.POLLIN, timeout);
1330 polled = (events != 0);
1331 } finally {
1332 endRead(blocking, polled);
1333 }
1334 return polled;
1335 } finally {
1336 readLock.unlock();
1337 }
1338 }
1339
1340 /**
1341 * Translates an interest operation set into a native poll event set
1342 */
1343 public int translateInterestOps(int ops) {
1344 int newOps = 0;
1345 if ((ops & SelectionKey.OP_READ) != 0)
1346 newOps |= Net.POLLIN;
1347 if ((ops & SelectionKey.OP_WRITE) != 0)
1348 newOps |= Net.POLLOUT;
1349 if ((ops & SelectionKey.OP_CONNECT) != 0)
1350 newOps |= Net.POLLIN;
1351 return newOps;
1352 }
1353
1354 public FileDescriptor getFD() {
1355 return fd;
1356 }
1357
1358 public int getFDVal() {
1359 return fdVal;
1360 }
1361
1362
1363 // -- Native methods --
1364
1365 private static native void initIDs();
1366
1367 private static native void disconnect0(FileDescriptor fd, boolean isIPv6)
1368 throws IOException;
1369
1370 private native int receive0(FileDescriptor fd, long address, int len,
1371 boolean connected)
1372 throws IOException;
1373
1374 private native int send0(boolean preferIPv6, FileDescriptor fd, long address,
1375 int len, InetAddress addr, int port)
1376 throws IOException;
1377
1378 static {
1379 IOUtil.load();
1380 initIDs();
1381 }
1382 }