/*
 * Copyright (c) 2008, 2009, 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.
/** * "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 { // Added parameter writing for AIX platform port. n = nd.force(fdObj, metaData, writing); } while ((n == IOStatus.INTERRUPTED) && isOpen()); } finally { end(n >= 0); } } finally { threads.remove(ti); } } @Override <A> Future<FileLock> 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<Integer> 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<Integer> 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; } }