Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

hadoop: reduce memory usage #2580

Merged
merged 6 commits into from
Aug 23, 2022
Merged

hadoop: reduce memory usage #2580

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3a63a23
hadoop: reduce memory usage
tangyoupeng Aug 22, 2022
2a6744e
add test
tangyoupeng Aug 22, 2022
2c52d3c
remove directbuffer in FileInputStream
tangyoupeng Aug 23, 2022
c21428d
buffer small read
tangyoupeng Aug 23, 2022
9dddafd
simplify
davies Aug 23, 2022
39b117b
fix
tangyoupeng Aug 23, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
103 changes: 22 additions & 81 deletions sdk/java/src/main/java/io/juicefs/JuiceFileSystemImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,8 @@
import jnr.ffi.Memory;
import jnr.ffi.Pointer;
import jnr.ffi.Runtime;
import jnr.ffi.annotations.In;
import jnr.ffi.annotations.Out;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.HadoopIllegalArgumentException;
Expand All @@ -39,11 +41,9 @@
import org.apache.hadoop.security.AccessControlException;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.util.DataChecksum;
import org.apache.hadoop.util.DirectBufferPool;
import org.apache.hadoop.util.Progressable;
import org.apache.hadoop.util.VersionInfo;
import org.json.JSONObject;
import sun.nio.ch.DirectBuffer;

