Browse Source

Hadoop: add BufferedByteBufferReadableInputStream

fix https://github.com/chrislusf/seaweedfs/issues/1645
pull/1650/head
Chris Lu 4 years ago
parent
commit
4d2855476c
  1. 25
      other/java/hdfs2/src/main/java/seaweed/hdfs/BufferedByteBufferReadableInputStream.java
  2. 2
      other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
  3. 25
      other/java/hdfs3/src/main/java/seaweed/hdfs/BufferedByteBufferReadableInputStream.java
  4. 2
      other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java
  5. 8
      other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedInputStream.java

25
other/java/hdfs2/src/main/java/seaweed/hdfs/BufferedByteBufferReadableInputStream.java

@ -0,0 +1,25 @@
package seaweed.hdfs;
import org.apache.hadoop.fs.*;
import java.io.IOException;
import java.nio.ByteBuffer;
public class BufferedByteBufferReadableInputStream extends BufferedFSInputStream implements ByteBufferReadable {
public BufferedByteBufferReadableInputStream(FSInputStream in, int size) {
super(in, size);
if (!(in instanceof Seekable) || !(in instanceof PositionedReadable)) {
throw new IllegalArgumentException("In is not an instance of Seekable or PositionedReadable");
}
}
@Override
public int read(ByteBuffer buf) throws IOException {
if (this.in instanceof ByteBufferReadable) {
return ((ByteBufferReadable)this.in).read(buf);
} else {
throw new UnsupportedOperationException("Byte-buffer read unsupported by input stream");
}
}
}

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

@ -82,7 +82,7 @@ public class SeaweedFileSystem extends FileSystem {
try { try {
int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE); int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics); FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics);
return new FSDataInputStream(new BufferedFSInputStream(inputStream, 4 * seaweedBufferSize));
return new FSDataInputStream(new BufferedByteBufferReadableInputStream(inputStream, 4 * seaweedBufferSize));
} catch (Exception ex) { } catch (Exception ex) {
LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex); LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex);
return null; return null;

25
other/java/hdfs3/src/main/java/seaweed/hdfs/BufferedByteBufferReadableInputStream.java

@ -0,0 +1,25 @@
package seaweed.hdfs;
import org.apache.hadoop.fs.*;
import java.io.IOException;
import java.nio.ByteBuffer;
public class BufferedByteBufferReadableInputStream extends BufferedFSInputStream implements ByteBufferReadable {
public BufferedByteBufferReadableInputStream(FSInputStream in, int size) {
super(in, size);
if (!(in instanceof Seekable) || !(in instanceof PositionedReadable)) {
throw new IllegalArgumentException("In is not an instance of Seekable or PositionedReadable");
}
}
@Override
public int read(ByteBuffer buf) throws IOException {
if (this.in instanceof ByteBufferReadable) {
return ((ByteBufferReadable)this.in).read(buf);
} else {
throw new UnsupportedOperationException("Byte-buffer read unsupported by input stream");
}
}
}

2
other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedFileSystem.java

@ -82,7 +82,7 @@ public class SeaweedFileSystem extends FileSystem {
try { try {
int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE); int seaweedBufferSize = this.getConf().getInt(FS_SEAWEED_BUFFER_SIZE, FS_SEAWEED_DEFAULT_BUFFER_SIZE);
FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics); FSInputStream inputStream = seaweedFileSystemStore.openFileForRead(path, statistics);
return new FSDataInputStream(new BufferedFSInputStream(inputStream, 4 * seaweedBufferSize));
return new FSDataInputStream(new BufferedByteBufferReadableInputStream(inputStream, 4 * seaweedBufferSize));
} catch (Exception ex) { } catch (Exception ex) {
LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex); LOG.warn("open path: {} bufferSize:{}", path, bufferSize, ex);
return null; return null;

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

@ -65,14 +65,8 @@ public class SeaweedInputStream extends FSInputStream implements ByteBufferReada
} }
@Override @Override
public synchronized int read(final byte[] b, final int off, final int len) throws IOException {
public int read(final byte[] b, final int off, final int len) throws IOException {
if (position < 0) {
throw new IllegalArgumentException("attempting to read from negative offset");
}
if (position >= contentLength) {
return -1; // Hadoop prefers -1 to EOFException
}
if (b == null) { if (b == null) {
throw new IllegalArgumentException("null byte array passed in to read() method"); throw new IllegalArgumentException("null byte array passed in to read() method");
} }

Loading…
Cancel
Save