test/jdk/java/nio/channels/AsynchronousFileChannel/Basic.java

/*
 * Copyright (c) 2008, 2022, 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 4607272 5041655 6822643 6830721 6842687
 * @summary Unit test for AsynchronousFileChannel
 * @key randomness
 */

import java.io.File;
import java.io.IOException;
import java.lang.foreign.Arena;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.CompletionHandler;
import java.nio.channels.FileLock;
import java.nio.channels.NonWritableChannelException;
import java.nio.channels.OverlappingFileLockException;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.List;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.CancellationException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import static java.nio.file.StandardOpenOption.*;

public class Basic {

    // Must be indeterministic
    private static final Random rand = new Random();

    public static void main(String[] args) throws IOException {
        // create temporary file
        File blah = File.createTempFile("blah", null);
        blah.deleteOnExit();

        AsynchronousFileChannel ch = AsynchronousFileChannel
            .open(blah.toPath(), READ, WRITE);
        try {
            // run tests
            testUsingCompletionHandlers(ch);
            testUsingWaitOnResult(ch);
            testInterruptHandlerThread(ch);
        } finally {
            ch.close();
        }

        // run test that expects channel to be closed
        testClosedChannel(ch);

        // these tests open the file themselves
        testLocking(blah.toPath());
        testCustomThreadPool(blah.toPath());
        testAsynchronousClose(blah.toPath());
        testCancel(blah.toPath());
        testTruncate(blah.toPath());

        // eagerly clean-up
        blah.delete();
    }

    /*
     * Generate buffer with random contents
     * Writes buffer to file using a CompletionHandler to consume the result
     *    of each write operation
     * Reads file to EOF to a new buffer using a CompletionHandler to consume
     *    the result of each read operation
     * Compares buffer contents
     */
    static void testUsingCompletionHandlers(AsynchronousFileChannel ch)
        throws IOException
    {
        System.out.println("testUsingCompletionHandlers");

        ch.truncate(0L);

        // generate buffer with random elements and write it to file
        ByteBuffer src = genBuffer();
        writeFully(ch, src, 0L);

        // read to EOF or buffer is full
        ByteBuffer dst = (rand.nextBoolean()) ?
            ByteBuffer.allocateDirect(src.capacity()) :
            ByteBuffer.allocate(src.capacity());
        readAll(ch, dst, 0L);

        // check buffers are the same
        src.flip();
        dst.flip();
        if (!src.equals(dst)) {
            throw new RuntimeException("Contents differ");
        }
    }

    /*
     * Generate buffer with random contents
     * Writes buffer to file, invoking the Future's get method to wait for
     *    each write operation to complete
     * Reads file to EOF to a new buffer, invoking the Future's get method to
     *    wait for each write operation to complete
     * Compares buffer contents
     */
    static void testUsingWaitOnResult(AsynchronousFileChannel ch)
        throws IOException
    {
        System.out.println("testUsingWaitOnResult");

        ch.truncate(0L);

        // generate buffer
        ByteBuffer src = genBuffer();

        // write buffer completely to file
        long position = 0L;
        while (src.hasRemaining()) {
            Future<Integer> result = ch.write(src, position);
            try {
                int n = result.get();
                // update position
                position += n;
            } catch (ExecutionException x) {
                throw new RuntimeException(x.getCause());
            } catch (InterruptedException x) {
                throw new RuntimeException(x);
            }
        }

        // read file into new buffer
        ByteBuffer dst = (rand.nextBoolean()) ?
            ByteBuffer.allocateDirect(src.capacity()) :
            ByteBuffer.allocate(src.capacity());
        position = 0L;
        int n;
        do {
            Future<Integer> result = ch.read(dst, position);
            try {
                n = result.get();

                // update position
                if (n > 0) position += n;
            } catch (ExecutionException x) {
                throw new RuntimeException(x.getCause());
            } catch (InterruptedException x) {
                throw new RuntimeException(x);
            }
        } while (n > 0);

        // check buffers are the same
        src.flip();
        dst.flip();
        if (!src.equals(dst)) {
            throw new RuntimeException("Contents differ");
        }
    }