import java.io.*;
import java.lang.reflect.Constructor;
Expand Down Expand Up @@ -89,7 +89,6 @@ public class JuiceFileSystemImpl extends FileSystem {
private ScheduledExecutorService nodesFetcherThread;
private ScheduledExecutorService refreshUidThread;
private Map<String, FileStatus> lastFileStatus = new HashMap<>();
private static final DirectBufferPool bufferPool = new DirectBufferPool();
private boolean metricsEnable = false;

/*
Expand All @@ -115,9 +114,9 @@ public static interface Libjfs {

long jfs_lseek(long pid, int fd, long pos, int whence);

int jfs_pread(long pid, int fd, Pointer b, int len, long offset);
int jfs_pread(long pid, int fd, @Out ByteBuffer b, int len, long offset);

int jfs_write(long pid, int fd, Pointer b, int len);
int jfs_write(long pid, int fd, @In byte[] b, int len);

int jfs_flush(long pid, int fd);

Expand Down Expand Up @@ -733,7 +732,7 @@ class FileInputStream extends FSInputStream implements ByteBufferReadable {
public FileInputStream(Path f, int fd, int size) throws IOException {
path = f;
this.fd = fd;
buf = bufferPool.getBuffer(size);
buf = ByteBuffer.allocate(size);
buf.limit(0);
position = 0;
}
Expand Down Expand Up @@ -780,34 +779,7 @@ public synchronized int read() throws IOException {
public synchronized int read(byte[] b, int off, int len) throws IOException {
if (off < 0 || len < 0 || b.length - off < len)
throw new IndexOutOfBoundsException();
if (len == 0)
return 0;
if (buf == null)
throw new IOException("stream was closed");
if (!buf.hasRemaining() && len <= buf.capacity() && !refill())
return -1; // No bytes were read before EOF.

int read = Math.min(buf.remaining(), len);
if (read > 0) {
buf.get(b, off, read);
statistics.incrementBytesRead(read);
off += read;
len -= read;
}
if (len == 0)
return read;
int more = read(position, b, off, len);
if (more <= 0) {
if (read > 0) {
return read;
} else {
return -1;
}
}
position += more;
buf.position(0);
buf.limit(0);
return read + more;
return read(ByteBuffer.wrap(b, off, len));
}

private boolean refill() throws IOException {
Expand All @@ -817,7 +789,6 @@ private boolean refill() throws IOException {
buf.limit(0);
return false; // EOF
}
statistics.incrementBytesRead(-read);
buf.position(0);
buf.limit(read);
position += read;
Expand All @@ -826,27 +797,10 @@ private boolean refill() throws IOException {

@Override
public synchronized int read(long pos, byte[] b, int off, int len) throws IOException {
if (len == 0)
return 0;
if (buf == null)
throw new IOException("stream was closed");
if (pos < 0)
throw new EOFException("position is negative");
if (b == null || off < 0 || len < 0 || b.length - off < len) {
throw new IllegalArgumentException("arguments: " + off + " " + len);
}
if (len > 128 << 20) {
len = 128 << 20;
}
Pointer tmp = Memory.allocate(Runtime.getRuntime(lib), len);
int got = lib.jfs_pread(Thread.currentThread().getId(), fd, tmp, len, pos);
if (got == 0)
return -1;
if (got == EINVAL)
throw new IOException("stream was closed");
if (got < 0)
throw error(got, path);
tmp.get(0, b, off, got);
int got = read(pos, ByteBuffer.wrap(b, off, len));
statistics.incrementBytesRead(got);
return got;
}
Expand All @@ -866,39 +820,31 @@ public synchronized int read(ByteBuffer b) throws IOException {
got++;
}
statistics.incrementBytesRead(got);
if (!b.hasRemaining())
return got;
int more = read(position, b);
if (more <= 0)
return got > 0 ? got : -1;
position += more;
statistics.incrementBytesRead(more);
buf.position(0);
buf.limit(0);
return got + more;
}

public synchronized int read(long pos, ByteBuffer b) throws IOException {
private synchronized int read(long pos, ByteBuffer b) throws IOException {
if (pos < 0)
throw new EOFException("position is negative");
if (!b.hasRemaining())
return 0;
int got;
if (b.hasArray()) {
got = read(pos, b.array(), b.position(), b.remaining());
if (got <= 0)
return got;
} else {
assert b.isDirect();
long address = ((DirectBuffer) b).address() + b.position();
Pointer destPtr = Runtime.getRuntime(lib).getMemoryManager().newPointer(address);
got = lib.jfs_pread(Thread.currentThread().getId(), fd, destPtr, b.remaining(), pos);
if (got == EINVAL)
throw new IOException("stream was closed");
if (got < 0)
throw error(got, path);
if (got == 0)
return -1;
statistics.incrementBytesRead(got);
}
b.position(b.position() + got);
int startPos = b.position();
got = lib.jfs_pread(Thread.currentThread().getId(), fd, b, b.remaining(), pos);
if (got == EINVAL)
throw new IOException("stream was closed");
if (got < 0)
throw error(got, path);
if (got == 0)
return -1;
b.position(startPos + got);
return got;
}

Expand Down Expand Up @@ -939,7 +885,6 @@ public synchronized void close() throws IOException {
if (buf == null) {
return; // already closed
}
bufferPool.returnBuffer(buf);
buf = null;
int r = lib.jfs_close(Thread.currentThread().getId(), fd);
fd = 0;
Expand Down Expand Up @@ -1009,9 +954,7 @@ public void write(byte[] b, int off, int len) throws IOException {
if (b.length - off < len) {
throw new IndexOutOfBoundsException();
}
Pointer buf = Memory.allocate(Runtime.getRuntime(lib), len);
buf.put(0, b, off, len);
int done = lib.jfs_write(Thread.currentThread().getId(), fd, buf, len);
int done = lib.jfs_write(Thread.currentThread().getId(), fd, b, len);
if (done == EINVAL)
throw new IOException("stream was closed");
if (done < 0)
Expand All @@ -1023,9 +966,7 @@ public void write(byte[] b, int off, int len) throws IOException {

@Override
public void write(int b) throws IOException {
Pointer buf = Memory.allocate(Runtime.getRuntime(lib), 1);
buf.putByte(0, (byte) b);
int done = lib.jfs_write(Thread.currentThread().getId(), fd, buf, 1);
int done = lib.jfs_write(Thread.currentThread().getId(), fd, new byte[]{(byte) b}, 1);
if (done == EINVAL)
throw new IOException("stream was closed");
if (done < 0)
Expand Down
19 changes: 19 additions & 0 deletions sdk/java/src/test/java/io/juicefs/JuiceFileSystemTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -349,6 +349,21 @@ public void testInputStream() throws Exception {
throw new IOException("Reached the end of stream. Still have: " + buf.remaining() + " bytes left");
}
}

Path directReadFile = new Path("/direct_file");
FSDataOutputStream ou = fs.create(directReadFile);
ou.write("hello world".getBytes());
ou.close();
FSDataInputStream dto = fs.open(directReadFile);
ByteBuffer directBuf = ByteBuffer.allocateDirect(11);
directBuf.put("hello ".getBytes());
dto.seek(6);
dto.read(directBuf);
byte[] rest = new byte[11];
directBuf.flip();
directBuf.get(rest, 0, rest.length);
assertEquals("hello world", new String(rest));

/*
* FSDataOutputStream out = fs.create(new Path("/bigfile"), true); byte[] arr =
* new byte[1<<20]; for (int i=0; i<1024; i++) { out.write(arr); } out.close();
Expand Down Expand Up @@ -407,6 +422,10 @@ public void testReadStats() throws IOException {
in.read(3000, new byte[6000], 0, 3000);
assertEquals(readSize * 2 + 3000 + 3000, statistics.getBytesRead());

in.read(new byte[3000], 0, 3000);
assertEquals(readSize * 2 + 3000 + 3000 + 3000, statistics.getBytesRead());

in.close();
}

public void testChecksum() throws IOException {
Expand Down