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_WR;
21
22 import java.io.FileDescriptor;
23 import java.io.IOException;
24 import java.net.DatagramPacket;
25 import java.net.DatagramSocketImpl;
26 import java.net.InetAddress;
27 import java.net.NetworkInterface;
28 import java.net.SocketAddress;
29 import java.net.SocketException;
30 import java.net.SocketTimeoutException;
31 import java.nio.ByteBuffer;
32 import java.nio.channels.ClosedChannelException;
33 import java.nio.channels.SelectionKey;
34 import java.util.Objects;
35 import java.util.concurrent.atomic.AtomicBoolean;
36 import java.util.concurrent.atomic.AtomicInteger;
37
38 import org.eclipse.jdt.annotation.NonNull;
39 import org.eclipse.jdt.annotation.Nullable;
40 import org.newsclub.net.unix.pool.MutableHolder;
41 import org.newsclub.net.unix.pool.ObjectPool.Lease;
42
43
44
45
46
47
48
49 @SuppressWarnings("PMD.CyclomaticComplexity")
50 public abstract class AFDatagramSocketImpl<A extends AFSocketAddress> extends
51 DatagramSocketImplShim {
52 private final AFSocketType socketType;
53 private final AFSocketCore core;
54 final AncillaryDataSupport ancillaryDataSupport = new AncillaryDataSupport();
55 private final AtomicBoolean connected = new AtomicBoolean(false);
56 private final AtomicBoolean bound = new AtomicBoolean(false);
57
58 private final AtomicInteger socketTimeout = new AtomicInteger(0);
59 private final AtomicInteger localPort = new AtomicInteger(0);
60 private final AtomicInteger remotePort = new AtomicInteger(0);
61 private final AFAddressFamily<@NonNull A> addressFamily;
62 private AFSocketImplExtensions<A> implExtensions = null;
63
64
65
66
67
68
69
70
71
72 @SuppressWarnings("this-escape")
73 protected AFDatagramSocketImpl(AFAddressFamily<@NonNull A> addressFamily, FileDescriptor fd,
74 AFSocketType socketType) {
75 super();
76 this.addressFamily = addressFamily;
77
78 this.socketType = socketType;
79 this.core = new AFSocketCore(this, fd, ancillaryDataSupport, getAddressFamily(), true);
80 this.fd = core.fd;
81 }
82
83 @Override
84 protected final void create() throws SocketException {
85 if (isClosed()) {
86 throw new SocketException("Already closed");
87 } else if (fd.valid()) {
88 return;
89 }
90 try {
91 NativeUnixSocket.createSocket(fd, getAddressFamily().getDomain(), socketType.getId());
92 } catch (SocketException e) {
93 throw e;
94 } catch (IOException e) {
95 throw (SocketException) new SocketException(e.getMessage()).initCause(e);
96 }
97 }
98
99 @Override
100 protected final void close() {
101 core.runCleaner();
102 }
103
104 @Override
105 protected final void connect(InetAddress address, int port) throws SocketException {
106
107 }
108
109 final void connect(AFSocketAddress socketAddress) throws IOException {
110 if (socketAddress == AFSocketAddress.INTERNAL_DUMMY_CONNECT) {
111 return;
112 }
113 try (Lease<ByteBuffer> abLease = socketAddress.getNativeAddressDirectBuffer()) {
114 ByteBuffer ab = abLease.get();
115 NativeUnixSocket.connect(ab, ab.limit(), fd, -1);
116 }
117 this.remotePort.set(socketAddress.getPort());
118 }
119
120 @Override
121 protected final void disconnect() {
122 try {
123 NativeUnixSocket.disconnect(fd);
124 connected.set(false);
125 this.remotePort.set(0);
126 } catch (IOException e) {
127 StackTraceUtil.printStackTrace(e);
128 }
129 }
130
131 final AFSocketCore getCore() {
132 return core;
133 }
134
135 @Override
136 protected final FileDescriptor getFileDescriptor() {
137 return core.fd;
138 }
139
140 final boolean isClosed() {
141 return core.isClosed();
142 }
143
144 @Override
145 protected final void bind(int lport, InetAddress laddr) throws SocketException {
146
147 }
148
149 final void bind(AFSocketAddress socketAddress) throws SocketException {
150 if (socketAddress == AFSocketAddress.INTERNAL_DUMMY_BIND) {
151 return;
152 }
153 try (Lease<ByteBuffer> abLease = socketAddress == null ? AFSocketAddress
154 .getNativeAddressDirectBuffer(0) : socketAddress.getNativeAddressDirectBuffer()) {
155 ByteBuffer ab = abLease.get();
156 NativeUnixSocket.bind(ab, ab.limit(), fd, NativeUnixSocket.OPT_DGRAM_MODE);
157 if (socketAddress == null) {
158 this.localPort.set(0);
159 this.bound.set(false);
160 } else {
161 this.localPort.set(socketAddress.getPort());
162 }
163 } catch (SocketException e) {
164 throw e;
165 } catch (IOException e) {
166 throw (SocketException) new SocketException(e.getMessage()).initCause(e);
167 }
168 }
169
170 @Override
171 protected final void receive(DatagramPacket p) throws IOException {
172 recv(p, 0);
173 }
174
175 @SuppressWarnings({"PMD.CognitiveComplexity", "PMD.CyclomaticComplexity"})
176 private void recv(DatagramPacket p, int options) throws IOException {
177 int len = p.getLength();
178 FileDescriptor fdesc = core.validFdOrException();
179
180 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
181 .isVirtualBlocking();
182 final long now;
183 if (virtualBlocking) {
184 now = System.currentTimeMillis();
185 } else {
186 now = 0;
187 }
188 if (virtualBlocking || !core.isBlocking()) {
189 options |= NativeUnixSocket.OPT_NON_BLOCKING;
190 }
191
192 boolean park = false;
193 virtualThreadLoop : do {
194 if (virtualBlocking) {
195 if (park) {
196 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_WRITE, now,
197 socketTimeout::get, this::close);
198 }
199 core.configureVirtualBlocking(true);
200 }
201
202 try (Lease<MutableHolder<ByteBuffer>> lease = core.getPrivateDirectByteBuffer(len);
203 Lease<ByteBuffer> socketAddressBufferLease = AFSocketAddress.SOCKETADDRESS_BUFFER_TL
204 .take()) {
205 ByteBuffer datagramPacketBuffer = Objects.requireNonNull(lease.get().get());
206 len = Math.min(len, datagramPacketBuffer.capacity());
207
208 ByteBuffer socketAddressBuffer = socketAddressBufferLease.get();
209 int count = NativeUnixSocket.receive(fdesc, datagramPacketBuffer, 0, len,
210 socketAddressBuffer, options, ancillaryDataSupport, socketTimeout.get());
211 if (count == 0 && virtualBlocking) {
212
213 park = true;
214 continue virtualThreadLoop;
215 }
216
217 if (count > len) {
218 throw new IllegalStateException("count > len: " + count + " > " + len);
219 } else if (count == -1) {
220 throw new SocketTimeoutException();
221 } else if (count < 0) {
222 throw new IllegalStateException("count: " + count + " < 0");
223 }
224 datagramPacketBuffer.limit(count);
225 datagramPacketBuffer.rewind();
226 datagramPacketBuffer.get(p.getData(), p.getOffset(), count);
227
228 p.setLength(count);
229
230 A addr = AFSocketAddress.ofInternal(socketAddressBuffer, getAddressFamily());
231 p.setAddress(addr == null ? null : addr.getInetAddress());
232 p.setPort(remotePort.get());
233 } catch (SocketTimeoutException e) {
234 if (virtualBlocking) {
235
236 park = true;
237 continue virtualThreadLoop;
238 } else {
239 throw e;
240 }
241 } finally {
242 if (virtualBlocking) {
243 core.configureVirtualBlocking(false);
244 }
245 }
246 break;
247 } while (true);
248 }
249
250 @SuppressWarnings({"PMD.CognitiveComplexity"})
251 @Override
252 protected final void send(DatagramPacket p) throws IOException {
253 InetAddress addr = p.getAddress();
254 ByteBuffer sendToBuf = null;
255 int sendToBufLen = 0;
256
257 byte[] addrBytes;
258 if (addr != null) {
259 addrBytes = AFInetAddress.unwrapAddress(addr, getAddressFamily());
260 } else {
261 addrBytes = null;
262 }
263
264 try (Lease<ByteBuffer> sendToBufLease = addrBytes == null ? null
265 : AFSocketAddress.SOCKETADDRESS_BUFFER_TL.take()) {
266 if (sendToBufLease != null) {
267 sendToBuf = sendToBufLease.get();
268 sendToBufLen = NativeUnixSocket.bytesToSockAddr(getAddressFamily().getDomain(), sendToBuf,
269 addrBytes);
270 sendToBuf.position(0);
271 if (sendToBufLen == -1) {
272 throw new SocketException("Unsupported domain");
273 }
274 }
275 }
276 FileDescriptor fdesc = core.validFdOrException();
277 int len = p.getLength();
278
279 final boolean virtualBlocking = (ThreadUtil.isVirtualThread() && core.isBlocking()) || core
280 .isVirtualBlocking();
281 final long now;
282 final int opt;
283 if (virtualBlocking) {
284 now = System.currentTimeMillis();
285 opt = NativeUnixSocket.OPT_DGRAM_MODE | NativeUnixSocket.OPT_NON_BLOCKING;
286 } else {
287 now = 0;
288 opt = NativeUnixSocket.OPT_DGRAM_MODE;
289 }
290
291 boolean park = false;
292 virtualThreadLoop : do {
293 if (virtualBlocking) {
294 if (park) {
295 VirtualThreadPoller.INSTANCE.parkThreadUntilReady(fdesc, SelectionKey.OP_WRITE, now,
296 socketTimeout::get, this::close);
297 }
298 core.configureVirtualBlocking(true);
299 }
300
301 try (Lease<MutableHolder<ByteBuffer>> lease = core.getPrivateDirectByteBuffer(len)) {
302 ByteBuffer datagramPacketBuffer = Objects.requireNonNull(lease.get().get());
303 datagramPacketBuffer.clear();
304 datagramPacketBuffer.put(p.getData(), p.getOffset(), p.getLength());
305 datagramPacketBuffer.flip();
306
307 int written = NativeUnixSocket.send(fdesc, datagramPacketBuffer, 0, len, sendToBuf,
308 sendToBufLen, opt, ancillaryDataSupport);
309 if (written == 0 && virtualBlocking) {
310
311 park = true;
312 continue virtualThreadLoop;
313 }
314 } catch (SocketTimeoutException e) {
315 if (virtualBlocking) {
316
317 park = true;
318 continue virtualThreadLoop;
319 } else {
320 throw e;
321 }
322 } finally {
323 if (virtualBlocking) {
324 core.configureVirtualBlocking(false);
325 }
326 }
327 break;
328 } while (true);
329 }
330
331 @Override
332 protected final int peek(InetAddress i) throws IOException {
333 throw new SocketException("Unsupported operation");
334 }
335
336 @Override
337 protected final int peekData(DatagramPacket p) throws IOException {
338 recv(p, NativeUnixSocket.OPT_PEEK);
339 return 0;
340 }
341
342 @Override
343 @Deprecated
344 @SuppressWarnings("doclint")
345 protected final byte getTTL() throws IOException {
346 return (byte) (getTimeToLive() & 0xFF);
347 }
348
349 @Override
350 @Deprecated
351 @SuppressWarnings("doclint")
352 protected final void setTTL(byte ttl) throws IOException {
353
354 }
355
356 @Override
357 protected final int getTimeToLive() throws IOException {
358 return 0;
359 }
360
361 @Override
362 protected final void setTimeToLive(int ttl) throws IOException {
363
364 }
365
366 @Override
367 protected final void join(InetAddress inetaddr) throws IOException {
368 throw new SocketException("Unsupported");
369 }
370
371 @Override
372 protected final void leave(InetAddress inetaddr) throws IOException {
373 throw new SocketException("Unsupported");
374 }
375
376 @Override
377 protected final void joinGroup(SocketAddress mcastaddr, NetworkInterface netIf)
378 throws IOException {
379 throw new SocketException("Unsupported");
380 }
381
382 @Override
383 protected final void leaveGroup(SocketAddress mcastaddr, NetworkInterface netIf)
384 throws IOException {
385 throw new SocketException("Unsupported");
386 }
387
388 @Override
389 public Object getOption(int optID) throws SocketException {
390 if (isClosed()) {
391 throw new SocketException("Socket is closed");
392 }
393
394 FileDescriptor fdesc = core.validFdOrException();
395 return AFSocketImpl.getOptionDefault(fdesc, optID, socketTimeout, getAddressFamily());
396 }
397
398 @Override
399 public void setOption(int optID, Object value) throws SocketException {
400 if (isClosed()) {
401 throw new SocketException("Socket is closed");
402 }
403
404 FileDescriptor fdesc = core.validFdOrException();
405 AFSocketImpl.setOptionDefault(fdesc, optID, value, socketTimeout);
406 }
407
408 @SuppressWarnings("unchecked")
409 final A receive(ByteBuffer dst) throws IOException {
410 try {
411 return (A) core.receive(dst, socketTimeout::get);
412 } catch (SocketClosedException e) {
413 throw (ClosedChannelException) new ClosedChannelException().initCause(e);
414 }
415 }
416
417 final int send(ByteBuffer src, SocketAddress target) throws IOException {
418 try {
419 return core.write(src, socketTimeout::get, target, 0);
420 } catch (SocketClosedException e) {
421 throw (ClosedChannelException) new ClosedChannelException().initCause(e);
422 }
423 }
424
425 final int read(ByteBuffer dst, ByteBuffer socketAddressBuffer) throws IOException {
426 try {
427 return core.read(dst, socketTimeout::get, socketAddressBuffer, 0);
428 } catch (SocketClosedException e) {
429 throw (ClosedChannelException) new ClosedChannelException().initCause(e);
430 }
431 }
432
433 final int write(ByteBuffer src) throws IOException {
434 try {
435 return core.write(src, socketTimeout::get);
436 } catch (SocketClosedException e) {
437 throw (ClosedChannelException) new ClosedChannelException().initCause(e);
438 }
439 }
440
441 final boolean isConnected() {
442 if (connected.get()) {
443 return true;
444 }
445 if (isClosed()) {
446 return false;
447 }
448 if (core.isConnected(false)) {
449 connected.set(true);
450 return true;
451 }
452 return false;
453 }
454
455 final boolean isBound() {
456 if (bound.get()) {
457 return true;
458 }
459 if (isClosed()) {
460 return false;
461 }
462 if (core.isConnected(true)) {
463 bound.set(true);
464 return true;
465 }
466 return false;
467 }
468
469 final void updatePorts(int local, int remote) {
470 this.localPort.set(local);
471 this.remotePort.set(remote);
472 }
473
474 final @Nullable A getLocalSocketAddress() {
475 return AFSocketAddress.getSocketAddress(getFileDescriptor(), false, localPort.get(),
476 getAddressFamily());
477 }
478
479 final @Nullable A getRemoteSocketAddress() {
480 return AFSocketAddress.getSocketAddress(getFileDescriptor(), true, remotePort.get(),
481 getAddressFamily());
482 }
483
484
485
486
487
488
489 protected final AFAddressFamily<@NonNull A> getAddressFamily() {
490 return addressFamily;
491 }
492
493
494
495
496
497
498
499
500 protected final synchronized AFSocketImplExtensions<A> getImplExtensions() {
501 if (implExtensions == null) {
502 implExtensions = addressFamily.initImplExtensions(ancillaryDataSupport);
503 }
504 return implExtensions;
505 }
506
507
508 @SuppressWarnings("Finally" )
509 final boolean accept0(AFDatagramSocketImpl<A> socket) throws IOException {
510 FileDescriptor fdesc = core.validFdOrException();
511 if (isClosed()) {
512 throw new SocketException("Socket is closed");
513 } else if (!isBound()) {
514 throw new SocketException("Socket is not bound");
515 }
516
517 AFSocketAddress socketAddress = core.socketAddress;
518 AFSocketAddress boundSocketAddress = getLocalSocketAddress();
519 if (boundSocketAddress != null) {
520
521 core.socketAddress = socketAddress = boundSocketAddress;
522 }
523
524 if (socketAddress == null) {
525 throw new SocketException("Socket is not bound");
526 }
527
528 final AFDatagramSocketImpl<A> si = socket;
529 core.incPendingAccepts();
530 try (Lease<ByteBuffer> abLease = socketAddress.getNativeAddressDirectBuffer()) {
531 ByteBuffer ab = abLease.get();
532
533 SocketException caught = null;
534 try {
535 if (!NativeUnixSocket.accept(ab, ab.limit(), fdesc, si.fd, core.inode.get(), socketTimeout
536 .get())) {
537 return false;
538 }
539 } catch (SocketException e) {
540 caught = e;
541 } finally {
542 if (!isBound() || isClosed()) {
543 if (getCore().isShutdownOnClose()) {
544 try {
545 NativeUnixSocket.shutdown(si.fd, SHUT_RD_WR);
546 } catch (Exception e) {
547
548 }
549 }
550 try {
551 NativeUnixSocket.close(si.fd);
552 } catch (Exception e) {
553
554 }
555 if (caught != null) {
556 throw caught;
557 } else {
558 throw new SocketClosedException("Socket is closed");
559 }
560 } else if (caught != null) {
561 throw caught;
562 }
563 }
564 } finally {
565 core.decPendingAccepts();
566 }
567 si.setSocketAddress(socketAddress);
568 si.connected.set(true);
569
570 return true;
571 }
572
573 final int getLocalPort1() {
574 return localPort.get();
575 }
576
577 final int getRemotePort() {
578 return remotePort.get();
579 }
580
581 final void setSocketAddress(AFSocketAddress socketAddress) {
582 if (socketAddress == null) {
583 this.core.socketAddress = null;
584 this.localPort.set(-1);
585 } else {
586 this.core.socketAddress = socketAddress;
587 if (this.localPort.get() <= 0) {
588 this.localPort.set(socketAddress.getPort());
589 }
590 }
591 }
592 }