    // exercise lock methods
    static void testLocking(Path file) throws IOException {
        System.out.println("testLocking");

        AsynchronousFileChannel ch = AsynchronousFileChannel
            .open(file, READ, WRITE);
        FileLock fl;
        try {
            // test 1 - acquire lock and check that tryLock throws
            // OverlappingFileLockException
            try {
                long pos = rand.nextInt(Integer.MAX_VALUE);
                fl = ch.lock(pos, 0, false).get();
                long expectedSize = Long.MAX_VALUE - pos;
                if(fl.size() != expectedSize)
                    throw new RuntimeException("Lock size " + fl.size() +
                        " != " + expectedSize + " for position " + pos);
            } catch (ExecutionException x) {
                throw new RuntimeException(x);
            } catch (InterruptedException x) {
                throw new RuntimeException("Should not be interrupted");
            }
            if (!fl.acquiredBy().equals(ch))
                throw new RuntimeException("FileLock#acquiredBy returned incorrect channel");
            try {
                ch.tryLock();
                throw new RuntimeException("OverlappingFileLockException expected");
            } catch (OverlappingFileLockException x) {
            }
            fl.release();

            // test 2 - acquire try and check that lock throws OverlappingFileLockException
            long pos = rand.nextInt(Integer.MAX_VALUE);
            fl = ch.tryLock(pos, 0, false);
            long expectedSize = Long.MAX_VALUE - pos;
            if(fl.size() != expectedSize)
                throw new RuntimeException("Lock size " + fl.size() + " != " +
                    expectedSize + " for position " + pos);
            if (fl == null)
                throw new RuntimeException("Unable to acquire lock");
            try {
                ch.lock((Void)null, new CompletionHandler<FileLock,Void> () {
                    public void completed(FileLock result, Void att) {
                    }
                    public void failed(Throwable exc, Void att) {
                    }
                });
                throw new RuntimeException("OverlappingFileLockException expected");
            } catch (OverlappingFileLockException x) {
            }
        } finally {
            ch.close();
        }

        // test 3 - channel is closed so FileLock should no longer be valid
        if (fl.isValid())
            throw new RuntimeException("FileLock expected to be invalid");
    }

    // interrupt should not close channel
    static void testInterruptHandlerThread(final AsynchronousFileChannel ch) {
        System.out.println("testInterruptHandlerThread");

        ByteBuffer buf = ByteBuffer.allocateDirect(100);
        final CountDownLatch latch = new CountDownLatch(1);

        ch.read(buf, 0L, (Void)null, new CompletionHandler<Integer,Void>() {
            public void completed(Integer result, Void att) {
                try {
                    Thread.currentThread().interrupt();
                    long size = ch.size();
                    latch.countDown();
                } catch (IOException x) {
                    x.printStackTrace();
                }
            }
            public void failed(Throwable exc, Void att) {
            }
        });

        // wait for handler to complete
        await(latch);
    }

    // invoke method on closed channel
    static void testClosedChannel(AsynchronousFileChannel ch) {
        System.out.println("testClosedChannel");

        if (ch.isOpen())
            throw new RuntimeException("Channel should be closed");

        ByteBuffer buf = ByteBuffer.allocateDirect(100);

        // check read fails with ClosedChannelException
        try {
            ch.read(buf, 0L).get();
            throw new RuntimeException("ExecutionException expected");
        } catch (ExecutionException x) {
            if (!(x.getCause() instanceof ClosedChannelException))
                throw new RuntimeException("Cause of ClosedChannelException expected");
        } catch (InterruptedException x) {
        }

        // check write fails with ClosedChannelException
        try {
            ch.write(buf, 0L).get();
            throw new RuntimeException("ExecutionException expected");
        } catch (ExecutionException x) {
            if (!(x.getCause() instanceof ClosedChannelException))
                throw new RuntimeException("Cause of ClosedChannelException expected");
        } catch (InterruptedException x) {
        }

        // check lock fails with ClosedChannelException
        try {
            ch.lock().get();
            throw new RuntimeException("ExecutionException expected");
        } catch (ExecutionException x) {
            if (!(x.getCause() instanceof ClosedChannelException))
                throw new RuntimeException("Cause of ClosedChannelException expected");
        } catch (InterruptedException x) {
        }
    }


