/*

 * Copyright (c) 2002, 2012, Oracle and/or its affiliates. All rights reserved.

 * 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

 * published by the Free Software Foundation.

 *

 * 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.

 *

 * 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.

 */

/* @test

 * @bug 4460583 4470470 4840199 6419424 6710579 6596323 6824135 6395224 7142919

 * @run main/othervm AsyncCloseAndInterrupt

 * @summary Comprehensive test of asynchronous closing and interruption

 * @author Mark Reinhold

 */

import java.io.*;

import java.net.*;

import java.nio.channels.*;

import java.nio.ByteBuffer;

import java.util.ArrayList;

import java.util.List;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Executors;

import java.util.concurrent.ThreadFactory;

import java.util.concurrent.Callable;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

public class AsyncCloseAndInterrupt {

    static PrintStream log = System.err;

    static void sleep(int ms) {

        try {

            Thread.sleep(ms);

        } catch (InterruptedException x) { }

    }

    // Wildcard address localized to this machine -- Windoze doesn't allow

    // connecting to a server socket that was previously bound to a true

    // wildcard, namely new InetSocketAddress((InetAddress)null, 0).

    //

    private static InetSocketAddress wildcardAddress;

    // Server socket that blindly accepts all connections

    static ServerSocketChannel acceptor;

    private static void initAcceptor() throws IOException {

        acceptor = ServerSocketChannel.open();

        acceptor.socket().bind(wildcardAddress);

        Thread th = new Thread("Acceptor") {

                public void run() {

                    try {

                        for (;;) {

                            SocketChannel sc = acceptor.accept();

                        }

                    } catch (IOException x) {

                        x.printStackTrace();

                    }

                }

            };

        th.setDaemon(true);

        th.start();

    }

    // Server socket that refuses all connections

    static ServerSocketChannel refuser;

    private static void initRefuser() throws IOException {

        refuser = ServerSocketChannel.open();

        refuser.socket().bind(wildcardAddress);

    }

    // Dead pipe source and sink

    static Pipe.SourceChannel deadSource;

    static Pipe.SinkChannel deadSink;

    private static void initPipes() throws IOException {

        if (deadSource != null)

            deadSource.close();

        deadSource = Pipe.open().source();

        if (deadSink != null)

            deadSink.close();

        deadSink = Pipe.open().sink();

    }

    // Files

    private static File fifoFile = null; // File that blocks on reads and writes

    private static File diskFile = null; // Disk file

    private static void initFile() throws Exception {

        diskFile = File.createTempFile("aci", ".tmp");

        diskFile.deleteOnExit();

        FileChannel fc = new FileOutputStream(diskFile).getChannel();

        buffer.clear();

        if (fc.write(buffer) != buffer.capacity())

            throw new RuntimeException("Cannot create disk file");

        fc.close();

        if (TestUtil.onWindows()) {

            log.println("WARNING: Cannot completely test FileChannels on Windows");

            return;

        }

        fifoFile = new File("x.fifo");

        if (fifoFile.exists()) {

            if (!fifoFile.delete())

                throw new IOException("Cannot delete existing fifo " + fifoFile);

        }

        Process p = Runtime.getRuntime().exec("mkfifo " + fifoFile);

        if (p.waitFor() != 0)

            throw new IOException("Error creating fifo");

        new RandomAccessFile(fifoFile, "rw").close();

    }

    // Channel factories

    static abstract class ChannelFactory {

        private final String name;

        ChannelFactory(String name) {

            this.name = name;

        }

        public String toString() {

            return name;

        }

        abstract InterruptibleChannel create() throws IOException;

    }

    static ChannelFactory socketChannelFactory

        = new ChannelFactory("SocketChannel") {

                InterruptibleChannel create() throws IOException {

                    return SocketChannel.open();

                }

            };

    static ChannelFactory connectedSocketChannelFactory

        = new ChannelFactory("SocketChannel") {

                InterruptibleChannel create() throws IOException {

                    SocketAddress sa = acceptor.socket().getLocalSocketAddress();

                    return SocketChannel.open(sa);

                }

            };

    static ChannelFactory serverSocketChannelFactory

        = new ChannelFactory("ServerSocketChannel") {

                InterruptibleChannel create() throws IOException {

                    ServerSocketChannel ssc = ServerSocketChannel.open();

                    ssc.socket().bind(wildcardAddress);

                    return ssc;

                }

            };

    static ChannelFactory datagramChannelFactory

        = new ChannelFactory("DatagramChannel") {

                InterruptibleChannel create() throws IOException {

                    DatagramChannel dc = DatagramChannel.open();

                    InetAddress lb = InetAddress.getByName("127.0.0.1");

                    dc.bind(new InetSocketAddress(lb, 0));

                    dc.connect(new InetSocketAddress(lb, 80));

                    return dc;

                }

            };

    static ChannelFactory pipeSourceChannelFactory

        = new ChannelFactory("Pipe.SourceChannel") {

                InterruptibleChannel create() throws IOException {

                    // ## arrange to close sink

                    return Pipe.open().source();

                }

            };

