Merge pull request #1 from chrislusf/master

sync

other/java/client/pom.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.3.5</version>
+    <version>1.3.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom.xml.deploy

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.3.5</version>
+    <version>1.3.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/pom_debug.xml

@@ -5,7 +5,7 @@
 
     <groupId>com.github.chrislusf</groupId>
     <artifactId>seaweedfs-client</artifactId>
-    <version>1.3.5</version>
+    <version>1.3.6</version>
 
     <parent>
         <groupId>org.sonatype.oss</groupId>

other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java

@@ -19,7 +19,7 @@ public class SeaweedRead {
 
     private static final Logger LOG = LoggerFactory.getLogger(SeaweedRead.class);
 
-    static ChunkCache chunkCache = new ChunkCache(1000);
+    static ChunkCache chunkCache = new ChunkCache(16);
 
     // returns bytesRead
     public static long read(FilerGrpcClient filerGrpcClient, List<VisibleInterval> visibleIntervals,

other/java/hdfs2/dependency-reduced-pom.xml

@@ -127,7 +127,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.3.5</seaweedfs.client.version>
+    <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
     <hadoop.version>2.9.2</hadoop.version>
   </properties>
 </project>

other/java/hdfs2/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.3.5</seaweedfs.client.version>
+        <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
         <hadoop.version>2.9.2</hadoop.version>
     </properties>
 

other/java/hdfs2/src/main/java/seaweed/hdfs/ReadBuffer.java

@@ -1,137 +0,0 @@
-/**
- * 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 seaweed.hdfs;
-
-import java.util.concurrent.CountDownLatch;
-
-class ReadBuffer {
-
-  private SeaweedInputStream 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
-  private byte[] buffer;                 // the buffer itself
-  private int bufferindex = -1;          // index in the buffers array in Buffer manager
-  private ReadBufferStatus status;             // status of the buffer
-  private CountDownLatch latch = null;   // signaled when the buffer is done reading, so any client
-  // waiting on this buffer gets unblocked
-
-  // fields to help with eviction logic
-  private long timeStamp = 0;  // tick at which buffer became available to read
-  private boolean isFirstByteConsumed = false;
-  private boolean isLastByteConsumed = false;
-  private boolean isAnyByteConsumed = false;
-
-  public SeaweedInputStream getStream() {
-    return stream;
-  }
-
-  public void setStream(SeaweedInputStream stream) {
-    this.stream = stream;
-  }
-
-  public long getOffset() {
-    return offset;
-  }
-
-  public void setOffset(long offset) {
-    this.offset = offset;
-  }
-
-  public int getLength() {
-    return length;
-  }
-
-  public void setLength(int length) {
-    this.length = length;
-  }
-
-  public int getRequestedLength() {
-    return requestedLength;
-  }
-
-  public void setRequestedLength(int requestedLength) {
-    this.requestedLength = requestedLength;
-  }
-
-  public byte[] getBuffer() {
-    return buffer;
-  }
-
-  public void setBuffer(byte[] buffer) {
-    this.buffer = buffer;
-  }
-
-  public int getBufferindex() {
-    return bufferindex;
-  }
-
-  public void setBufferindex(int bufferindex) {
-    this.bufferindex = bufferindex;
-  }
-
-  public ReadBufferStatus getStatus() {
-    return status;
-  }
-
-  public void setStatus(ReadBufferStatus status) {
-    this.status = status;
-  }
-
-  public CountDownLatch getLatch() {
-    return latch;
-  }
-
-  public void setLatch(CountDownLatch latch) {
-    this.latch = latch;
-  }
-
-  public long getTimeStamp() {
-    return timeStamp;
-  }
-
-  public void setTimeStamp(long timeStamp) {
-    this.timeStamp = timeStamp;
-  }
-
-  public boolean isFirstByteConsumed() {
-    return isFirstByteConsumed;
-  }
-
-  public void setFirstByteConsumed(boolean isFirstByteConsumed) {
-    this.isFirstByteConsumed = isFirstByteConsumed;
-  }
-
-  public boolean isLastByteConsumed() {
-    return isLastByteConsumed;
-  }
-
-  public void setLastByteConsumed(boolean isLastByteConsumed) {
-    this.isLastByteConsumed = isLastByteConsumed;
-  }
-
-  public boolean isAnyByteConsumed() {
-    return isAnyByteConsumed;
-  }
-
-  public void setAnyByteConsumed(boolean isAnyByteConsumed) {
-    this.isAnyByteConsumed = isAnyByteConsumed;
-  }
-
-}

other/java/hdfs2/src/main/java/seaweed/hdfs/ReadBufferManager.java

@@ -1,394 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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 seaweed.hdfs;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.Stack;
-import java.util.concurrent.CountDownLatch;
-
-/**
- * The Read Buffer Manager for Rest AbfsClient.
- */
-final class ReadBufferManager {
-  private static final Logger LOGGER = LoggerFactory.getLogger(ReadBufferManager.class);
-
-  private static final int NUM_BUFFERS = 16;
-  private static final int BLOCK_SIZE = 4 * 1024 * 1024;
-  private static final int NUM_THREADS = 8;
-  private static final int THRESHOLD_AGE_MILLISECONDS = 3000; // have to see if 3 seconds is a good threshold
-
-  private Thread[] threads = new Thread[NUM_THREADS];
-  private byte[][] buffers;    // array of byte[] buffers, to hold the data that is read
-  private Stack<Integer> freeList = new Stack<>();   // indices in buffers[] array that are available
-
-  private Queue<ReadBuffer> readAheadQueue = new LinkedList<>(); // queue of requests that are not picked up by any worker thread yet
-  private LinkedList<ReadBuffer> inProgressList = new LinkedList<>(); // requests being processed by worker threads
-  private LinkedList<ReadBuffer> completedReadList = new LinkedList<>(); // buffers available for reading
-  private static final ReadBufferManager BUFFER_MANAGER; // singleton, initialized in static initialization block
-
-  static {
-    BUFFER_MANAGER = new ReadBufferManager();
-    BUFFER_MANAGER.init();
-  }
-
-  static ReadBufferManager getBufferManager() {
-    return BUFFER_MANAGER;
-  }
-
-  private void init() {
-    buffers = new byte[NUM_BUFFERS][];
-    for (int i = 0; i < NUM_BUFFERS; i++) {
-      buffers[i] = new byte[BLOCK_SIZE];  // same buffers are reused. The byte array never goes back to GC
-      freeList.add(i);
-    }
-    for (int i = 0; i < NUM_THREADS; i++) {
-      Thread t = new Thread(new ReadBufferWorker(i));
-      t.setDaemon(true);
-      threads[i] = t;
-      t.setName("SeaweedFS-prefetch-" + i);
-      t.start();
-    }
-    ReadBufferWorker.UNLEASH_WORKERS.countDown();
-  }
-
-  // hide instance constructor
-  private ReadBufferManager() {
-  }
-
-
-  /*
-   *
-   *  SeaweedInputStream-facing methods
-   *
-   */
-
-
-  /**
-   * {@link SeaweedInputStream} calls this method to queue read-aheads.
-   *
-   * @param stream          The {@link SeaweedInputStream} for which to do the read-ahead
-   * @param requestedOffset The offset in the file which shoukd be read
-   * @param requestedLength The length to read
-   */
-  void queueReadAhead(final SeaweedInputStream stream, final long requestedOffset, final int requestedLength) {
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("Start Queueing readAhead for {} offset {} length {}",
-          stream.getPath(), requestedOffset, requestedLength);
-    }
-    ReadBuffer buffer;
-    synchronized (this) {
-      if (isAlreadyQueued(stream, requestedOffset)) {
-        return; // already queued, do not queue again
-      }
-      if (freeList.isEmpty() && !tryEvict()) {
-        return; // no buffers available, cannot queue anything
-      }
-
-      buffer = new ReadBuffer();
-      buffer.setStream(stream);
-      buffer.setOffset(requestedOffset);
-      buffer.setLength(0);
-      buffer.setRequestedLength(requestedLength);
-      buffer.setStatus(ReadBufferStatus.NOT_AVAILABLE);
-      buffer.setLatch(new CountDownLatch(1));
-
-      Integer bufferIndex = freeList.pop();  // will return a value, since we have checked size > 0 already
-
-      buffer.setBuffer(buffers[bufferIndex]);
-      buffer.setBufferindex(bufferIndex);
-      readAheadQueue.add(buffer);
-      notifyAll();
-    }
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("Done q-ing readAhead for file {} offset {} buffer idx {}",
-          stream.getPath(), requestedOffset, buffer.getBufferindex());
-    }
-  }
-
-
-  /**
-   * {@link SeaweedInputStream} calls this method read any bytes already available in a buffer (thereby saving a
-   * remote read). This returns the bytes if the data already exists in buffer. If there is a buffer that is reading
-   * the requested offset, then this method blocks until that read completes. If the data is queued in a read-ahead
-   * but not picked up by a worker thread yet, then it cancels that read-ahead and reports cache miss. This is because
-   * depending on worker thread availability, the read-ahead may take a while - the calling thread can do it's own
-   * read to get the data faster (copmared to the read waiting in queue for an indeterminate amount of time).
-   *
-   * @param stream   the file to read bytes for
-   * @param position the offset in the file to do a read for
-   * @param length   the length to read
-   * @param buffer   the buffer to read data into. Note that the buffer will be written into from offset 0.
-   * @return the number of bytes read
-   */
-  int getBlock(final SeaweedInputStream stream, final long position, final int length, final byte[] buffer) {
-    // not synchronized, so have to be careful with locking
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("getBlock for file {}  position {}  thread {}",
-          stream.getPath(), position, Thread.currentThread().getName());
-    }
-
-    waitForProcess(stream, position);
-
-    int bytesRead = 0;
-    synchronized (this) {
-      bytesRead = getBlockFromCompletedQueue(stream, position, length, buffer);
-    }
-    if (bytesRead > 0) {
-      if (LOGGER.isTraceEnabled()) {
-        LOGGER.trace("Done read from Cache for {} position {} length {}",
-            stream.getPath(), position, bytesRead);
-      }
-      return bytesRead;
-    }
-
-    // otherwise, just say we got nothing - calling thread can do its own read
-    return 0;
-  }
-
-  /*
-   *
-   *  Internal methods
-   *
-   */
-
-  private void waitForProcess(final SeaweedInputStream stream, final long position) {
-    ReadBuffer readBuf;
-    synchronized (this) {
-      clearFromReadAheadQueue(stream, position);
-      readBuf = getFromList(inProgressList, stream, position);
-    }
-    if (readBuf != null) {         // if in in-progress queue, then block for it
-      try {
-        if (LOGGER.isTraceEnabled()) {
-          LOGGER.trace("got a relevant read buffer for file {} offset {} buffer idx {}",
-              stream.getPath(), readBuf.getOffset(), readBuf.getBufferindex());
-        }
-        readBuf.getLatch().await();  // blocking wait on the caller stream's thread
-        // Note on correctness: readBuf gets out of inProgressList only in 1 place: after worker thread
-        // is done processing it (in doneReading). There, the latch is set after removing the buffer from
-        // inProgressList. So this latch is safe to be outside the synchronized block.
-        // Putting it in synchronized would result in a deadlock, since this thread would be holding the lock
-        // while waiting, so no one will be able to  change any state. If this becomes more complex in the future,
-        // then the latch cane be removed and replaced with wait/notify whenever inProgressList is touched.
-      } catch (InterruptedException ex) {
-        Thread.currentThread().interrupt();
-      }
-      if (LOGGER.isTraceEnabled()) {
-        LOGGER.trace("latch done for file {} buffer idx {} length {}",
-            stream.getPath(), readBuf.getBufferindex(), readBuf.getLength());
-      }
-    }
-  }
-
-  /**
-   * If any buffer in the completedlist can be reclaimed then reclaim it and return the buffer to free list.
-   * The objective is to find just one buffer - there is no advantage to evicting more than one.
-   *
-   * @return whether the eviction succeeeded - i.e., were we able to free up one buffer
-   */
-  private synchronized boolean tryEvict() {
-    ReadBuffer nodeToEvict = null;
-    if (completedReadList.size() <= 0) {
-      return false;  // there are no evict-able buffers
-    }
-
-    // first, try buffers where all bytes have been consumed (approximated as first and last bytes consumed)
-    for (ReadBuffer buf : completedReadList) {
-      if (buf.isFirstByteConsumed() && buf.isLastByteConsumed()) {
-        nodeToEvict = buf;
-        break;
-      }
-    }
-    if (nodeToEvict != null) {
-      return evict(nodeToEvict);
-    }
-
-    // next, try buffers where any bytes have been consumed (may be a bad idea? have to experiment and see)
-    for (ReadBuffer buf : completedReadList) {
-      if (buf.isAnyByteConsumed()) {
-        nodeToEvict = buf;
-        break;
-      }
-    }
-
-    if (nodeToEvict != null) {
-      return evict(nodeToEvict);
-    }
-
-    // next, try any old nodes that have not been consumed
-    long earliestBirthday = Long.MAX_VALUE;
-    for (ReadBuffer buf : completedReadList) {
-      if (buf.getTimeStamp() < earliestBirthday) {
-        nodeToEvict = buf;
-        earliestBirthday = buf.getTimeStamp();
-      }
-    }
-    if ((currentTimeMillis() - earliestBirthday > THRESHOLD_AGE_MILLISECONDS) && (nodeToEvict != null)) {
-      return evict(nodeToEvict);
-    }
-
-    // nothing can be evicted
-    return false;
-  }
-
-  private boolean evict(final ReadBuffer buf) {
-    freeList.push(buf.getBufferindex());
-    completedReadList.remove(buf);
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("Evicting buffer idx {}; was used for file {} offset {} length {}",
-          buf.getBufferindex(), buf.getStream().getPath(), buf.getOffset(), buf.getLength());
-    }
-    return true;
-  }
-
-  private boolean isAlreadyQueued(final SeaweedInputStream 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 SeaweedInputStream stream, final long requestedOffset) {
-    return (getFromList(list, stream, requestedOffset) != null);
-  }
-
-  private ReadBuffer getFromList(final Collection<ReadBuffer> list, final SeaweedInputStream stream, final long requestedOffset) {
-    for (ReadBuffer buffer : list) {
-      if (buffer.getStream() == stream) {
-        if (buffer.getStatus() == ReadBufferStatus.AVAILABLE
-            && requestedOffset >= buffer.getOffset()
-            && requestedOffset < buffer.getOffset() + buffer.getLength()) {
-          return buffer;
-        } else if (requestedOffset >= buffer.getOffset()
-            && requestedOffset < buffer.getOffset() + buffer.getRequestedLength()) {
-          return buffer;
-        }
-      }
-    }
-    return null;
-  }
-
-  private void clearFromReadAheadQueue(final SeaweedInputStream stream, final long requestedOffset) {
-    ReadBuffer buffer = getFromList(readAheadQueue, stream, requestedOffset);
-    if (buffer != null) {
-      readAheadQueue.remove(buffer);
-      notifyAll();   // lock is held in calling method
-      freeList.push(buffer.getBufferindex());
-    }
-  }
-
-  private int getBlockFromCompletedQueue(final SeaweedInputStream stream, final long position, final int length,
-                                         final byte[] buffer) {
-    ReadBuffer buf = getFromList(completedReadList, stream, position);
-    if (buf == null || position >= buf.getOffset() + buf.getLength()) {
-      return 0;
-    }
-    int cursor = (int) (position - buf.getOffset());
-    int availableLengthInBuffer = buf.getLength() - cursor;
-    int lengthToCopy = Math.min(length, availableLengthInBuffer);
-    System.arraycopy(buf.getBuffer(), cursor, buffer, 0, lengthToCopy);
-    if (cursor == 0) {
-      buf.setFirstByteConsumed(true);
-    }
-    if (cursor + lengthToCopy == buf.getLength()) {
-      buf.setLastByteConsumed(true);
-    }
-    buf.setAnyByteConsumed(true);
-    return lengthToCopy;
-  }
-
-  /*
-   *
-   *  ReadBufferWorker-thread-facing methods
-   *
-   */
-
-  /**
-   * ReadBufferWorker thread calls this to get the next buffer that it should work on.
-   *
-   * @return {@link ReadBuffer}
-   * @throws InterruptedException if thread is interrupted
-   */
-  ReadBuffer getNextBlockToRead() throws InterruptedException {
-    ReadBuffer buffer = null;
-    synchronized (this) {
-      //buffer = readAheadQueue.take();  // blocking method
-      while (readAheadQueue.size() == 0) {
-        wait();
-      }
-      buffer = readAheadQueue.remove();
-      notifyAll();
-      if (buffer == null) {
-        return null;            // should never happen
-      }
-      buffer.setStatus(ReadBufferStatus.READING_IN_PROGRESS);
-      inProgressList.add(buffer);
-    }
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("ReadBufferWorker picked file {} for offset {}",
-          buffer.getStream().getPath(), buffer.getOffset());
-    }
-    return buffer;
-  }
-
-  /**
-   * ReadBufferWorker thread calls this method to post completion.
-   *
-   * @param buffer            the buffer whose read was completed
-   * @param result            the {@link ReadBufferStatus} after the read operation in the worker thread
-   * @param bytesActuallyRead the number of bytes that the worker thread was actually able to read
-   */
-  void doneReading(final ReadBuffer buffer, final ReadBufferStatus result, final int bytesActuallyRead) {
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("ReadBufferWorker completed file {} for offset {} bytes {}",
-          buffer.getStream().getPath(),  buffer.getOffset(), bytesActuallyRead);
-    }
-    synchronized (this) {
-      inProgressList.remove(buffer);
-      if (result == ReadBufferStatus.AVAILABLE && bytesActuallyRead > 0) {
-        buffer.setStatus(ReadBufferStatus.AVAILABLE);
-        buffer.setTimeStamp(currentTimeMillis());
-        buffer.setLength(bytesActuallyRead);
-        completedReadList.add(buffer);
-      } else {
-        freeList.push(buffer.getBufferindex());
-        // buffer should go out of scope after the end of the calling method in ReadBufferWorker, and eligible for GC
-      }
-    }
-    //outside the synchronized, since anyone receiving a wake-up from the latch must see safe-published results
-    buffer.getLatch().countDown(); // wake up waiting threads (if any)
-  }
-
-  /**
-   * Similar to System.currentTimeMillis, except implemented with System.nanoTime().
-   * System.currentTimeMillis can go backwards when system clock is changed (e.g., with NTP time synchronization),
-   * making it unsuitable for measuring time intervals. nanotime is strictly monotonically increasing per CPU core.
-   * Note: it is not monotonic across Sockets, and even within a CPU, its only the
-   * more recent parts which share a clock across all cores.
-   *
-   * @return current time in milliseconds
-   */
-  private long currentTimeMillis() {
-    return System.nanoTime() / 1000 / 1000;
-  }
-}

