ojluni/src/main/java/sun/nio/ch/SimpleAsynchronousFileChannelImpl.java - platform/libcore.git - Git at Google

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

 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<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;
     }
 }
	/*
	* 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.
	*/

	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<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;
	}
	}