1 /*
2 * Copyright (c) 1995, 2018, 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 rdma.ch;
27
28 import java.io.IOException;
29 import java.io.InputStream;
30 import java.io.OutputStream;
31 import java.io.FileDescriptor;
32
33 import java.util.Set;
34 import java.util.HashSet;
35 import java.util.Collections;
36 import java.net.Socket;
37 import java.net.ServerSocket;
38 import java.net.SocketImpl;
39 import java.net.SocketOption;
40 import java.net.SocketException;
41 import java.net.UnknownHostException;
42 import java.net.InetAddress;
43 import java.net.SocketAddress;
44 import java.net.InetSocketAddress;
45 import java.net.StandardSocketOptions;
46 import java.net.SocketOptions;
47 import java.lang.reflect.Field;
48 import sun.net.ConnectionResetException;
49 import java.security.AccessController;
50 import java.security.PrivilegedAction;
51 import sun.net.ext.RdmaSocketOptions;
52
53 public class RdmaSocketImpl extends SocketImpl
54 {
55 Socket socket = null;
56 ServerSocket serverSocket = null;
57
58 protected FileDescriptor fd;
59
60 int timeout; // timeout in millisec
61
62 int trafficClass;
63
64 protected InetAddress address;
65
66 protected int port;
67
68 protected int localport;
69
70 private boolean shut_rd = false;
71 private boolean shut_wr = false;
72
73 private RdmaSocketInputStream socketInputStream = null;
74 private RdmaSocketOutputStream socketOutputStream = null;
75
76 /* number of threads using the FileDescriptor */
77 protected int fdUseCount = 0;
78
79 /* lock when increment/decrementing fdUseCount */
80 protected final Object fdLock = new Object();
81
82 /* indicates a close is pending on the file descriptor */
83 protected boolean closePending = false;
84
85 /* indicates connection reset state */
86 private int CONNECTION_NOT_RESET = 0;
87 private int CONNECTION_RESET_PENDING = 1;
88 private int CONNECTION_RESET = 2;
89 private int resetState;
90 private final Object resetLock = new Object();
91
92 /* whether this Socket is a stream (TCP) socket or not (UDP)
93 */
94 protected boolean stream;
95
96 static final sun.net.ext.RdmaSocketOptions rdmaOptions =
97 sun.net.ext.RdmaSocketOptions.getInstance();
98
99 private static PlatformRdmaSocketImpl platformRdmaSocketImpl =
100 PlatformRdmaSocketImpl.get();
101
102 private Field sCreateState;
103 private Field sBoundState;
104 private Field sConnectState;
105 private Field ssCreateState;
106 private Field ssBoundState;
107 private Class<?> sClass;
108 private Class<?> ssClass;
109
110 boolean isRdmaAvailable() {
111 return platformRdmaSocketImpl.isRdmaAvailable();
112 }
113
114 void setSocket(Socket soc) throws NoSuchFieldException {
115 this.socket = soc;
116 try {
117 this.sClass = socket.getClass();
118 this.sCreateState = sClass.getDeclaredField("created");
119 this.sBoundState = sClass.getDeclaredField("bound");
120 this.sConnectState = sClass.getDeclaredField("connect");
121 } catch (NoSuchFieldException e) {
122 throw e;
123 }
124 }
125
126 Socket getSocket() {
127 return socket;
128 }
129
130 void setServerSocket(ServerSocket soc) throws NoSuchFieldException {
131 this.serverSocket = soc;
132 try {
133 this.ssClass = serverSocket.getClass();
134 this.ssCreateState = ssClass.getDeclaredField("created");
135 this.ssBoundState = ssClass.getDeclaredField("bound");
136 } catch (NoSuchFieldException e) {
137 throw e;
138 }
139 }
140
141 ServerSocket getServerSocket() {
142 return serverSocket;
143 }
144
145 @Override
146 protected Set<SocketOption<?>> supportedOptions() {
147 Set<SocketOption<?>> options;
148 if (isRdmaAvailable()) {
149 options = new HashSet<>();
150 options.addAll(super.supportedOptions());
151 options.add(jdk.net.RdmaSocketOptions.RDMA_SQSIZE);
152 options.add(jdk.net.RdmaSocketOptions.RDMA_RQSIZE);
153 options.add(jdk.net.RdmaSocketOptions.RDMA_INLINE);
154 options = Collections.unmodifiableSet(options);
155 } else {
156 options = super.supportedOptions();
157 }
158 return options;
159 }
160
161 protected synchronized void create(boolean stream) throws IOException {
162 this.stream = stream;
163 if (stream) {
164 fd = new FileDescriptor();
165 platformRdmaSocketImpl.rdmaSocketCreate(true, this);
166 }
167 try {
168 if (socket != null)
169 sCreateState.setBoolean(socket, true);
170 if (serverSocket != null)
171 ssCreateState.setBoolean(serverSocket, true);
172 } catch (IllegalAccessException e) {
173 throw new AssertionError(e);
174 }
175 }
176
177 protected void connect(String host, int port)
178 throws UnknownHostException, IOException
179 {
180 boolean connected = false;
181 try {
182 InetAddress address = InetAddress.getByName(host);
183 this.port = port;
184 this.address = address;
185
186 connectToAddress(address, port, timeout);
187 connected = true;
188 } finally {
189 if (!connected) {
190 try {
191 close();
192 } catch (IOException ioe) {
193 }
194 }
195 }
196 }
197
198 protected void connect(InetAddress address, int port) throws IOException {
199 this.port = port;
200 this.address = address;
201
202 try {
203 connectToAddress(address, port, timeout);
204 return;
205 } catch (IOException e) {
206 close();
207 throw e;
208 }
209 }
210
211 protected void connect(SocketAddress address, int timeout)
212 throws IOException {
213 boolean connected = false;
214 try {
215 if (address == null || !(address instanceof InetSocketAddress))
216 throw new IllegalArgumentException("unsupported address type");
217 InetSocketAddress addr = (InetSocketAddress) address;
218 if (addr.isUnresolved())
219 throw new UnknownHostException(addr.getHostName());
220 this.port = addr.getPort();
221 this.address = addr.getAddress();
222
223 connectToAddress(this.address, port, timeout);
224 connected = true;
225 } finally {
226 if (!connected) {
227 try {
228 close();
229 } catch (IOException ioe) {
230 }
231 }
232 }
233 }
234
235 private void connectToAddress(InetAddress address, int port, int timeout) throws IOException {
236 if (address.isAnyLocalAddress()) {
237 doConnect(InetAddress.getLocalHost(), port, timeout);
238 } else {
239 doConnect(address, port, timeout);
240 }
241 }
242
243 protected <T> void setOption(SocketOption<T> name, T value) throws IOException {
244 if (!rdmaOptions.isOptionSupported(name)) {
245 int opt;
246 if (name == StandardSocketOptions.SO_KEEPALIVE) {
247 opt = SocketOptions.SO_KEEPALIVE;
248 } else if (name == StandardSocketOptions.SO_SNDBUF) {
249 opt = SocketOptions.SO_SNDBUF;
250 } else if (name == StandardSocketOptions.SO_RCVBUF) {
251 opt = SocketOptions.SO_RCVBUF;
252 } else if (name == StandardSocketOptions.SO_REUSEADDR) {
253 opt = SocketOptions.SO_REUSEADDR;
254 } else if (name == StandardSocketOptions.TCP_NODELAY) {
255 opt = SocketOptions.TCP_NODELAY;
256 } else {
257 throw new UnsupportedOperationException("unsupported option");
258 }
259 setOption(opt, value);
260 } else {
261 if (((socket != null) && !socket.isConnected())
262 || ((serverSocket != null) && !serverSocket.isBound())) {
263 rdmaOptions.setOption(fd, name, value);
264 } else {
265 throw new UnsupportedOperationException("RDMA socket option must be set before connect!");
266 }
267 }
268 }
269
270 @SuppressWarnings("unchecked")
271 protected <T> T getOption(SocketOption<T> name) throws IOException {
272 if (!rdmaOptions.isOptionSupported(name)) {
273 int opt;
274 if (name == StandardSocketOptions.SO_KEEPALIVE) {
275 opt = SocketOptions.SO_KEEPALIVE;
276 } else if (name == StandardSocketOptions.SO_SNDBUF) {
277 opt = SocketOptions.SO_SNDBUF;
278 } else if (name == StandardSocketOptions.SO_RCVBUF) {
279 opt = SocketOptions.SO_RCVBUF;
280 } else if (name == StandardSocketOptions.SO_REUSEADDR) {
281 opt = SocketOptions.SO_REUSEADDR;
282 } else if (name == StandardSocketOptions.TCP_NODELAY) {
283 opt = SocketOptions.TCP_NODELAY;
284 } else {
285 throw new UnsupportedOperationException("unsupported option");
286 }
287 return (T) getOption(opt);
288 } else {
289 return (T) rdmaOptions.getOption(fd, name);
290 }
291 }
292
293 public void setOption(int opt, Object val) throws SocketException {
294 if (isClosedOrPending()) {
295 throw new SocketException("Socket Closed");
296 }
297 boolean on = true;
298 switch (opt) {
299 case SO_TIMEOUT:
300 if (val == null || (!(val instanceof Integer)))
301 throw new SocketException("Bad parameter for SO_TIMEOUT");
302 int tmp = ((Integer) val).intValue();
303 if (tmp < 0)
304 throw new IllegalArgumentException("timeout < 0");
305 timeout = tmp;
306 break;
307 case SO_BINDADDR:
308 throw new SocketException("Cannot re-bind socket");
309 case TCP_NODELAY:
310 if (val == null || !(val instanceof Boolean))
311 throw new SocketException("bad parameter for TCP_NODELAY");
312 on = ((Boolean)val).booleanValue();
313 break;
314 case SO_SNDBUF:
315 case SO_RCVBUF:
316 if (val == null || !(val instanceof Integer) ||
317 !(((Integer)val).intValue() > 0)) {
318 throw new SocketException("bad parameter for SO_SNDBUF " +
319 "or SO_RCVBUF");
320 }
321 break;
322 case SO_KEEPALIVE:
323 if (val == null || !(val instanceof Boolean))
324 throw new SocketException("bad parameter for SO_KEEPALIVE");
325 on = ((Boolean)val).booleanValue();
326 break;
327 case SO_REUSEADDR:
328 if (val == null || !(val instanceof Boolean))
329 throw new SocketException("bad parameter for SO_REUSEADDR");
330 on = ((Boolean)val).booleanValue();
331 break;
332 default:
333 throw new SocketException("unrecognized TCP option: " + opt);
334 }
335 socketSetOption(opt, on, val);
336 }
337
338 public Object getOption(int opt) throws SocketException {
339 if (isClosedOrPending()) {
340 throw new SocketException("Socket Closed");
341 }
342 if (opt == SO_TIMEOUT) {
343 return timeout;
344 }
345 int ret = 0;
346
347 switch (opt) {
348 case TCP_NODELAY:
349 ret = platformRdmaSocketImpl.rdmaSocketGetOption(this, opt, null);
350 return Boolean.valueOf(ret != -1);
351 case SO_REUSEADDR:
352 ret = platformRdmaSocketImpl.rdmaSocketGetOption(this, opt, null);
353 return Boolean.valueOf(ret != -1);
354 case SO_SNDBUF:
355 case SO_RCVBUF:
356 ret = platformRdmaSocketImpl.rdmaSocketGetOption(this, opt, null);
357 return ret;
358 case SO_KEEPALIVE:
359 ret = platformRdmaSocketImpl.rdmaSocketGetOption(this, opt, null);
360 return Boolean.valueOf(ret != -1);
361 default:
362 return null;
363 }
364 }
365
366 protected void socketSetOption(int opt, boolean b, Object val) throws SocketException {
367 if (opt == SocketOptions.SO_REUSEPORT &&
368 !supportedOptions().contains(StandardSocketOptions.SO_REUSEPORT)) {
369 throw new UnsupportedOperationException("unsupported option");
370 }
371 try {
372 platformRdmaSocketImpl.rdmaSocketSetOption(this, opt, b, val);
373 } catch (SocketException se) {
374 if (socket == null || !socket.isConnected())
375 throw se;
376 }
377 }
378
379 synchronized void doConnect(InetAddress address, int port, int timeout) throws IOException {
380 try {
381 acquireFD();
382 try {
383 platformRdmaSocketImpl.rdmaSocketConnect(this, address, port, timeout);
384 synchronized (fdLock) {
385 if (closePending) {
386 throw new SocketException ("Socket closed");
387 }
388 }
389 try {
390 if (socket != null) {
391 sBoundState.setBoolean(socket, true);
392 sConnectState.setBoolean(socket, true);
393 }
394 } catch (IllegalAccessException e) {
395 throw new AssertionError(e);
396 }
397 } finally {
398 releaseFD();
399 }
400 } catch (IOException e) {
401 close();
402 throw e;
403 }
404 }
405
406 protected synchronized void bind(InetAddress address, int lport)
407 throws IOException
408 {
409 platformRdmaSocketImpl.rdmaSocketBind(this, address, lport);
410 try {
411 if (socket != null)
412 sBoundState.setBoolean(socket, true);
413 if (serverSocket != null)
414 sBoundState.setBoolean(serverSocket, true);
415 } catch (IllegalAccessException e) {
416 throw new AssertionError(e);
417 }
418 }
419
420 protected synchronized void listen(int count) throws IOException {
421 platformRdmaSocketImpl.rdmaSocketListen(this, count);
422 }
423
424 protected void accept(SocketImpl s) throws IOException {
425 acquireFD();
426 try {
427 platformRdmaSocketImpl.rdmaSocketAccept(s, this);
428 } finally {
429 releaseFD();
430 }
431 }
432
433 protected synchronized InputStream getInputStream() throws IOException {
434 synchronized (fdLock) {
435 if (isClosedOrPending())
436 throw new IOException("Socket Closed");
437 if (shut_rd)
438 throw new IOException("Socket input is shutdown");
439 if (socketInputStream == null)
440 socketInputStream = new RdmaSocketInputStream(this);
441 }
442 return socketInputStream;
443 }
444
445 void setInputStream(RdmaSocketInputStream in) {
446 socketInputStream = in;
447 }
448
449 protected synchronized OutputStream getOutputStream() throws IOException {
450 synchronized (fdLock) {
451 if (isClosedOrPending())
452 throw new IOException("Socket Closed");
453 if (shut_wr)
454 throw new IOException("Socket output is shutdown");
455 if (socketOutputStream == null)
456 socketOutputStream = new RdmaSocketOutputStream(this);
457 }
458 return socketOutputStream;
459 }
460
461 protected FileDescriptor getFileDescriptor() {
462 return fd;
463 }
464
465 protected void setFileDescriptor(FileDescriptor fd) {
466 this.fd = fd;
467 }
468
469 protected void setAddress(InetAddress address) {
470 this.address = address;
471 }
472
473 void setPort(int port) {
474 this.port = port;
475 }
476
477 void setLocalPort(int localport) {
478 this.localport = localport;
479 }
480
481 protected synchronized int available() throws IOException {
482 if (isClosedOrPending()) {
483 throw new IOException("Stream closed.");
484 }
485
486 if (isConnectionReset() || shut_rd) {
487 return 0;
488 }
489
490 int n = 0;
491 try {
492 n = platformRdmaSocketImpl.rdmaSocketAvailable(this);
493 if (n == 0 && isConnectionResetPending()) {
494 setConnectionReset();
495 }
496 } catch (ConnectionResetException exc1) {
497 setConnectionResetPending();
498 try {
499 n = platformRdmaSocketImpl.rdmaSocketAvailable(this);
500 if (n == 0) {
501 setConnectionReset();
502 }
503 } catch (ConnectionResetException exc2) {
504 }
505 }
506 return n;
507 }
508
509 protected void close() throws IOException {
510 synchronized(fdLock) {
511 if (fd != null) {
512 if (fdUseCount == 0) {
513 if (closePending) {
514 return;
515 }
516 closePending = true;
517 try {
518 platformRdmaSocketImpl.rdmaSocketClose(true, this);
519 } finally {
520 platformRdmaSocketImpl.rdmaSocketClose(false, this);
521 }
522 fd = null;
523 return;
524 } else {
525 if (!closePending) {
526 closePending = true;
527 fdUseCount--;
528 platformRdmaSocketImpl.rdmaSocketClose(true, this);
529 }
530 }
531 }
532 }
533 }
534
535 void reset() throws IOException {
536 if (fd != null) {
537 platformRdmaSocketImpl.rdmaSocketClose(false, this);
538 }
539 fd = null;
540 postReset();
541 }
542
543 void postReset() throws IOException {
544 address = null;
545 port = 0;
546 localport = 0;
547 }
548
549 protected void shutdownInput() throws IOException {
550 if (fd != null) {
551 platformRdmaSocketImpl.rdmaSocketShutdown(SHUT_RD, this);
552 if (socketInputStream != null) {
553 socketInputStream.setEOF(true);
554 }
555 shut_rd = true;
556 }
557 }
558
559 protected void shutdownOutput() throws IOException {
560 if (fd != null) {
561 platformRdmaSocketImpl.rdmaSocketShutdown(SHUT_WR, this);
562 shut_wr = true;
563 }
564 }
565
566 protected boolean supportsUrgentData () {
567 return true;
568 }
569
570 protected void sendUrgentData (int data) throws IOException {
571 if (fd == null) {
572 throw new IOException("Socket Closed");
573 }
574 platformRdmaSocketImpl.rdmaSocketSendUrgentData(this, data);
575 }
576
577 FileDescriptor acquireFD() {
578 synchronized (fdLock) {
579 fdUseCount++;
580 return fd;
581 }
582 }
583
584 void releaseFD() {
585 synchronized (fdLock) {
586 fdUseCount--;
587 if (fdUseCount == -1) {
588 if (fd != null) {
589 try {
590 platformRdmaSocketImpl.rdmaSocketClose(false, this);
591 } catch (IOException e) {
592 } finally {
593 fd = null;
594 }
595 }
596 }
597 }
598 }
599
600 public boolean isConnectionReset() {
601 synchronized (resetLock) {
602 return (resetState == CONNECTION_RESET);
603 }
604 }
605
606 public boolean isConnectionResetPending() {
607 synchronized (resetLock) {
608 return (resetState == CONNECTION_RESET_PENDING);
609 }
610 }
611
612 public void setConnectionReset() {
613 synchronized (resetLock) {
614 resetState = CONNECTION_RESET;
615 }
616 }
617
618 public void setConnectionResetPending() {
619 synchronized (resetLock) {
620 if (resetState == CONNECTION_NOT_RESET) {
621 resetState = CONNECTION_RESET_PENDING;
622 }
623 }
624
625 }
626
627 public boolean isClosedOrPending() {
628 synchronized (fdLock) {
629 if (closePending || (fd == null)) {
630 return true;
631 } else {
632 return false;
633 }
634 }
635 }
636
637 public int getTimeout() {
638 return timeout;
639 }
640
641 protected InetAddress getInetAddress() {
642 return address;
643 }
644
645 protected int getPort() {
646 return port;
647 }
648
649 protected int getLocalPort() {
650 return localport;
651 }
652
653 public static final int SHUT_RD = 0;
654 public static final int SHUT_WR = 1;
655
656 static class PlatformRdmaSocketImpl {
657
658 @SuppressWarnings("unchecked")
659 private static PlatformRdmaSocketImpl newInstance(String cn) {
660 Class<PlatformRdmaSocketImpl> c;
661 try {
662 c = (Class<PlatformRdmaSocketImpl>)Class.forName(cn);
663 return c.getConstructor(new Class<?>[] {}).newInstance();
664 } catch (ReflectiveOperationException x) {
665 throw new AssertionError(x);
666 }
667 }
668
669 private static PlatformRdmaSocketImpl create() {
670 String osname = AccessController.doPrivileged(
671 new PrivilegedAction<String>() {
672 public String run() {
673 return System.getProperty("os.name");
674 }
675 });
676 if ("Linux".equals(osname))
677 return newInstance("rdma.ch.LinuxRdmaSocketImpl");
678 return new PlatformRdmaSocketImpl();
679 }
680
681 private static final PlatformRdmaSocketImpl instance = create();
682
683 static PlatformRdmaSocketImpl get() {
684 return instance;
685 }
686
687 boolean isRdmaAvailable() {
688 return false;
689 }
690
691 void rdmaSocketClose(boolean useDeferredClose, RdmaSocketImpl impl) throws IOException {
692 throw new UnsupportedOperationException("unsupported socket option");
693 }
694
695 void rdmaSocketCreate(boolean isServer, RdmaSocketImpl impl) throws IOException {
696 throw new UnsupportedOperationException("unsupported socket option");
697 }
698
699 void rdmaSocketConnect(RdmaSocketImpl impl, InetAddress address, int port, int timeout)
700 throws IOException {
701 throw new UnsupportedOperationException("unsupported socket option");
702 }
703
704 void rdmaSocketBind(RdmaSocketImpl impl, InetAddress address, int port)
705 throws IOException {
706 throw new UnsupportedOperationException("unsupported socket option");
707 }
708
709 void rdmaSocketListen(RdmaSocketImpl impl, int count)
710 throws IOException {
711 throw new UnsupportedOperationException("unsupported socket option");
712 }
713
714 void rdmaSocketAccept(SocketImpl s, RdmaSocketImpl impl)
715 throws IOException {
716 throw new UnsupportedOperationException("unsupported socket option");
717 }
718
719 int rdmaSocketAvailable(RdmaSocketImpl impl)
720 throws IOException {
721 throw new UnsupportedOperationException("unsupported socket option");
722 }
723
724 void rdmaSocketShutdown(int howto, RdmaSocketImpl impl)
725 throws IOException {
726 throw new UnsupportedOperationException("unsupported socket option");
727 }
728
729 void rdmaSocketSetOption(RdmaSocketImpl impl, int cmd, boolean on, Object value)
730 throws SocketException {
731 throw new UnsupportedOperationException("unsupported socket option");
732 }
733
734 int rdmaSocketGetOption(RdmaSocketImpl impl, int opt, Object iaContainerObj)
735 throws SocketException {
736 throw new UnsupportedOperationException("unsupported socket option");
737 }
738
739 void rdmaSocketSendUrgentData(RdmaSocketImpl impl, int data)
740 throws IOException {
741 throw new UnsupportedOperationException("unsupported socket option");
742 }
743 }
744 }