other/java/hdfs2/src/main/java/seaweed/hdfs/ReadBufferStatus.java

@@ -1,29 +0,0 @@
-/**
- * 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 seaweed.hdfs;
-
-/**
- * The ReadBufferStatus for Rest AbfsClient
- */
-public enum ReadBufferStatus {
-  NOT_AVAILABLE,  // buffers sitting in readaheadqueue have this stats
-  READING_IN_PROGRESS,  // reading is in progress on this buffer. Buffer should be in inProgressList
-  AVAILABLE,  // data is available in buffer. It should be in completedList
-  READ_FAILED  // read completed, but failed.
-}

other/java/hdfs2/src/main/java/seaweed/hdfs/ReadBufferWorker.java

@@ -1,70 +0,0 @@
-/**
- * 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 seaweed.hdfs;
-
-import java.util.concurrent.CountDownLatch;
-
-class ReadBufferWorker implements Runnable {
-
-  protected static final CountDownLatch UNLEASH_WORKERS = new CountDownLatch(1);
-  private int id;
-
-  ReadBufferWorker(final int id) {
-    this.id = id;
-  }
-
-  /**
-   * return the ID of ReadBufferWorker.
-   */
-  public int getId() {
-    return this.id;
-  }
-
-  /**
-   * Waits until a buffer becomes available in ReadAheadQueue.
-   * Once a buffer becomes available, reads the file specified in it and then posts results back to buffer manager.
-   * Rinse and repeat. Forever.
-   */
-  public void run() {
-    try {
-      UNLEASH_WORKERS.await();
-    } catch (InterruptedException ex) {
-      Thread.currentThread().interrupt();
-    }
-    ReadBufferManager bufferManager = ReadBufferManager.getBufferManager();
-    ReadBuffer buffer;
-    while (true) {
-      try {
-        buffer = bufferManager.getNextBlockToRead();   // blocks, until a buffer is available for this thread
-      } catch (InterruptedException ex) {
-        Thread.currentThread().interrupt();
-        return;
-      }
-      if (buffer != null) {
-        try {
-          // do the actual read, from the file.
-          int bytesRead = buffer.getStream().readRemote(buffer.getOffset(), buffer.getBuffer(), 0, buffer.getRequestedLength());
-          bufferManager.doneReading(buffer, ReadBufferStatus.AVAILABLE, bytesRead);  // post result back to ReadBufferManager
-        } catch (Exception ex) {
-          bufferManager.doneReading(buffer, ReadBufferStatus.READ_FAILED, 0);
-        }
-      }
-    }
-  }
-}

