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