    // exercise custom thread pool
    static void testCustomThreadPool(Path file) throws IOException {
        System.out.println("testCustomThreadPool");

        // records threads that are created
        final List<Thread> threads = new ArrayList<Thread>();

        ThreadFactory threadFactory = new ThreadFactory() {
             @Override
             public Thread newThread(Runnable r) {
                 Thread t = new Thread(r);
                 t.setDaemon(true);
                 synchronized (threads) {
                     threads.add(t);
                 }
                 return t;
             }
        };

        // exercise tests with varied number of threads
        for (int nThreads=1; nThreads<=5; nThreads++) {
            synchronized (threads) {
                threads.clear();
            }
            ExecutorService executor = Executors.newFixedThreadPool(nThreads, threadFactory);
            Set<StandardOpenOption> opts = EnumSet.of(WRITE);
            AsynchronousFileChannel ch = AsynchronousFileChannel.open(file, opts, executor);
            try {
                for (int i=0; i<10; i++) {
                    // do I/O operation to see which thread invokes the completion handler
                    final AtomicReference<Thread> invoker = new AtomicReference<Thread>();
                    final CountDownLatch latch = new CountDownLatch(1);

                    ch.write(genBuffer(), 0L, (Void)null, new CompletionHandler<Integer,Void>() {
                        public void completed(Integer result, Void att) {
                            invoker.set(Thread.currentThread());
                            latch.countDown();
                        }
                        public void failed(Throwable exc, Void att) {
                        }
                    });
                    await(latch);

                    // check invoker
                    boolean found = false;
                    synchronized (threads) {
                        for (Thread t: threads) {
                            if (t == invoker.get()) {
                                found = true;
                                break;
                            }
                        }
                    }
                    if (!found)
                        throw new RuntimeException("Invoker thread not found");
                }
            } finally {
                ch.close();
                executor.shutdown();
            }
        }


        // test sharing a thread pool between many channels
        ExecutorService executor = Executors
            .newFixedThreadPool(1+rand.nextInt(10), threadFactory);
        final int n = 50 + rand.nextInt(50);
        AsynchronousFileChannel[] channels = new AsynchronousFileChannel[n];
        try {
            for (int i=0; i<n; i++) {
                Set<StandardOpenOption> opts = EnumSet.of(WRITE);
                channels[i] = AsynchronousFileChannel.open(file, opts, executor);
                final CountDownLatch latch = new CountDownLatch(1);
                channels[i].write(genBuffer(), 0L, (Void)null, new CompletionHandler<Integer,Void>() {
                    public void completed(Integer result, Void att) {
                        latch.countDown();
                    }
                    public void failed(Throwable exc, Void att) {
                    }
                });
                await(latch);

                // close ~half the channels
                if (rand.nextBoolean())
                    channels[i].close();
            }
        } finally {
            // close remaining channels
            for (int i=0; i<n; i++) {
                if (channels[i] != null) channels[i].close();
            }
            executor.shutdown();
        }
    }

    // exercise asynchronous close
    static void testAsynchronousClose(Path file) throws IOException {
        System.out.println("testAsynchronousClose");

        // create file
        AsynchronousFileChannel ch = AsynchronousFileChannel
            .open(file, WRITE, TRUNCATE_EXISTING);
        long size = 0L;
        do {
            ByteBuffer buf = genBuffer();
            int n = buf.remaining();
            writeFully(ch, buf, size);
            size += n;
        } while (size < (50L * 1024L * 1024L));

        ch.close();

        ch = AsynchronousFileChannel.open(file, WRITE, SYNC);

        // randomize number of writers, buffer size, and positions

        int nwriters = 1 + rand.nextInt(8);
        ByteBuffer[] buf = new ByteBuffer[nwriters];
        long[] position = new long[nwriters];
        for (int i=0; i<nwriters; i++) {
            buf[i] = genBuffer();
            position[i] = rand.nextInt((int)size);
        }

        // initiate I/O
        Future[] result = new Future[nwriters];
        for (int i=0; i<nwriters; i++) {
            result[i] = ch.write(buf[i], position[i]);
        }

        // close file
        ch.close();

        // write operations should complete or fail with AsynchronousCloseException
        for (int i=0; i<nwriters; i++) {
            try {
                result[i].get();
            } catch (ExecutionException x) {
                Throwable cause = x.getCause();
                if (!(cause instanceof AsynchronousCloseException))
                    throw new RuntimeException(cause);
            } catch (CancellationException  x) {
                throw new RuntimeException(x);   // should not happen
            } catch (InterruptedException x) {
                throw new RuntimeException(x);   // should not happen
            }
        }
    }