other/java/hdfs3/src/main/java/seaweed/hdfs/ReadBufferStatus.java → other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedAbstractFileSystem.java

@@ -18,12 +18,18 @@
 
 package seaweed.hdfs;
 
-/**
- * The ReadBufferStatus for Rest AbfsClient
- */
-public enum ReadBufferStatus {
-  NOT_AVAILABLE,  // buffers sitting in readaheadqueue have this stats
-  READING_IN_PROGRESS,  // reading is in progress on this buffer. Buffer should be in inProgressList
-  AVAILABLE,  // data is available in buffer. It should be in completedList
-  READ_FAILED  // read completed, but failed.
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.DelegateToFileSystem;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+
+public class SeaweedAbstractFileSystem extends DelegateToFileSystem {
+
+    SeaweedAbstractFileSystem(final URI uri, final Configuration conf)
+            throws IOException, URISyntaxException {
+        super(uri, new SeaweedFileSystem(), conf, "seaweedfs", false);
+    }
+
 }

other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java

@@ -10,6 +10,7 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Progressable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import seaweedfs.client.FilerProto;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
@@ -22,7 +23,7 @@ import java.util.Map;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.IO_FILE_BUFFER_SIZE_KEY;
 
-public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
+public class SeaweedFileSystem extends FileSystem {
 
     public static final int FS_SEAWEED_DEFAULT_PORT = 8888;
     public static final String FS_SEAWEED_FILER_HOST = "fs.seaweed.filer.host";
@@ -144,7 +145,7 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
     }
 
     @Override
-    public boolean rename(Path src, Path dst) {
+    public boolean rename(Path src, Path dst) throws IOException {
 
         LOG.debug("rename path: {} => {}", src, dst);
 
@@ -155,12 +156,13 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
         if (src.equals(dst)) {
             return true;
         }
-        FileStatus dstFileStatus = getFileStatus(dst);
+        FilerProto.Entry entry = seaweedFileSystemStore.lookupEntry(dst);
 
-        String sourceFileName = src.getName();
         Path adjustedDst = dst;
 
-        if (dstFileStatus != null) {
+        if (entry != null) {
+            FileStatus dstFileStatus = getFileStatus(dst);
+            String sourceFileName = src.getName();
             if (!dstFileStatus.isDirectory()) {
                 return false;
             }
@@ -175,18 +177,20 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
     }
 
     @Override
-    public boolean delete(Path path, boolean recursive) {
+    public boolean delete(Path path, boolean recursive) throws IOException {
 
         LOG.debug("delete path: {} recursive:{}", path, recursive);
 
         path = qualify(path);
 
-        FileStatus fileStatus = getFileStatus(path);
+        FilerProto.Entry entry = seaweedFileSystemStore.lookupEntry(path);
 
-        if (fileStatus == null) {
+        if (entry == null) {
             return true;
         }
 
+        FileStatus fileStatus = getFileStatus(path);
+
         return seaweedFileSystemStore.deleteEntries(path, fileStatus.isDirectory(), recursive);
 
     }
@@ -222,9 +226,9 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
 
         path = qualify(path);
 
-        FileStatus fileStatus = getFileStatus(path);
+        FilerProto.Entry entry = seaweedFileSystemStore.lookupEntry(path);
 
-        if (fileStatus == null) {
+        if (entry == null) {
 
             UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
             return seaweedFileSystemStore.createDirectory(path, currentUser,
@@ -233,6 +237,8 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
 
         }
 
+        FileStatus fileStatus = getFileStatus(path);
+
         if (fileStatus.isDirectory()) {
             return true;
         } else {
@@ -241,7 +247,7 @@ public class SeaweedFileSystem extends org.apache.hadoop.fs.FileSystem {
     }
 
     @Override
-    public FileStatus getFileStatus(Path path) {
+    public FileStatus getFileStatus(Path path) throws IOException {
 
         LOG.debug("getFileStatus path: {}", path);
 

other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java

@@ -61,7 +61,7 @@ public class SeaweedFileSystemStore {
         );
     }
 
-    public FileStatus[] listEntries(final Path path) {
+    public FileStatus[] listEntries(final Path path) throws IOException {
         LOG.debug("listEntries path: {}", path);
 
         FileStatus pathStatus = getFileStatus(path);
@@ -89,11 +89,11 @@ public class SeaweedFileSystemStore {
 
     }
 
-    public FileStatus getFileStatus(final Path path) {
+    public FileStatus getFileStatus(final Path path) throws IOException {
 
         FilerProto.Entry entry = lookupEntry(path);
         if (entry == null) {
-            return null;
+            throw new FileNotFoundException("File does not exist: " + path);
         }
         LOG.debug("doGetFileStatus path:{} entry:{}", path, entry);
 
@@ -136,7 +136,7 @@ public class SeaweedFileSystemStore {
             modification_time, access_time, permission, owner, group, null, path);
     }
 
-    private FilerProto.Entry lookupEntry(Path path) {
+    public FilerProto.Entry lookupEntry(Path path) {
 
         return filerClient.lookupEntry(getParentDirectory(path), path.getName());
 
@@ -212,7 +212,6 @@ public class SeaweedFileSystemStore {
 
         LOG.debug("openFileForRead path:{} bufferSize:{}", path, bufferSize);
 
-        int readAheadQueueDepth = 2;
         FilerProto.Entry entry = lookupEntry(path);
 
         if (entry == null) {
@@ -223,8 +222,7 @@ public class SeaweedFileSystemStore {
             statistics,
             path.toUri().getPath(),
             entry,
-            bufferSize,
-            readAheadQueueDepth);
+            bufferSize);
     }
 
     public void setOwner(Path path, String owner, String group) {

other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedInputStream.java

@@ -27,16 +27,9 @@ public class SeaweedInputStream extends FSInputStream {
     private final List<SeaweedRead.VisibleInterval> visibleIntervalList;
     private final long contentLength;
     private final int bufferSize; // default buffer size
-    private final int readAheadQueueDepth;         // initialized in constructor
-    private final boolean readAheadEnabled; // whether enable readAhead;
 
-    private byte[] buffer = null;            // will be initialized on first use
+    private long position = 0;  // cursor of the file
 
-    private long fCursor = 0;  // cursor of buffer within file - offset of next byte to read from remote server
-    private long fCursorAfterLastRead = -1;
-    private int bCursor = 0;   // cursor of read within buffer - offset of next byte to be returned from buffer
-    private int limit = 0;     // offset of next byte to be read into buffer from service (i.e., upper marker+1
-    //                                                      of valid bytes in buffer)
     private boolean closed = false;
 
     public SeaweedInputStream(
@@ -44,16 +37,13 @@ public class SeaweedInputStream extends FSInputStream {
             final Statistics statistics,
             final String path,
             final FilerProto.Entry entry,
-        final int bufferSize,
-        final int readAheadQueueDepth) {
+            final int bufferSize) {
         this.filerGrpcClient = filerGrpcClient;
         this.statistics = statistics;
         this.path = path;
         this.entry = entry;
         this.contentLength = SeaweedRead.totalSize(entry.getChunksList());
         this.bufferSize = bufferSize;
-        this.readAheadQueueDepth = (readAheadQueueDepth >= 0) ? readAheadQueueDepth : Runtime.getRuntime().availableProcessors();
-        this.readAheadEnabled = true;
 
         this.visibleIntervalList = SeaweedRead.nonOverlappingVisibleIntervals(entry.getChunksList());
 
@@ -78,122 +68,7 @@ public class SeaweedInputStream extends FSInputStream {
 
     @Override
     public synchronized int read(final byte[] b, final int off, final int len) throws IOException {
-        int currentOff = off;
-        int currentLen = len;
-        int lastReadBytes;
-        int totalReadBytes = 0;
-        do {
-            lastReadBytes = readOneBlock(b, currentOff, currentLen);
-            if (lastReadBytes > 0) {
-                currentOff += lastReadBytes;
-                currentLen -= lastReadBytes;
-                totalReadBytes += lastReadBytes;
-            }
-            if (currentLen <= 0 || currentLen > b.length - currentOff) {
-                break;
-            }
-        } while (lastReadBytes > 0);
-        return totalReadBytes > 0 ? totalReadBytes : lastReadBytes;
-    }
-
-    private int readOneBlock(final byte[] b, final int off, final int len) throws IOException {
-        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-
-        Preconditions.checkNotNull(b);
-
-        if (len == 0) {
-            return 0;
-        }
-
-        if (this.available() == 0) {
-            return -1;
-        }
-
-        if (off < 0 || len < 0 || len > b.length - off) {
-            throw new IndexOutOfBoundsException();
-        }
-
-        //If buffer is empty, then fill the buffer.
-        if (bCursor == limit) {
-            //If EOF, then return -1
-            if (fCursor >= contentLength) {
-                return -1;
-            }
-
-            long bytesRead = 0;
-            //reset buffer to initial state - i.e., throw away existing data
-            bCursor = 0;
-            limit = 0;
-            if (buffer == null) {
-                buffer = new byte[bufferSize];
-            }
-
-            // Enable readAhead when reading sequentially
-            if (-1 == fCursorAfterLastRead || fCursorAfterLastRead == fCursor || b.length >= bufferSize) {
-                bytesRead = readInternal(fCursor, buffer, 0, bufferSize, false);
-            } else {
-                bytesRead = readInternal(fCursor, buffer, 0, b.length, true);
-            }
-
-            if (bytesRead == -1) {
-                return -1;
-            }
 
-            limit += bytesRead;
-            fCursor += bytesRead;
-            fCursorAfterLastRead = fCursor;
-        }
-
-        //If there is anything in the buffer, then return lesser of (requested bytes) and (bytes in buffer)
-        //(bytes returned may be less than requested)
-        int bytesRemaining = limit - bCursor;
-        int bytesToRead = Math.min(len, bytesRemaining);
-        System.arraycopy(buffer, bCursor, b, off, bytesToRead);
-        bCursor += bytesToRead;
-        if (statistics != null) {
-            statistics.incrementBytesRead(bytesToRead);
-        }
-        return bytesToRead;
-    }
-
-
-    private int readInternal(final long position, final byte[] b, final int offset, final int length,
-                             final boolean bypassReadAhead) throws IOException {
-        if (readAheadEnabled && !bypassReadAhead) {
-            // try reading from read-ahead
-            if (offset != 0) {
-                throw new IllegalArgumentException("readahead buffers cannot have non-zero buffer offsets");
-            }
-            int receivedBytes;
-
-            // queue read-aheads
-            int numReadAheads = this.readAheadQueueDepth;
-            long nextSize;
-            long nextOffset = position;
-            while (numReadAheads > 0 && nextOffset < contentLength) {
-                nextSize = Math.min((long) bufferSize, contentLength - nextOffset);
-                ReadBufferManager.getBufferManager().queueReadAhead(this, nextOffset, (int) nextSize);
-                nextOffset = nextOffset + nextSize;
-                numReadAheads--;
-            }
-
-            // try reading from buffers first
-            receivedBytes = ReadBufferManager.getBufferManager().getBlock(this, position, length, b);
-            if (receivedBytes > 0) {
-                return receivedBytes;
-            }
-
-            // got nothing from read-ahead, do our own read now
-            receivedBytes = readRemote(position, b, offset, length);
-            return receivedBytes;
-        } else {
-            return readRemote(position, b, offset, length);
-        }
-    }
-
-    int readRemote(long position, byte[] b, int offset, int length) throws IOException {
         if (position < 0) {
             throw new IllegalArgumentException("attempting to read from negative offset");
         }
@@ -203,21 +78,30 @@ public class SeaweedInputStream extends FSInputStream {
         if (b == null) {
             throw new IllegalArgumentException("null byte array passed in to read() method");
         }
-        if (offset >= b.length) {
+        if (off >= b.length) {
             throw new IllegalArgumentException("offset greater than length of array");
         }
-        if (length < 0) {
+        if (len < 0) {
             throw new IllegalArgumentException("requested read length is less than zero");
         }
-        if (length > (b.length - offset)) {
+        if (len > (b.length - off)) {
             throw new IllegalArgumentException("requested read length is more than will fit after requested offset in buffer");
         }
 
-        long bytesRead = SeaweedRead.read(filerGrpcClient, visibleIntervalList, position, b, offset, length);
+        long bytesRead = SeaweedRead.read(this.filerGrpcClient, this.visibleIntervalList, this.position, b, off, len);
         if (bytesRead > Integer.MAX_VALUE) {
             throw new IOException("Unexpected Content-Length");
         }
+
+        if (bytesRead > 0) {
+            this.position += bytesRead;
+            if (statistics != null) {
+                statistics.incrementBytesRead(bytesRead);
+            }
+        }
+
         return (int)bytesRead;
+
     }
 
     /**
@@ -239,17 +123,8 @@ public class SeaweedInputStream extends FSInputStream {
             throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
         }
 
-        if (n >= fCursor - limit && n <= fCursor) { // within buffer
-            bCursor = (int) (n - (fCursor - limit));
-            return;
-        }
-
-        // next read will read from here
-        fCursor = n;
+        this.position = n;
 
-        //invalidate buffer
-        limit = 0;
-        bCursor = 0;
     }
 
     @Override
@@ -257,20 +132,19 @@ public class SeaweedInputStream extends FSInputStream {
         if (closed) {
             throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
         }
-        long currentPos = getPos();
-        if (currentPos == contentLength) {
+        if (this.position == contentLength) {
             if (n > 0) {
                 throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
             }
         }
-        long newPos = currentPos + n;
+        long newPos = this.position + n;
         if (newPos < 0) {
             newPos = 0;
-            n = newPos - currentPos;
+            n = newPos - this.position;
         }
         if (newPos > contentLength) {
             newPos = contentLength;
-            n = newPos - currentPos;
+            n = newPos - this.position;
         }
         seek(newPos);
         return n;
@@ -321,7 +195,7 @@ public class SeaweedInputStream extends FSInputStream {
         if (closed) {
             throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
         }
-        return fCursor - limit + bCursor;
+        return position;
     }
 
     /**
@@ -338,7 +212,6 @@ public class SeaweedInputStream extends FSInputStream {
     @Override
     public synchronized void close() throws IOException {
         closed = true;
-        buffer = null; // de-reference the buffer so it can be GC'ed sooner
     }
 
     /**

other/java/hdfs3/dependency-reduced-pom.xml

@@ -127,7 +127,7 @@
     </snapshotRepository>
   </distributionManagement>
   <properties>
-    <seaweedfs.client.version>1.3.5</seaweedfs.client.version>
+    <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
     <hadoop.version>3.1.1</hadoop.version>
   </properties>
 </project>

other/java/hdfs3/pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
 
     <properties>
-        <seaweedfs.client.version>1.3.5</seaweedfs.client.version>
+        <seaweedfs.client.version>1.3.6</seaweedfs.client.version>
         <hadoop.version>3.1.1</hadoop.version>
     </properties>
 

other/java/hdfs3/src/main/java/seaweed/hdfs/ReadBuffer.java

@@ -1,137 +0,0 @@
-/**
- * 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 seaweed.hdfs;
-
-import java.util.concurrent.CountDownLatch;
-
-class ReadBuffer {
-
-  private SeaweedInputStream 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
-  private byte[] buffer;                 // the buffer itself
-  private int bufferindex = -1;          // index in the buffers array in Buffer manager
-  private ReadBufferStatus status;             // status of the buffer
-  private CountDownLatch latch = null;   // signaled when the buffer is done reading, so any client
-  // waiting on this buffer gets unblocked
-
-  // fields to help with eviction logic
-  private long timeStamp = 0;  // tick at which buffer became available to read
-  private boolean isFirstByteConsumed = false;
-  private boolean isLastByteConsumed = false;
-  private boolean isAnyByteConsumed = false;
-
-  public SeaweedInputStream getStream() {
-    return stream;
-  }
-
-  public void setStream(SeaweedInputStream stream) {
-    this.stream = stream;
-  }
-
-  public long getOffset() {
-    return offset;
-  }
-
-  public void setOffset(long offset) {
-    this.offset = offset;
-  }
-
-  public int getLength() {
-    return length;
-  }
-
-  public void setLength(int length) {
-    this.length = length;
-  }
-
-  public int getRequestedLength() {
-    return requestedLength;
-  }
-
-  public void setRequestedLength(int requestedLength) {
-    this.requestedLength = requestedLength;
-  }
-
-  public byte[] getBuffer() {
-    return buffer;
-  }
-
-  public void setBuffer(byte[] buffer) {
-    this.buffer = buffer;
-  }
-
-  public int getBufferindex() {
-    return bufferindex;
-  }
-
-  public void setBufferindex(int bufferindex) {
-    this.bufferindex = bufferindex;
-  }
-
-  public ReadBufferStatus getStatus() {
-    return status;
-  }
-
-  public void setStatus(ReadBufferStatus status) {
-    this.status = status;
-  }
-
-  public CountDownLatch getLatch() {
-    return latch;
-  }
-
-  public void setLatch(CountDownLatch latch) {
-    this.latch = latch;
-  }
-
-  public long getTimeStamp() {
-    return timeStamp;
-  }
-
-  public void setTimeStamp(long timeStamp) {
-    this.timeStamp = timeStamp;
-  }
-
-  public boolean isFirstByteConsumed() {
-    return isFirstByteConsumed;
-  }
-
-  public void setFirstByteConsumed(boolean isFirstByteConsumed) {
-    this.isFirstByteConsumed = isFirstByteConsumed;
-  }
-
-  public boolean isLastByteConsumed() {
-    return isLastByteConsumed;
-  }
-
-  public void setLastByteConsumed(boolean isLastByteConsumed) {
-    this.isLastByteConsumed = isLastByteConsumed;
-  }
-
-  public boolean isAnyByteConsumed() {
-    return isAnyByteConsumed;
-  }
-
-  public void setAnyByteConsumed(boolean isAnyByteConsumed) {
-    this.isAnyByteConsumed = isAnyByteConsumed;
-  }
-
-}

other/java/hdfs3/src/main/java/seaweed/hdfs/ReadBufferManager.java

@@ -1,394 +0,0 @@
-/**
- * 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
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * 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 seaweed.hdfs;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.Queue;
-import java.util.Stack;
-import java.util.concurrent.CountDownLatch;
-
-/**
- * The Read Buffer Manager for Rest AbfsClient.
- */
-final class ReadBufferManager {
-  private static final Logger LOGGER = LoggerFactory.getLogger(ReadBufferManager.class);
-
-  private static final int NUM_BUFFERS = 16;
-  private static final int BLOCK_SIZE = 4 * 1024 * 1024;
-  private static final int NUM_THREADS = 8;
-  private static final int THRESHOLD_AGE_MILLISECONDS = 3000; // have to see if 3 seconds is a good threshold
-
-  private Thread[] threads = new Thread[NUM_THREADS];
-  private byte[][] buffers;    // array of byte[] buffers, to hold the data that is read
-  private Stack<Integer> freeList = new Stack<>();   // indices in buffers[] array that are available
-
-  private Queue<ReadBuffer> readAheadQueue = new LinkedList<>(); // queue of requests that are not picked up by any worker thread yet
-  private LinkedList<ReadBuffer> inProgressList = new LinkedList<>(); // requests being processed by worker threads
-  private LinkedList<ReadBuffer> completedReadList = new LinkedList<>(); // buffers available for reading
-  private static final ReadBufferManager BUFFER_MANAGER; // singleton, initialized in static initialization block
-
-  static {
-    BUFFER_MANAGER = new ReadBufferManager();
-    BUFFER_MANAGER.init();
-  }
-
-  static ReadBufferManager getBufferManager() {
-    return BUFFER_MANAGER;
-  }
-
-  private void init() {
-    buffers = new byte[NUM_BUFFERS][];
-    for (int i = 0; i < NUM_BUFFERS; i++) {
-      buffers[i] = new byte[BLOCK_SIZE];  // same buffers are reused. The byte array never goes back to GC
-      freeList.add(i);
-    }
-    for (int i = 0; i < NUM_THREADS; i++) {
-      Thread t = new Thread(new ReadBufferWorker(i));
-      t.setDaemon(true);
-      threads[i] = t;
-      t.setName("SeaweedFS-prefetch-" + i);
-      t.start();
-    }
-    ReadBufferWorker.UNLEASH_WORKERS.countDown();
-  }
-
-  // hide instance constructor
-  private ReadBufferManager() {
-  }
-
-
-  /*
-   *
-   *  SeaweedInputStream-facing methods
-   *
-   */
-
-
-  /**
-   * {@link SeaweedInputStream} calls this method to queue read-aheads.
-   *
-   * @param stream          The {@link SeaweedInputStream} for which to do the read-ahead
-   * @param requestedOffset The offset in the file which shoukd be read
-   * @param requestedLength The length to read
-   */
-  void queueReadAhead(final SeaweedInputStream stream, final long requestedOffset, final int requestedLength) {
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("Start Queueing readAhead for {} offset {} length {}",
-          stream.getPath(), requestedOffset, requestedLength);
-    }
-    ReadBuffer buffer;
-    synchronized (this) {
-      if (isAlreadyQueued(stream, requestedOffset)) {
-        return; // already queued, do not queue again
-      }
-      if (freeList.isEmpty() && !tryEvict()) {
-        return; // no buffers available, cannot queue anything
-      }
-
-      buffer = new ReadBuffer();
-      buffer.setStream(stream);
-      buffer.setOffset(requestedOffset);
-      buffer.setLength(0);
-      buffer.setRequestedLength(requestedLength);
-      buffer.setStatus(ReadBufferStatus.NOT_AVAILABLE);
-      buffer.setLatch(new CountDownLatch(1));
-
-      Integer bufferIndex = freeList.pop();  // will return a value, since we have checked size > 0 already
-
-      buffer.setBuffer(buffers[bufferIndex]);
-      buffer.setBufferindex(bufferIndex);
-      readAheadQueue.add(buffer);
-      notifyAll();
-    }
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("Done q-ing readAhead for file {} offset {} buffer idx {}",
-          stream.getPath(), requestedOffset, buffer.getBufferindex());
-    }
-  }
-
-
-  /**
-   * {@link SeaweedInputStream} calls this method read any bytes already available in a buffer (thereby saving a
-   * remote read). This returns the bytes if the data already exists in buffer. If there is a buffer that is reading
-   * the requested offset, then this method blocks until that read completes. If the data is queued in a read-ahead
-   * but not picked up by a worker thread yet, then it cancels that read-ahead and reports cache miss. This is because
-   * depending on worker thread availability, the read-ahead may take a while - the calling thread can do it's own
-   * read to get the data faster (copmared to the read waiting in queue for an indeterminate amount of time).
-   *
-   * @param stream   the file to read bytes for
-   * @param position the offset in the file to do a read for
-   * @param length   the length to read
-   * @param buffer   the buffer to read data into. Note that the buffer will be written into from offset 0.
-   * @return the number of bytes read
-   */
-  int getBlock(final SeaweedInputStream stream, final long position, final int length, final byte[] buffer) {
-    // not synchronized, so have to be careful with locking
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("getBlock for file {}  position {}  thread {}",
-          stream.getPath(), position, Thread.currentThread().getName());
-    }
-
-    waitForProcess(stream, position);
-
-    int bytesRead = 0;
-    synchronized (this) {
-      bytesRead = getBlockFromCompletedQueue(stream, position, length, buffer);
-    }
-    if (bytesRead > 0) {
-      if (LOGGER.isTraceEnabled()) {
-        LOGGER.trace("Done read from Cache for {} position {} length {}",
-            stream.getPath(), position, bytesRead);
-      }
-      return bytesRead;
-    }
-
-    // otherwise, just say we got nothing - calling thread can do its own read
-    return 0;
-  }
-
-  /*
-   *
-   *  Internal methods
-   *
-   */
-
-  private void waitForProcess(final SeaweedInputStream stream, final long position) {
-    ReadBuffer readBuf;
-    synchronized (this) {
-      clearFromReadAheadQueue(stream, position);
-      readBuf = getFromList(inProgressList, stream, position);
-    }
-    if (readBuf != null) {         // if in in-progress queue, then block for it
-      try {
-        if (LOGGER.isTraceEnabled()) {
-          LOGGER.trace("got a relevant read buffer for file {} offset {} buffer idx {}",
-              stream.getPath(), readBuf.getOffset(), readBuf.getBufferindex());
-        }
-        readBuf.getLatch().await();  // blocking wait on the caller stream's thread
-        // Note on correctness: readBuf gets out of inProgressList only in 1 place: after worker thread
-        // is done processing it (in doneReading). There, the latch is set after removing the buffer from
-        // inProgressList. So this latch is safe to be outside the synchronized block.
-        // Putting it in synchronized would result in a deadlock, since this thread would be holding the lock
-        // while waiting, so no one will be able to  change any state. If this becomes more complex in the future,
-        // then the latch cane be removed and replaced with wait/notify whenever inProgressList is touched.
-      } catch (InterruptedException ex) {
-        Thread.currentThread().interrupt();
-      }
-      if (LOGGER.isTraceEnabled()) {
-        LOGGER.trace("latch done for file {} buffer idx {} length {}",
-            stream.getPath(), readBuf.getBufferindex(), readBuf.getLength());
-      }
-    }
-  }
-
-  /**
-   * If any buffer in the completedlist can be reclaimed then reclaim it and return the buffer to free list.
-   * The objective is to find just one buffer - there is no advantage to evicting more than one.
-   *
-   * @return whether the eviction succeeeded - i.e., were we able to free up one buffer
-   */
-  private synchronized boolean tryEvict() {
-    ReadBuffer nodeToEvict = null;
-    if (completedReadList.size() <= 0) {
-      return false;  // there are no evict-able buffers
-    }
-
-    // first, try buffers where all bytes have been consumed (approximated as first and last bytes consumed)
-    for (ReadBuffer buf : completedReadList) {
-      if (buf.isFirstByteConsumed() && buf.isLastByteConsumed()) {
-        nodeToEvict = buf;
-        break;
-      }
-    }
-    if (nodeToEvict != null) {
-      return evict(nodeToEvict);
-    }
-
-    // next, try buffers where any bytes have been consumed (may be a bad idea? have to experiment and see)
-    for (ReadBuffer buf : completedReadList) {
-      if (buf.isAnyByteConsumed()) {
-        nodeToEvict = buf;
-        break;
-      }
-    }
-
-    if (nodeToEvict != null) {
-      return evict(nodeToEvict);
-    }
-
-    // next, try any old nodes that have not been consumed
-    long earliestBirthday = Long.MAX_VALUE;
-    for (ReadBuffer buf : completedReadList) {
-      if (buf.getTimeStamp() < earliestBirthday) {
-        nodeToEvict = buf;
-        earliestBirthday = buf.getTimeStamp();
-      }
-    }
-    if ((currentTimeMillis() - earliestBirthday > THRESHOLD_AGE_MILLISECONDS) && (nodeToEvict != null)) {
-      return evict(nodeToEvict);
-    }
-
-    // nothing can be evicted
-    return false;
-  }
-
-  private boolean evict(final ReadBuffer buf) {
-    freeList.push(buf.getBufferindex());
-    completedReadList.remove(buf);
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("Evicting buffer idx {}; was used for file {} offset {} length {}",
-          buf.getBufferindex(), buf.getStream().getPath(), buf.getOffset(), buf.getLength());
-    }
-    return true;
-  }
-
-  private boolean isAlreadyQueued(final SeaweedInputStream 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 SeaweedInputStream stream, final long requestedOffset) {
-    return (getFromList(list, stream, requestedOffset) != null);
-  }
-
-  private ReadBuffer getFromList(final Collection<ReadBuffer> list, final SeaweedInputStream stream, final long requestedOffset) {
-    for (ReadBuffer buffer : list) {
-      if (buffer.getStream() == stream) {
-        if (buffer.getStatus() == ReadBufferStatus.AVAILABLE
-            && requestedOffset >= buffer.getOffset()
-            && requestedOffset < buffer.getOffset() + buffer.getLength()) {
-          return buffer;
-        } else if (requestedOffset >= buffer.getOffset()
-            && requestedOffset < buffer.getOffset() + buffer.getRequestedLength()) {
-          return buffer;
-        }
-      }
-    }
-    return null;
-  }
-
-  private void clearFromReadAheadQueue(final SeaweedInputStream stream, final long requestedOffset) {
-    ReadBuffer buffer = getFromList(readAheadQueue, stream, requestedOffset);
-    if (buffer != null) {
-      readAheadQueue.remove(buffer);
-      notifyAll();   // lock is held in calling method
-      freeList.push(buffer.getBufferindex());
-    }
-  }
-
-  private int getBlockFromCompletedQueue(final SeaweedInputStream stream, final long position, final int length,
-                                         final byte[] buffer) {
-    ReadBuffer buf = getFromList(completedReadList, stream, position);
-    if (buf == null || position >= buf.getOffset() + buf.getLength()) {
-      return 0;
-    }
-    int cursor = (int) (position - buf.getOffset());
-    int availableLengthInBuffer = buf.getLength() - cursor;
-    int lengthToCopy = Math.min(length, availableLengthInBuffer);
-    System.arraycopy(buf.getBuffer(), cursor, buffer, 0, lengthToCopy);
-    if (cursor == 0) {
-      buf.setFirstByteConsumed(true);
-    }
-    if (cursor + lengthToCopy == buf.getLength()) {
-      buf.setLastByteConsumed(true);
-    }
-    buf.setAnyByteConsumed(true);
-    return lengthToCopy;
-  }
-
-  /*
-   *
-   *  ReadBufferWorker-thread-facing methods
-   *
-   */
-
-  /**
-   * ReadBufferWorker thread calls this to get the next buffer that it should work on.
-   *
-   * @return {@link ReadBuffer}
-   * @throws InterruptedException if thread is interrupted
-   */
-  ReadBuffer getNextBlockToRead() throws InterruptedException {
-    ReadBuffer buffer = null;
-    synchronized (this) {
-      //buffer = readAheadQueue.take();  // blocking method
-      while (readAheadQueue.size() == 0) {
-        wait();
-      }
-      buffer = readAheadQueue.remove();
-      notifyAll();
-      if (buffer == null) {
-        return null;            // should never happen
-      }
-      buffer.setStatus(ReadBufferStatus.READING_IN_PROGRESS);
-      inProgressList.add(buffer);
-    }
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("ReadBufferWorker picked file {} for offset {}",
-          buffer.getStream().getPath(), buffer.getOffset());
-    }
-    return buffer;
-  }
-
-  /**
-   * ReadBufferWorker thread calls this method to post completion.
-   *
-   * @param buffer            the buffer whose read was completed
-   * @param result            the {@link ReadBufferStatus} after the read operation in the worker thread
-   * @param bytesActuallyRead the number of bytes that the worker thread was actually able to read
-   */
-  void doneReading(final ReadBuffer buffer, final ReadBufferStatus result, final int bytesActuallyRead) {
-    if (LOGGER.isTraceEnabled()) {
-      LOGGER.trace("ReadBufferWorker completed file {} for offset {} bytes {}",
-          buffer.getStream().getPath(),  buffer.getOffset(), bytesActuallyRead);
-    }
-    synchronized (this) {
-      inProgressList.remove(buffer);
-      if (result == ReadBufferStatus.AVAILABLE && bytesActuallyRead > 0) {
-        buffer.setStatus(ReadBufferStatus.AVAILABLE);
-        buffer.setTimeStamp(currentTimeMillis());
-        buffer.setLength(bytesActuallyRead);
-        completedReadList.add(buffer);
-      } else {
-        freeList.push(buffer.getBufferindex());
-        // buffer should go out of scope after the end of the calling method in ReadBufferWorker, and eligible for GC
-      }
-    }
-    //outside the synchronized, since anyone receiving a wake-up from the latch must see safe-published results
-    buffer.getLatch().countDown(); // wake up waiting threads (if any)
-  }
-
-  /**
-   * Similar to System.currentTimeMillis, except implemented with System.nanoTime().
-   * System.currentTimeMillis can go backwards when system clock is changed (e.g., with NTP time synchronization),
-   * making it unsuitable for measuring time intervals. nanotime is strictly monotonically increasing per CPU core.
-   * Note: it is not monotonic across Sockets, and even within a CPU, its only the
-   * more recent parts which share a clock across all cores.
-   *
-   * @return current time in milliseconds
-   */
-  private long currentTimeMillis() {
-    return System.nanoTime() / 1000 / 1000;
-  }
-}

other/java/hdfs3/src/main/java/seaweed/hdfs/ReadBufferWorker.java

@@ -1,70 +0,0 @@
-/**
- * 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 seaweed.hdfs;
-
-import java.util.concurrent.CountDownLatch;
-
-class ReadBufferWorker implements Runnable {
-
-  protected static final CountDownLatch UNLEASH_WORKERS = new CountDownLatch(1);
-  private int id;
-
-  ReadBufferWorker(final int id) {
-    this.id = id;
-  }
-
-  /**
-   * return the ID of ReadBufferWorker.
-   */
-  public int getId() {
-    return this.id;
-  }
-
-  /**
-   * Waits until a buffer becomes available in ReadAheadQueue.
-   * Once a buffer becomes available, reads the file specified in it and then posts results back to buffer manager.
-   * Rinse and repeat. Forever.
-   */
-  public void run() {
-    try {
-      UNLEASH_WORKERS.await();
-    } catch (InterruptedException ex) {
-      Thread.currentThread().interrupt();
-    }
-    ReadBufferManager bufferManager = ReadBufferManager.getBufferManager();
-    ReadBuffer buffer;
-    while (true) {
-      try {
-        buffer = bufferManager.getNextBlockToRead();   // blocks, until a buffer is available for this thread
-      } catch (InterruptedException ex) {
-        Thread.currentThread().interrupt();
-        return;
-      }
-      if (buffer != null) {
-        try {
-          // do the actual read, from the file.
-          int bytesRead = buffer.getStream().readRemote(buffer.getOffset(), buffer.getBuffer(), 0, buffer.getRequestedLength());
-          bufferManager.doneReading(buffer, ReadBufferStatus.AVAILABLE, bytesRead);  // post result back to ReadBufferManager
-        } catch (Exception ex) {
-          bufferManager.doneReading(buffer, ReadBufferStatus.READ_FAILED, 0);
-        }
-      }
-    }
-  }
-}

other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystemStore.java

@@ -212,7 +212,6 @@ public class SeaweedFileSystemStore {
 
         LOG.debug("openFileForRead path:{} bufferSize:{}", path, bufferSize);
 
-        int readAheadQueueDepth = 2;
         FilerProto.Entry entry = lookupEntry(path);
 
         if (entry == null) {
@@ -223,8 +222,7 @@ public class SeaweedFileSystemStore {
             statistics,
             path.toUri().getPath(),
             entry,
-            bufferSize,
-            readAheadQueueDepth);
+            bufferSize);
     }
 
     public void setOwner(Path path, String owner, String group) {

other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedInputStream.java

@@ -27,16 +27,9 @@ public class SeaweedInputStream extends FSInputStream {
     private final List<SeaweedRead.VisibleInterval> visibleIntervalList;
     private final long contentLength;
     private final int bufferSize; // default buffer size
-    private final int readAheadQueueDepth;         // initialized in constructor
-    private final boolean readAheadEnabled; // whether enable readAhead;
 
-    private byte[] buffer = null;            // will be initialized on first use
+    private long position = 0;  // cursor of the file
 
-    private long fCursor = 0;  // cursor of buffer within file - offset of next byte to read from remote server
-    private long fCursorAfterLastRead = -1;
-    private int bCursor = 0;   // cursor of read within buffer - offset of next byte to be returned from buffer
-    private int limit = 0;     // offset of next byte to be read into buffer from service (i.e., upper marker+1
-    //                                                      of valid bytes in buffer)
     private boolean closed = false;
 
     public SeaweedInputStream(
@@ -44,16 +37,13 @@ public class SeaweedInputStream extends FSInputStream {
             final Statistics statistics,
             final String path,
             final FilerProto.Entry entry,
-        final int bufferSize,
-        final int readAheadQueueDepth) {
+            final int bufferSize) {
         this.filerGrpcClient = filerGrpcClient;
         this.statistics = statistics;
         this.path = path;
         this.entry = entry;
         this.contentLength = SeaweedRead.totalSize(entry.getChunksList());
         this.bufferSize = bufferSize;
-        this.readAheadQueueDepth = (readAheadQueueDepth >= 0) ? readAheadQueueDepth : Runtime.getRuntime().availableProcessors();
-        this.readAheadEnabled = true;
 
         this.visibleIntervalList = SeaweedRead.nonOverlappingVisibleIntervals(entry.getChunksList());
 
@@ -78,122 +68,7 @@ public class SeaweedInputStream extends FSInputStream {
 
     @Override
     public synchronized int read(final byte[] b, final int off, final int len) throws IOException {
-        int currentOff = off;
-        int currentLen = len;
-        int lastReadBytes;
-        int totalReadBytes = 0;
-        do {
-            lastReadBytes = readOneBlock(b, currentOff, currentLen);
-            if (lastReadBytes > 0) {
-                currentOff += lastReadBytes;
-                currentLen -= lastReadBytes;
-                totalReadBytes += lastReadBytes;
-            }
-            if (currentLen <= 0 || currentLen > b.length - currentOff) {
-                break;
-            }
-        } while (lastReadBytes > 0);
-        return totalReadBytes > 0 ? totalReadBytes : lastReadBytes;
-    }
-
-    private int readOneBlock(final byte[] b, final int off, final int len) throws IOException {
-        if (closed) {
-            throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
-        }
-
-        Preconditions.checkNotNull(b);
-
-        if (len == 0) {
-            return 0;
-        }
-
-        if (this.available() == 0) {
-            return -1;
-        }
-
-        if (off < 0 || len < 0 || len > b.length - off) {
-            throw new IndexOutOfBoundsException();
-        }
-
-        //If buffer is empty, then fill the buffer.
-        if (bCursor == limit) {
-            //If EOF, then return -1
-            if (fCursor >= contentLength) {
-                return -1;
-            }
-
-            long bytesRead = 0;
-            //reset buffer to initial state - i.e., throw away existing data
-            bCursor = 0;
-            limit = 0;
-            if (buffer == null) {
-                buffer = new byte[bufferSize];
-            }
-
-            // Enable readAhead when reading sequentially
-            if (-1 == fCursorAfterLastRead || fCursorAfterLastRead == fCursor || b.length >= bufferSize) {
-                bytesRead = readInternal(fCursor, buffer, 0, bufferSize, false);
-            } else {
-                bytesRead = readInternal(fCursor, buffer, 0, b.length, true);
-            }
-
-            if (bytesRead == -1) {
-                return -1;
-            }
 
-            limit += bytesRead;
-            fCursor += bytesRead;
-            fCursorAfterLastRead = fCursor;
-        }
-
-        //If there is anything in the buffer, then return lesser of (requested bytes) and (bytes in buffer)
-        //(bytes returned may be less than requested)
-        int bytesRemaining = limit - bCursor;
-        int bytesToRead = Math.min(len, bytesRemaining);
-        System.arraycopy(buffer, bCursor, b, off, bytesToRead);
-        bCursor += bytesToRead;
-        if (statistics != null) {
-            statistics.incrementBytesRead(bytesToRead);
-        }
-        return bytesToRead;
-    }
-
-
-    private int readInternal(final long position, final byte[] b, final int offset, final int length,
-                             final boolean bypassReadAhead) throws IOException {
-        if (readAheadEnabled && !bypassReadAhead) {
-            // try reading from read-ahead
-            if (offset != 0) {
-                throw new IllegalArgumentException("readahead buffers cannot have non-zero buffer offsets");
-            }
-            int receivedBytes;
-
-            // queue read-aheads
-            int numReadAheads = this.readAheadQueueDepth;
-            long nextSize;
-            long nextOffset = position;
-            while (numReadAheads > 0 && nextOffset < contentLength) {
-                nextSize = Math.min((long) bufferSize, contentLength - nextOffset);
-                ReadBufferManager.getBufferManager().queueReadAhead(this, nextOffset, (int) nextSize);
-                nextOffset = nextOffset + nextSize;
-                numReadAheads--;
-            }
-
-            // try reading from buffers first
-            receivedBytes = ReadBufferManager.getBufferManager().getBlock(this, position, length, b);
-            if (receivedBytes > 0) {
-                return receivedBytes;
-            }
-
-            // got nothing from read-ahead, do our own read now
-            receivedBytes = readRemote(position, b, offset, length);
-            return receivedBytes;
-        } else {
-            return readRemote(position, b, offset, length);
-        }
-    }
-
-    int readRemote(long position, byte[] b, int offset, int length) throws IOException {
         if (position < 0) {
             throw new IllegalArgumentException("attempting to read from negative offset");
         }
@@ -203,21 +78,30 @@ public class SeaweedInputStream extends FSInputStream {
         if (b == null) {
             throw new IllegalArgumentException("null byte array passed in to read() method");
         }
-        if (offset >= b.length) {
+        if (off >= b.length) {
             throw new IllegalArgumentException("offset greater than length of array");
         }
-        if (length < 0) {
+        if (len < 0) {
             throw new IllegalArgumentException("requested read length is less than zero");
         }
-        if (length > (b.length - offset)) {
+        if (len > (b.length - off)) {
             throw new IllegalArgumentException("requested read length is more than will fit after requested offset in buffer");
         }
 
-        long bytesRead = SeaweedRead.read(filerGrpcClient, visibleIntervalList, position, b, offset, length);
+        long bytesRead = SeaweedRead.read(this.filerGrpcClient, this.visibleIntervalList, this.position, b, off, len);
         if (bytesRead > Integer.MAX_VALUE) {
             throw new IOException("Unexpected Content-Length");
         }
+
+        if (bytesRead > 0) {
+            this.position += bytesRead;
+            if (statistics != null) {
+                statistics.incrementBytesRead(bytesRead);
+            }
+        }
+
         return (int)bytesRead;
+
     }
 
     /**
@@ -239,17 +123,8 @@ public class SeaweedInputStream extends FSInputStream {
             throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
         }
 
-        if (n >= fCursor - limit && n <= fCursor) { // within buffer
-            bCursor = (int) (n - (fCursor - limit));
-            return;
-        }
-
-        // next read will read from here
-        fCursor = n;
+        this.position = n;
 
-        //invalidate buffer
-        limit = 0;
-        bCursor = 0;
     }
 
     @Override
@@ -257,20 +132,19 @@ public class SeaweedInputStream extends FSInputStream {
         if (closed) {
             throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
         }
-        long currentPos = getPos();
-        if (currentPos == contentLength) {
+        if (this.position == contentLength) {
             if (n > 0) {
                 throw new EOFException(FSExceptionMessages.CANNOT_SEEK_PAST_EOF);
             }
         }
-        long newPos = currentPos + n;
+        long newPos = this.position + n;
         if (newPos < 0) {
             newPos = 0;
-            n = newPos - currentPos;
+            n = newPos - this.position;
         }
         if (newPos > contentLength) {
             newPos = contentLength;
-            n = newPos - currentPos;
+            n = newPos - this.position;
         }
         seek(newPos);
         return n;
@@ -321,7 +195,7 @@ public class SeaweedInputStream extends FSInputStream {
         if (closed) {
             throw new IOException(FSExceptionMessages.STREAM_IS_CLOSED);
         }
-        return fCursor - limit + bCursor;
+        return position;
     }
 
     /**
@@ -338,7 +212,6 @@ public class SeaweedInputStream extends FSInputStream {
     @Override
     public synchronized void close() throws IOException {
         closed = true;
-        buffer = null; // de-reference the buffer so it can be GC'ed sooner
     }
 
     /**

weed/command/mount_std.go

@@ -13,6 +13,9 @@ import (
 	"strings"
 	"time"
 
+	"github.com/seaweedfs/fuse"
+	"github.com/seaweedfs/fuse/fs"
+
 	"github.com/chrislusf/seaweedfs/weed/filesys"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb"
@@ -20,8 +23,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/security"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/util/grace"
-	"github.com/seaweedfs/fuse"
-	"github.com/seaweedfs/fuse/fs"
 )
 
 func runMount(cmd *Command, args []string) bool {
@@ -97,6 +98,9 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 		mountMode = os.ModeDir | fileInfo.Mode()
 		uid, gid = util.GetFileUidGid(fileInfo)
 		fmt.Printf("mount point owner uid=%d gid=%d mode=%s\n", uid, gid, fileInfo.Mode())
+	} else {
+		fmt.Printf("can not stat %s\n", dir)
+		return false
 	}
 
 	if uid == 0 {