1 /*
   2  * Copyright (c) 2009, 2011, 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 package sun.nio.ch.sctp;
  26 
  27 import java.net.InetAddress;
  28 import java.net.SocketAddress;
  29 import java.net.SocketException;
  30 import java.net.InetSocketAddress;
  31 import java.io.FileDescriptor;
  32 import java.io.IOException;
  33 import java.util.Collections;
  34 import java.util.Set;
  35 import java.util.HashSet;
  36 import java.nio.ByteBuffer;
  37 import java.nio.channels.SelectionKey;
  38 import java.nio.channels.ClosedChannelException;
  39 import java.nio.channels.ConnectionPendingException;
  40 import java.nio.channels.NoConnectionPendingException;
  41 import java.nio.channels.AlreadyConnectedException;
  42 import java.nio.channels.NotYetBoundException;
  43 import java.nio.channels.NotYetConnectedException;
  44 import java.nio.channels.spi.SelectorProvider;
  45 import com.sun.nio.sctp.AbstractNotificationHandler;
  46 import com.sun.nio.sctp.Association;
  47 import com.sun.nio.sctp.AssociationChangeNotification;
  48 import com.sun.nio.sctp.HandlerResult;
  49 import com.sun.nio.sctp.IllegalReceiveException;
  50 import com.sun.nio.sctp.InvalidStreamException;
  51 import com.sun.nio.sctp.IllegalUnbindException;
  52 import com.sun.nio.sctp.MessageInfo;
  53 import com.sun.nio.sctp.NotificationHandler;
  54 import com.sun.nio.sctp.SctpChannel;
  55 import com.sun.nio.sctp.SctpSocketOption;
  56 import sun.nio.ch.DirectBuffer;
  57 import sun.nio.ch.IOStatus;
  58 import sun.nio.ch.IOUtil;
  59 import sun.nio.ch.NativeThread;
  60 import sun.nio.ch.Net;
  61 import sun.nio.ch.PollArrayWrapper;
  62 import sun.nio.ch.SelChImpl;
  63 import sun.nio.ch.SelectionKeyImpl;
  64 import sun.nio.ch.Util;
  65 import static com.sun.nio.sctp.SctpStandardSocketOptions.*;
  66 import static sun.nio.ch.sctp.ResultContainer.SEND_FAILED;
  67 import static sun.nio.ch.sctp.ResultContainer.ASSOCIATION_CHANGED;
  68 import static sun.nio.ch.sctp.ResultContainer.PEER_ADDRESS_CHANGED;
  69 import static sun.nio.ch.sctp.ResultContainer.SHUTDOWN;
  70 
  71 /**
  72  * An implementation of an SctpChannel
  73  */
  74 public class SctpChannelImpl extends SctpChannel
  75     implements SelChImpl
  76 {
  77     private final FileDescriptor fd;
  78 
  79     private final int fdVal;
  80 
  81     /* IDs of native threads doing send and receivess, for signalling */
  82     private volatile long receiverThread = 0;
  83     private volatile long senderThread = 0;
  84 
  85     /* Lock held by current receiving or connecting thread */
  86     private final Object receiveLock = new Object();
  87 
  88     /* Lock held by current sending or connecting thread */
  89     private final Object sendLock = new Object();
  90 
  91     private final ThreadLocal<Boolean> receiveInvoked =
  92         new ThreadLocal<Boolean>() {
  93              @Override protected Boolean initialValue() {
  94                  return Boolean.FALSE;
  95             }
  96     };
  97 
  98     /* Lock held by any thread that modifies the state fields declared below
  99        DO NOT invoke a blocking I/O operation while holding this lock! */
 100     private final Object stateLock = new Object();
 101 
 102     private enum ChannelState {
 103         UNINITIALIZED,
 104         UNCONNECTED,
 105         PENDING,
 106         CONNECTED,
 107         KILLPENDING,
 108         KILLED,
 109     }
 110     /* -- The following fields are protected by stateLock -- */
 111     private ChannelState state = ChannelState.UNINITIALIZED;
 112 
 113     /* Binding; Once bound the port will remain constant. */
 114     int port = -1;
 115     private HashSet<InetSocketAddress> localAddresses = new HashSet<InetSocketAddress>();
 116     /* Has the channel been bound to the wildcard address */
 117     private boolean wildcard; /* false */
 118     //private InetSocketAddress remoteAddress = null;
 119 
 120     /* Input/Output open */
 121     private boolean readyToConnect;
 122 
 123     /* Shutdown */
 124     private boolean isShutdown;
 125 
 126     private Association association;
 127 
 128     private Set<SocketAddress> remoteAddresses = Collections.emptySet();
 129 
 130     /* -- End of fields protected by stateLock -- */
 131 
 132     /**
 133      * Constructor for normal connecting sockets
 134      */
 135     public SctpChannelImpl(SelectorProvider provider) throws IOException {
 136         //TODO: update provider remove public modifier
 137         super(provider);
 138         this.fd = SctpNet.socket(true);
 139         this.fdVal = IOUtil.fdVal(fd);
 140         this.state = ChannelState.UNCONNECTED;
 141     }
 142 
 143     /**
 144      * Constructor for sockets obtained from server sockets
 145      */
 146     public SctpChannelImpl(SelectorProvider provider, FileDescriptor fd)
 147          throws IOException {
 148         this(provider, fd, null);
 149     }
 150 
 151     /**
 152      * Constructor for sockets obtained from branching
 153      */
 154     public SctpChannelImpl(SelectorProvider provider,
 155                            FileDescriptor fd,
 156                            Association association)
 157             throws IOException {
 158         super(provider);
 159         this.fd = fd;
 160         this.fdVal = IOUtil.fdVal(fd);
 161         this.state = ChannelState.CONNECTED;
 162         port = (Net.localAddress(fd)).getPort();
 163 
 164         if (association != null) { /* branched */
 165             this.association = association;
 166         } else { /* obtained from server channel */
 167             /* Receive COMM_UP */
 168             ByteBuffer buf = Util.getTemporaryDirectBuffer(50);
 169             try {
 170                 receive(buf, null, null, true);
 171             } finally {
 172                 Util.releaseTemporaryDirectBuffer(buf);
 173             }
 174         }
 175     }
 176 
 177     /**
 178      * Binds the channel's socket to a local address.
 179      */
 180     @Override
 181     public SctpChannel bind(SocketAddress local) throws IOException {
 182         synchronized (receiveLock) {
 183             synchronized (sendLock) {
 184                 synchronized (stateLock) {
 185                     ensureOpenAndUnconnected();
 186                     if (isBound())
 187                         SctpNet.throwAlreadyBoundException();
 188                     InetSocketAddress isa = (local == null) ?
 189                         new InetSocketAddress(0) : Net.checkAddress(local);
 190                     Net.bind(fd, isa.getAddress(), isa.getPort());
 191                     InetSocketAddress boundIsa = Net.localAddress(fd);
 192                     port = boundIsa.getPort();
 193                     localAddresses.add(isa);
 194                     if (isa.getAddress().isAnyLocalAddress())
 195                         wildcard = true;
 196                 }
 197             }
 198         }
 199         return this;
 200     }
 201 
 202     @Override
 203     public SctpChannel bindAddress(InetAddress address)
 204             throws IOException {
 205         bindUnbindAddress(address, true);
 206         localAddresses.add(new InetSocketAddress(address, port));
 207         return this;
 208     }
 209 
 210     @Override
 211     public SctpChannel unbindAddress(InetAddress address)
 212             throws IOException {
 213         bindUnbindAddress(address, false);
 214         localAddresses.remove(new InetSocketAddress(address, port));
 215         return this;
 216     }
 217 
 218     private SctpChannel bindUnbindAddress(InetAddress address, boolean add)
 219             throws IOException {
 220         if (address == null)
 221             throw new IllegalArgumentException();
 222 
 223         synchronized (receiveLock) {
 224             synchronized (sendLock) {
 225                 synchronized (stateLock) {
 226                     if (!isOpen())
 227                         throw new ClosedChannelException();
 228                     if (!isBound())
 229                         throw new NotYetBoundException();
 230                     if (wildcard)
 231                         throw new IllegalStateException(
 232                                 "Cannot add or remove addresses from a channel that is bound to the wildcard address");
 233                     if (address.isAnyLocalAddress())
 234                         throw new IllegalArgumentException(
 235                                 "Cannot add or remove the wildcard address");
 236                     if (add) {
 237                         for (InetSocketAddress addr : localAddresses) {
 238                             if (addr.getAddress().equals(address)) {
 239                                 SctpNet.throwAlreadyBoundException();
 240                             }
 241                         }
 242                     } else { /*removing */
 243                         /* Verify that there is more than one address
 244                          * and that address is already bound */
 245                         if (localAddresses.size() <= 1)
 246                             throw new IllegalUnbindException("Cannot remove address from a channel with only one address bound");
 247                         boolean foundAddress = false;
 248                         for (InetSocketAddress addr : localAddresses) {
 249                             if (addr.getAddress().equals(address)) {
 250                                 foundAddress = true;
 251                                 break;
 252                             }
 253                         }
 254                         if (!foundAddress )
 255                             throw new IllegalUnbindException("Cannot remove address from a channel that is not bound to that address");
 256                     }
 257 
 258                     SctpNet.bindx(fdVal, new InetAddress[]{address}, port, add);
 259 
 260                     /* Update our internal Set to reflect the addition/removal */
 261                     if (add)
 262                         localAddresses.add(new InetSocketAddress(address, port));
 263                     else {
 264                         for (InetSocketAddress addr : localAddresses) {
 265                             if (addr.getAddress().equals(address)) {
 266                                 localAddresses.remove(addr);
 267                                 break;
 268                             }
 269                         }
 270                     }
 271                 }
 272             }
 273         }
 274         return this;
 275     }
 276 
 277     private boolean isBound() {
 278         synchronized (stateLock) {
 279             return port == -1 ? false : true;
 280         }
 281     }
 282 
 283     private boolean isConnected() {
 284         synchronized (stateLock) {
 285             return (state == ChannelState.CONNECTED);
 286         }
 287     }
 288 
 289     private void ensureOpenAndUnconnected() throws IOException {
 290         synchronized (stateLock) {
 291             if (!isOpen())
 292                 throw new ClosedChannelException();
 293             if (isConnected())
 294                 throw new AlreadyConnectedException();
 295             if (state == ChannelState.PENDING)
 296                 throw new ConnectionPendingException();
 297         }
 298     }
 299 
 300     private boolean ensureReceiveOpen() throws ClosedChannelException {
 301         synchronized (stateLock) {
 302             if (!isOpen())
 303                 throw new ClosedChannelException();
 304             if (!isConnected())
 305                 throw new NotYetConnectedException();
 306             else
 307                 return true;
 308         }
 309     }
 310 
 311     private void ensureSendOpen() throws ClosedChannelException {
 312         synchronized (stateLock) {
 313             if (!isOpen())
 314                 throw new ClosedChannelException();
 315             if (isShutdown)
 316                 throw new ClosedChannelException();
 317             if (!isConnected())
 318                 throw new NotYetConnectedException();
 319         }
 320     }
 321 
 322     private void receiverCleanup() throws IOException {
 323         synchronized (stateLock) {
 324             receiverThread = 0;
 325             if (state == ChannelState.KILLPENDING)
 326                 kill();
 327         }
 328     }
 329 
 330     private void senderCleanup() throws IOException {
 331         synchronized (stateLock) {
 332             senderThread = 0;
 333             if (state == ChannelState.KILLPENDING)
 334                 kill();
 335         }
 336     }
 337 
 338     @Override
 339     public Association association() throws ClosedChannelException {
 340         synchronized (stateLock) {
 341             if (!isOpen())
 342                 throw new ClosedChannelException();
 343             if (!isConnected())
 344                 return null;
 345 
 346             return association;
 347         }
 348     }
 349 
 350     @Override
 351     public boolean connect(SocketAddress endpoint) throws IOException {
 352         synchronized (receiveLock) {
 353             synchronized (sendLock) {
 354                 ensureOpenAndUnconnected();
 355                 InetSocketAddress isa = Net.checkAddress(endpoint);
 356                 SecurityManager sm = System.getSecurityManager();
 357                 if (sm != null)
 358                     sm.checkConnect(isa.getAddress().getHostAddress(),
 359                                     isa.getPort());
 360                 synchronized (blockingLock()) {
 361                     int n = 0;
 362                     try {
 363                         try {
 364                             begin();
 365                             synchronized (stateLock) {
 366                                 if (!isOpen()) {
 367                                     return false;
 368                                 }
 369                                 receiverThread = NativeThread.current();
 370                             }
 371                             for (;;) {
 372                                 InetAddress ia = isa.getAddress();
 373                                 if (ia.isAnyLocalAddress())
 374                                     ia = InetAddress.getLocalHost();
 375                                 n = SctpNet.connect(fdVal, ia, isa.getPort());
 376                                 if (  (n == IOStatus.INTERRUPTED)
 377                                       && isOpen())
 378                                     continue;
 379                                 break;
 380                             }
 381                         } finally {
 382                             receiverCleanup();
 383                             end((n > 0) || (n == IOStatus.UNAVAILABLE));
 384                             assert IOStatus.check(n);
 385                         }
 386                     } catch (IOException x) {
 387                         /* If an exception was thrown, close the channel after
 388                          * invoking end() so as to avoid bogus
 389                          * AsynchronousCloseExceptions */
 390                         close();
 391                         throw x;
 392                     }
 393 
 394                     if (n > 0) {
 395                         synchronized (stateLock) {
 396                             /* Connection succeeded */
 397                             state = ChannelState.CONNECTED;
 398                             if (!isBound()) {
 399                                 InetSocketAddress boundIsa =
 400                                         Net.localAddress(fd);
 401                                 port = boundIsa.getPort();
 402                             }
 403 
 404                             /* Receive COMM_UP */
 405                             ByteBuffer buf = Util.getTemporaryDirectBuffer(50);
 406                             try {
 407                                 receive(buf, null, null, true);
 408                             } finally {
 409                                 Util.releaseTemporaryDirectBuffer(buf);
 410                             }
 411 
 412                             /* cache remote addresses */
 413                             try {
 414                                 remoteAddresses = getRemoteAddresses();
 415                             } catch (IOException unused) { /* swallow exception */ }
 416 
 417                             return true;
 418                         }
 419                     } else  {
 420                         synchronized (stateLock) {
 421                             /* If nonblocking and no exception then connection
 422                              * pending; disallow another invocation */
 423                             if (!isBlocking())
 424                                 state = ChannelState.PENDING;
 425                             else
 426                                 assert false;
 427                         }
 428                     }
 429                 }
 430                 return false;
 431             }
 432         }
 433     }
 434 
 435     @Override
 436     public boolean connect(SocketAddress endpoint,
 437                            int maxOutStreams,
 438                            int maxInStreams)
 439             throws IOException {
 440         ensureOpenAndUnconnected();
 441         return setOption(SCTP_INIT_MAXSTREAMS, InitMaxStreams.
 442                 create(maxInStreams, maxOutStreams)).connect(endpoint);
 443 
 444     }
 445 
 446     @Override
 447     public boolean isConnectionPending() {
 448         synchronized (stateLock) {
 449             return (state == ChannelState.PENDING);
 450         }
 451     }
 452 
 453     @Override
 454     public boolean finishConnect() throws IOException {
 455         synchronized (receiveLock) {
 456             synchronized (sendLock) {
 457                 synchronized (stateLock) {
 458                     if (!isOpen())
 459                         throw new ClosedChannelException();
 460                     if (isConnected())
 461                         return true;
 462                     if (state != ChannelState.PENDING)
 463                         throw new NoConnectionPendingException();
 464                 }
 465                 int n = 0;
 466                 try {
 467                     try {
 468                         begin();
 469                         synchronized (blockingLock()) {
 470                             synchronized (stateLock) {
 471                                 if (!isOpen()) {
 472                                     return false;
 473                                 }
 474                                 receiverThread = NativeThread.current();
 475                             }
 476                             if (!isBlocking()) {
 477                                 for (;;) {
 478                                     n = checkConnect(fd, false, readyToConnect);
 479                                     if (  (n == IOStatus.INTERRUPTED)
 480                                           && isOpen())
 481                                         continue;
 482                                     break;
 483                                 }
 484                             } else {
 485                                 for (;;) {
 486                                     n = checkConnect(fd, true, readyToConnect);
 487                                     if (n == 0) {
 488                                         // Loop in case of
 489                                         // spurious notifications
 490                                         continue;
 491                                     }
 492                                     if (  (n == IOStatus.INTERRUPTED)
 493                                           && isOpen())
 494                                         continue;
 495                                     break;
 496                                 }
 497                             }
 498                         }
 499                     } finally {
 500                         synchronized (stateLock) {
 501                             receiverThread = 0;
 502                             if (state == ChannelState.KILLPENDING) {
 503                                 kill();
 504                                 /* poll()/getsockopt() does not report
 505                                  * error (throws exception, with n = 0)
 506                                  * on Linux platform after dup2 and
 507                                  * signal-wakeup. Force n to 0 so the
 508                                  * end() can throw appropriate exception */
 509                                 n = 0;
 510                             }
 511                         }
 512                         end((n > 0) || (n == IOStatus.UNAVAILABLE));
 513                         assert IOStatus.check(n);
 514                     }
 515                 } catch (IOException x) {
 516                     /* If an exception was thrown, close the channel after
 517                      * invoking end() so as to avoid bogus
 518                      * AsynchronousCloseExceptions */
 519                     close();
 520                     throw x;
 521                 }
 522 
 523                 if (n > 0) {
 524                     synchronized (stateLock) {
 525                         state = ChannelState.CONNECTED;
 526                         if (!isBound()) {
 527                             InetSocketAddress boundIsa =
 528                                     Net.localAddress(fd);
 529                             port = boundIsa.getPort();
 530                         }
 531 
 532                         /* Receive COMM_UP */
 533                         ByteBuffer buf = Util.getTemporaryDirectBuffer(50);
 534                         try {
 535                             receive(buf, null, null, true);
 536                         } finally {
 537                             Util.releaseTemporaryDirectBuffer(buf);
 538                         }
 539 
 540                         /* cache remote addresses */
 541                         try {
 542                             remoteAddresses = getRemoteAddresses();
 543                         } catch (IOException unused) { /* swallow exception */ }
 544 
 545                         return true;
 546                     }
 547                 }
 548             }
 549         }
 550         return false;
 551     }
 552 
 553     @Override
 554     protected void implConfigureBlocking(boolean block) throws IOException {
 555         IOUtil.configureBlocking(fd, block);
 556     }
 557 
 558     @Override
 559     public void implCloseSelectableChannel() throws IOException {
 560         synchronized (stateLock) {
 561             SctpNet.preClose(fdVal);
 562 
 563             if (receiverThread != 0)
 564                 NativeThread.signal(receiverThread);
 565 
 566             if (senderThread != 0)
 567                 NativeThread.signal(senderThread);
 568 
 569             if (!isRegistered())
 570                 kill();
 571         }
 572     }
 573 
 574     @Override
 575     public FileDescriptor getFD() {
 576         return fd;
 577     }
 578 
 579     @Override
 580     public int getFDVal() {
 581         return fdVal;
 582     }
 583 
 584     /**
 585      * Translates native poll revent ops into a ready operation ops
 586      */
 587     private boolean translateReadyOps(int ops, int initialOps, SelectionKeyImpl sk) {
 588         int intOps = sk.nioInterestOps();
 589         int oldOps = sk.nioReadyOps();
 590         int newOps = initialOps;
 591 
 592         if ((ops & PollArrayWrapper.POLLNVAL) != 0) {
 593             /* This should only happen if this channel is pre-closed while a
 594              * selection operation is in progress
 595              * ## Throw an error if this channel has not been pre-closed */
 596             return false;
 597         }
 598 
 599         if ((ops & (PollArrayWrapper.POLLERR
 600                     | PollArrayWrapper.POLLHUP)) != 0) {
 601             newOps = intOps;
 602             sk.nioReadyOps(newOps);
 603             /* No need to poll again in checkConnect,
 604              * the error will be detected there */
 605             readyToConnect = true;
 606             return (newOps & ~oldOps) != 0;
 607         }
 608 
 609         if (((ops & PollArrayWrapper.POLLIN) != 0) &&
 610             ((intOps & SelectionKey.OP_READ) != 0) &&
 611             isConnected())
 612             newOps |= SelectionKey.OP_READ;
 613 
 614         if (((ops & PollArrayWrapper.POLLCONN) != 0) &&
 615             ((intOps & SelectionKey.OP_CONNECT) != 0) &&
 616             ((state == ChannelState.UNCONNECTED) || (state == ChannelState.PENDING))) {
 617             newOps |= SelectionKey.OP_CONNECT;
 618             readyToConnect = true;
 619         }
 620 
 621         if (((ops & PollArrayWrapper.POLLOUT) != 0) &&
 622             ((intOps & SelectionKey.OP_WRITE) != 0) &&
 623             isConnected())
 624             newOps |= SelectionKey.OP_WRITE;
 625 
 626         sk.nioReadyOps(newOps);
 627         return (newOps & ~oldOps) != 0;
 628     }
 629 
 630     @Override
 631     public boolean translateAndUpdateReadyOps(int ops, SelectionKeyImpl sk) {
 632         return translateReadyOps(ops, sk.nioReadyOps(), sk);
 633     }
 634 
 635     @Override
 636     @SuppressWarnings("all")
 637     public boolean translateAndSetReadyOps(int ops, SelectionKeyImpl sk) {
 638         return translateReadyOps(ops, 0, sk);
 639     }
 640 
 641     @Override
 642     public void translateAndSetInterestOps(int ops, SelectionKeyImpl sk) {
 643         int newOps = 0;
 644         if ((ops & SelectionKey.OP_READ) != 0)
 645             newOps |= PollArrayWrapper.POLLIN;
 646         if ((ops & SelectionKey.OP_WRITE) != 0)
 647             newOps |= PollArrayWrapper.POLLOUT;
 648         if ((ops & SelectionKey.OP_CONNECT) != 0)
 649             newOps |= PollArrayWrapper.POLLCONN;
 650         sk.selector.putEventOps(sk, newOps);
 651     }
 652 
 653     @Override
 654     public void kill() throws IOException {
 655         synchronized (stateLock) {
 656             if (state == ChannelState.KILLED)
 657                 return;
 658             if (state == ChannelState.UNINITIALIZED) {
 659                 state = ChannelState.KILLED;
 660                 return;
 661             }
 662             assert !isOpen() && !isRegistered();
 663 
 664             /* Postpone the kill if there is a waiting reader
 665              * or writer thread. */
 666             if (receiverThread == 0 && senderThread == 0) {
 667                 SctpNet.close(fdVal);
 668                 state = ChannelState.KILLED;
 669             } else {
 670                 state = ChannelState.KILLPENDING;
 671             }
 672         }
 673     }
 674 
 675     @Override
 676     public <T> SctpChannel setOption(SctpSocketOption<T> name, T value)
 677             throws IOException {
 678         if (name == null)
 679             throw new NullPointerException();
 680         if (!supportedOptions().contains(name))
 681             throw new UnsupportedOperationException("'" + name + "' not supported");
 682 
 683         synchronized (stateLock) {
 684             if (!isOpen())
 685                 throw new ClosedChannelException();
 686 
 687             SctpNet.setSocketOption(fdVal, name, value, 0 /*oneToOne*/);
 688         }
 689         return this;
 690     }
 691 
 692     @Override
 693     @SuppressWarnings("unchecked")
 694     public <T> T getOption(SctpSocketOption<T> name) throws IOException {
 695         if (name == null)
 696             throw new NullPointerException();
 697         if (!supportedOptions().contains(name))
 698             throw new UnsupportedOperationException("'" + name + "' not supported");
 699 
 700         synchronized (stateLock) {
 701             if (!isOpen())
 702                 throw new ClosedChannelException();
 703 
 704             return (T)SctpNet.getSocketOption(fdVal, name, 0 /*oneToOne*/);
 705         }
 706     }
 707 
 708     private static class DefaultOptionsHolder {
 709         static final Set<SctpSocketOption<?>> defaultOptions = defaultOptions();
 710 
 711         private static Set<SctpSocketOption<?>> defaultOptions() {
 712             HashSet<SctpSocketOption<?>> set = new HashSet<SctpSocketOption<?>>(10);
 713             set.add(SCTP_DISABLE_FRAGMENTS);
 714             set.add(SCTP_EXPLICIT_COMPLETE);
 715             set.add(SCTP_FRAGMENT_INTERLEAVE);
 716             set.add(SCTP_INIT_MAXSTREAMS);
 717             set.add(SCTP_NODELAY);
 718             set.add(SCTP_PRIMARY_ADDR);
 719             set.add(SCTP_SET_PEER_PRIMARY_ADDR);
 720             set.add(SO_SNDBUF);
 721             set.add(SO_RCVBUF);
 722             set.add(SO_LINGER);
 723             return Collections.unmodifiableSet(set);
 724         }
 725     }
 726 
 727     @Override
 728     public final Set<SctpSocketOption<?>> supportedOptions() {
 729         return DefaultOptionsHolder.defaultOptions;
 730     }
 731 
 732     @Override
 733     public <T> MessageInfo receive(ByteBuffer buffer,
 734                                    T attachment,
 735                                    NotificationHandler<T> handler)
 736             throws IOException {
 737         return receive(buffer, attachment, handler, false);
 738     }
 739 
 740     private <T> MessageInfo receive(ByteBuffer buffer,
 741                                     T attachment,
 742                                     NotificationHandler<T> handler,
 743                                     boolean fromConnect)
 744             throws IOException {
 745         if (buffer == null)
 746             throw new IllegalArgumentException("buffer cannot be null");
 747 
 748         if (buffer.isReadOnly())
 749             throw new IllegalArgumentException("Read-only buffer");
 750 
 751         if (receiveInvoked.get())
 752             throw new IllegalReceiveException(
 753                     "cannot invoke receive from handler");
 754         receiveInvoked.set(Boolean.TRUE);
 755 
 756         try {
 757             ResultContainer resultContainer = new ResultContainer();
 758             do {
 759                 resultContainer.clear();
 760                 synchronized (receiveLock) {
 761                     if (!ensureReceiveOpen())
 762                         return null;
 763 
 764                     int n = 0;
 765                     try {
 766                         begin();
 767 
 768                         synchronized (stateLock) {
 769                             if(!isOpen())
 770                                 return null;
 771                             receiverThread = NativeThread.current();
 772                         }
 773 
 774                         do {
 775                             n = receive(fdVal, buffer, resultContainer, fromConnect);
 776                         } while ((n == IOStatus.INTERRUPTED) && isOpen());
 777                     } finally {
 778                         receiverCleanup();
 779                         end((n > 0) || (n == IOStatus.UNAVAILABLE));
 780                         assert IOStatus.check(n);
 781                     }
 782 
 783                     if (!resultContainer.isNotification()) {
 784                         /* message or nothing */
 785                         if (resultContainer.hasSomething()) {
 786                             /* Set the association before returning */
 787                             MessageInfoImpl info =
 788                                     resultContainer.getMessageInfo();
 789                             synchronized (stateLock) {
 790                                 assert association != null;
 791                                 info.setAssociation(association);
 792                             }
 793                             return info;
 794                         } else
 795                             /* Non-blocking may return null if nothing available*/
 796                             return null;
 797                     } else { /* notification */
 798                         synchronized (stateLock) {
 799                             handleNotificationInternal(
 800                                     resultContainer);
 801                         }
 802                     }
 803 
 804                     if (fromConnect)  {
 805                         /* If we reach here, then it was connect that invoked
 806                          * receive and received the COMM_UP. We have already
 807                          * handled the COMM_UP with the internal notification
 808                          * handler. Simply return. */
 809                         return null;
 810                     }
 811                 }  /* receiveLock */
 812             } while (handler == null ? true :
 813                 (invokeNotificationHandler(resultContainer, handler, attachment)
 814                  == HandlerResult.CONTINUE));
 815 
 816             return null;
 817         } finally {
 818             receiveInvoked.set(Boolean.FALSE);
 819         }
 820     }
 821 
 822     private int receive(int fd,
 823                         ByteBuffer dst,
 824                         ResultContainer resultContainer,
 825                         boolean peek)
 826             throws IOException {
 827         int pos = dst.position();
 828         int lim = dst.limit();
 829         assert (pos <= lim);
 830         int rem = (pos <= lim ? lim - pos : 0);
 831         if (dst instanceof DirectBuffer && rem > 0)
 832             return receiveIntoNativeBuffer(fd, resultContainer, dst, rem, pos, peek);
 833 
 834         /* Substitute a native buffer */
 835         int newSize = Math.max(rem, 1);
 836         ByteBuffer bb = Util.getTemporaryDirectBuffer(newSize);
 837         try {
 838             int n = receiveIntoNativeBuffer(fd, resultContainer, bb, newSize, 0, peek);
 839             bb.flip();
 840             if (n > 0 && rem > 0)
 841                 dst.put(bb);
 842             return n;
 843         } finally {
 844             Util.releaseTemporaryDirectBuffer(bb);
 845         }
 846     }
 847 
 848     private int receiveIntoNativeBuffer(int fd,
 849                                         ResultContainer resultContainer,
 850                                         ByteBuffer bb,
 851                                         int rem,
 852                                         int pos,
 853                                         boolean peek)
 854         throws IOException
 855     {
 856         int n = receive0(fd, resultContainer, ((DirectBuffer)bb).address() + pos, rem, peek);
 857 
 858         if (n > 0)
 859             bb.position(pos + n);
 860         return n;
 861     }
 862 
 863     private InternalNotificationHandler internalNotificationHandler =
 864             new InternalNotificationHandler();
 865 
 866     private void handleNotificationInternal(ResultContainer resultContainer)
 867     {
 868         invokeNotificationHandler(resultContainer,
 869                 internalNotificationHandler, null);
 870     }
 871 
 872     private class InternalNotificationHandler
 873             extends AbstractNotificationHandler<Object>
 874     {
 875         @Override
 876         public HandlerResult handleNotification(
 877                 AssociationChangeNotification not, Object unused) {
 878             if (not.event().equals(
 879                     AssociationChangeNotification.AssocChangeEvent.COMM_UP) &&
 880                     association == null) {
 881                 AssociationChange sac = (AssociationChange) not;
 882                 association = new AssociationImpl
 883                        (sac.assocId(), sac.maxInStreams(), sac.maxOutStreams());
 884             }
 885             return HandlerResult.CONTINUE;
 886         }
 887     }
 888 
 889     private <T> HandlerResult invokeNotificationHandler
 890                                  (ResultContainer resultContainer,
 891                                   NotificationHandler<T> handler,
 892                                   T attachment) {
 893         SctpNotification notification = resultContainer.notification();
 894         synchronized (stateLock) {
 895             notification.setAssociation(association);
 896         }
 897 
 898         if (!(handler instanceof AbstractNotificationHandler)) {
 899             return handler.handleNotification(notification, attachment);
 900         }
 901 
 902         /* AbstractNotificationHandler */
 903         AbstractNotificationHandler<T> absHandler =
 904                 (AbstractNotificationHandler<T>)handler;
 905         switch(resultContainer.type()) {
 906             case ASSOCIATION_CHANGED :
 907                 return absHandler.handleNotification(
 908                         resultContainer.getAssociationChanged(), attachment);
 909             case PEER_ADDRESS_CHANGED :
 910                 return absHandler.handleNotification(
 911                         resultContainer.getPeerAddressChanged(), attachment);
 912             case SEND_FAILED :
 913                 return absHandler.handleNotification(
 914                         resultContainer.getSendFailed(), attachment);
 915             case SHUTDOWN :
 916                 return absHandler.handleNotification(
 917                         resultContainer.getShutdown(), attachment);
 918             default :
 919                 /* implementation specific handlers */
 920                 return absHandler.handleNotification(
 921                         resultContainer.notification(), attachment);
 922         }
 923     }
 924 
 925     private void checkAssociation(Association sendAssociation) {
 926         synchronized (stateLock) {
 927             if (sendAssociation != null && !sendAssociation.equals(association)) {
 928                 throw new IllegalArgumentException(
 929                         "Cannot send to another association");
 930             }
 931         }
 932     }
 933 
 934     private void checkStreamNumber(int streamNumber) {
 935         synchronized (stateLock) {
 936             if (association != null) {
 937                 if (streamNumber < 0 ||
 938                       streamNumber >= association.maxOutboundStreams())
 939                     throw new InvalidStreamException();
 940             }
 941         }
 942     }
 943 
 944     /* TODO: Add support for ttl and isComplete to both 121 12M
 945      *       SCTP_EOR not yet supported on reference platforms
 946      *       TTL support limited...
 947      */
 948     @Override
 949     public int send(ByteBuffer buffer, MessageInfo messageInfo)
 950             throws IOException {
 951         if (buffer == null)
 952             throw new IllegalArgumentException("buffer cannot be null");
 953 
 954         if (messageInfo == null)
 955             throw new IllegalArgumentException("messageInfo cannot be null");
 956 
 957         checkAssociation(messageInfo.association());
 958         checkStreamNumber(messageInfo.streamNumber());
 959 
 960         synchronized (sendLock) {
 961             ensureSendOpen();
 962 
 963             int n = 0;
 964             try {
 965                 begin();
 966 
 967                 synchronized (stateLock) {
 968                     if(!isOpen())
 969                         return 0;
 970                     senderThread = NativeThread.current();
 971                 }
 972 
 973                 do {
 974                     n = send(fdVal, buffer, messageInfo);
 975                 } while ((n == IOStatus.INTERRUPTED) && isOpen());
 976 
 977                 return IOStatus.normalize(n);
 978             } finally {
 979                 senderCleanup();
 980                 end((n > 0) || (n == IOStatus.UNAVAILABLE));
 981                 assert IOStatus.check(n);
 982             }
 983         }
 984     }
 985 
 986     private int send(int fd, ByteBuffer src, MessageInfo messageInfo)
 987             throws IOException {
 988         int streamNumber = messageInfo.streamNumber();
 989         SocketAddress target = messageInfo.address();
 990         boolean unordered = messageInfo.isUnordered();
 991         int ppid = messageInfo.payloadProtocolID();
 992 
 993         if (src instanceof DirectBuffer)
 994             return sendFromNativeBuffer(fd, src, target, streamNumber,
 995                     unordered, ppid);
 996 
 997         /* Substitute a native buffer */
 998         int pos = src.position();
 999         int lim = src.limit();
1000         assert (pos <= lim && streamNumber >= 0);
1001 
1002         int rem = (pos <= lim ? lim - pos : 0);
1003         ByteBuffer bb = Util.getTemporaryDirectBuffer(rem);
1004         try {
1005             bb.put(src);
1006             bb.flip();
1007             /* Do not update src until we see how many bytes were written */
1008             src.position(pos);
1009 
1010             int n = sendFromNativeBuffer(fd, bb, target, streamNumber,
1011                     unordered, ppid);
1012             if (n > 0) {
1013                 /* now update src */
1014                 src.position(pos + n);
1015             }
1016             return n;
1017         } finally {
1018             Util.releaseTemporaryDirectBuffer(bb);
1019         }
1020     }
1021 
1022     private int sendFromNativeBuffer(int fd,
1023                                      ByteBuffer bb,
1024                                      SocketAddress target,
1025                                      int streamNumber,
1026                                      boolean unordered,
1027                                      int ppid)
1028             throws IOException {
1029         int pos = bb.position();
1030         int lim = bb.limit();
1031         assert (pos <= lim);
1032         int rem = (pos <= lim ? lim - pos : 0);
1033 
1034         int written = send0(fd, ((DirectBuffer)bb).address() + pos,
1035                             rem, target, -1 /*121*/, streamNumber, unordered, ppid);
1036         if (written > 0)
1037             bb.position(pos + written);
1038         return written;
1039     }
1040 
1041     @Override
1042     public SctpChannel shutdown() throws IOException {
1043         synchronized(stateLock) {
1044             if (isShutdown)
1045                 return this;
1046 
1047             ensureSendOpen();
1048             SctpNet.shutdown(fdVal, -1);
1049             if (senderThread != 0)
1050                 NativeThread.signal(senderThread);
1051             isShutdown = true;
1052         }
1053         return this;
1054     }
1055 
1056     @Override
1057     public Set<SocketAddress> getAllLocalAddresses()
1058             throws IOException {
1059         synchronized (stateLock) {
1060             if (!isOpen())
1061                 throw new ClosedChannelException();
1062             if (!isBound())
1063                 return Collections.emptySet();
1064 
1065             return SctpNet.getLocalAddresses(fdVal);
1066         }
1067     }
1068 
1069     @Override
1070     public Set<SocketAddress> getRemoteAddresses()
1071             throws IOException {
1072         synchronized (stateLock) {
1073             if (!isOpen())
1074                 throw new ClosedChannelException();
1075             if (!isConnected() || isShutdown)
1076                 return Collections.emptySet();
1077 
1078             try {
1079                 return SctpNet.getRemoteAddresses(fdVal, 0/*unused*/);
1080             } catch (SocketException unused) {
1081                 /* an open connected channel should always have remote addresses */
1082                 return remoteAddresses;
1083             }
1084         }
1085     }
1086 
1087     /* Native */
1088     private static native void initIDs();
1089 
1090     static native int receive0(int fd, ResultContainer resultContainer,
1091             long address, int length, boolean peek) throws IOException;
1092 
1093     static native int send0(int fd, long address, int length,
1094             SocketAddress target, int assocId, int streamNumber,
1095             boolean unordered, int ppid) throws IOException;
1096 
1097     private static native int checkConnect(FileDescriptor fd, boolean block,
1098             boolean ready) throws IOException;
1099 
1100     static {
1101         Util.load();   /* loads nio & net native libraries */
1102         java.security.AccessController.doPrivileged(
1103                 new sun.security.action.LoadLibraryAction("sctp"));
1104         initIDs();
1105     }
1106 }