1 /*
   2  * Copyright (c) 2008, 2018, Oracle and/or its affiliates. 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.  Oracle designates this
   8  * particular file as subject to the "Classpath" exception as provided
   9  * by Oracle 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 Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  22  * or visit www.oracle.com if you need additional information or have any
  23  * 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.nio.BufferOverflowException;
  32 import java.io.IOException;
  33 import java.io.FileDescriptor;
  34 import jdk.internal.access.SharedSecrets;
  35 import jdk.internal.access.JavaIOFileDescriptorAccess;
  36 
  37 /**
  38  * Windows implementation of AsynchronousFileChannel using overlapped I/O.
  39  */
  40 
  41 public class WindowsAsynchronousFileChannelImpl
  42     extends AsynchronousFileChannelImpl
  43     implements Iocp.OverlappedChannel, Groupable
  44 {
  45     private static final JavaIOFileDescriptorAccess fdAccess =
  46         SharedSecrets.getJavaIOFileDescriptorAccess();
  47 
  48     // error when EOF is detected asynchronously.
  49     private static final int ERROR_HANDLE_EOF = 38;
  50 
  51     // Lazy initialization of default I/O completion port
  52     private static class DefaultIocpHolder {
  53         static final Iocp defaultIocp = defaultIocp();
  54         private static Iocp defaultIocp() {
  55             try {
  56                 return new Iocp(null, ThreadPool.createDefault()).start();
  57             } catch (IOException ioe) {
  58                 throw new InternalError(ioe);
  59             }
  60         }
  61     }
  62 
  63     // Used for force/truncate/size methods
  64     private static final FileDispatcher nd = new FileDispatcherImpl();
  65 
  66     // The handle is extracted for use in native methods invoked from this class.
  67     private final long handle;
  68 
  69     // The key that identifies the channel's association with the I/O port
  70     private final int completionKey;
  71 
  72     // I/O completion port (group)
  73     private final Iocp iocp;
  74 
  75     private final boolean isDefaultIocp;
  76 
  77     // Caches OVERLAPPED structure for each outstanding I/O operation
  78     private final PendingIoCache ioCache;
  79 
  80 
  81     private WindowsAsynchronousFileChannelImpl(FileDescriptor fdObj,
  82                                                boolean reading,
  83                                                boolean writing,
  84                                                Iocp iocp,
  85                                                boolean isDefaultIocp)
  86         throws IOException
  87     {
  88         super(fdObj, reading, writing, iocp.executor());
  89         this.handle = fdAccess.getHandle(fdObj);
  90         this.iocp = iocp;
  91         this.isDefaultIocp = isDefaultIocp;
  92         this.ioCache = new PendingIoCache();
  93         this.completionKey = iocp.associate(this, handle);
  94     }
  95 
  96     public static AsynchronousFileChannel open(FileDescriptor fdo,
  97                                                boolean reading,
  98                                                boolean writing,
  99                                                ThreadPool pool)
 100         throws IOException
 101     {
 102         Iocp iocp;
 103         boolean isDefaultIocp;
 104         if (pool == null) {
 105             iocp = DefaultIocpHolder.defaultIocp;
 106             isDefaultIocp = true;
 107         } else {
 108             iocp = new Iocp(null, pool).start();
 109             isDefaultIocp = false;
 110         }
 111         try {
 112             return new
 113                 WindowsAsynchronousFileChannelImpl(fdo, reading, writing, iocp, isDefaultIocp);
 114         } catch (IOException x) {
 115             // error binding to port so need to close it (if created for this channel)
 116             if (!isDefaultIocp)
 117                 iocp.implClose();
 118             throw x;
 119         }
 120     }
 121 
 122     @Override
 123     public <V,A> PendingFuture<V,A> getByOverlapped(long overlapped) {
 124         return ioCache.remove(overlapped);
 125     }
 126 
 127     @Override
 128     public void close() throws IOException {
 129         closeLock.writeLock().lock();
 130         try {
 131             if (closed)
 132                 return;     // already closed
 133             closed = true;
 134         } finally {
 135             closeLock.writeLock().unlock();
 136         }
 137 
 138         // invalidate all locks held for this channel
 139         invalidateAllLocks();
 140 
 141         // close the file
 142         nd.close(fdObj);
 143 
 144         // waits until all I/O operations have completed
 145         ioCache.close();
 146 
 147         // disassociate from port
 148         iocp.disassociate(completionKey);
 149 
 150         // for the non-default group close the port
 151         if (!isDefaultIocp)
 152             iocp.detachFromThreadPool();
 153     }
 154 
 155     @Override
 156     public AsynchronousChannelGroupImpl group() {
 157         return iocp;
 158     }
 159 
 160     /**
 161      * Translates Throwable to IOException
 162      */
 163     private static IOException toIOException(Throwable x) {
 164         if (x instanceof IOException) {
 165             if (x instanceof ClosedChannelException)
 166                 x = new AsynchronousCloseException();
 167             return (IOException)x;
 168         }
 169         return new IOException(x);
 170     }
 171 
 172     @Override
 173     public long size() throws IOException {
 174         try {
 175             begin();
 176             return nd.size(fdObj);
 177         } finally {
 178             end();
 179         }
 180     }
 181 
 182     @Override
 183     public AsynchronousFileChannel truncate(long size) throws IOException {
 184         if (size < 0)
 185             throw new IllegalArgumentException("Negative size");
 186         if (!writing)
 187             throw new NonWritableChannelException();
 188         try {
 189             begin();
 190             if (size > nd.size(fdObj))
 191                 return this;
 192             nd.truncate(fdObj, size);
 193         } finally {
 194             end();
 195         }
 196         return this;
 197     }
 198 
 199     @Override
 200     public void force(boolean metaData) throws IOException {
 201         try {
 202             begin();
 203             nd.force(fdObj, metaData);
 204         } finally {
 205             end();
 206         }
 207     }
 208 
 209     // -- file locking --
 210 
 211     /**
 212      * Task that initiates locking operation and handles completion result.
 213      */
 214     private class LockTask<A> implements Runnable, Iocp.ResultHandler {
 215         private final long position;
 216         private final FileLockImpl fli;
 217         private final PendingFuture<FileLock,A> result;
 218 
 219         LockTask(long position,
 220                  FileLockImpl fli,
 221                  PendingFuture<FileLock,A> result)
 222         {
 223             this.position = position;
 224             this.fli = fli;
 225             this.result = result;
 226         }
 227 
 228         @Override
 229         public void run() {
 230             long overlapped = 0L;
 231             boolean pending = false;
 232             try {
 233                 begin();
 234 
 235                 // allocate OVERLAPPED structure
 236                 overlapped = ioCache.add(result);
 237 
 238                 // synchronize on result to avoid race with handler thread
 239                 // when lock is acquired immediately.
 240                 synchronized (result) {
 241                     int n = lockFile(handle, position, fli.size(), fli.isShared(),
 242                                      overlapped);
 243                     if (n == IOStatus.UNAVAILABLE) {
 244                         // I/O is pending
 245                         pending = true;
 246                         return;
 247                     }
 248                     // acquired lock immediately
 249                     result.setResult(fli);
 250                 }
 251 
 252             } catch (Throwable x) {
 253                 // lock failed or channel closed
 254                 removeFromFileLockTable(fli);
 255                 result.setFailure(toIOException(x));
 256             } finally {
 257                 if (!pending && overlapped != 0L)
 258                     ioCache.remove(overlapped);
 259                 end();
 260             }
 261 
 262             // invoke completion handler
 263             Invoker.invoke(result);
 264         }
 265 
 266         @Override
 267         public void completed(int bytesTransferred, boolean canInvokeDirect) {
 268             // release waiters and invoke completion handler
 269             result.setResult(fli);
 270             if (canInvokeDirect) {
 271                 Invoker.invokeUnchecked(result);
 272             } else {
 273                 Invoker.invoke(result);
 274             }
 275         }
 276 
 277         @Override
 278         public void failed(int error, IOException x) {
 279             // lock not acquired so remove from lock table
 280             removeFromFileLockTable(fli);
 281 
 282             // release waiters
 283             if (isOpen()) {
 284                 result.setFailure(x);
 285             } else {
 286                 result.setFailure(new AsynchronousCloseException());
 287             }
 288             Invoker.invoke(result);
 289         }
 290     }
 291 
 292     @Override
 293     <A> Future<FileLock> implLock(final long position,
 294                                   final long size,
 295                                   final boolean shared,
 296                                   A attachment,
 297                                   final CompletionHandler<FileLock,? super A> handler)
 298     {
 299         if (shared && !reading)
 300             throw new NonReadableChannelException();
 301         if (!shared && !writing)
 302             throw new NonWritableChannelException();
 303 
 304         // add to lock table
 305         FileLockImpl fli = addToFileLockTable(position, size, shared);
 306         if (fli == null) {
 307             Throwable exc = new ClosedChannelException();
 308             if (handler == null)
 309                 return CompletedFuture.withFailure(exc);
 310             Invoker.invoke(this, handler, attachment, null, exc);
 311             return null;
 312         }
 313 
 314         // create Future and task that will be invoked to acquire lock
 315         PendingFuture<FileLock,A> result =
 316             new PendingFuture<FileLock,A>(this, handler, attachment);
 317         LockTask<A> lockTask = new LockTask<A>(position, fli, result);
 318         result.setContext(lockTask);
 319 
 320         // initiate I/O
 321         lockTask.run();
 322         return result;
 323     }
 324 
 325     static final int NO_LOCK = -1;       // Failed to lock
 326     static final int LOCKED = 0;         // Obtained requested lock
 327 
 328     @Override
 329     public FileLock tryLock(long position, long size, boolean shared)
 330         throws IOException
 331     {
 332         if (shared && !reading)
 333             throw new NonReadableChannelException();
 334         if (!shared && !writing)
 335             throw new NonWritableChannelException();
 336 
 337         // add to lock table
 338         final FileLockImpl fli = addToFileLockTable(position, size, shared);
 339         if (fli == null)
 340             throw new ClosedChannelException();
 341 
 342         boolean gotLock = false;
 343         try {
 344             begin();
 345             // try to acquire the lock
 346             int res = nd.lock(fdObj, false, position, size, shared);
 347             if (res == NO_LOCK)
 348                 return null;
 349             gotLock = true;
 350             return fli;
 351         } finally {
 352             if (!gotLock)
 353                 removeFromFileLockTable(fli);
 354             end();
 355         }
 356     }
 357 
 358     @Override
 359     protected void implRelease(FileLockImpl fli) throws IOException {
 360         nd.release(fdObj, fli.position(), fli.size());
 361     }
 362 
 363     /**
 364      * Task that initiates read operation and handles completion result.
 365      */
 366     private class ReadTask<A> implements Runnable, Iocp.ResultHandler {
 367         private final ByteBuffer dst;
 368         private final int pos, rem;     // buffer position/remaining
 369         private final long position;    // file position
 370         private final PendingFuture<Integer,A> result;
 371 
 372         // set to dst if direct; otherwise set to substituted direct buffer
 373         private volatile ByteBuffer buf;
 374 
 375         ReadTask(ByteBuffer dst,
 376                  int pos,
 377                  int rem,
 378                  long position,
 379                  PendingFuture<Integer,A> result)
 380         {
 381             this.dst = dst;
 382             this.pos = pos;
 383             this.rem = rem;
 384             this.position = position;
 385             this.result = result;
 386         }
 387 
 388         void releaseBufferIfSubstituted() {
 389             if (buf != dst)
 390                 Util.releaseTemporaryDirectBuffer(buf);
 391         }
 392 
 393         void updatePosition(int bytesTransferred) {
 394             // if the I/O succeeded then adjust buffer position
 395             if (bytesTransferred > 0) {
 396                 if (buf == dst) {
 397                     try {
 398                         dst.position(pos + bytesTransferred);
 399                     } catch (IllegalArgumentException x) {
 400                         // someone has changed the position; ignore
 401                     }
 402                 } else {
 403                     // had to substitute direct buffer
 404                     buf.position(bytesTransferred).flip();
 405                     try {
 406                         dst.put(buf);
 407                     } catch (BufferOverflowException x) {
 408                         // someone has changed the position; ignore
 409                     }
 410                 }
 411             }
 412         }
 413 
 414         @Override
 415         public void run() {
 416             int n = -1;
 417             long overlapped = 0L;
 418             long address;
 419 
 420             // Substitute a native buffer if not direct
 421             if (dst instanceof DirectBuffer) {
 422                 buf = dst;
 423                 address = ((DirectBuffer)dst).address() + pos;
 424             } else {
 425                 buf = Util.getTemporaryDirectBuffer(rem);
 426                 address = ((DirectBuffer)buf).address();
 427             }
 428 
 429             boolean pending = false;
 430             try {
 431                 begin();
 432 
 433                 // allocate OVERLAPPED
 434                 overlapped = ioCache.add(result);
 435 
 436                 // initiate read
 437                 n = readFile(handle, address, rem, position, overlapped);
 438                 if (n == IOStatus.UNAVAILABLE) {
 439                     // I/O is pending
 440                     pending = true;
 441                     return;
 442                 } else if (n == IOStatus.EOF) {
 443                     result.setResult(n);
 444                 } else {
 445                     throw new InternalError("Unexpected result: " + n);
 446                 }
 447 
 448             } catch (Throwable x) {
 449                 // failed to initiate read
 450                 result.setFailure(toIOException(x));
 451             } finally {
 452                 if (!pending) {
 453                     // release resources
 454                     if (overlapped != 0L)
 455                         ioCache.remove(overlapped);
 456                     releaseBufferIfSubstituted();
 457                 }
 458                 end();
 459             }
 460 
 461             // invoke completion handler
 462             Invoker.invoke(result);
 463         }
 464 
 465         /**
 466          * Executed when the I/O has completed
 467          */
 468         @Override
 469         public void completed(int bytesTransferred, boolean canInvokeDirect) {
 470             updatePosition(bytesTransferred);
 471 
 472             // return direct buffer to cache if substituted
 473             releaseBufferIfSubstituted();
 474 
 475             // release waiters and invoke completion handler
 476             result.setResult(bytesTransferred);
 477             if (canInvokeDirect) {
 478                 Invoker.invokeUnchecked(result);
 479             } else {
 480                 Invoker.invoke(result);
 481             }
 482         }
 483 
 484         @Override
 485         public void failed(int error, IOException x) {
 486             // if EOF detected asynchronously then it is reported as error
 487             if (error == ERROR_HANDLE_EOF) {
 488                 completed(-1, false);
 489             } else {
 490                 // return direct buffer to cache if substituted
 491                 releaseBufferIfSubstituted();
 492 
 493                 // release waiters
 494                 if (isOpen()) {
 495                     result.setFailure(x);
 496                 } else {
 497                     result.setFailure(new AsynchronousCloseException());
 498                 }
 499                 Invoker.invoke(result);
 500             }
 501         }
 502     }
 503 
 504     @Override
 505     <A> Future<Integer> implRead(ByteBuffer dst,
 506                                  long position,
 507                                  A attachment,
 508                                  CompletionHandler<Integer,? super A> handler)
 509     {
 510         if (!reading)
 511             throw new NonReadableChannelException();
 512         if (position < 0)
 513             throw new IllegalArgumentException("Negative position");
 514         if (dst.isReadOnly())
 515             throw new IllegalArgumentException("Read-only buffer");
 516 
 517         // check if channel is closed
 518         if (!isOpen()) {
 519             Throwable exc = new ClosedChannelException();
 520             if (handler == null)
 521                 return CompletedFuture.withFailure(exc);
 522             Invoker.invoke(this, handler, attachment, null, exc);
 523             return null;
 524         }
 525 
 526         int pos = dst.position();
 527         int lim = dst.limit();
 528         assert (pos <= lim);
 529         int rem = (pos <= lim ? lim - pos : 0);
 530 
 531         // no space remaining
 532         if (rem == 0) {
 533             if (handler == null)
 534                 return CompletedFuture.withResult(0);
 535             Invoker.invoke(this, handler, attachment, 0, null);
 536             return null;
 537         }
 538 
 539         // create Future and task that initiates read
 540         PendingFuture<Integer,A> result =
 541             new PendingFuture<Integer,A>(this, handler, attachment);
 542         ReadTask<A> readTask = new ReadTask<A>(dst, pos, rem, position, result);
 543         result.setContext(readTask);
 544 
 545         // initiate I/O
 546         readTask.run();
 547         return result;
 548     }
 549 
 550     /**
 551      * Task that initiates write operation and handles completion result.
 552      */
 553     private class WriteTask<A> implements Runnable, Iocp.ResultHandler {
 554         private final ByteBuffer src;
 555         private final int pos, rem;     // buffer position/remaining
 556         private final long position;    // file position
 557         private final PendingFuture<Integer,A> result;
 558 
 559         // set to src if direct; otherwise set to substituted direct buffer
 560         private volatile ByteBuffer buf;
 561 
 562         WriteTask(ByteBuffer src,
 563                   int pos,
 564                   int rem,
 565                   long position,
 566                   PendingFuture<Integer,A> result)
 567         {
 568             this.src = src;
 569             this.pos = pos;
 570             this.rem = rem;
 571             this.position = position;
 572             this.result = result;
 573         }
 574 
 575         void releaseBufferIfSubstituted() {
 576             if (buf != src)
 577                 Util.releaseTemporaryDirectBuffer(buf);
 578         }
 579 
 580         void updatePosition(int bytesTransferred) {
 581             // if the I/O succeeded then adjust buffer position
 582             if (bytesTransferred > 0) {
 583                 try {
 584                     src.position(pos + bytesTransferred);
 585                 } catch (IllegalArgumentException x) {
 586                     // someone has changed the position
 587                 }
 588             }
 589         }
 590 
 591         @Override
 592         public void run() {
 593             int n = -1;
 594             long overlapped = 0L;
 595             long address;
 596 
 597             // Substitute a native buffer if not direct
 598             if (src instanceof DirectBuffer) {
 599                 buf = src;
 600                 address = ((DirectBuffer)src).address() + pos;
 601             } else {
 602                 buf = Util.getTemporaryDirectBuffer(rem);
 603                 buf.put(src);
 604                 buf.flip();
 605                 // temporarily restore position as we don't know how many bytes
 606                 // will be written
 607                 src.position(pos);
 608                 address = ((DirectBuffer)buf).address();
 609             }
 610 
 611             try {
 612                 begin();
 613 
 614                 // allocate an OVERLAPPED structure
 615                 overlapped = ioCache.add(result);
 616 
 617                 // initiate the write
 618                 n = writeFile(handle, address, rem, position, overlapped);
 619                 if (n == IOStatus.UNAVAILABLE) {
 620                     // I/O is pending
 621                     return;
 622                 } else {
 623                     throw new InternalError("Unexpected result: " + n);
 624                 }
 625 
 626             } catch (Throwable x) {
 627                 // failed to initiate read:
 628                 result.setFailure(toIOException(x));
 629 
 630                 // release resources
 631                 if (overlapped != 0L)
 632                     ioCache.remove(overlapped);
 633                 releaseBufferIfSubstituted();
 634 
 635             } finally {
 636                 end();
 637             }
 638 
 639             // invoke completion handler
 640             Invoker.invoke(result);
 641         }
 642 
 643         /**
 644          * Executed when the I/O has completed
 645          */
 646         @Override
 647         public void completed(int bytesTransferred, boolean canInvokeDirect) {
 648             updatePosition(bytesTransferred);
 649 
 650             // return direct buffer to cache if substituted
 651             releaseBufferIfSubstituted();
 652 
 653             // release waiters and invoke completion handler
 654             result.setResult(bytesTransferred);
 655             if (canInvokeDirect) {
 656                 Invoker.invokeUnchecked(result);
 657             } else {
 658                 Invoker.invoke(result);
 659             }
 660         }
 661 
 662         @Override
 663         public void failed(int error, IOException x) {
 664             // return direct buffer to cache if substituted
 665             releaseBufferIfSubstituted();
 666 
 667             // release waiters and invoker completion handler
 668             if (isOpen()) {
 669                 result.setFailure(x);
 670             } else {
 671                 result.setFailure(new AsynchronousCloseException());
 672             }
 673             Invoker.invoke(result);
 674         }
 675     }
 676 
 677     <A> Future<Integer> implWrite(ByteBuffer src,
 678                                   long position,
 679                                   A attachment,
 680                                   CompletionHandler<Integer,? super A> handler)
 681     {
 682         if (!writing)
 683             throw new NonWritableChannelException();
 684         if (position < 0)
 685             throw new IllegalArgumentException("Negative position");
 686 
 687         // check if channel is closed
 688         if (!isOpen()) {
 689            Throwable exc = new ClosedChannelException();
 690             if (handler == null)
 691                 return CompletedFuture.withFailure(exc);
 692             Invoker.invoke(this, handler, attachment, null, exc);
 693             return null;
 694         }
 695 
 696         int pos = src.position();
 697         int lim = src.limit();
 698         assert (pos <= lim);
 699         int rem = (pos <= lim ? lim - pos : 0);
 700 
 701         // nothing to write
 702         if (rem == 0) {
 703             if (handler == null)
 704                 return CompletedFuture.withResult(0);
 705             Invoker.invoke(this, handler, attachment, 0, null);
 706             return null;
 707         }
 708 
 709         // create Future and task to initiate write
 710         PendingFuture<Integer,A> result =
 711             new PendingFuture<Integer,A>(this, handler, attachment);
 712         WriteTask<A> writeTask = new WriteTask<A>(src, pos, rem, position, result);
 713         result.setContext(writeTask);
 714 
 715         // initiate I/O
 716         writeTask.run();
 717         return result;
 718     }
 719 
 720     // -- Native methods --
 721 
 722     private static native int readFile(long handle, long address, int len,
 723         long offset, long overlapped) throws IOException;
 724 
 725     private static native int writeFile(long handle, long address, int len,
 726         long offset, long overlapped) throws IOException;
 727 
 728     private static native int lockFile(long handle, long position, long size,
 729         boolean shared, long overlapped) throws IOException;
 730 
 731     static {
 732         IOUtil.load();
 733     }
 734 }