    static ChannelFactory pipeSinkChannelFactory

        = new ChannelFactory("Pipe.SinkChannel") {

                InterruptibleChannel create() throws IOException {

                    // ## arrange to close source

                    return Pipe.open().sink();

                }

            };

    static ChannelFactory fifoFileChannelFactory

        = new ChannelFactory("FileChannel") {

                InterruptibleChannel create() throws IOException {

                    return new RandomAccessFile(fifoFile, "rw").getChannel();

                }

            };

    static ChannelFactory diskFileChannelFactory

        = new ChannelFactory("FileChannel") {

                InterruptibleChannel create() throws IOException {

                    return new RandomAccessFile(diskFile, "rw").getChannel();

                }

            };

    // I/O operations

    static abstract class Op {

        private final String name;

        protected Op(String name) {

            this.name = name;

        }

        abstract void doIO(InterruptibleChannel ich) throws IOException;

        void setup() throws IOException { }

        public String toString() { return name; }

    }

    static ByteBuffer buffer = ByteBuffer.allocateDirect(1 << 20);

    static ByteBuffer[] buffers = new ByteBuffer[] {

        ByteBuffer.allocateDirect(1 << 19),

        ByteBuffer.allocateDirect(1 << 19)

    };

    static void clearBuffers() {

        buffers[0].clear();

        buffers[1].clear();

    }

    static void show(Channel ch) {

        log.print("Channel " + (ch.isOpen() ? "open" : "closed"));

        if (ch.isOpen() && (ch instanceof SocketChannel)) {

            SocketChannel sc = (SocketChannel)ch;

            if (sc.socket().isInputShutdown())

                log.print(", input shutdown");

            if (sc.socket().isOutputShutdown())

                log.print(", output shutdown");

        }

        log.println();

    }

    static final Op READ = new Op("read") {

            void doIO(InterruptibleChannel ich) throws IOException {

                ReadableByteChannel rbc = (ReadableByteChannel)ich;

                buffer.clear();

                int n = rbc.read(buffer);

                log.println("Read returned " + n);

                show(rbc);

                if     (rbc.isOpen()

                        && (n == -1)

                        && (rbc instanceof SocketChannel)

                        && ((SocketChannel)rbc).socket().isInputShutdown()) {

                    return;

                }

                throw new RuntimeException("Read succeeded");

            }

        };

    static final Op READV = new Op("readv") {

            void doIO(InterruptibleChannel ich) throws IOException {

                ScatteringByteChannel sbc = (ScatteringByteChannel)ich;

                clearBuffers();

                int n = (int)sbc.read(buffers);

                log.println("Read returned " + n);

                show(sbc);

                if     (sbc.isOpen()

                        && (n == -1)

                        && (sbc instanceof SocketChannel)

                        && ((SocketChannel)sbc).socket().isInputShutdown()) {

                    return;

                }

                throw new RuntimeException("Read succeeded");

            }

        };

    static final Op RECEIVE = new Op("receive") {

            void doIO(InterruptibleChannel ich) throws IOException {

                DatagramChannel dc = (DatagramChannel)ich;

                buffer.clear();

                dc.receive(buffer);

                show(dc);

                throw new RuntimeException("Read succeeded");

            }

        };

    static final Op WRITE = new Op("write") {

            void doIO(InterruptibleChannel ich) throws IOException {

                WritableByteChannel wbc = (WritableByteChannel)ich;

                SocketChannel sc = null;

                if (wbc instanceof SocketChannel)

                    sc = (SocketChannel)wbc;

                int n = 0;

                for (;;) {

                    buffer.clear();

                    int d = wbc.write(buffer);

                    n += d;

                    if (!wbc.isOpen())

                        break;

                    if ((sc != null) && sc.socket().isOutputShutdown())

                        break;

                }

                log.println("Wrote " + n + " bytes");

                show(wbc);

            }

        };

    static final Op WRITEV = new Op("writev") {

            void doIO(InterruptibleChannel ich) throws IOException {

                GatheringByteChannel gbc = (GatheringByteChannel)ich;

                SocketChannel sc = null;

                if (gbc instanceof SocketChannel)

                    sc = (SocketChannel)gbc;

                int n = 0;

                for (;;) {

                    clearBuffers();

                    int d = (int)gbc.write(buffers);

                    n += d;

                    if (!gbc.isOpen())

                        break;

                    if ((sc != null) && sc.socket().isOutputShutdown())

                        break;

                }

                log.println("Wrote " + n + " bytes");

                show(gbc);

            }

        };

    static final Op CONNECT = new Op("connect") {

            void setup() {

                waitPump("connect wait for pumping refuser ...");

            }

            void doIO(InterruptibleChannel ich) throws IOException {

                SocketChannel sc = (SocketChannel)ich;

                if (sc.connect(refuser.socket().getLocalSocketAddress()))

                    throw new RuntimeException("Connection succeeded");

                throw new RuntimeException("Connection did not block");

            }

        };

