HADOOP-18521. isolating read buffer invocations on stream for testing

This should now be set up for unit tests to simulate the failure conditions

Change-Id: I25660e7067c95d0f3626d63e6f8f8156d2a1fc30
mkcsv enhancements

Author: steveloughran

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/AbfsInputStream.java

@@ -53,14 +53,16 @@
 
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.ONE_KB;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.STREAM_ID_LEN;
+import static org.apache.hadoop.fs.statistics.IOStatisticsLogging.ioStatisticsToPrettyString;
+import static org.apache.hadoop.fs.statistics.IOStatisticsSupport.retrieveIOStatistics;
 import static org.apache.hadoop.fs.statistics.impl.IOStatisticsBinding.emptyStatisticsStore;
 import static org.apache.hadoop.util.StringUtils.toLowerCase;
 
 /**
  * The AbfsInputStream for AbfsClient.
  */
 public class AbfsInputStream extends FSInputStream implements CanUnbuffer,
-        StreamCapabilities, IOStatisticsSource {
+        StreamCapabilities, IOStatisticsSource, ReadBufferStreamOperations {
   private static final Logger LOG = LoggerFactory.getLogger(AbfsInputStream.class);
   //  Footer size is set to qualify for both ORC and parquet files
   public static final int FOOTER_SIZE = 16 * ONE_KB;
@@ -523,7 +525,12 @@ private int readInternal(final long position, final byte[] b, final int offset,
     }
   }
 
-  int readRemote(long position, byte[] b, int offset, int length, TracingContext tracingContext) throws IOException {
+  @Override
+  public int readRemote(long position,
+      byte[] b,
+      int offset,
+      int length,
+      TracingContext tracingContext) throws IOException {
     if (position < 0) {
       throw new IllegalArgumentException("attempting to read from negative offset");
     }
@@ -709,12 +716,7 @@ public synchronized void close() throws IOException {
     }
   }
 
-  /**
-   * Is the stream closed?
-   * This must be thread safe as prefetch operations in
-   * different threads probe this before closure.
-   * @return true if the stream has been closed.
-   */
+  @Override
   @InterfaceAudience.Private
   public boolean isClosed() {
     return closed.get();
@@ -789,6 +791,7 @@ protected void setCachedSasToken(final CachedSASToken cachedSasToken) {
     this.cachedSasToken = cachedSasToken;
   }
 
+  @Override
   @VisibleForTesting
   public String getStreamID() {
     return inputStreamId;
@@ -867,7 +870,7 @@ public String toString() {
         ", bufferedPreadDisabled=" + bufferedPreadDisabled +
         ", firstRead=" + firstRead +
         ", fCursor=" + fCursor +
-        ", " + streamStatistics.toString() +
+        ", " + ioStatisticsToPrettyString(retrieveIOStatistics(streamStatistics)) +
         "}";
   }
 

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBuffer.java

@@ -23,16 +23,22 @@
 
 import org.apache.hadoop.fs.azurebfs.contracts.services.ReadBufferStatus;
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+import org.apache.hadoop.fs.statistics.DurationTracker;
 import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
 
 import static org.apache.hadoop.fs.azurebfs.contracts.services.ReadBufferStatus.READ_FAILED;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ACTIVE_PREFETCH_OPERATIONS;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_BLOCKS_USED;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_BYTES_DISCARDED;
 import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_BYTES_USED;
+import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS;
 
 class ReadBuffer {
 
-  private AbfsInputStream stream;
+  /**
+   * Stream operations.
+   */
+  private ReadBufferStreamOperations stream;
   private long offset;                   // offset within the file for the buffer
   private int length;                    // actual length, set after the buffer is filles
   private int requestedLength;           // requested length of the read
@@ -51,11 +57,11 @@ class ReadBuffer {
 
   private IOException errException = null;
 
-  public AbfsInputStream getStream() {
+  public ReadBufferStreamOperations getStream() {
     return stream;
   }
 
-  public void setStream(AbfsInputStream stream) {
+  public void setStream(ReadBufferStreamOperations stream) {
     this.stream = stream;
   }
 
@@ -212,7 +218,7 @@ public IOStatisticsStore getStreamIOStatistics() {
    * @param offset offset in buffer where copy began
    * @param bytesCopied bytes copied.
    */
-  public void dataConsumedByStream(int offset, int bytesCopied) {
+  void dataConsumedByStream(int offset, int bytesCopied) {
     setAnyByteConsumed(true);
     if (offset == 0) {
       setFirstByteConsumed(true);
@@ -225,40 +231,11 @@ public void dataConsumedByStream(int offset, int bytesCopied) {
     iostats.incrementCounter(STREAM_READ_PREFETCH_BYTES_USED, bytesCopied);
   }
 
-
-  /**
-   * The read completed
-   * @param result read result
-   * @param bytesActuallyRead the number of bytes actually read.
-   * @param timestampMillis timestamp of completion
-   */
-  void readCompleted(
-      final ReadBufferStatus result,
-      final int bytesActuallyRead,
-      final long timestampMillis) {
-    setStatus(status);
-    if (status == ReadBufferStatus.AVAILABLE) {
-      setLength(bytesActuallyRead);
-    } else {
-      setLength(0);
-    }
-    setTimeStamp(timestampMillis);
-  }
-
-  /**
-   * Has the read succeeded wth valid data
-   * @return true if there is data from a successful read
-   */
-  boolean readSucceededWithData() {
-    return status == ReadBufferStatus.AVAILABLE
-        && getLength() > 0;
-  }
-
   /**
    * The (completed) buffer was evicted; update stream statistics
    * as appropriate.
    */
-  public void evicted() {
+  void evicted() {
     if (getBufferindex() >= 0) {
       IOStatisticsStore iostats = getStreamIOStatistics();
       iostats.incrementCounter(STREAM_READ_PREFETCH_BYTES_DISCARDED, 1);
@@ -273,4 +250,28 @@ void releaseBuffer() {
     setBuffer(null);
     setBufferindex(-1);
   }
+
+
+  /**
+   * Prefetch started -update stream statistics.
+   */
+  void prefetchStarted()  {
+    getStreamIOStatistics().incrementGauge(STREAM_READ_ACTIVE_PREFETCH_OPERATIONS, 1);
+  }
+
+  /**
+   * Prefetch started -update stream statistics.
+   */
+  void prefetchFinished() {
+    getStreamIOStatistics().incrementGauge(STREAM_READ_ACTIVE_PREFETCH_OPERATIONS, -1);
+  }
+
+  /**
+   * Get a duration tracker for the prefetch.
+   * @return a duration tracker.
+   */
+  DurationTracker trackPrefetchOperation() {
+    return getStreamIOStatistics().trackDuration(STREAM_READ_PREFETCH_OPERATIONS);
+  }
+
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferManager.java

@@ -36,7 +36,6 @@
 import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
 import org.apache.hadoop.classification.VisibleForTesting;
 
-import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_ACTIVE_PREFETCH_OPERATIONS;
 import static org.apache.hadoop.util.Preconditions.checkState;
 
 /**
@@ -144,7 +143,7 @@ private ReadBufferManager() {
    * @param requestedOffset The offset in the file which shoukd be read
    * @param requestedLength The length to read
    */
-  void queueReadAhead(final AbfsInputStream stream, final long requestedOffset, final int requestedLength,
+  void queueReadAhead(final ReadBufferStreamOperations stream, final long requestedOffset, final int requestedLength,
                       TracingContext tracingContext) {
     if (LOGGER.isTraceEnabled()) {
       LOGGER.trace("Start Queueing readAhead for {} offset {} length {}",
@@ -341,18 +340,18 @@ private boolean evict(final ReadBuffer buf) {
     return true;
   }
 
-  private boolean isAlreadyQueued(final AbfsInputStream stream, final long requestedOffset) {
+  private boolean isAlreadyQueued(final ReadBufferStreamOperations stream, final long requestedOffset) {
     // returns true if any part of the buffer is already queued
     return (isInList(readAheadQueue, stream, requestedOffset)
         || isInList(inProgressList, stream, requestedOffset)
         || isInList(completedReadList, stream, requestedOffset));
   }
 
-  private boolean isInList(final Collection<ReadBuffer> list, final AbfsInputStream stream, final long requestedOffset) {
+  private boolean isInList(final Collection<ReadBuffer> list, final ReadBufferStreamOperations stream, final long requestedOffset) {
     return (getFromList(list, stream, requestedOffset) != null);
   }
 
-  private ReadBuffer getFromList(final Collection<ReadBuffer> list, final AbfsInputStream stream, final long requestedOffset) {
+  private ReadBuffer getFromList(final Collection<ReadBuffer> list, final ReadBufferStreamOperations stream, final long requestedOffset) {
     for (ReadBuffer buffer : list) {
       if (buffer.getStream() == stream) {
         if (buffer.getStatus() == ReadBufferStatus.AVAILABLE
@@ -374,7 +373,7 @@ private ReadBuffer getFromList(final Collection<ReadBuffer> list, final AbfsInpu
    * @param requestedOffset
    * @return
    */
-  private ReadBuffer getBufferFromCompletedQueue(final AbfsInputStream stream, final long requestedOffset) {
+  private ReadBuffer getBufferFromCompletedQueue(final ReadBufferStreamOperations stream, final long requestedOffset) {
     for (ReadBuffer buffer : completedReadList) {
       // Buffer is returned if the requestedOffset is at or above buffer's
       // offset but less than buffer's length or the actual requestedLength
@@ -389,7 +388,7 @@ private ReadBuffer getBufferFromCompletedQueue(final AbfsInputStream stream, fin
     return null;
   }
 
-  private void clearFromReadAheadQueue(final AbfsInputStream stream, final long requestedOffset) {
+  private void clearFromReadAheadQueue(final ReadBufferStreamOperations stream, final long requestedOffset) {
     ReadBuffer buffer = getFromList(readAheadQueue, stream, requestedOffset);
     if (buffer != null) {
       readAheadQueue.remove(buffer);
@@ -509,7 +508,7 @@ public synchronized void validateReadManagerState() {
 
   }
 
-  private int getBlockFromCompletedQueue(final AbfsInputStream stream, final long position, final int length,
+  private int getBlockFromCompletedQueue(final ReadBufferStreamOperations stream, final long position, final int length,
                                          final byte[] buffer) throws IOException {
     ReadBuffer buf = getBufferFromCompletedQueue(stream, position);
 
@@ -576,7 +575,7 @@ ReadBuffer getNextBlockToRead() throws InterruptedException {
     }
     validateReadManagerState();
     // update stream gauge.
-    buffer.getStreamIOStatistics().incrementGauge(STREAM_READ_ACTIVE_PREFETCH_OPERATIONS, 1);
+    buffer.prefetchStarted();
     return buffer;
   }
 
@@ -593,7 +592,7 @@ void doneReading(final ReadBuffer buffer, final ReadBufferStatus result, final i
           buffer.getStream().getPath(),  buffer.getOffset(), bytesActuallyRead, buffer);
     }
     // decrement counter.
-    buffer.getStreamIOStatistics().incrementGauge(STREAM_READ_ACTIVE_PREFETCH_OPERATIONS, -1);
+    buffer.prefetchFinished();
 
     try {
       synchronized (this) {
@@ -612,15 +611,18 @@ void doneReading(final ReadBuffer buffer, final ReadBufferStatus result, final i
 
         boolean shouldFreeBuffer = false;
         String freeBufferReason = "";
-        buffer.readCompleted(result, bytesActuallyRead, currentTimeMillis());
-        if (!buffer.readSucceededWithData()) {
-          // reead failed or there was no data, -the buffer can be returned to the free list.
+        if (result == ReadBufferStatus.AVAILABLE && bytesActuallyRead > 0) {
+          buffer.setStatus(ReadBufferStatus.AVAILABLE);
+          buffer.setLength(bytesActuallyRead);
+        } else {
+          // read failed or there was no data, -the buffer can be returned to the free list.
           shouldFreeBuffer = true;
           freeBufferReason = "failed read";
         }
-
         // completed list also contains FAILED read buffers
         // for sending exception message to clients.
+        buffer.setStatus(result);
+        buffer.setTimeStamp(currentTimeMillis());
         if (!buffer.isStreamClosed()) {
           // completed reads are added to the list.
           LOGGER.trace("Adding buffer to completed list {}", buffer);
@@ -708,7 +710,7 @@ void callTryEvict() {
    * still in use.
    * @param stream input stream.
    */
-  public synchronized void purgeBuffersForStream(AbfsInputStream stream) {
+  public synchronized void purgeBuffersForStream(ReadBufferStreamOperations stream) {
     LOGGER.debug("Purging stale buffers for AbfsInputStream {}/{}",
         stream.getStreamID(), stream.getPath());
 
@@ -737,7 +739,7 @@ public synchronized void purgeBuffersForStream(AbfsInputStream stream) {
    * @param list list of buffers like {@link this#completedReadList}
    *             or {@link this#inProgressList}.
    */
-  private int purgeList(AbfsInputStream stream, LinkedList<ReadBuffer> list) {
+  private int purgeList(ReadBufferStreamOperations stream, LinkedList<ReadBuffer> list) {
     int purged = 0;
     for (Iterator<ReadBuffer> it = list.iterator(); it.hasNext();) {
       ReadBuffer readBuffer = it.next();

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferStreamOperations.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.services;
+
+import java.io.IOException;
+
+import org.apache.hadoop.fs.azurebfs.utils.TracingContext;
+import org.apache.hadoop.fs.statistics.impl.IOStatisticsStore;
+
+/**
+ * Interface which is required for read buffer stream
+ * calls.
+ * Extracted from {@code AbfsInputStream} to make testing
+ * easier and to isolate what operations the read buffer
+ * makes of the streams using it.
+ */
+interface ReadBufferStreamOperations {
+
+  /**
+   * Read a block from the store.
+   * @param position position in file
+   * @param b destination buffer.
+   * @param offset offset in buffer
+   * @param length length of read
+   * @param tracingContext trace context
+   * @return count of bytes read.
+   * @throws IOException failure.
+   */
+  int readRemote(long position,
+      byte[] b,
+      int offset,
+      int length,
+      TracingContext tracingContext) throws IOException;
+
+  /**
+   * Is the stream closed?
+   * This must be thread safe as prefetch operations in
+   * different threads probe this before closure.
+   * @return true if the stream has been closed.
+   */
+  boolean isClosed();
+
+  String getStreamID();
+
+  IOStatisticsStore getIOStatistics();
+
+  /**
+   * Get the stream path as a string.
+   * @return path string.
+   */
+  String getPath();
+
+
+}

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/services/ReadBufferWorker.java

@@ -28,8 +28,6 @@
 import org.apache.hadoop.fs.azurebfs.contracts.services.ReadBufferStatus;
 import org.apache.hadoop.fs.statistics.DurationTracker;
 
-import static org.apache.hadoop.fs.statistics.StreamStatisticNames.STREAM_READ_PREFETCH_OPERATIONS;
-
 class ReadBufferWorker implements Runnable {
 
   private static final Logger LOGGER =
@@ -71,8 +69,7 @@ public void run() {
       }
       if (buffer != null) {
         // input stream is updated with count/duration of prefetching
-        DurationTracker tracker =
-            buffer.getStreamIOStatistics().trackDuration(STREAM_READ_PREFETCH_OPERATIONS);
+        DurationTracker tracker = buffer.trackPrefetchOperation();
         try {
           // do the actual read, from the file.
           LOGGER.trace("Reading {}", buffer);