    // exercise cancel method
    static void testCancel(Path file) throws IOException {
        System.out.println("testCancel");

        for (int i=0; i<2; i++) {
            boolean mayInterruptIfRunning = (i == 0) ? false : true;

            // open with SYNC option to improve chances that write will not
            // complete immediately
            AsynchronousFileChannel ch = AsynchronousFileChannel
                .open(file, WRITE, SYNC);

            // start write operation
            Future<Integer> res = ch.write(genBuffer(), 0L);

            // cancel operation
            boolean cancelled = res.cancel(mayInterruptIfRunning);

            // check post-conditions
            if (!res.isDone())
                throw new RuntimeException("isDone should return true");
            if (res.isCancelled() != cancelled)
                throw new RuntimeException("isCancelled not consistent");
            try {
                res.get();
                if (cancelled)
                    throw new RuntimeException("CancellationException expected");
            } catch (CancellationException x) {
                if (!cancelled)
                    throw new RuntimeException("CancellationException not expected");
            } catch (ExecutionException x) {
                throw new RuntimeException(x);
            } catch (InterruptedException x) {
                throw new RuntimeException(x);
            }
            try {
                res.get(1, TimeUnit.SECONDS);
                if (cancelled)
                    throw new RuntimeException("CancellationException expected");
            } catch (CancellationException x) {
                if (!cancelled)
                    throw new RuntimeException("CancellationException not expected");
            } catch (ExecutionException x) {
                throw new RuntimeException(x);
            } catch (TimeoutException x) {
                throw new RuntimeException(x);
            } catch (InterruptedException x) {
                throw new RuntimeException(x);
            }

            ch.close();
        }
    }

    // exercise truncate method
    static void testTruncate(Path file) throws IOException {
        System.out.println("testTruncate");

        // basic tests
        AsynchronousFileChannel ch = AsynchronousFileChannel
            .open(file, CREATE, WRITE, TRUNCATE_EXISTING);
        try {
            writeFully(ch, genBuffer(), 0L);
            long size = ch.size();

            // attempt to truncate to a size greater than the current size
            if (ch.truncate(size + 1L).size() != size)
                throw new RuntimeException("Unexpected size after truncation");

            // truncate file
            if (ch.truncate(size - 1L).size() != (size - 1L))
                throw new RuntimeException("Unexpected size after truncation");

            // invalid size
            try {
                ch.truncate(-1L);
                throw new RuntimeException("IllegalArgumentException expected");
            } catch (IllegalArgumentException e) { }

        } finally {
            ch.close();
        }

        // channel is closed
        try {
            ch.truncate(0L);
            throw new RuntimeException("ClosedChannelException expected");
        } catch (ClosedChannelException  e) { }

        // channel is read-only
        ch = AsynchronousFileChannel.open(file, READ);
        try {
            try {
            ch.truncate(0L);
                throw new RuntimeException("NonWritableChannelException expected");
            } catch (NonWritableChannelException  e) { }
        } finally {
            ch.close();
        }
    }

    // returns ByteBuffer with random bytes
    static ByteBuffer genBuffer() {
        int size = 1024 + rand.nextInt(16000);
        byte[] buf = new byte[size];
        return switch (rand.nextInt(3)) {
            case 0 -> ByteBuffer.allocateDirect(buf.length)
                    .put(buf)
                    .flip();
            case 1 -> ByteBuffer.wrap(buf);
            case 2 -> Arena.ofAuto().allocate(buf.length).asByteBuffer()
                    .put(buf)
                    .flip();
            default -> throw new InternalError("Should not reach here");
        };
    }

    // writes all remaining bytes in the buffer to the given channel at the
    // given position
    static void writeFully(final AsynchronousFileChannel ch,
                           final ByteBuffer src,
                           long position)
    {
        final CountDownLatch latch = new CountDownLatch(1);

        // use position as attachment
        ch.write(src, position, position, new CompletionHandler<Integer,Long>() {
            public void completed(Integer result, Long position) {
                int n = result;
                if (src.hasRemaining()) {
                    long p = position + n;
                    ch.write(src, p, p, this);
                } else {
                    latch.countDown();
                }
            }
            public void failed(Throwable exc, Long position) {
            }
        });

        // wait for writes to complete
        await(latch);
    }

    static void readAll(final AsynchronousFileChannel ch,
                        final ByteBuffer dst,
                       long position)
    {
        final CountDownLatch latch = new CountDownLatch(1);

        // use position as attachment
        ch.read(dst, position, position, new CompletionHandler<Integer,Long>() {
            public void completed(Integer result, Long position) {
                int n = result;
                if (n > 0) {
                    long p = position + n;
                    ch.read(dst, p, p, this);
                } else {
                    latch.countDown();
                }
            }
            public void failed(Throwable exc, Long position) {
            }
        });

        // wait for reads to complete
        await(latch);
    }

    static void await(CountDownLatch latch) {
        // wait until done
        boolean done = false;
        while (!done) {
            try {
                latch.await();
                done = true;
            } catch (InterruptedException x) { }
        }
    }
}