    static final Op FINISH_CONNECT = new Op("finishConnect") {

            void setup() {

                waitPump("finishConnect wait for pumping refuser ...");

            }

            void doIO(InterruptibleChannel ich) throws IOException {

                SocketChannel sc = (SocketChannel)ich;

                sc.configureBlocking(false);

                SocketAddress sa = refuser.socket().getLocalSocketAddress();

                if (sc.connect(sa))

                    throw new RuntimeException("Connection succeeded");

                sc.configureBlocking(true);

                if (sc.finishConnect())

                    throw new RuntimeException("Connection succeeded");

                throw new RuntimeException("Connection did not block");

            }

        };

    static final Op ACCEPT = new Op("accept") {

            void doIO(InterruptibleChannel ich) throws IOException {

                ServerSocketChannel ssc = (ServerSocketChannel)ich;

                ssc.accept();

                throw new RuntimeException("Accept succeeded");

            }

        };

    // Use only with diskFileChannelFactory

    static final Op TRANSFER_TO = new Op("transferTo") {

            void doIO(InterruptibleChannel ich) throws IOException {

                FileChannel fc = (FileChannel)ich;

                long n = fc.transferTo(0, fc.size(), deadSink);

                log.println("Transferred " + n + " bytes");

                show(fc);

            }

        };

    // Use only with diskFileChannelFactory

    static final Op TRANSFER_FROM = new Op("transferFrom") {

            void doIO(InterruptibleChannel ich) throws IOException {

                FileChannel fc = (FileChannel)ich;

                long n = fc.transferFrom(deadSource, 0, 1 << 20);

                log.println("Transferred " + n + " bytes");

                show(fc);

            }

        };

    // Test modes

    static final int TEST_PREINTR = 0;  // Interrupt thread before I/O

    static final int TEST_INTR = 1;     // Interrupt thread during I/O

    static final int TEST_CLOSE = 2;    // Close channel during I/O

    static final int TEST_SHUTI = 3;    // Shutdown input during I/O

    static final int TEST_SHUTO = 4;    // Shutdown output during I/O

    static final String[] testName = new String[] {

        "pre-interrupt", "interrupt", "close",

        "shutdown-input", "shutdown-output"

    };

    static class Tester extends TestThread {

        private InterruptibleChannel ch;

        private Op op;

        private int test;

        volatile boolean ready = false;

        protected Tester(ChannelFactory cf, InterruptibleChannel ch,

                         Op op, int test)

        {

            super(cf + "/" + op + "/" + testName[test]);

            this.ch = ch;

            this.op = op;

            this.test = test;

        }

        @SuppressWarnings("fallthrough")

        private void caught(Channel ch, IOException x) {

            String xn = x.getClass().getName();

            switch (test) {

            case TEST_PREINTR:

            case TEST_INTR:

                if (!xn.equals("java.nio.channels.ClosedByInterruptException"))

                    throw new RuntimeException("Wrong exception thrown: " + x);

                break;

            case TEST_CLOSE:

            case TEST_SHUTO:

                if (!xn.equals("java.nio.channels.AsynchronousCloseException"))

                    throw new RuntimeException("Wrong exception thrown: " + x);

                break;

            case TEST_SHUTI:

                if (TestUtil.onWindows())

                    break;

                // FALL THROUGH

            default:

                throw new Error(x);

            }

            if (ch.isOpen()) {

                if (test == TEST_SHUTO) {

                    SocketChannel sc = (SocketChannel)ch;

                    if (!sc.socket().isOutputShutdown())

                        throw new RuntimeException("Output not shutdown");

                } else if ((test == TEST_INTR) && (op == TRANSFER_FROM)) {

                    // Let this case pass -- CBIE applies to other channel

                } else {

                    throw new RuntimeException("Channel still open");

                }

            }

            log.println("Thrown as expected: " + x);

        }

        final void go() throws Exception {

            if (test == TEST_PREINTR)

                Thread.currentThread().interrupt();

            ready = true;

            try {

                op.doIO(ch);

            } catch (ClosedByInterruptException x) {

                caught(ch, x);

            } catch (AsynchronousCloseException x) {

                caught(ch, x);

            } finally {

                ch.close();

            }

        }

    }

    private static volatile boolean pumpDone = false;

    private static volatile boolean pumpReady = false;

    private static void waitPump(String msg){

        pumpReady = false;

        log.println(msg);

        while (!pumpReady){

            sleep(200);

        }

    }

    // Create a pump thread dedicated to saturate refuser's connection backlog

    private static Future<Integer> pumpRefuser(ExecutorService pumperExecutor) {

        Callable<Integer> pumpTask = new Callable<Integer>() {

            @Override

            public Integer call() throws IOException {

                // Can't reliably saturate connection backlog on Windows Server editions

                assert !TestUtil.onWindows();

                log.println("Start pumping refuser ...");

                List<SocketChannel> refuserClients = new ArrayList<>();

                // Saturate the refuser's connection backlog so that further connection

                // attempts will be blocked

                while (!pumpDone) {

                    SocketChannel sc = SocketChannel.open();