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