Java example source code file (SimpleAsynchronousFileChannelImpl.java)

This example Java source code file (SimpleAsynchronousFileChannelImpl.java) is included in the alvinalexander.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Learn more about this Java project at its project page.

Java - Java tags/keywords

asynchronouscloseexception, closedchannelexception, completionhandler, executorservice, illegalargumentexception, ioexception, nio, nonreadablechannelexception, nonwritablechannelexception, override, pendingfuture, runnable, security, simpleasynchronousfilechannelimpl, threading, threads, throwable

The SimpleAsynchronousFileChannelImpl.java Java example source code

/*
 * Copyright (c) 2008, 2013, 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.  Oracle designates this
 * particular file as subject to the "Classpath" exception as provided
 * by Oracle in the LICENSE file that accompanied this code.
 *
 * 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.
 */

package sun.nio.ch;

import java.nio.channels.*;
import java.util.concurrent.*;
import java.nio.ByteBuffer;
import java.security.AccessController;
import java.security.PrivilegedAction;
import java.io.FileDescriptor;
import java.io.IOException;

/**
 * "Portable" implementation of AsynchronousFileChannel for use on operating
 * systems that don't support asynchronous file I/O.
 */

public class SimpleAsynchronousFileChannelImpl
    extends AsynchronousFileChannelImpl
{
    // lazy initialization of default thread pool for file I/O
    private static class DefaultExecutorHolder {
        static final ExecutorService defaultExecutor =
            ThreadPool.createDefault().executor();
    }

    // Used to make native read and write calls
    private static final FileDispatcher nd = new FileDispatcherImpl();

    // Thread-safe set of IDs of native threads, for signalling
    private final NativeThreadSet threads = new NativeThreadSet(2);


    SimpleAsynchronousFileChannelImpl(FileDescriptor fdObj,
                                      boolean reading,
                                      boolean writing,
                                      ExecutorService executor)
    {
        super(fdObj, reading, writing, executor);
    }

    public static AsynchronousFileChannel open(FileDescriptor fdo,
                                               boolean reading,
                                               boolean writing,
                                               ThreadPool pool)
    {
        // Executor is either default or based on pool parameters
        ExecutorService executor = (pool == null) ?
            DefaultExecutorHolder.defaultExecutor : pool.executor();
        return new SimpleAsynchronousFileChannelImpl(fdo, reading, writing, executor);
    }

    @Override
    public void close() throws IOException {
        // mark channel as closed
        synchronized (fdObj) {
            if (closed)
                return;     // already closed
            closed = true;
            // from this point on, if another thread invokes the begin() method
            // then it will throw ClosedChannelException
        }

        // Invalidate and release any locks that we still hold
        invalidateAllLocks();

        // signal any threads blocked on this channel
        nd.preClose(fdObj);
        threads.signalAndWait();

        // wait until all async I/O operations have completely gracefully
        closeLock.writeLock().lock();
        try {
            // do nothing
        } finally {
            closeLock.writeLock().unlock();
        }

        // close file
        nd.close(fdObj);
    }

    @Override
    public long size() throws IOException {
        int ti = threads.add();
        try {
            long n = 0L;
            try {
                begin();
                do {
                    n = nd.size(fdObj);
                } while ((n == IOStatus.INTERRUPTED) && isOpen());
                return n;
            } finally {
                end(n >= 0L);
            }
        } finally {
            threads.remove(ti);
        }
    }

    @Override
    public AsynchronousFileChannel truncate(long size) throws IOException {
        if (size < 0L)
            throw new IllegalArgumentException("Negative size");
        if (!writing)
            throw new NonWritableChannelException();
        int ti = threads.add();
        try {
            long n = 0L;
            try {
                begin();
                do {
                    n = nd.size(fdObj);
                } while ((n == IOStatus.INTERRUPTED) && isOpen());

                // truncate file if 'size' less than current size
                if (size < n && isOpen()) {
                    do {
                        n = nd.truncate(fdObj, size);
                    } while ((n == IOStatus.INTERRUPTED) && isOpen());
                }
                return this;
            } finally {
                end(n > 0);
            }
        } finally {
            threads.remove(ti);
        }
    }

    @Override
    public void force(boolean metaData) throws IOException {
        int ti = threads.add();
        try {
            int n = 0;
            try {
                begin();
                do {
                    n = nd.force(fdObj, metaData);
                } while ((n == IOStatus.INTERRUPTED) && isOpen());
            } finally {
                end(n >= 0);
            }
        } finally {
            threads.remove(ti);
        }
    }

    @Override
    <A> Future implLock(final long position,
                                  final long size,
                                  final boolean shared,
                                  final A attachment,
                                  final CompletionHandler<FileLock,? super A> handler)
    {
        if (shared && !reading)
            throw new NonReadableChannelException();
        if (!shared && !writing)
            throw new NonWritableChannelException();

        // add to lock table
        final FileLockImpl fli = addToFileLockTable(position, size, shared);
        if (fli == null) {
            Throwable exc = new ClosedChannelException();
            if (handler == null)
                return CompletedFuture.withFailure(exc);
            Invoker.invokeIndirectly(handler, attachment, null, exc, executor);
            return null;
        }

        final PendingFuture<FileLock,A> result = (handler == null) ?
            new PendingFuture<FileLock,A>(this) : null;
        Runnable task = new Runnable() {
            public void run() {
                Throwable exc = null;

                int ti = threads.add();
                try {
                    int n;
                    try {
                        begin();
                        do {
                            n = nd.lock(fdObj, true, position, size, shared);
                        } while ((n == FileDispatcher.INTERRUPTED) && isOpen());
                        if (n != FileDispatcher.LOCKED || !isOpen()) {
                            throw new AsynchronousCloseException();
                        }
                    } catch (IOException x) {
                        removeFromFileLockTable(fli);
                        if (!isOpen())
                            x = new AsynchronousCloseException();
                        exc = x;
                    } finally {
                        end();
                    }
                } finally {
                    threads.remove(ti);
                }
                if (handler == null) {
                    result.setResult(fli, exc);
                } else {
                    Invoker.invokeUnchecked(handler, attachment, fli, exc);
                }
            }
        };
        boolean executed = false;
        try {
            executor.execute(task);
            executed = true;
        } finally {
            if (!executed) {
                // rollback
                removeFromFileLockTable(fli);
            }
        }
        return result;
    }

    @Override
    public FileLock tryLock(long position, long size, boolean shared)
        throws IOException
    {
        if (shared && !reading)
            throw new NonReadableChannelException();
        if (!shared && !writing)
            throw new NonWritableChannelException();

        // add to lock table
        FileLockImpl fli = addToFileLockTable(position, size, shared);
        if (fli == null)
            throw new ClosedChannelException();

        int ti = threads.add();
        boolean gotLock = false;
        try {
            begin();
            int n;
            do {
                n = nd.lock(fdObj, false, position, size, shared);
            } while ((n == FileDispatcher.INTERRUPTED) && isOpen());
            if (n == FileDispatcher.LOCKED && isOpen()) {
                gotLock = true;
                return fli;    // lock acquired
            }
            if (n == FileDispatcher.NO_LOCK)
                return null;    // locked by someone else
            if (n == FileDispatcher.INTERRUPTED)
                throw new AsynchronousCloseException();
            // should not get here
            throw new AssertionError();
        } finally {
            if (!gotLock)
                removeFromFileLockTable(fli);
            end();
            threads.remove(ti);
        }
    }

    @Override
    protected void implRelease(FileLockImpl fli) throws IOException {
        nd.release(fdObj, fli.position(), fli.size());
    }

    @Override
    <A> Future implRead(final ByteBuffer dst,
                                 final long position,
                                 final A attachment,
                                 final CompletionHandler<Integer,? super A> handler)
    {
        if (position < 0)
            throw new IllegalArgumentException("Negative position");
        if (!reading)
            throw new NonReadableChannelException();
        if (dst.isReadOnly())
            throw new IllegalArgumentException("Read-only buffer");

        // complete immediately if channel closed or no space remaining
        if (!isOpen() || (dst.remaining() == 0)) {
            Throwable exc = (isOpen()) ? null : new ClosedChannelException();
            if (handler == null)
                return CompletedFuture.withResult(0, exc);
            Invoker.invokeIndirectly(handler, attachment, 0, exc, executor);
            return null;
        }

        final PendingFuture<Integer,A> result = (handler == null) ?
            new PendingFuture<Integer,A>(this) : null;
        Runnable task = new Runnable() {
            public void run() {
                int n = 0;
                Throwable exc = null;

                int ti = threads.add();
                try {
                    begin();
                    do {
                        n = IOUtil.read(fdObj, dst, position, nd);
                    } while ((n == IOStatus.INTERRUPTED) && isOpen());
                    if (n < 0 && !isOpen())
                        throw new AsynchronousCloseException();
                } catch (IOException x) {
                    if (!isOpen())
                        x = new AsynchronousCloseException();
                    exc = x;
                } finally {
                    end();
                    threads.remove(ti);
                }
                if (handler == null) {
                    result.setResult(n, exc);
                } else {
                    Invoker.invokeUnchecked(handler, attachment, n, exc);
                }
            }
        };
        executor.execute(task);
        return result;
    }

    @Override
    <A> Future implWrite(final ByteBuffer src,
                                  final long position,
                                  final A attachment,
                                  final CompletionHandler<Integer,? super A> handler)
    {
        if (position < 0)
            throw new IllegalArgumentException("Negative position");
        if (!writing)
            throw new NonWritableChannelException();

        // complete immediately if channel is closed or no bytes remaining
        if (!isOpen() || (src.remaining() == 0)) {
            Throwable exc = (isOpen()) ? null : new ClosedChannelException();
            if (handler == null)
                return CompletedFuture.withResult(0, exc);
            Invoker.invokeIndirectly(handler, attachment, 0, exc, executor);
            return null;
        }

        final PendingFuture<Integer,A> result = (handler == null) ?
            new PendingFuture<Integer,A>(this) : null;
        Runnable task = new Runnable() {
            public void run() {
                int n = 0;
                Throwable exc = null;

                int ti = threads.add();
                try {
                    begin();
                    do {
                        n = IOUtil.write(fdObj, src, position, nd);
                    } while ((n == IOStatus.INTERRUPTED) && isOpen());
                    if (n < 0 && !isOpen())
                        throw new AsynchronousCloseException();
                } catch (IOException x) {
                    if (!isOpen())
                        x = new AsynchronousCloseException();
                    exc = x;
                } finally {
                    end();
                    threads.remove(ti);
                }
                if (handler == null) {
                    result.setResult(n, exc);
                } else {
                    Invoker.invokeUnchecked(handler, attachment, n, exc);
                }
            }
        };
        executor.execute(task);
        return result;
    }
}