1 /*

     2  * Copyright 2008-2009 Sun Microsystems, Inc.  All Rights Reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.  Sun designates this

     8  * particular file as subject to the "Classpath" exception as provided

     9  * by Sun in the LICENSE file that accompanied this code.

    10  *

    11  * This code is distributed in the hope that it will be useful, but WITHOUT

    12  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    13  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    14  * version 2 for more details (a copy is included in the LICENSE file that

    15  * accompanied this code).

    16  *

    17  * You should have received a copy of the GNU General Public License version

    18  * 2 along with this work; if not, write to the Free Software Foundation,

    19  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    20  *

    21  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    22  * CA 95054 USA or visit www.sun.com if you need additional information or

    23  * have any questions.

    24  */

    25 

    26 package sun.nio.ch;

    27 

    28 import java.nio.channels.*;

    29 import java.util.concurrent.*;

    30 import java.nio.ByteBuffer;

    31 import java.security.AccessController;

    32 import java.security.PrivilegedAction;

    33 import java.io.FileDescriptor;

    34 import java.io.IOException;

    35 

    36 /**

    37  * "Portable" implementation of AsynchronousFileChannel for use on operating

    38  * systems that don't support asynchronous file I/O.

    39  */

    40 

    41 public class SimpleAsynchronousFileChannelImpl

    42     extends AsynchronousFileChannelImpl

    43 {

    44     // lazy initialization of default thread pool for file I/O

    45     private static class DefaultExecutorHolder {

    46         static final ExecutorService defaultExecutor =

    47             ThreadPool.createDefault().executor();

    48     }

    49 

    50     // Used to make native read and write calls

    51     private static final FileDispatcher nd = new FileDispatcherImpl();

    52 

    53     // indicates if the associated thread pool is the default thread pool

    54     private final boolean isDefaultExecutor;

    55 

    56     // Thread-safe set of IDs of native threads, for signalling

    57     private final NativeThreadSet threads = new NativeThreadSet(2);

    58 

    59 

    60     SimpleAsynchronousFileChannelImpl(FileDescriptor fdObj,

    61                                       boolean reading,

    62                                       boolean writing,

    63                                       ExecutorService executor,

    64                                       boolean isDefaultexecutor)

    65     {

    66         super(fdObj, reading, writing, executor);

    67         this.isDefaultExecutor = isDefaultexecutor;

    68     }

    69 

    70     public static AsynchronousFileChannel open(FileDescriptor fdo,

    71                                                boolean reading,

    72                                                boolean writing,

    73                                                ThreadPool pool)

    74     {

    75         // Executor is either default or based on pool parameters

    76         ExecutorService executor;

    77         boolean isDefaultexecutor;

    78         if (pool == null) {

    79             executor = DefaultExecutorHolder.defaultExecutor;

    80             isDefaultexecutor = true;

    81         } else {

    82             executor = pool.executor();

    83             isDefaultexecutor = false;

    84         }

    85         return new SimpleAsynchronousFileChannelImpl(fdo,

    86             reading, writing, executor, isDefaultexecutor);

    87     }

    88 

    89     @Override

    90     public void close() throws IOException {

    91         // mark channel as closed

    92         synchronized (fdObj) {

    93             if (closed)

    94                 return;     // already closed

    95             closed = true;

    96             // from this point on, if another thread invokes the begin() method

    97             // then it will throw ClosedChannelException

    98         }

    99 

   100         // signal any threads blocked on this channel

   101         nd.preClose(fdObj);

   102         threads.signalAndWait();

   103 

   104         // wait until all async I/O operations have completely gracefully

   105         closeLock.writeLock().lock();

   106         try {

   107             // do nothing

   108         } finally {

   109             closeLock.writeLock().unlock();

   110         }

   111 

   112         // Invalidate and release any locks that we still hold

   113         invalidateAllLocks();

   114 

   115         // close file

   116         nd.close(fdObj);

   117 

   118         // shutdown executor if specific to this channel

   119         if (!isDefaultExecutor) {

   120             AccessController.doPrivileged(new PrivilegedAction<Void>() {

   121                 public Void run() {

   122                     executor.shutdown();

   123                     return null;

   124                 }

   125             });

   126         }

   127     }

   128 

   129     @Override

   130     public long size() throws IOException {

   131         int ti = threads.add();

   132         try {

   133             long n = 0L;

   134             try {

   135                 begin();

   136                 do {

   137                     n = nd.size(fdObj);

   138                 } while ((n == IOStatus.INTERRUPTED) && isOpen());

   139                 return n;

   140             } finally {

   141                 end(n >= 0L);

   142             }

   143         } finally {

   144             threads.remove(ti);

   145         }

   146     }

   147 

   148     @Override

   149     public AsynchronousFileChannel truncate(long size) throws IOException {

   150         if (size < 0L)

   151             throw new IllegalArgumentException("Negative size");

   152         if (!writing)

   153             throw new NonWritableChannelException();

   154         int ti = threads.add();

   155         try {

   156             long n = 0L;

   157             try {

   158                 begin();

   159                 do {

   160                     n = nd.size(fdObj);

   161                 } while ((n == IOStatus.INTERRUPTED) && isOpen());

   162 

   163                 // truncate file if 'size' less than current size

   164                 if (size < n && isOpen()) {

   165                     do {

   166                         n = nd.truncate(fdObj, size);

   167                     } while ((n == IOStatus.INTERRUPTED) && isOpen());

   168                 }

   169                 return this;

   170             } finally {

   171                 end(n > 0);

   172             }

   173         } finally {

   174             threads.remove(ti);

   175         }

   176     }

   177 

   178     @Override

   179     public void force(boolean metaData) throws IOException {

   180         int ti = threads.add();

   181         try {

   182             int n = 0;

   183             try {

   184                 begin();

   185                 do {

   186                     n = nd.force(fdObj, metaData);

   187                 } while ((n == IOStatus.INTERRUPTED) && isOpen());

   188             } finally {

   189                 end(n >= 0);

   190             }

   191         } finally {

   192             threads.remove(ti);

   193         }

   194     }

   195 

   196     @Override

   197     public <A> Future<FileLock> lock(final long position,

   198                                      final long size,

   199                                      final boolean shared,

   200                                      A attachment,

   201                                      final CompletionHandler<FileLock,? super A> handler)

   202     {

   203         if (shared && !reading)

   204             throw new NonReadableChannelException();

   205         if (!shared && !writing)

   206             throw new NonWritableChannelException();

   207 

   208         // add to lock table

   209         final FileLockImpl fli = addToFileLockTable(position, size, shared);

   210         if (fli == null) {

   211             CompletedFuture<FileLock,A> result = CompletedFuture

   212                 .withFailure(this, new ClosedChannelException(), attachment);

   213             Invoker.invokeIndirectly(handler, result, executor);

   214             return result;

   215         }

   216 

   217         final PendingFuture<FileLock,A> result =

   218             new PendingFuture<FileLock,A>(this, handler, attachment);

   219         Runnable task = new Runnable() {

   220             public void run() {

   221                 int ti = threads.add();

   222                 try {

   223                     int n;

   224                     try {

   225                         begin();

   226                         do {

   227                             n = nd.lock(fdObj, true, position, size, shared);

   228                         } while ((n == FileDispatcher.INTERRUPTED) && isOpen());

   229                         if (n == FileDispatcher.LOCKED) {

   230                             result.setResult(fli);

   231                         } else {

   232                             if (n != FileDispatcher.INTERRUPTED)

   233                                 throw new AssertionError();

   234                             throw new AsynchronousCloseException();

   235                         }

   236                     } catch (IOException x) {

   237                         removeFromFileLockTable(fli);

   238                         if (!isOpen())

   239                             x = new AsynchronousCloseException();

   240                         result.setFailure(x);

   241                     } finally {

   242                         end();

   243                     }

   244                 } finally {

   245                     threads.remove(ti);

   246                 }

   247                 Invoker.invokeUnchecked(handler, result);

   248             }

   249         };

   250         try {

   251             executor.execute(task);

   252         } catch (RejectedExecutionException ree) {

   253             // rollback

   254             removeFromFileLockTable(fli);

   255             throw new ShutdownChannelGroupException();

   256         }

   257         return result;

   258     }

   259 

   260     @Override

   261     public FileLock tryLock(long position, long size, boolean shared)

   262         throws IOException

   263     {

   264         if (shared && !reading)

   265             throw new NonReadableChannelException();

   266         if (!shared && !writing)

   267             throw new NonWritableChannelException();

   268 

   269         // add to lock table

   270         FileLockImpl fli = addToFileLockTable(position, size, shared);

   271         if (fli == null)

   272             throw new ClosedChannelException();

   273 

   274         int ti = threads.add();

   275         boolean gotLock = false;

   276         try {

   277             begin();

   278             int n;

   279             do {

   280                 n = nd.lock(fdObj, false, position, size, shared);

   281             } while ((n == FileDispatcher.INTERRUPTED) && isOpen());

   282             if (n != FileDispatcher.LOCKED) {

   283                 if (n == FileDispatcher.NO_LOCK)

   284                     return null;    // locked by someone else

   285                 if (n == FileDispatcher.INTERRUPTED)

   286                     throw new AsynchronousCloseException();

   287                 // should not get here

   288                 throw new AssertionError();

   289             }

   290             gotLock = true;

   291             return fli;

   292         } finally {

   293             if (!gotLock)

   294                 removeFromFileLockTable(fli);

   295             end();

   296             threads.remove(ti);

   297         }

   298     }

   299 

   300     @Override

   301     void release(FileLockImpl fli) throws IOException {

   302         try {

   303             begin();

   304             nd.release(fdObj, fli.position(), fli.size());

   305             removeFromFileLockTable(fli);

   306         } finally {

   307             end();

   308         }

   309     }

   310 

   311     @Override

   312     public <A> Future<Integer> read(final ByteBuffer dst,

   313                                     final long position,

   314                                     A attachment,

   315                                     final CompletionHandler<Integer,? super A> handler)

   316     {

   317         if (position < 0)

   318             throw new IllegalArgumentException("Negative position");

   319         if (!reading)

   320             throw new NonReadableChannelException();

   321         if (dst.isReadOnly())

   322             throw new IllegalArgumentException("Read-only buffer");

   323 

   324         // complete immediately if channel closed or no space remaining

   325         if (!isOpen() || (dst.remaining() == 0)) {

   326             CompletedFuture<Integer,A> result;

   327             if (isOpen()) {

   328                 result = CompletedFuture.withResult(this, 0, attachment);

   329             } else {

   330                 result = CompletedFuture.withFailure(this,

   331                     new ClosedChannelException(), attachment);

   332             }

   333             Invoker.invokeIndirectly(handler, result, executor);

   334             return result;

   335         }

   336 

   337         final PendingFuture<Integer,A> result =

   338             new PendingFuture<Integer,A>(this, handler, attachment);

   339         Runnable task = new Runnable() {

   340             public void run() {

   341                 int ti = threads.add();

   342                 try {

   343                     begin();

   344                     int n;

   345                     do {

   346                         n = IOUtil.read(fdObj, dst, position, nd, null);

   347                     } while ((n == IOStatus.INTERRUPTED) && isOpen());

   348                     if (n < 0 && !isOpen())

   349                         throw new AsynchronousCloseException();

   350                     result.setResult(n);

   351                 } catch (IOException x) {

   352                     if (!isOpen())

   353                         x = new AsynchronousCloseException();

   354                     result.setFailure(x);

   355                 } finally {

   356                     end();

   357                     threads.remove(ti);

   358                 }

   359                 Invoker.invokeUnchecked(handler, result);

   360             }

   361         };

   362         try {

   363             executor.execute(task);

   364         } catch (RejectedExecutionException ree) {

   365             throw new ShutdownChannelGroupException();

   366         }

   367         return result;

   368     }

   369 

   370     @Override

   371     public <A> Future<Integer> write(final ByteBuffer src,

   372                                      final long position,

   373                                      A attachment,

   374                                      final CompletionHandler<Integer,? super A> handler)

   375     {

   376         if (position < 0)

   377             throw new IllegalArgumentException("Negative position");

   378         if (!writing)

   379             throw new NonWritableChannelException();

   380 

   381         // complete immediately if channel is closed or no bytes remaining

   382         if (!isOpen() || (src.remaining() == 0)) {

   383             CompletedFuture<Integer,A> result;

   384             if (isOpen()) {

   385                 result = CompletedFuture.withResult(this, 0, attachment);

   386             } else {

   387                 result = CompletedFuture.withFailure(this,

   388                     new ClosedChannelException(), attachment);

   389             }

   390             Invoker.invokeIndirectly(handler, result, executor);

   391             return result;

   392         }

   393 

   394         final PendingFuture<Integer,A> result =

   395             new PendingFuture<Integer,A>(this, handler, attachment);

   396         Runnable task = new Runnable() {

   397             public void run() {

   398                 int ti = threads.add();

   399                 try {

   400                     begin();

   401                     int n;

   402                     do {

   403                         n = IOUtil.write(fdObj, src, position, nd, null);

   404                     } while ((n == IOStatus.INTERRUPTED) && isOpen());

   405                     if (n < 0 && !isOpen())

   406                         throw new AsynchronousCloseException();

   407                     result.setResult(n);

   408                 } catch (IOException x) {

   409                     if (!isOpen())

   410                         x = new AsynchronousCloseException();

   411                     result.setFailure(x);

   412                 } finally {

   413                     end();

   414                     threads.remove(ti);

   415                 }

   416                 Invoker.invokeUnchecked(handler, result);

   417             }

   418         };

   419         try {

   420             executor.execute(task);

   421         } catch (RejectedExecutionException ree) {

   422             throw new ShutdownChannelGroupException();

   423         }

   424         return result;

   425     }

   426 }