1 /*
2 * Copyright (c) 2018, 2023, 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.
8 *
9 * This code is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12 * version 2 for more details (a copy is included in the LICENSE file that
13 * accompanied this code).
14 *
15 * You should have received a copy of the GNU General Public License version
16 * 2 along with this work; if not, write to the Free Software Foundation,
17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18 *
19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20 * or visit www.oracle.com if you need additional information or have any
21 * questions.
22 */
23
24 /**
25 * @test id=default
26 * @bug 8284161
27 * @summary Test virtual threads doing blocking I/O on java.net Sockets
28 * @library /test/lib
29 * @run junit BlockingSocketOps
30 */
31
32 /**
33 * @test id=poller-modes
34 * @requires (os.family == "linux") | (os.family == "mac")
35 * @library /test/lib
36 * @run junit/othervm -Djdk.pollerMode=1 BlockingSocketOps
37 * @run junit/othervm -Djdk.pollerMode=2 BlockingSocketOps
38 */
39
40 /**
41 * @test id=no-vmcontinuations
42 * @requires vm.continuations
43 * @library /test/lib
44 * @run junit/othervm -XX:+UnlockExperimentalVMOptions -XX:-VMContinuations BlockingSocketOps
45 */
46
47 import java.io.Closeable;
48 import java.io.IOException;
49 import java.io.InputStream;
50 import java.io.OutputStream;
51 import java.net.DatagramPacket;
52 import java.net.DatagramSocket;
53 import java.net.InetAddress;
54 import java.net.InetSocketAddress;
55 import java.net.ServerSocket;
56 import java.net.Socket;
57 import java.net.SocketAddress;
58 import java.net.SocketException;
59 import java.net.SocketTimeoutException;
60
61 import jdk.test.lib.thread.VThreadRunner;
62 import org.junit.jupiter.api.Test;
63 import static org.junit.jupiter.api.Assertions.*;
64
65 class BlockingSocketOps {
66
67 /**
68 * Socket read/write, no blocking.
69 */
70 @Test
71 void testSocketReadWrite1() throws Exception {
72 VThreadRunner.run(() -> {
73 try (var connection = new Connection()) {
74 Socket s1 = connection.socket1();
75 Socket s2 = connection.socket2();
76
77 // write should not block
78 byte[] ba = "XXX".getBytes("UTF-8");
79 s1.getOutputStream().write(ba);
80
81 // read should not block
82 ba = new byte[10];
83 int n = s2.getInputStream().read(ba);
84 assertTrue(n > 0);
85 assertTrue(ba[0] == 'X');
86 }
87 });
88 }
89
90 /**
91 * Virtual thread blocks in read.
92 */
93 @Test
94 void testSocketRead1() throws Exception {
95 testSocketRead(0);
96 }
97
98 /**
99 * Virtual thread blocks in timed read.
100 */
101 @Test
102 void testSocketRead2() throws Exception {
103 testSocketRead(60_000);
104 }
105
106 void testSocketRead(int timeout) throws Exception {
107 VThreadRunner.run(() -> {
108 try (var connection = new Connection()) {
109 Socket s1 = connection.socket1();
110 Socket s2 = connection.socket2();
111
112 // delayed write from sc1
113 byte[] ba1 = "XXX".getBytes("UTF-8");
114 runAfterParkedAsync(() -> s1.getOutputStream().write(ba1));
115
116 // read from sc2 should block
117 if (timeout > 0) {
118 s2.setSoTimeout(timeout);
119 }
120 byte[] ba2 = new byte[10];
121 int n = s2.getInputStream().read(ba2);
122 assertTrue(n > 0);
123 assertTrue(ba2[0] == 'X');
124 }
125 });
126 }
127
128 /**
129 * Virtual thread blocks in write.
130 */
131 @Test
132 void testSocketWrite1() throws Exception {
133 VThreadRunner.run(() -> {
134 try (var connection = new Connection()) {
135 Socket s1 = connection.socket1();
136 Socket s2 = connection.socket2();
137
138 // delayed read from s2 to EOF
139 InputStream in = s2.getInputStream();
140 Thread reader = runAfterParkedAsync(() ->
141 in.transferTo(OutputStream.nullOutputStream()));
142
143 // write should block
144 byte[] ba = new byte[100*1024];
145 try (OutputStream out = s1.getOutputStream()) {
146 for (int i = 0; i < 1000; i++) {
147 out.write(ba);
148 }
149 }
150
151 // wait for reader to finish
152 reader.join();
153 }
154 });
155 }
156
157 /**
158 * Virtual thread blocks in read, peer closes connection gracefully.
159 */
160 @Test
161 void testSocketReadPeerClose1() throws Exception {
162 VThreadRunner.run(() -> {
163 try (var connection = new Connection()) {
164 Socket s1 = connection.socket1();
165 Socket s2 = connection.socket2();
166
167 // delayed close of s2
168 runAfterParkedAsync(s2::close);
169
170 // read from s1 should block, then read -1
171 int n = s1.getInputStream().read();
172 assertTrue(n == -1);
173 }
174 });
175 }
176
177 /**
178 * Virtual thread blocks in read, peer closes connection abruptly.
179 */
180 @Test
181 void testSocketReadPeerClose2() throws Exception {
182 VThreadRunner.run(() -> {
183 try (var connection = new Connection()) {
184 Socket s1 = connection.socket1();
185 Socket s2 = connection.socket2();
186
187 // delayed abrupt close of s2
188 s2.setSoLinger(true, 0);
189 runAfterParkedAsync(s2::close);
190
191 // read from s1 should block, then throw
192 try {
193 int n = s1.getInputStream().read();
194 fail("read " + n);
195 } catch (IOException ioe) {
196 // expected
197 }
198 }
199 });
200 }
201
202 /**
203 * Socket close while virtual thread blocked in read.
204 */
205 @Test
206 void testSocketReadAsyncClose1() throws Exception {
207 testSocketReadAsyncClose(0);
208 }
209
210 /**
211 * Socket close while virtual thread blocked in timed read.
212 */
213 @Test
214 void testSocketReadAsyncClose2() throws Exception {
215 testSocketReadAsyncClose(60_000);
216 }
217
218 void testSocketReadAsyncClose(int timeout) throws Exception {
219 VThreadRunner.run(() -> {
220 try (var connection = new Connection()) {
221 Socket s = connection.socket1();
222
223 // delayed close of s
224 runAfterParkedAsync(s::close);
225
226 // read from s should block, then throw
227 if (timeout > 0) {
228 s.setSoTimeout(timeout);
229 }
230 try {
231 int n = s.getInputStream().read();
232 fail("read " + n);
233 } catch (SocketException expected) { }
234 }
235 });
236 }
237
238 /**
239 * Socket shutdownInput while virtual thread blocked in read.
240 */
241 @Test
242 void testSocketReadAsyncShutdownInput1() throws Exception {
243 testSocketReadAsyncShutdownInput(0);
244 }
245
246 /**
247 * Socket shutdownInput while virtual thread blocked in timed read.
248 */
249 @Test
250 void testSocketReadAsyncShutdownInput2() throws Exception {
251 testSocketReadAsyncShutdownInput(60_000);
252 }
253
254 void testSocketReadAsyncShutdownInput(int timeout) throws Exception {
255 VThreadRunner.run(() -> {
256 try (var connection = new Connection()) {
257 Socket s = connection.socket1();
258
259 // delayed shutdown of s
260 runAfterParkedAsync(s::shutdownInput);
261
262 // read from s should block, then throw
263 if (timeout > 0) {
264 s.setSoTimeout(timeout);
265 }
266
267 // -1 or SocketException
268 try {
269 int n = s.getInputStream().read();
270 assertEquals(-1, n);
271 } catch (SocketException e) { }
272
273 assertTrue(s.isInputShutdown());
274 assertFalse(s.isClosed());
275 }
276 });
277 }
278
279 /**
280 * Virtual thread interrupted while blocked in Socket read.
281 */
282 @Test
283 void testSocketReadInterrupt1() throws Exception {
284 testSocketReadInterrupt(0);
285 }
286
287 /**
288 * Virtual thread interrupted while blocked in Socket read with timeout
289 */
290 @Test
291 void testSocketReadInterrupt2() throws Exception {
292 testSocketReadInterrupt(60_000);
293 }
294
295 void testSocketReadInterrupt(int timeout) throws Exception {
296 VThreadRunner.run(() -> {
297 try (var connection = new Connection()) {
298 Socket s = connection.socket1();
299
300
301 // delayed interrupt of current thread
302 Thread thisThread = Thread.currentThread();
303 runAfterParkedAsync(thisThread::interrupt);
304
305 // read from s should block, then throw
306 if (timeout > 0) {
307 s.setSoTimeout(timeout);
308 }
309 try {
310 int n = s.getInputStream().read();
311 fail("read " + n);
312 } catch (SocketException expected) {
313 assertTrue(Thread.interrupted());
314 assertTrue(s.isClosed());
315 }
316 }
317 });
318 }
319
320 /**
321 * Socket close while virtual thread blocked in write.
322 */
323 @Test
324 void testSocketWriteAsyncClose() throws Exception {
325 VThreadRunner.run(() -> {
326 try (var connection = new Connection()) {
327 Socket s = connection.socket1();
328
329 // delayed close of s
330 runAfterParkedAsync(s::close);
331
332 // write to s should block, then throw
333 try {
334 byte[] ba = new byte[100*1024];
335 OutputStream out = s.getOutputStream();
336 for (;;) {
337 out.write(ba);
338 }
339 } catch (SocketException expected) { }
340 }
341 });
342 }
343
344 /**
345 * Socket shutdownOutput while virtual thread blocked in write.
346 */
347 @Test
348 void testSocketWriteAsyncShutdownOutput() throws Exception {
349 VThreadRunner.run(() -> {
350 try (var connection = new Connection()) {
351 Socket s = connection.socket1();
352
353 // delayed shutdown of s
354 runAfterParkedAsync(s::shutdownOutput);
355
356 // write to s should block, then throw
357 try {
358 byte[] ba = new byte[100*1024];
359 OutputStream out = s.getOutputStream();
360 for (;;) {
361 out.write(ba);
362 }
363 } catch (SocketException expected) { }
364
365 assertTrue(s.isOutputShutdown());
366 assertFalse(s.isClosed());
367 }
368 });
369 }
370
371 /**
372 * Virtual thread interrupted while blocked in Socket write.
373 */
374 @Test
375 void testSocketWriteInterrupt() throws Exception {
376 VThreadRunner.run(() -> {
377 try (var connection = new Connection()) {
378 Socket s = connection.socket1();
379
380 // delayed interrupt of current thread
381 Thread thisThread = Thread.currentThread();
382 runAfterParkedAsync(thisThread::interrupt);
383
384 // write to s should block, then throw
385 try {
386 byte[] ba = new byte[100*1024];
387 OutputStream out = s.getOutputStream();
388 for (;;) {
389 out.write(ba);
390 }
391 } catch (SocketException expected) {
392 assertTrue(Thread.interrupted());
393 assertTrue(s.isClosed());
394 }
395 }
396 });
397 }
398
399 /**
400 * Virtual thread reading urgent data when SO_OOBINLINE is enabled.
401 */
402 @Test
403 void testSocketReadUrgentData() throws Exception {
404 VThreadRunner.run(() -> {
405 try (var connection = new Connection()) {
406 Socket s1 = connection.socket1();
407 Socket s2 = connection.socket2();
408
409 // urgent data should be received
410 runAfterParkedAsync(() -> s2.sendUrgentData('X'));
411
412 // read should block, then read the OOB byte
413 s1.setOOBInline(true);
414 byte[] ba = new byte[10];
415 int n = s1.getInputStream().read(ba);
416 assertTrue(n == 1);
417 assertTrue(ba[0] == 'X');
418
419 // urgent data should not be received
420 s1.setOOBInline(false);
421 s1.setSoTimeout(500);
422 s2.sendUrgentData('X');
423 try {
424 s1.getInputStream().read(ba);
425 fail();
426 } catch (SocketTimeoutException expected) { }
427 }
428 });
429 }
430
431 /**
432 * ServerSocket accept, no blocking.
433 */
434 @Test
435 void testServerSocketAccept1() throws Exception {
436 VThreadRunner.run(() -> {
437 try (var listener = new ServerSocket()) {
438 InetAddress loopback = InetAddress.getLoopbackAddress();
439 listener.bind(new InetSocketAddress(loopback, 0));
440
441 // establish connection
442 var socket1 = new Socket(loopback, listener.getLocalPort());
443
444 // accept should not block
445 var socket2 = listener.accept();
446 socket1.close();
447 socket2.close();
448 }
449 });
450 }
451
452 /**
453 * Virtual thread blocks in accept.
454 */
455 @Test
456 void testServerSocketAccept2() throws Exception {
457 testServerSocketAccept(0);
458 }
459
460 /**
461 * Virtual thread blocks in timed accept.
462 */
463 @Test
464 void testServerSocketAccept3() throws Exception {
465 testServerSocketAccept(60_000);
466 }
467
468 void testServerSocketAccept(int timeout) throws Exception {
469 VThreadRunner.run(() -> {
470 try (var listener = new ServerSocket()) {
471 InetAddress loopback = InetAddress.getLoopbackAddress();
472 listener.bind(new InetSocketAddress(loopback, 0));
473
474 // schedule connect
475 var socket1 = new Socket();
476 SocketAddress remote = listener.getLocalSocketAddress();
477 runAfterParkedAsync(() -> socket1.connect(remote));
478
479 // accept should block
480 if (timeout > 0) {
481 listener.setSoTimeout(timeout);
482 }
483 var socket2 = listener.accept();
484 socket1.close();
485 socket2.close();
486 }
487 });
488 }
489
490 /**
491 * ServerSocket close while virtual thread blocked in accept.
492 */
493 @Test
494 void testServerSocketAcceptAsyncClose1() throws Exception {
495 testServerSocketAcceptAsyncClose(0);
496 }
497
498 /**
499 * ServerSocket close while virtual thread blocked in timed accept.
500 */
501 @Test
502 void testServerSocketAcceptAsyncClose2() throws Exception {
503 testServerSocketAcceptAsyncClose(60_000);
504 }
505
506 void testServerSocketAcceptAsyncClose(int timeout) throws Exception {
507 VThreadRunner.run(() -> {
508 try (var listener = new ServerSocket()) {
509 InetAddress loopback = InetAddress.getLoopbackAddress();
510 listener.bind(new InetSocketAddress(loopback, 0));
511
512 // delayed close of listener
513 runAfterParkedAsync(listener::close);
514
515 // accept should block, then throw
516 if (timeout > 0) {
517 listener.setSoTimeout(timeout);
518 }
519 try {
520 listener.accept().close();
521 fail("connection accepted???");
522 } catch (SocketException expected) { }
523 }
524 });
525 }
526
527 /**
528 * Virtual thread interrupted while blocked in ServerSocket accept.
529 */
530 @Test
531 void testServerSocketAcceptInterrupt1() throws Exception {
532 testServerSocketAcceptInterrupt(0);
533 }
534
535 /**
536 * Virtual thread interrupted while blocked in ServerSocket accept with timeout.
537 */
538 @Test
539 void testServerSocketAcceptInterrupt2() throws Exception {
540 testServerSocketAcceptInterrupt(60_000);
541 }
542
543 void testServerSocketAcceptInterrupt(int timeout) throws Exception {
544 VThreadRunner.run(() -> {
545 try (var listener = new ServerSocket()) {
546 InetAddress loopback = InetAddress.getLoopbackAddress();
547 listener.bind(new InetSocketAddress(loopback, 0));
548
549 // delayed interrupt of current thread
550 Thread thisThread = Thread.currentThread();
551 runAfterParkedAsync(thisThread::interrupt);
552
553 // accept should block, then throw
554 if (timeout > 0) {
555 listener.setSoTimeout(timeout);
556 }
557 try {
558 listener.accept().close();
559 fail("connection accepted???");
560 } catch (SocketException expected) {
561 assertTrue(Thread.interrupted());
562 assertTrue(listener.isClosed());
563 }
564 }
565 });
566 }
567
568 /**
569 * DatagramSocket receive/send, no blocking.
570 */
571 @Test
572 void testDatagramSocketSendReceive1() throws Exception {
573 VThreadRunner.run(() -> {
574 try (DatagramSocket s1 = new DatagramSocket(null);
575 DatagramSocket s2 = new DatagramSocket(null)) {
576
577 InetAddress lh = InetAddress.getLoopbackAddress();
578 s1.bind(new InetSocketAddress(lh, 0));
579 s2.bind(new InetSocketAddress(lh, 0));
580
581 // send should not block
582 byte[] bytes = "XXX".getBytes("UTF-8");
583 DatagramPacket p1 = new DatagramPacket(bytes, bytes.length);
584 p1.setSocketAddress(s2.getLocalSocketAddress());
585 s1.send(p1);
586
587 // receive should not block
588 byte[] ba = new byte[100];
589 DatagramPacket p2 = new DatagramPacket(ba, ba.length);
590 s2.receive(p2);
591 assertEquals(s1.getLocalSocketAddress(), p2.getSocketAddress());
592 assertTrue(ba[0] == 'X');
593 }
594 });
595 }
596
597 /**
598 * Virtual thread blocks in DatagramSocket receive.
599 */
600 @Test
601 void testDatagramSocketSendReceive2() throws Exception {
602 testDatagramSocketSendReceive(0);
603 }
604
605 /**
606 * Virtual thread blocks in DatagramSocket receive with timeout.
607 */
608 @Test
609 void testDatagramSocketSendReceive3() throws Exception {
610 testDatagramSocketSendReceive(60_000);
611 }
612
613 private void testDatagramSocketSendReceive(int timeout) throws Exception {
614 VThreadRunner.run(() -> {
615 try (DatagramSocket s1 = new DatagramSocket(null);
616 DatagramSocket s2 = new DatagramSocket(null)) {
617
618 InetAddress lh = InetAddress.getLoopbackAddress();
619 s1.bind(new InetSocketAddress(lh, 0));
620 s2.bind(new InetSocketAddress(lh, 0));
621
622 // delayed send
623 byte[] bytes = "XXX".getBytes("UTF-8");
624 DatagramPacket p1 = new DatagramPacket(bytes, bytes.length);
625 p1.setSocketAddress(s2.getLocalSocketAddress());
626 runAfterParkedAsync(() -> s1.send(p1));
627
628 // receive should block
629 if (timeout > 0) {
630 s2.setSoTimeout(timeout);
631 }
632 byte[] ba = new byte[100];
633 DatagramPacket p2 = new DatagramPacket(ba, ba.length);
634 s2.receive(p2);
635 assertEquals(s1.getLocalSocketAddress(), p2.getSocketAddress());
636 assertTrue(ba[0] == 'X');
637 }
638 });
639 }
640
641 /**
642 * Virtual thread blocks in DatagramSocket receive that times out.
643 */
644 @Test
645 void testDatagramSocketReceiveTimeout() throws Exception {
646 VThreadRunner.run(() -> {
647 try (DatagramSocket s = new DatagramSocket(null)) {
648 InetAddress lh = InetAddress.getLoopbackAddress();
649 s.bind(new InetSocketAddress(lh, 0));
650 s.setSoTimeout(500);
651 byte[] ba = new byte[100];
652 DatagramPacket p = new DatagramPacket(ba, ba.length);
653 try {
654 s.receive(p);
655 fail();
656 } catch (SocketTimeoutException expected) { }
657 }
658 });
659 }
660
661 /**
662 * DatagramSocket close while virtual thread blocked in receive.
663 */
664 @Test
665 void testDatagramSocketReceiveAsyncClose1() throws Exception {
666 testDatagramSocketReceiveAsyncClose(0);
667 }
668
669 /**
670 * DatagramSocket close while virtual thread blocked with timeout.
671 */
672 @Test
673 void testDatagramSocketReceiveAsyncClose2() throws Exception {
674 testDatagramSocketReceiveAsyncClose(60_000);
675 }
676
677 private void testDatagramSocketReceiveAsyncClose(int timeout) throws Exception {
678 VThreadRunner.run(() -> {
679 try (DatagramSocket s = new DatagramSocket(null)) {
680 InetAddress lh = InetAddress.getLoopbackAddress();
681 s.bind(new InetSocketAddress(lh, 0));
682
683 // delayed close of s
684 runAfterParkedAsync(s::close);
685
686 // receive should block, then throw
687 if (timeout > 0) {
688 s.setSoTimeout(timeout);
689 }
690 try {
691 byte[] ba = new byte[100];
692 DatagramPacket p = new DatagramPacket(ba, ba.length);
693 s.receive(p);
694 fail();
695 } catch (SocketException expected) { }
696 }
697 });
698 }
699
700 /**
701 * Virtual thread interrupted while blocked in DatagramSocket receive.
702 */
703 @Test
704 void testDatagramSocketReceiveInterrupt1() throws Exception {
705 testDatagramSocketReceiveInterrupt(0);
706 }
707
708 /**
709 * Virtual thread interrupted while blocked in DatagramSocket receive with timeout.
710 */
711 @Test
712 void testDatagramSocketReceiveInterrupt2() throws Exception {
713 testDatagramSocketReceiveInterrupt(60_000);
714 }
715
716 private void testDatagramSocketReceiveInterrupt(int timeout) throws Exception {
717 VThreadRunner.run(() -> {
718 try (DatagramSocket s = new DatagramSocket(null)) {
719 InetAddress lh = InetAddress.getLoopbackAddress();
720 s.bind(new InetSocketAddress(lh, 0));
721
722 // delayed interrupt of current thread
723 Thread thisThread = Thread.currentThread();
724 runAfterParkedAsync(thisThread::interrupt);
725
726 // receive should block, then throw
727 if (timeout > 0) {
728 s.setSoTimeout(timeout);
729 }
730 try {
731 byte[] ba = new byte[100];
732 DatagramPacket p = new DatagramPacket(ba, ba.length);
733 s.receive(p);
734 fail();
735 } catch (SocketException expected) {
736 assertTrue(Thread.interrupted());
737 assertTrue(s.isClosed());
738 }
739 }
740 });
741 }
742
743 /**
744 * Creates a loopback connection
745 */
746 static class Connection implements Closeable {
747 private final Socket s1;
748 private final Socket s2;
749 Connection() throws IOException {
750 var lh = InetAddress.getLoopbackAddress();
751 try (var listener = new ServerSocket()) {
752 listener.bind(new InetSocketAddress(lh, 0));
753 Socket s1 = new Socket();
754 Socket s2;
755 try {
756 s1.connect(listener.getLocalSocketAddress());
757 s2 = listener.accept();
758 } catch (IOException ioe) {
759 s1.close();
760 throw ioe;
761 }
762 this.s1 = s1;
763 this.s2 = s2;
764 }
765
766 }
767 Socket socket1() {
768 return s1;
769 }
770 Socket socket2() {
771 return s2;
772 }
773 @Override
774 public void close() throws IOException {
775 s1.close();
776 s2.close();
777 }
778 }
779
780 @FunctionalInterface
781 interface ThrowingRunnable {
782 void run() throws Exception;
783 }
784
785 /**
786 * Runs the given task asynchronously after the current virtual thread has parked.
787 * @return the thread started to run the task
788 */
789 static Thread runAfterParkedAsync(ThrowingRunnable task) {
790 Thread target = Thread.currentThread();
791 if (!target.isVirtual())
792 throw new WrongThreadException();
793 return Thread.ofPlatform().daemon().start(() -> {
794 try {
795 Thread.State state = target.getState();
796 while (state != Thread.State.WAITING
797 && state != Thread.State.TIMED_WAITING) {
798 Thread.sleep(20);
799 state = target.getState();
800 }
801 Thread.sleep(20); // give a bit more time to release carrier
802 task.run();
803 } catch (Exception e) {
804 e.printStackTrace();
805 }
806 });
807 }
808 }