AsynchronousSocketChannelImpl.java 20.2 KB
Newer Older
1
/*
2
 * Copyright (c) 2008, 2011, Oracle and/or its affiliates. All rights reserved.
3 4 5 6
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * 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 12 13 14 15 16 17 18 19 20
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
21 22 23
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
24 25 26 27 28 29 30
 */

package sun.nio.ch;

import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.net.SocketOption;
31
import java.net.StandardSocketOptions;
32 33 34 35 36 37 38 39 40
import java.net.SocketAddress;
import java.net.InetSocketAddress;
import java.io.IOException;
import java.io.FileDescriptor;
import java.util.Set;
import java.util.HashSet;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.concurrent.locks.*;
A
alanb 已提交
41
import sun.net.NetHooks;
42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81

/**
 * Base implementation of AsynchronousSocketChannel
 */

abstract class AsynchronousSocketChannelImpl
    extends AsynchronousSocketChannel
    implements Cancellable, Groupable
{
    protected final FileDescriptor fd;

    // protects state, localAddress, and remoteAddress
    protected final Object stateLock = new Object();

    protected volatile SocketAddress localAddress = null;
    protected volatile SocketAddress remoteAddress = null;

    // State, increases monotonically
    static final int ST_UNINITIALIZED = -1;
    static final int ST_UNCONNECTED = 0;
    static final int ST_PENDING = 1;
    static final int ST_CONNECTED = 2;
    protected volatile int state = ST_UNINITIALIZED;

    // reading state
    private final Object readLock = new Object();
    private boolean reading;
    private boolean readShutdown;
    private boolean readKilled;     // further reading disallowed due to timeout

    // writing state
    private final Object writeLock = new Object();
    private boolean writing;
    private boolean writeShutdown;
    private boolean writeKilled;    // further writing disallowed due to timeout

    // close support
    private final ReadWriteLock closeLock = new ReentrantReadWriteLock();
    private volatile boolean open = true;

82 83 84
    // set true when exclusive binding is on and SO_REUSEADDR is emulated
    private boolean isReuseAddress;

85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188
    AsynchronousSocketChannelImpl(AsynchronousChannelGroupImpl group)
        throws IOException
    {
        super(group.provider());
        this.fd = Net.socket(true);
        this.state = ST_UNCONNECTED;
    }

    // Constructor for sockets obtained from AsynchronousServerSocketChannelImpl
    AsynchronousSocketChannelImpl(AsynchronousChannelGroupImpl group,
                                  FileDescriptor fd,
                                  InetSocketAddress remote)
        throws IOException
    {
        super(group.provider());
        this.fd = fd;
        this.state = ST_CONNECTED;
        this.localAddress = Net.localAddress(fd);
        this.remoteAddress = remote;
    }

    @Override
    public final boolean isOpen() {
        return open;
    }

    /**
     * Marks beginning of access to file descriptor/handle
     */
    final void begin() throws IOException {
        closeLock.readLock().lock();
        if (!isOpen())
            throw new ClosedChannelException();
    }

    /**
     * Marks end of access to file descriptor/handle
     */
    final void end() {
        closeLock.readLock().unlock();
    }

    /**
     * Invoked to close socket and release other resources.
     */
    abstract void implClose() throws IOException;

    @Override
    public final void close() throws IOException {
        // synchronize with any threads initiating asynchronous operations
        closeLock.writeLock().lock();
        try {
            if (!open)
                return;     // already closed
            open = false;
        } finally {
            closeLock.writeLock().unlock();
        }
        implClose();
    }

    final void enableReading(boolean killed) {
        synchronized (readLock) {
            reading = false;
            if (killed)
                readKilled = true;
        }
    }

    final void enableReading() {
        enableReading(false);
    }

    final void enableWriting(boolean killed) {
        synchronized (writeLock) {
            writing = false;
            if (killed)
                writeKilled = true;
        }
    }

    final void enableWriting() {
        enableWriting(false);
    }

    final void killReading() {
        synchronized (readLock) {
            readKilled = true;
        }
    }

    final void killWriting() {
        synchronized (writeLock) {
            writeKilled = true;
        }
    }

    final void killConnect() {
        // when a connect is cancelled then the connection may have been
        // established so prevent reading or writing.
        killReading();
        killWriting();
    }

189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210
    /**
     * Invoked by connect to initiate the connect operation.
     */
    abstract <A> Future<Void> implConnect(SocketAddress remote,
                                          A attachment,
                                          CompletionHandler<Void,? super A> handler);

    @Override
    public final Future<Void> connect(SocketAddress remote) {
        return implConnect(remote, null, null);
    }

    @Override
    public final <A> void connect(SocketAddress remote,
                                  A attachment,
                                  CompletionHandler<Void,? super A> handler)
    {
        if (handler == null)
            throw new NullPointerException("'handler' is null");
        implConnect(remote, attachment, handler);
    }

211 212 213
    /**
     * Invoked by read to initiate the I/O operation.
     */
214 215 216
    abstract <V extends Number,A> Future<V> implRead(boolean isScatteringRead,
                                                     ByteBuffer dst,
                                                     ByteBuffer[] dsts,
217 218 219 220 221 222
                                                     long timeout,
                                                     TimeUnit unit,
                                                     A attachment,
                                                     CompletionHandler<V,? super A> handler);

    @SuppressWarnings("unchecked")
223 224 225
    private <V extends Number,A> Future<V> read(boolean isScatteringRead,
                                                ByteBuffer dst,
                                                ByteBuffer[] dsts,
226 227
                                                long timeout,
                                                TimeUnit unit,
228
                                                A att,
229 230 231
                                                CompletionHandler<V,? super A> handler)
    {
        if (!isOpen()) {
232 233 234 235 236
            Throwable e = new ClosedChannelException();
            if (handler == null)
                return CompletedFuture.withFailure(e);
            Invoker.invoke(this, handler, att, null, e);
            return null;
237 238 239 240 241
        }

        if (remoteAddress == null)
            throw new NotYetConnectedException();

242
        boolean hasSpaceToRead = isScatteringRead || dst.hasRemaining();
243 244 245 246 247
        boolean shutdown = false;

        // check and update state
        synchronized (readLock) {
            if (readKilled)
248
                throw new IllegalStateException("Reading not allowed due to timeout or cancellation");
249 250 251 252 253 254 255 256 257 258 259 260 261 262
            if (reading)
                throw new ReadPendingException();
            if (readShutdown) {
                shutdown = true;
            } else {
                if (hasSpaceToRead) {
                    reading = true;
                }
            }
        }

        // immediately complete with -1 if shutdown for read
        // immediately complete with 0 if no space remaining
        if (shutdown || !hasSpaceToRead) {
263
            Number result;
264
            if (isScatteringRead) {
265
                result = (shutdown) ? Long.valueOf(-1L) : Long.valueOf(0L);
266
            } else {
267
                result = (shutdown) ? -1 : 0;
268
            }
269 270 271 272
            if (handler == null)
                return CompletedFuture.withResult((V)result);
            Invoker.invoke(this, handler, att, (V)result, null);
            return null;
273 274
        }

275
        return implRead(isScatteringRead, dst, dsts, timeout, unit, att, handler);
276 277 278
    }

    @Override
279 280 281 282 283 284 285 286 287 288 289 290
    public final Future<Integer> read(ByteBuffer dst) {
        if (dst.isReadOnly())
            throw new IllegalArgumentException("Read-only buffer");
        return read(false, dst, null, 0L, TimeUnit.MILLISECONDS, null, null);
    }

    @Override
    public final <A> void read(ByteBuffer dst,
                               long timeout,
                               TimeUnit unit,
                               A attachment,
                               CompletionHandler<Integer,? super A> handler)
291
    {
292 293
        if (handler == null)
            throw new NullPointerException("'handler' is null");
294 295
        if (dst.isReadOnly())
            throw new IllegalArgumentException("Read-only buffer");
296
        read(false, dst, null, timeout, unit, attachment, handler);
297 298 299
    }

    @Override
300 301 302 303 304 305 306
    public final <A> void read(ByteBuffer[] dsts,
                               int offset,
                               int length,
                               long timeout,
                               TimeUnit unit,
                               A attachment,
                               CompletionHandler<Long,? super A> handler)
307
    {
308 309
        if (handler == null)
            throw new NullPointerException("'handler' is null");
310 311 312 313 314 315 316
        if ((offset < 0) || (length < 0) || (offset > dsts.length - length))
            throw new IndexOutOfBoundsException();
        ByteBuffer[] bufs = Util.subsequence(dsts, offset, length);
        for (int i=0; i<bufs.length; i++) {
            if (bufs[i].isReadOnly())
                throw new IllegalArgumentException("Read-only buffer");
        }
317
        read(true, null, bufs, timeout, unit, attachment, handler);
318 319 320 321 322
    }

    /**
     * Invoked by write to initiate the I/O operation.
     */
323 324 325
    abstract <V extends Number,A> Future<V> implWrite(boolean isGatheringWrite,
                                                      ByteBuffer src,
                                                      ByteBuffer[] srcs,
326 327 328 329 330 331
                                                      long timeout,
                                                      TimeUnit unit,
                                                      A attachment,
                                                      CompletionHandler<V,? super A> handler);

    @SuppressWarnings("unchecked")
332 333 334
    private <V extends Number,A> Future<V> write(boolean isGatheringWrite,
                                                 ByteBuffer src,
                                                 ByteBuffer[] srcs,
335 336
                                                 long timeout,
                                                 TimeUnit unit,
337
                                                 A att,
338 339
                                                 CompletionHandler<V,? super A> handler)
    {
340
        boolean hasDataToWrite = isGatheringWrite || src.hasRemaining();
341 342 343 344 345 346 347 348

        boolean closed = false;
        if (isOpen()) {
            if (remoteAddress == null)
                throw new NotYetConnectedException();
            // check and update state
            synchronized (writeLock) {
                if (writeKilled)
349
                    throw new IllegalStateException("Writing not allowed due to timeout or cancellation");
350 351 352 353 354 355 356 357 358 359 360 361 362 363 364
                if (writing)
                    throw new WritePendingException();
                if (writeShutdown) {
                    closed = true;
                } else {
                    if (hasDataToWrite)
                        writing = true;
                }
            }
        } else {
            closed = true;
        }

        // channel is closed or shutdown for write
        if (closed) {
365 366 367 368 369
            Throwable e = new ClosedChannelException();
            if (handler == null)
                return CompletedFuture.withFailure(e);
            Invoker.invoke(this, handler, att, null, e);
            return null;
370 371 372 373
        }

        // nothing to write so complete immediately
        if (!hasDataToWrite) {
374 375 376 377 378
            Number result = (isGatheringWrite) ? (Number)0L : (Number)0;
            if (handler == null)
                return CompletedFuture.withResult((V)result);
            Invoker.invoke(this, handler, att, (V)result, null);
            return null;
379 380
        }

381
        return implWrite(isGatheringWrite, src, srcs, timeout, unit, att, handler);
382 383 384
    }

    @Override
385 386 387 388 389 390 391 392 393 394
    public final Future<Integer> write(ByteBuffer src) {
        return write(false, src, null, 0L, TimeUnit.MILLISECONDS, null, null);
    }

    @Override
    public final <A> void write(ByteBuffer src,
                                long timeout,
                                TimeUnit unit,
                                A attachment,
                                CompletionHandler<Integer,? super A> handler)
395
    {
396 397 398
        if (handler == null)
            throw new NullPointerException("'handler' is null");
        write(false, src, null, timeout, unit, attachment, handler);
399 400 401
    }

    @Override
402 403 404 405 406 407 408
    public final <A> void  write(ByteBuffer[] srcs,
                                 int offset,
                                 int length,
                                 long timeout,
                                 TimeUnit unit,
                                 A attachment,
                                 CompletionHandler<Long,? super A> handler)
409
    {
410 411
        if (handler == null)
            throw new NullPointerException("'handler' is null");
412 413 414
        if ((offset < 0) || (length < 0) || (offset > srcs.length - length))
            throw new IndexOutOfBoundsException();
        srcs = Util.subsequence(srcs, offset, length);
415
        write(true, null, srcs, timeout, unit, attachment, handler);
416 417 418 419 420 421 422 423 424 425 426 427 428 429 430
    }

    @Override
    public final AsynchronousSocketChannel bind(SocketAddress local)
        throws IOException
    {
        try {
            begin();
            synchronized (stateLock) {
                if (state == ST_PENDING)
                    throw new ConnectionPendingException();
                if (localAddress != null)
                    throw new AlreadyBoundException();
                InetSocketAddress isa = (local == null) ?
                    new InetSocketAddress(0) : Net.checkAddress(local);
A
alanb 已提交
431
                NetHooks.beforeTcpBind(fd, isa.getAddress(), isa.getPort());
432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460
                Net.bind(fd, isa.getAddress(), isa.getPort());
                localAddress = Net.localAddress(fd);
            }
        } finally {
            end();
        }
        return this;
    }

    @Override
    public final SocketAddress getLocalAddress() throws IOException {
        if (!isOpen())
            throw new ClosedChannelException();
        return localAddress;
    }

    @Override
    public final <T> AsynchronousSocketChannel setOption(SocketOption<T> name, T value)
        throws IOException
    {
        if (name == null)
            throw new NullPointerException();
        if (!supportedOptions().contains(name))
            throw new UnsupportedOperationException("'" + name + "' not supported");

        try {
            begin();
            if (writeShutdown)
                throw new IOException("Connection has been shutdown for writing");
461 462 463 464 465 466 467 468
            if (name == StandardSocketOptions.SO_REUSEADDR &&
                    Net.useExclusiveBind())
            {
                // SO_REUSEADDR emulated when using exclusive bind
                isReuseAddress = (Boolean)value;
            } else {
                Net.setSocketOption(fd, Net.UNSPEC, name, value);
            }
469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484
            return this;
        } finally {
            end();
        }
    }

    @Override
    @SuppressWarnings("unchecked")
    public final <T> T getOption(SocketOption<T> name) throws IOException {
        if (name == null)
            throw new NullPointerException();
        if (!supportedOptions().contains(name))
            throw new UnsupportedOperationException("'" + name + "' not supported");

        try {
            begin();
485 486 487 488 489 490
            if (name == StandardSocketOptions.SO_REUSEADDR &&
                    Net.useExclusiveBind())
            {
                // SO_REUSEADDR emulated when using exclusive bind
                return (T)Boolean.valueOf(isReuseAddress);
            }
491 492 493 494 495 496 497 498 499 500 501
            return (T) Net.getSocketOption(fd, Net.UNSPEC, name);
        } finally {
            end();
        }
    }

    private static class DefaultOptionsHolder {
        static final Set<SocketOption<?>> defaultOptions = defaultOptions();

        private static Set<SocketOption<?>> defaultOptions() {
            HashSet<SocketOption<?>> set = new HashSet<SocketOption<?>>(5);
502 503 504 505 506
            set.add(StandardSocketOptions.SO_SNDBUF);
            set.add(StandardSocketOptions.SO_RCVBUF);
            set.add(StandardSocketOptions.SO_KEEPALIVE);
            set.add(StandardSocketOptions.SO_REUSEADDR);
            set.add(StandardSocketOptions.TCP_NODELAY);
507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596
            return Collections.unmodifiableSet(set);
        }
    }

    @Override
    public final Set<SocketOption<?>> supportedOptions() {
        return DefaultOptionsHolder.defaultOptions;
    }

    @Override
    public final SocketAddress getRemoteAddress() throws IOException {
        if (!isOpen())
            throw new ClosedChannelException();
        return remoteAddress;
    }

    @Override
    public final AsynchronousSocketChannel shutdownInput() throws IOException {
        try {
            begin();
            if (remoteAddress == null)
                throw new NotYetConnectedException();
            synchronized (readLock) {
                if (!readShutdown) {
                    Net.shutdown(fd, Net.SHUT_RD);
                    readShutdown = true;
                }
            }
        } finally {
            end();
        }
        return this;
    }

    @Override
    public final AsynchronousSocketChannel shutdownOutput() throws IOException {
        try {
            begin();
            if (remoteAddress == null)
                throw new NotYetConnectedException();
            synchronized (writeLock) {
                if (!writeShutdown) {
                    Net.shutdown(fd, Net.SHUT_WR);
                    writeShutdown = true;
                }
            }
        } finally {
            end();
        }
        return this;
    }

    @Override
    public final String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append(this.getClass().getName());
        sb.append('[');
        synchronized (stateLock) {
            if (!isOpen()) {
                sb.append("closed");
            } else {
                switch (state) {
                case ST_UNCONNECTED:
                    sb.append("unconnected");
                    break;
                case ST_PENDING:
                    sb.append("connection-pending");
                    break;
                case ST_CONNECTED:
                    sb.append("connected");
                    if (readShutdown)
                        sb.append(" ishut");
                    if (writeShutdown)
                        sb.append(" oshut");
                    break;
                }
                if (localAddress != null) {
                    sb.append(" local=");
                    sb.append(localAddress.toString());
                }
                if (remoteAddress != null) {
                    sb.append(" remote=");
                    sb.append(remoteAddress.toString());
                }
            }
        }
        sb.append(']');
        return sb.toString();
    }
}