1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.newsclub.net.unix;
19
20 import static org.newsclub.net.unix.NativeUnixSocket.SHUT_RD;
21 import static org.newsclub.net.unix.NativeUnixSocket.SHUT_RD_WR;
22 import static org.newsclub.net.unix.NativeUnixSocket.SHUT_WR;
23
24 import java.io.EOFException;
25 import java.io.FileDescriptor;
26 import java.io.IOException;
27 import java.io.InputStream;
28 import java.io.InterruptedIOException;
29 import java.io.OutputStream;
30 import java.net.InetAddress;
31 import java.net.SocketAddress;
32 import java.net.SocketException;
33 import java.net.SocketImpl;
34 import java.net.SocketOption;
35 import java.net.SocketOptions;
36 import java.net.SocketTimeoutException;
37 import java.nio.ByteBuffer;
38 import java.nio.channels.SelectionKey;
39 import java.util.Objects;
40 import java.util.Set;
41 import java.util.concurrent.TimeUnit;
42 import java.util.concurrent.atomic.AtomicBoolean;
43 import java.util.concurrent.atomic.AtomicInteger;
44
45 import org.eclipse.jdt.annotation.NonNull;
46 import org.eclipse.jdt.annotation.Nullable;
47 import org.newsclub.net.unix.pool.MutableHolder;
48 import org.newsclub.net.unix.pool.ObjectPool.Lease;
49
50
51
52
53
54
55
56 @SuppressWarnings({
57 "PMD.CyclomaticComplexity", "PMD.CouplingBetweenObjects",
58 "UnsafeFinalization" })
59 public abstract class AFSocketImpl<A extends AFSocketAddress> extends SocketImplShim {
60 private static final int SHUTDOWN_RD_WR = (1 << SHUT_RD) | (1 << SHUT_WR);
61
62 private final AFSocketStreamCore core;
63 final AncillaryDataSupport ancillaryDataSupport = new AncillaryDataSupport();
64
65 private final AtomicBoolean bound = new AtomicBoolean(false);
66 private Boolean createType = null;
67 private final AtomicBoolean connected = new AtomicBoolean(false);
68
69 private volatile boolean closedInputStream = false;
70 private volatile boolean closedOutputStream = false;
71
72 private final AFInputStream in;
73 private final AFOutputStream out;
74
75 private final AtomicBoolean reuseAddr = new AtomicBoolean(true);
76
77 private final AtomicInteger socketTimeout = new AtomicInteger(0);
78 private final AFAddressFamily<A> addressFamily;
79
80 private int shutdownState = 0;
81
82 private AFSocketImplExtensions<A> implExtensions = null;
83
84 private final AtomicBoolean closed = new AtomicBoolean(false);
85
86
87
88
89
90
91
92 static final class AFSocketStreamCore extends AFSocketCore {
93 AFSocketStreamCore(AFSocketImpl<?> observed, FileDescriptor fd,
94 AncillaryDataSupport ancillaryDataSupport, AFAddressFamily<?> af) {
95 super(observed, fd, ancillaryDataSupport, af, false);
96 }
97
98 void createSocket(FileDescriptor fdTarget, AFSocketType type) throws IOException {
99 NativeUnixSocket.createSocket(fdTarget, addressFamily().getDomain(), type.getId());
100 }
101
102
103
104
105
106 @Override
107 protected void unblockAccepts() {
108 if (socketAddress == null || socketAddress.getBytes() == null || inode.get() < 0) {
109 return;
110 }
111 if (!hasPendingAccepts()) {
112 return;
113 }
114 try {
115 ThreadUtil.runOnSystemThread(this::unblockAccepts0);
116 } catch (InterruptedException e) {
117
118 }
119 }
120
121 private void unblockAccepts0() {
122 while (hasPendingAccepts()) {
123 try {
124 FileDescriptor tmpFd = new FileDescriptor();
125
126 try (Lease<ByteBuffer> abLease = socketAddress.getNativeAddressDirectBuffer()) {
127 createSocket(tmpFd, AFSocketType.SOCK_STREAM);
128 ByteBuffer ab = abLease.get();
129 NativeUnixSocket.connect(ab, ab.limit(), tmpFd, inode.get());
130 } catch (IOException e) {
131
132
133 return;
134 }
135 if (isShutdownOnClose()) {
136 try {
137 NativeUnixSocket.shutdown(tmpFd, SHUT_RD_WR);
138 } catch (Exception e) {
139
140 }
141 }
142 try {
143 NativeUnixSocket.close(tmpFd);
144 } catch (Exception e) {
145
146 }
147 } catch (RuntimeException e) {
148
149 }
150
151
152 try {
153 Thread.sleep(5);
154 } catch (InterruptedException e) {
155
156 }
157 }
158 }
159 }
160
161
162
163
164
165
166
167 protected AFSocketImpl(AFAddressFamily<@NonNull A> addressFamily, FileDescriptor fdObj) {
168 super();
169 this.addressFamily = addressFamily;
170 this.address = InetAddress.getLoopbackAddress();
171 this.core = new AFSocketStreamCore(this, fdObj, ancillaryDataSupport, addressFamily);
172 this.fd = core.fd;
173 this.in = newInputStream();
174 this.out = newOutputStream();
175 }
176
177
178
179
180
181
182 protected final AFInputStream newInputStream() {
183 return new AFInputStreamImpl();
184 }
185
186
187
188
189
190
191 protected final AFOutputStream newOutputStream() {
192 return new AFOutputStreamImpl();
193 }
194
195 final FileDescriptor getFD() {
196 return fd;
197 }
198
199
200 final boolean isConnected() {
201 if (connected.get()) {
202 return true;
203 }
204 if (isClosed()) {
205 return false;
206 }
207 if (core.isConnected(false)) {
208 connected.set(true);
209 return true;
210 }
211 return false;
212 }
213
214 final boolean isBound() {
215 if (bound.get()) {
216 return true;
217 }
218 if (isClosed()) {
219 return false;
220 }
221 if (core.isConnected(true)) {
222 bound.set(true);
223 return true;
224 }
225 return false;
226 }
227
228 final AFSocketCore getCore() {
229 return core;
230 }
231
232 boolean isClosed() {
233 return closed.get() || core.isClosed();
234 }
235
236
237 @Override
238 protected final void accept(SocketImpl socket) throws IOException {
239 accept0(socket);
240 }
241
242 @SuppressWarnings({
243 "Finally" ,
244 "PMD.CognitiveComplexity", "PMD.NcssCount"})
245 final boolean accept0(SocketImpl socket) throws IOException {
246 FileDescriptor fdesc = core.validFdOrException();
247 if (isClosed()) {
248 throw new SocketClosedException();
249 } else if (!isBound()) {
250 throw new NotBoundSocketException();
251 }
252
253 AFSocketAddress socketAddress = core.socketAddress;
254 AFSocketAddress boundSocketAddress = getLocalSocketAddress();
255 if (boundSocketAddress != null) {
256
257 core.socketAddress = socketAddress = boundSocketAddress;
258 }
259
260 if (socketAddress == null) {
261 throw new NotBoundSocketException();
262 }
263
264 @SuppressWarnings("unchecked")
265 final AFSocketImpl<A> si = (AFSocketImpl<A>) socket;
266 core.incPendingAccepts();
267
268 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
269 .isVirtualBlocking();
270
271 long now = virtualBlocking ? System.currentTimeMillis() : 0;
272 boolean park = false;
273 virtualThreadLoop : do {
274 if (virtualBlocking) {
275 if (park) {
276 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_ACCEPT, now,
277 socketTimeout::get, this::close);
278 }
279 }
280
281 try (Lease<ByteBuffer> abLease = socketAddress.getNativeAddressDirectBuffer()) {
282 ByteBuffer ab = abLease.get();
283
284 SocketException caught = null;
285 try {
286 boolean success;
287 if (virtualBlocking) {
288 core.configureVirtualBlocking(true);
289 }
290 try {
291 success = NativeUnixSocket.accept(ab, ab.limit(), fdesc, si.fd, core.inode.get(),
292 socketTimeout.get());
293 } catch (SocketTimeoutException e) {
294 if (virtualBlocking) {
295
296 park = true;
297 continue virtualThreadLoop;
298 } else {
299 throw e;
300 }
301 } finally {
302 if (virtualBlocking) {
303 core.configureVirtualBlocking(false);
304 }
305 }
306
307 if (virtualBlocking) {
308 if (success) {
309
310 NativeUnixSocket.configureBlocking(si.fd, core.isBlocking());
311 } else {
312
313 park = true;
314 continue virtualThreadLoop;
315 }
316 }
317 } catch (NotConnectedSocketException | SocketClosedException
318 | BrokenPipeSocketException e) {
319 try {
320 close();
321 } catch (Exception e2) {
322 e.addSuppressed(e2);
323 }
324 caught = e;
325 throw e;
326 } catch (SocketException e) {
327 caught = e;
328 } finally {
329 if (!isBound() || isClosed()) {
330 if (getCore().isShutdownOnClose()) {
331 try {
332 NativeUnixSocket.shutdown(si.fd, SHUT_RD_WR);
333 } catch (Exception e) {
334
335 }
336 }
337 try {
338 NativeUnixSocket.close(si.fd);
339 } catch (Exception e) {
340
341 }
342 if (caught != null) {
343 throw caught;
344 } else {
345 throw new BrokenPipeSocketException("Socket is closed");
346 }
347 } else if (caught != null) {
348 throw caught;
349 }
350 }
351 } finally {
352 core.decPendingAccepts();
353 }
354 break;
355 } while (true);
356
357 if (!si.fd.valid()) {
358 return false;
359 }
360
361 si.setSocketAddress(socketAddress);
362 si.connected.set(true);
363
364 return true;
365 }
366
367 final void setSocketAddress(AFSocketAddress socketAddress) {
368 if (socketAddress == null) {
369 this.core.socketAddress = null;
370 this.address = null;
371 this.localport = -1;
372 } else {
373 this.core.socketAddress = socketAddress;
374 this.address = socketAddress.getAddress();
375 if (this.localport <= 0) {
376 this.localport = socketAddress.getPort();
377 }
378 }
379 }
380
381 @Override
382 protected final int available() throws IOException {
383 FileDescriptor fdesc = core.validFdOrException();
384 try (Lease<MutableHolder<ByteBuffer>> lease = core.getPrivateDirectByteBuffer(0)) {
385 return NativeUnixSocket.available(fdesc, lease.get().get());
386 }
387 }
388
389 final void bind(SocketAddress addr, int options) throws IOException {
390 if (addr == null) {
391 addr = addressFamily.nullBindAddress();
392 if (addr == null) {
393 throw new UnsupportedOperationException("Cannot bind to null address");
394 }
395 }
396
397 if (addr == AFSocketAddress.INTERNAL_DUMMY_BIND) {
398 bound.set(true);
399 core.inode.set(0);
400 return;
401 }
402
403 addr = AFSocketAddress.mapOrFail(addr, addressFamily.getSocketAddressClass());
404 bound.set(true);
405
406 AFSocketAddress socketAddress = Objects.requireNonNull((AFSocketAddress) addr);
407
408 this.setSocketAddress(socketAddress);
409 try (Lease<ByteBuffer> abLease = socketAddress.getNativeAddressDirectBuffer()) {
410 ByteBuffer ab = abLease.get();
411 long inode = NativeUnixSocket.bind(ab, ab.limit(), fd, options);
412 core.inode.set(inode);
413 }
414 core.validFdOrException();
415 }
416
417 @Override
418 @SuppressWarnings("hiding")
419 protected final void bind(InetAddress host, int port) throws IOException {
420
421 }
422
423 private void checkClose() throws IOException {
424 if (closedInputStream && closedOutputStream) {
425 close();
426 }
427 }
428
429 @Override
430 protected final void close() throws IOException {
431 this.closed.set(true);
432 try {
433 shutdown();
434 } catch (NotConnectedSocketException | SocketClosedException e) {
435
436 }
437
438 core.runCleaner();
439 }
440
441 @Override
442 @SuppressWarnings("hiding")
443 protected final void connect(String host, int port) throws IOException {
444 throw new SocketException("Cannot bind to this type of address: " + InetAddress.class);
445 }
446
447 @Override
448 @SuppressWarnings("hiding")
449 protected final void connect(InetAddress address, int port) throws IOException {
450 throw new SocketException("Cannot bind to this type of address: " + InetAddress.class);
451 }
452
453 @Override
454 protected final void connect(SocketAddress addr, int connectTimeoutMs) throws IOException {
455 connect0(addr, connectTimeoutMs);
456 }
457
458 final boolean connect0(SocketAddress addr, int connectTimeoutMs) throws IOException {
459 if (connectTimeoutMs <= 0) {
460 return connect1(addr, 0);
461 }
462
463
464 int remainingTimeout = connectTimeoutMs;
465 SocketTimeoutException ste = null;
466 while (remainingTimeout > 0) {
467 long time = System.nanoTime();
468 try {
469 return connect1(addr, remainingTimeout);
470 } catch (SocketTimeoutException e) {
471 if (ste == null) {
472 ste = e;
473 } else {
474 ste.addSuppressed(ste);
475 }
476 }
477 time = System.nanoTime() - time;
478 remainingTimeout -= TimeUnit.NANOSECONDS.toMillis(time);
479 }
480 if (ste == null) {
481 ste = new SocketTimeoutException();
482 }
483 throw ste;
484 }
485
486 @SuppressWarnings({"PMD.CognitiveComplexity", "PMD.NcssCount"})
487 final boolean connect1(SocketAddress addr, int connectTimeout) throws IOException {
488 if (addr == AFSocketAddress.INTERNAL_DUMMY_CONNECT) {
489 this.connected.set(true);
490 return true;
491 } else if (addr == AFSocketAddress.INTERNAL_DUMMY_DONT_CONNECT) {
492 return false;
493 }
494
495 addr = AFSocketAddress.mapOrFail(addr, addressFamily.getSocketAddressClass());
496 AFSocketAddress socketAddress = Objects.requireNonNull((AFSocketAddress) addr);
497
498 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
499 .isVirtualBlocking();
500 long now = virtualBlocking ? System.currentTimeMillis() : 0;
501
502
503 AFSupplier<Integer> virtualConnectTimeout = null;
504
505 if (virtualBlocking) {
506 core.configureVirtualBlocking(true);
507 }
508 boolean park = false;
509 try {
510 virtualThreadLoop : do {
511 try (Lease<ByteBuffer> abLease = socketAddress.getNativeAddressDirectBuffer()) {
512 ByteBuffer ab = abLease.get();
513 boolean success = false;
514 boolean ignoreSpuriousTimeout = true;
515 do {
516 if (virtualBlocking) {
517 if (virtualConnectTimeout != null) {
518 if (park) {
519 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fd, SelectionKey.OP_CONNECT,
520 now, virtualConnectTimeout, this::close);
521 }
522 } else {
523 Thread.yield();
524 }
525 }
526
527 if (virtualBlocking) {
528 core.configureVirtualBlocking(true);
529 }
530 try {
531 success = NativeUnixSocket.connect(ab, ab.limit(), fd, -2);
532 if (!success && virtualBlocking) {
533
534 if (virtualConnectTimeout == null) {
535 virtualConnectTimeout = () -> connectTimeout;
536 }
537 park = true;
538 continue virtualThreadLoop;
539 }
540 break;
541 } catch (SocketTimeoutException e) {
542
543
544 if (ignoreSpuriousTimeout) {
545 Object o = getOption(SocketOptions.SO_TIMEOUT);
546 if (o instanceof Integer) {
547 if (((Integer) o) == 0) {
548 ignoreSpuriousTimeout = false;
549 continue;
550 }
551 } else if (o == null) {
552 ignoreSpuriousTimeout = false;
553 continue;
554 }
555 }
556 throw e;
557 } catch (ConnectionRefusedSocketException e) {
558 if (virtualBlocking) {
559
560 if (connectTimeout == 0 || ((System.currentTimeMillis() - now) < connectTimeout)) {
561 Thread.yield();
562 continue;
563 }
564 }
565 throw e;
566 } catch (NotConnectedSocketException | SocketClosedException
567 | BrokenPipeSocketException e) {
568 try {
569 close();
570 } catch (Exception e2) {
571 e.addSuppressed(e2);
572 }
573 throw e;
574 } catch (SocketException e) {
575 if (virtualBlocking) {
576 Thread.yield();
577 }
578 throw e;
579 } finally {
580 if (virtualBlocking) {
581 core.configureVirtualBlocking(false);
582 }
583 }
584 } while (ThreadUtil.checkNotInterruptedOrThrow());
585 if (success) {
586 setSocketAddress(socketAddress);
587 this.connected.set(true);
588 }
589 core.validFdOrException();
590 return success;
591 }
592 } while (true);
593 } finally {
594 if (virtualBlocking) {
595 core.configureVirtualBlocking(true);
596 }
597 }
598 }
599
600 @Override
601 protected final void create(boolean stream) throws IOException {
602 if (isClosed()) {
603 throw new SocketException("Already closed");
604 }
605 if (fd.valid()) {
606 if (createType != null) {
607 if (createType.booleanValue() != stream) {
608 throw new IllegalStateException("Already created with different mode");
609 }
610 } else {
611 createType = stream;
612 }
613 return;
614 }
615 createType = stream;
616 createSocket(fd, stream ? AFSocketType.SOCK_STREAM : AFSocketType.SOCK_DGRAM);
617 }
618
619 @Override
620 protected final AFInputStream getInputStream() throws IOException {
621 if (!isConnected() && !isBound()) {
622 close();
623 throw new SocketClosedException("Not connected/not bound");
624 }
625 core.validFdOrException();
626 return in;
627 }
628
629 @Override
630 protected final AFOutputStream getOutputStream() throws IOException {
631 if (!isClosed() && !isBound()) {
632 close();
633 throw new SocketClosedException("Not connected/not bound");
634 }
635 core.validFdOrException();
636 return out;
637 }
638
639 @Override
640 protected final void listen(int backlog) throws IOException {
641 FileDescriptor fdesc = core.validFdOrException();
642 if (backlog <= 0) {
643 backlog = 50;
644 }
645 NativeUnixSocket.listen(fdesc, backlog);
646 }
647
648 @Override
649 protected final boolean supportsUrgentData() {
650 return false;
651 }
652
653 @Override
654 protected final void sendUrgentData(int data) throws IOException {
655 throw new UnsupportedOperationException();
656 }
657
658 private final class AFInputStreamImpl extends AFInputStream {
659 private volatile boolean streamClosed = false;
660 private final AtomicBoolean eofReached = new AtomicBoolean(false);
661
662 private final int defaultOpt = (core.isBlocking() ? 0 : NativeUnixSocket.OPT_NON_BLOCKING);
663
664 @SuppressWarnings("PMD.CognitiveComplexity")
665 @Override
666 public int read(byte[] buf, int off, int len) throws IOException {
667 if (streamClosed) {
668 throw new SocketClosedException("This InputStream has already been closed.");
669 }
670 if (eofReached.get()) {
671 return -1;
672 }
673
674 FileDescriptor fdesc = core.validFdOrException();
675 if (len == 0) {
676 return 0;
677 } else if (off < 0 || len < 0 || (len > buf.length - off)) {
678 throw new IndexOutOfBoundsException();
679 }
680
681 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
682 .isVirtualBlocking();
683 final long now;
684 final int opt;
685 if (virtualBlocking) {
686 now = System.currentTimeMillis();
687 opt = defaultOpt | NativeUnixSocket.OPT_NON_BLOCKING;
688 } else {
689 now = 0;
690 opt = defaultOpt;
691 }
692
693 int read;
694
695 boolean park = false;
696 virtualThreadLoop : do {
697 if (virtualBlocking) {
698 if (park) {
699 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_READ, now,
700 socketTimeout::get, this::forceCloseSocket);
701 }
702 core.configureVirtualBlocking(true);
703 }
704
705 try {
706 read = NativeUnixSocket.read(fdesc, buf, off, len, opt, ancillaryDataSupport,
707 socketTimeout.get());
708 if (read == -2) {
709 if (virtualBlocking) {
710
711 park = true;
712 continue virtualThreadLoop;
713 } else {
714 read = 0;
715 }
716 }
717 } catch (SocketTimeoutException e) {
718 if (virtualBlocking) {
719
720 park = true;
721 continue virtualThreadLoop;
722 } else {
723 throw e;
724 }
725 } catch (EOFException e) {
726 eofReached.set(true);
727 throw e;
728 } finally {
729 if (virtualBlocking) {
730 core.configureVirtualBlocking(false);
731 }
732 }
733 break;
734 } while (true);
735
736 return read;
737 }
738
739 @SuppressWarnings("PMD.CognitiveComplexity")
740 @Override
741 public int read() throws IOException {
742 FileDescriptor fdesc = core.validFdOrException();
743
744 if (eofReached.get()) {
745 return -1;
746 }
747
748
749 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
750 .isVirtualBlocking();
751 final long now;
752 final int opt;
753 if (virtualBlocking) {
754 now = System.currentTimeMillis();
755 opt = defaultOpt | NativeUnixSocket.OPT_NON_BLOCKING;
756 } else {
757 now = 0;
758 opt = defaultOpt;
759 }
760
761 boolean park = false;
762 virtualThreadLoop : do {
763 if (virtualBlocking) {
764 if (park) {
765 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_READ, now,
766 socketTimeout::get, this::forceCloseSocket);
767 }
768 core.configureVirtualBlocking(true);
769 }
770
771 try {
772 int byteRead = NativeUnixSocket.read(fdesc, null, 0, 1, opt, ancillaryDataSupport,
773 socketTimeout.get());
774 if (byteRead < 0) {
775 if (byteRead == -2) {
776 if (virtualBlocking) {
777
778 park = true;
779 continue virtualThreadLoop;
780 } else {
781 byteRead = -1;
782 }
783 }
784 eofReached.set(true);
785 return -1;
786 } else {
787 return byteRead;
788 }
789 } catch (SocketTimeoutException e) {
790 if (virtualBlocking) {
791
792 park = true;
793 continue virtualThreadLoop;
794 } else {
795 throw e;
796 }
797 } finally {
798 if (virtualBlocking) {
799 core.configureVirtualBlocking(false);
800 }
801 }
802 } while (true);
803
804
805 }
806
807 private void forceCloseSocket() throws IOException {
808 closedOutputStream = true;
809 close();
810 }
811
812 @Override
813 public synchronized void close() throws IOException {
814 if (streamClosed || isClosed()) {
815 return;
816 }
817 streamClosed = true;
818 FileDescriptor fdesc = core.validFd();
819 if (fdesc != null && getCore().isShutdownOnClose()) {
820 NativeUnixSocket.shutdown(fdesc, SHUT_RD);
821 }
822
823 closedInputStream = true;
824 checkClose();
825 }
826
827 @Override
828 public int available() throws IOException {
829 if (streamClosed) {
830 throw new SocketClosedException("This InputStream has already been closed.");
831 }
832
833 return AFSocketImpl.this.available();
834 }
835
836 @Override
837 public FileDescriptor getFileDescriptor() throws IOException {
838 return getFD();
839 }
840
841 }
842
843 private static boolean checkWriteInterruptedException(int bytesTransferred)
844 throws InterruptedIOException {
845 if (Thread.currentThread().isInterrupted()) {
846 InterruptedIOException ex = new InterruptedIOException("write");
847 ex.bytesTransferred = bytesTransferred;
848 throw ex;
849 }
850 return true;
851 }
852
853 private final class AFOutputStreamImpl extends AFOutputStream {
854 private volatile boolean streamClosed = false;
855
856 private final int defaultOpt = (core.isBlocking() ? 0 : NativeUnixSocket.OPT_NON_BLOCKING);
857
858 @SuppressWarnings("PMD.CognitiveComplexity")
859 @Override
860 public void write(int oneByte) throws IOException {
861 FileDescriptor fdesc = core.validFdOrException();
862
863 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
864 .isVirtualBlocking();
865 final long now;
866 final int opt;
867 if (virtualBlocking) {
868 now = System.currentTimeMillis();
869 opt = defaultOpt | NativeUnixSocket.OPT_NON_BLOCKING;
870 } else {
871 now = 0;
872 opt = defaultOpt;
873 }
874
875 boolean park = false;
876 virtualThreadLoop : do {
877 if (virtualBlocking) {
878 if (park) {
879 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_WRITE, now,
880 socketTimeout::get, this::forceCloseSocket);
881 }
882 core.configureVirtualBlocking(true);
883 }
884
885 try {
886 int written;
887 do {
888 written = NativeUnixSocket.write(fdesc, null, oneByte, 1, opt, ancillaryDataSupport);
889 if (written != 0) {
890 break;
891 }
892 if (virtualBlocking) {
893 park = true;
894 continue virtualThreadLoop;
895 }
896 } while (checkWriteInterruptedException(0));
897 } catch (NotConnectedSocketException | SocketClosedException
898 | BrokenPipeSocketException e) {
899 try {
900 forceCloseSocket();
901 } catch (Exception e2) {
902 e.addSuppressed(e2);
903 }
904 throw e;
905 } catch (SocketTimeoutException e) {
906 if (virtualBlocking) {
907
908 park = true;
909 continue virtualThreadLoop;
910 } else {
911 throw e;
912 }
913 } finally {
914 if (virtualBlocking) {
915 core.configureVirtualBlocking(false);
916 }
917 }
918 break;
919 } while (true);
920 }
921
922 @SuppressWarnings({"PMD.CognitiveComplexity"})
923 @Override
924 public void write(byte[] buf, int off, int len) throws IOException {
925 if (streamClosed) {
926 throw new SocketException("This OutputStream has already been closed.");
927 }
928 if (len < 0 || off < 0 || len > buf.length - off) {
929 throw new IndexOutOfBoundsException();
930 }
931 FileDescriptor fdesc = core.validFdOrException();
932
933
934
935 if (len == 0 && !AFSocket.supports(AFSocketCapability.CAPABILITY_ZERO_LENGTH_SEND)) {
936 return;
937 }
938
939 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
940 .isVirtualBlocking();
941 final long now;
942 final int opt;
943 if (virtualBlocking) {
944 now = System.currentTimeMillis();
945 opt = defaultOpt | NativeUnixSocket.OPT_NON_BLOCKING;
946 } else {
947 now = 0;
948 opt = defaultOpt;
949 }
950
951 int writtenTotal = 0;
952 do {
953 boolean park = false;
954 virtualThreadLoop : do {
955 if (virtualBlocking) {
956 if (park) {
957 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_WRITE, now,
958 socketTimeout::get, this::forceCloseSocket);
959 }
960 core.configureVirtualBlocking(true);
961 }
962
963 final int written;
964 try {
965 written = NativeUnixSocket.write(fdesc, buf, off, len, opt, ancillaryDataSupport);
966 if (written == 0 && virtualBlocking) {
967
968 park = true;
969 continue virtualThreadLoop;
970 }
971 if (written < 0) {
972 if (len == 0) {
973
974
975
976
977
978
979
980
981 return;
982 } else {
983 throw new IOException("Unspecific error while writing");
984 }
985 }
986 } catch (NotConnectedSocketException | SocketClosedException
987 | BrokenPipeSocketException e) {
988 try {
989 forceCloseSocket();
990 } catch (Exception e2) {
991 e.addSuppressed(e2);
992 }
993 throw e;
994 } catch (SocketTimeoutException e) {
995 if (virtualBlocking) {
996
997 park = true;
998 continue virtualThreadLoop;
999 } else {
1000 throw e;
1001 }
1002 } finally {
1003 if (virtualBlocking) {
1004 core.configureVirtualBlocking(false);
1005 }
1006 }
1007
1008 len -= written;
1009 off += written;
1010 writtenTotal += written;
1011 break;
1012 } while (true);
1013
1014 } while (len > 0 && checkWriteInterruptedException(writtenTotal));
1015 }
1016
1017 private void forceCloseSocket() throws IOException {
1018 closedInputStream = true;
1019 close();
1020 }
1021
1022 @Override
1023 public synchronized void close() throws IOException {
1024 if (streamClosed || isClosed()) {
1025 return;
1026 }
1027 streamClosed = true;
1028 FileDescriptor fdesc = core.validFd();
1029 if (fdesc != null && getCore().isShutdownOnClose()) {
1030 NativeUnixSocket.shutdown(fdesc, SHUT_WR);
1031 }
1032 closedOutputStream = true;
1033 checkClose();
1034 }
1035
1036 @Override
1037 public FileDescriptor getFileDescriptor() throws IOException {
1038 return getFD();
1039 }
1040 }
1041
1042 @Override
1043 public final String toString() {
1044 return super.toString() + "[fd=" + fd + "; addr=" + this.core.socketAddress + "; connected="
1045 + connected + "; bound=" + bound + "]";
1046 }
1047
1048 private static int expectInteger(Object value) throws SocketException {
1049 if (value == null) {
1050 throw (SocketException) new SocketException("Value must not be null").initCause(
1051 new NullPointerException());
1052 }
1053 try {
1054 return (Integer) value;
1055 } catch (final ClassCastException e) {
1056 throw (SocketException) new SocketException("Unsupported value: " + value).initCause(e);
1057 }
1058 }
1059
1060 private static int expectBoolean(Object value) throws SocketException {
1061 if (value == null) {
1062 throw (SocketException) new SocketException("Value must not be null").initCause(
1063 new NullPointerException());
1064 }
1065 try {
1066 return ((Boolean) value) ? 1 : 0;
1067 } catch (final ClassCastException e) {
1068 throw (SocketException) new SocketException("Unsupported value: " + value).initCause(e);
1069 }
1070 }
1071
1072 @Override
1073 public Object getOption(int optID) throws SocketException {
1074 return getOption0(optID);
1075 }
1076
1077 private Object getOption0(int optID) throws SocketException {
1078 if (isClosed()) {
1079 throw new SocketException("Socket is closed");
1080 }
1081 if (optID == SocketOptions.SO_REUSEADDR) {
1082 return reuseAddr.get();
1083 }
1084
1085 FileDescriptor fdesc = core.validFdOrException();
1086 return getOptionDefault(fdesc, optID, socketTimeout, addressFamily);
1087 }
1088
1089 static final Object getOptionDefault(FileDescriptor fdesc, int optID, AtomicInteger acceptTimeout,
1090 AFAddressFamily<?> af) throws SocketException {
1091 try {
1092 switch (optID) {
1093 case SocketOptions.SO_KEEPALIVE:
1094 try {
1095 return (NativeUnixSocket.getSocketOptionInt(fdesc, optID) != 0);
1096 } catch (SocketException e) {
1097
1098 return false;
1099 }
1100 case SocketOptions.TCP_NODELAY:
1101 return (NativeUnixSocket.getSocketOptionInt(fdesc, optID) != 0);
1102 case SocketOptions.SO_TIMEOUT:
1103 int v = Math.max(NativeUnixSocket.getSocketOptionInt(fdesc, 0x1005), NativeUnixSocket
1104 .getSocketOptionInt(fdesc, 0x1006));
1105 if (v == -1) {
1106
1107 return 0;
1108 }
1109 return Math.max((acceptTimeout == null ? 0 : acceptTimeout.get()), v);
1110 case SocketOptions.SO_LINGER:
1111 case SocketOptions.SO_RCVBUF:
1112 case SocketOptions.SO_SNDBUF:
1113 return NativeUnixSocket.getSocketOptionInt(fdesc, optID);
1114 case SocketOptions.IP_TOS:
1115 return 0;
1116 case SocketOptions.SO_BINDADDR:
1117 return AFSocketAddress.getInetAddress(fdesc, false, af);
1118 case SocketOptions.SO_REUSEADDR:
1119 return false;
1120 default:
1121 throw new SocketException("Unsupported option: " + optID);
1122 }
1123 } catch (final SocketException e) {
1124 throw e;
1125 } catch (final Exception e) {
1126 throw (SocketException) new SocketException("Could not get option").initCause(e);
1127 }
1128 }
1129
1130 @Override
1131 public void setOption(int optID, Object value) throws SocketException {
1132 setOption0(optID, value);
1133 }
1134
1135 private void setOption0(int optID, Object value) throws SocketException {
1136 if (isClosed()) {
1137 throw new SocketException("Socket is closed");
1138 }
1139 if (optID == SocketOptions.SO_REUSEADDR) {
1140 reuseAddr.set((expectBoolean(value) != 0));
1141 return;
1142 }
1143
1144 FileDescriptor fdesc = core.validFdOrException();
1145 setOptionDefault(fdesc, optID, value, socketTimeout);
1146 }
1147
1148
1149
1150
1151
1152
1153
1154
1155 protected final Object getOptionLenient(int optID) throws SocketException {
1156 try {
1157 return getOption0(optID);
1158 } catch (SocketException e) {
1159 switch (optID) {
1160 case SocketOptions.TCP_NODELAY:
1161 case SocketOptions.SO_KEEPALIVE:
1162 return false;
1163 default:
1164 throw e;
1165 }
1166 }
1167 }
1168
1169
1170
1171
1172
1173
1174
1175
1176 protected final void setOptionLenient(int optID, Object value) throws SocketException {
1177 try {
1178 setOption0(optID, value);
1179 } catch (SocketException e) {
1180 switch (optID) {
1181 case SocketOptions.TCP_NODELAY:
1182 return;
1183 default:
1184 throw e;
1185 }
1186 }
1187 }
1188
1189 static final void setOptionDefault(FileDescriptor fdesc, int optID, Object value,
1190 AtomicInteger acceptTimeout) throws SocketException {
1191 try {
1192 switch (optID) {
1193 case SocketOptions.SO_LINGER:
1194
1195 if (value instanceof Boolean) {
1196 final boolean b = (Boolean) value;
1197 if (b) {
1198 throw new SocketException("Only accepting Boolean.FALSE here");
1199 }
1200 NativeUnixSocket.setSocketOptionInt(fdesc, optID, -1);
1201 return;
1202 }
1203 NativeUnixSocket.setSocketOptionInt(fdesc, optID, expectInteger(value));
1204 return;
1205 case SocketOptions.SO_TIMEOUT: {
1206 int timeout = expectInteger(value);
1207 try {
1208 NativeUnixSocket.setSocketOptionInt(fdesc, 0x1005, timeout);
1209 } catch (InvalidArgumentSocketException e) {
1210
1211 }
1212 try {
1213 NativeUnixSocket.setSocketOptionInt(fdesc, 0x1006, timeout);
1214 } catch (InvalidArgumentSocketException e) {
1215
1216 }
1217 if (acceptTimeout != null) {
1218 acceptTimeout.set(timeout);
1219 }
1220 return;
1221 }
1222 case SocketOptions.SO_RCVBUF:
1223 case SocketOptions.SO_SNDBUF:
1224 NativeUnixSocket.setSocketOptionInt(fdesc, optID, expectInteger(value));
1225 return;
1226 case SocketOptions.SO_KEEPALIVE:
1227 try {
1228 NativeUnixSocket.setSocketOptionInt(fdesc, optID, expectBoolean(value));
1229 } catch (SocketException e) {
1230
1231 }
1232 return;
1233 case SocketOptions.TCP_NODELAY:
1234 NativeUnixSocket.setSocketOptionInt(fdesc, optID, expectBoolean(value));
1235 return;
1236 case SocketOptions.IP_TOS:
1237
1238 return;
1239 case SocketOptions.SO_REUSEADDR:
1240
1241 return;
1242 default:
1243 throw new SocketException("Unsupported option: " + optID);
1244 }
1245 } catch (final SocketException e) {
1246 throw e;
1247 } catch (final Exception e) {
1248 throw (SocketException) new SocketException("Error while setting option").initCause(e);
1249 }
1250 }
1251
1252
1253
1254
1255
1256
1257
1258 protected final synchronized void shutdown() throws IOException {
1259 FileDescriptor fdesc = core.validFd();
1260 if (fdesc != null) {
1261 NativeUnixSocket.shutdown(fdesc, SHUT_RD_WR);
1262 shutdownState = 0;
1263 }
1264 }
1265
1266 @Override
1267 protected final synchronized void shutdownInput() throws IOException {
1268 FileDescriptor fdesc = core.validFd();
1269 if (fdesc != null) {
1270 NativeUnixSocket.shutdown(fdesc, SHUT_RD);
1271 shutdownState |= 1 << (SHUT_RD);
1272 if (shutdownState == SHUTDOWN_RD_WR) {
1273 NativeUnixSocket.shutdown(fdesc, SHUT_RD_WR);
1274 shutdownState = 0;
1275 }
1276 }
1277 }
1278
1279 @Override
1280 protected final synchronized void shutdownOutput() throws IOException {
1281 FileDescriptor fdesc = core.validFd();
1282 if (fdesc != null) {
1283 NativeUnixSocket.shutdown(fdesc, SHUT_WR);
1284 shutdownState |= 1 << (SHUT_RD_WR);
1285 if (shutdownState == SHUTDOWN_RD_WR) {
1286 NativeUnixSocket.shutdown(fdesc, SHUT_RD_WR);
1287 shutdownState = 0;
1288 }
1289 }
1290 }
1291
1292 final int getAncillaryReceiveBufferSize() {
1293 return ancillaryDataSupport.getAncillaryReceiveBufferSize();
1294 }
1295
1296 final void setAncillaryReceiveBufferSize(int size) {
1297 ancillaryDataSupport.setAncillaryReceiveBufferSize(size);
1298 }
1299
1300 final void ensureAncillaryReceiveBufferSize(int minSize) {
1301 ancillaryDataSupport.ensureAncillaryReceiveBufferSize(minSize);
1302 }
1303
1304 AncillaryDataSupport getAncillaryDataSupport() {
1305 return ancillaryDataSupport;
1306 }
1307
1308 final SocketAddress receive(ByteBuffer dst) throws IOException {
1309 return core.receive(dst, socketTimeout::get);
1310 }
1311
1312 final int send(ByteBuffer src, SocketAddress target) throws IOException {
1313 return core.write(src, socketTimeout::get, target, 0);
1314 }
1315
1316 final int read(ByteBuffer dst, ByteBuffer socketAddressBuffer) throws IOException {
1317 return core.read(dst, socketTimeout::get, socketAddressBuffer, 0);
1318 }
1319
1320 final int write(ByteBuffer src) throws IOException {
1321 return core.write(src, socketTimeout::get);
1322 }
1323
1324 @Override
1325 protected final FileDescriptor getFileDescriptor() {
1326 return core.fd;
1327 }
1328
1329 final void updatePorts(int local, int remote) {
1330 this.localport = local;
1331 if (remote >= 0) {
1332 this.port = remote;
1333 }
1334 }
1335
1336 final @Nullable A getLocalSocketAddress() {
1337 return AFSocketAddress.getSocketAddress(getFileDescriptor(), false, localport, addressFamily);
1338 }
1339
1340 final byte[] getLocalSocketAddressBytes() {
1341 return AFSocketAddress.getSocketAddressBytes(getFileDescriptor(), false, addressFamily);
1342 }
1343
1344 final @Nullable A getRemoteSocketAddress() {
1345 return AFSocketAddress.getSocketAddress(getFileDescriptor(), true, port, addressFamily);
1346 }
1347
1348 final int getLocalPort1() {
1349 return localport;
1350 }
1351
1352 final int getRemotePort() {
1353 return port;
1354 }
1355
1356 @Override
1357 protected final InetAddress getInetAddress() {
1358 @Nullable
1359 A rsa = getRemoteSocketAddress();
1360 if (rsa == null) {
1361 return InetAddress.getLoopbackAddress();
1362 } else {
1363 return rsa.getInetAddress();
1364 }
1365 }
1366
1367 final void createSocket(FileDescriptor fdTarget, AFSocketType type) throws IOException {
1368 NativeUnixSocket.createSocket(fdTarget, addressFamily.getDomain(), type.getId());
1369 }
1370
1371 final AFAddressFamily<A> getAddressFamily() {
1372 return addressFamily;
1373 }
1374
1375 @Override
1376 protected <T> void setOption(SocketOption<T> name, T value) throws IOException {
1377 if (name instanceof AFSocketOption<?>) {
1378 getCore().setOption((AFSocketOption<T>) name, value);
1379 return;
1380 }
1381 Integer optionId = SocketOptionsMapper.resolve(name);
1382 if (optionId == null) {
1383 super.setOption(name, value);
1384 } else {
1385 setOption(optionId, value);
1386 }
1387 }
1388
1389 @SuppressWarnings("unchecked")
1390 @Override
1391 protected <T> T getOption(SocketOption<T> name) throws IOException {
1392 if (name instanceof AFSocketOption<?>) {
1393 return getCore().getOption((AFSocketOption<T>) name);
1394 }
1395 Integer optionId = SocketOptionsMapper.resolve(name);
1396 if (optionId == null) {
1397 return super.getOption(name);
1398 } else {
1399 return (T) getOption(optionId);
1400 }
1401 }
1402
1403 @Override
1404 protected Set<SocketOption<?>> supportedOptions() {
1405 return SocketOptionsMapper.SUPPORTED_SOCKET_OPTIONS;
1406 }
1407
1408
1409
1410
1411
1412
1413
1414
1415 protected final synchronized AFSocketImplExtensions<A> getImplExtensions() {
1416 if (implExtensions == null) {
1417 implExtensions = addressFamily.initImplExtensions(ancillaryDataSupport);
1418 }
1419 return implExtensions;
1420 }
1421 }