DfsBlockCache: Consolidate where ReadableChannel is opened
Opening a readable channel can be expensive and the number of channels can be limited in DFS. Ensure that caller of BlockBasedFile.readOneBlock() is responsible for opening and closing the file, and that the ReadableChannel is reused in the request. As a side effect, this makes the code easier to read, with better use of try-with-resources. The downside is that this means a readable channel is always opened, even when the entire pack is already available for copying from cache. This should be an acceptable cost: it's rare enough not to overload the server and from a client latency perspective, the latency cost is in the noise relative to the data transfer cost involved in a clone. If this turns out to be a problem in practice, we can reintroduce that optimization in a followup change. Change-Id: I340428ee4bacd2dce019d5616ef12339a0c85f0b Signed-off-by: Minh Thai <mthai@google.com> Signed-off-by: Jonathan Nieder <jrn@google.com>
This commit is contained in:
parent
8bc9acf264
commit
4207b8d6ae
|
@ -48,7 +48,6 @@
|
|||
import java.nio.ByteBuffer;
|
||||
import java.text.MessageFormat;
|
||||
|
||||
import org.eclipse.jgit.annotations.Nullable;
|
||||
import org.eclipse.jgit.errors.PackInvalidException;
|
||||
import org.eclipse.jgit.internal.storage.pack.PackExt;
|
||||
|
||||
|
@ -130,18 +129,18 @@ else if (size < cache.getBlockSize())
|
|||
}
|
||||
|
||||
DfsBlock getOrLoadBlock(long pos, DfsReader ctx) throws IOException {
|
||||
return cache.getOrLoad(this, pos, ctx, null);
|
||||
try (ReadableChannel rc = ctx.db.openFile(desc, ext)) {
|
||||
return cache.getOrLoad(this, pos, ctx, () -> rc);
|
||||
}
|
||||
}
|
||||
|
||||
DfsBlock readOneBlock(long pos, DfsReader ctx,
|
||||
@Nullable ReadableChannel fileChannel) throws IOException {
|
||||
DfsBlock readOneBlock(long pos, DfsReader ctx, ReadableChannel rc)
|
||||
throws IOException {
|
||||
if (invalid)
|
||||
throw new PackInvalidException(getFileName());
|
||||
|
||||
ctx.stats.readBlock++;
|
||||
long start = System.nanoTime();
|
||||
ReadableChannel rc = fileChannel != null ? fileChannel
|
||||
: ctx.db.openFile(desc, ext);
|
||||
try {
|
||||
int size = blockSize(rc);
|
||||
pos = (pos / size) * size;
|
||||
|
@ -189,9 +188,6 @@ DfsBlock readOneBlock(long pos, DfsReader ctx,
|
|||
|
||||
return new DfsBlock(key, pos, buf);
|
||||
} finally {
|
||||
if (rc != fileChannel) {
|
||||
rc.close();
|
||||
}
|
||||
ctx.stats.readBlockMicros += elapsedMicros(start);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -52,7 +52,6 @@
|
|||
import java.util.function.Consumer;
|
||||
import java.util.stream.LongStream;
|
||||
|
||||
import org.eclipse.jgit.annotations.Nullable;
|
||||
import org.eclipse.jgit.internal.JGitText;
|
||||
import org.eclipse.jgit.internal.storage.pack.PackExt;
|
||||
|
||||
|
@ -375,13 +374,13 @@ private static int tableSize(DfsBlockCacheConfig cfg) {
|
|||
* @param ctx
|
||||
* current thread's reader.
|
||||
* @param fileChannel
|
||||
* optional channel to read {@code pack}.
|
||||
* supplier for channel to read {@code pack}.
|
||||
* @return the object reference.
|
||||
* @throws IOException
|
||||
* the reference was not in the cache and could not be loaded.
|
||||
*/
|
||||
DfsBlock getOrLoad(BlockBasedFile file, long position, DfsReader ctx,
|
||||
@Nullable ReadableChannel fileChannel) throws IOException {
|
||||
ReadableChannelSupplier fileChannel) throws IOException {
|
||||
final long requestedPosition = position;
|
||||
position = file.alignToBlock(position);
|
||||
|
||||
|
@ -413,7 +412,8 @@ DfsBlock getOrLoad(BlockBasedFile file, long position, DfsReader ctx,
|
|||
getStat(statMiss, key).incrementAndGet();
|
||||
boolean credit = true;
|
||||
try {
|
||||
v = file.readOneBlock(requestedPosition, ctx, fileChannel);
|
||||
v = file.readOneBlock(requestedPosition, ctx,
|
||||
fileChannel.get());
|
||||
credit = false;
|
||||
} finally {
|
||||
if (credit) {
|
||||
|
@ -749,4 +749,16 @@ boolean has() {
|
|||
interface RefLoader<T> {
|
||||
Ref<T> load() throws IOException;
|
||||
}
|
||||
|
||||
/**
|
||||
* Supplier for readable channel
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface ReadableChannelSupplier {
|
||||
/**
|
||||
* @return ReadableChannel
|
||||
* @throws IOException
|
||||
*/
|
||||
ReadableChannel get() throws IOException;
|
||||
}
|
||||
}
|
||||
|
|
|
@ -438,63 +438,42 @@ private byte[] decompress(long position, int sz, DfsReader ctx)
|
|||
return dstbuf;
|
||||
}
|
||||
|
||||
void copyPackAsIs(PackOutputStream out, DfsReader ctx)
|
||||
throws IOException {
|
||||
void copyPackAsIs(PackOutputStream out, DfsReader ctx) throws IOException {
|
||||
// If the length hasn't been determined yet, pin to set it.
|
||||
if (length == -1) {
|
||||
ctx.pin(this, 0);
|
||||
ctx.unpin();
|
||||
}
|
||||
if (cache.shouldCopyThroughCache(length)) {
|
||||
copyPackThroughCache(out, ctx);
|
||||
} else {
|
||||
copyPackBypassCache(out, ctx);
|
||||
}
|
||||
}
|
||||
|
||||
private void copyPackThroughCache(PackOutputStream out, DfsReader ctx)
|
||||
throws IOException {
|
||||
@SuppressWarnings("resource") // Explicitly closed in finally block
|
||||
ReadableChannel rc = null;
|
||||
try {
|
||||
long position = 12;
|
||||
long remaining = length - (12 + 20);
|
||||
while (0 < remaining) {
|
||||
DfsBlock b;
|
||||
if (rc != null) {
|
||||
b = cache.getOrLoad(this, position, ctx, rc);
|
||||
} else {
|
||||
b = cache.get(key, alignToBlock(position));
|
||||
if (b == null) {
|
||||
rc = ctx.db.openFile(desc, PACK);
|
||||
try (ReadableChannel rc = ctx.db.openFile(desc, PACK)) {
|
||||
int sz = ctx.getOptions().getStreamPackBufferSize();
|
||||
if (sz > 0) {
|
||||
rc.setReadAheadBytes(sz);
|
||||
}
|
||||
b = cache.getOrLoad(this, position, ctx, rc);
|
||||
if (cache.shouldCopyThroughCache(length)) {
|
||||
copyPackThroughCache(out, ctx, rc);
|
||||
} else {
|
||||
copyPackBypassCache(out, rc);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void copyPackThroughCache(PackOutputStream out, DfsReader ctx,
|
||||
ReadableChannel rc) throws IOException {
|
||||
long position = 12;
|
||||
long remaining = length - (12 + 20);
|
||||
while (0 < remaining) {
|
||||
DfsBlock b = cache.getOrLoad(this, position, ctx, () -> rc);
|
||||
int ptr = (int) (position - b.start);
|
||||
int n = (int) Math.min(b.size() - ptr, remaining);
|
||||
b.write(out, position, n);
|
||||
position += n;
|
||||
remaining -= n;
|
||||
}
|
||||
} finally {
|
||||
if (rc != null) {
|
||||
rc.close();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private long copyPackBypassCache(PackOutputStream out, DfsReader ctx)
|
||||
private long copyPackBypassCache(PackOutputStream out, ReadableChannel rc)
|
||||
throws IOException {
|
||||
try (ReadableChannel rc = ctx.db.openFile(desc, PACK)) {
|
||||
ByteBuffer buf = newCopyBuffer(out, rc);
|
||||
if (ctx.getOptions().getStreamPackBufferSize() > 0) {
|
||||
rc.setReadAheadBytes(ctx.getOptions().getStreamPackBufferSize());
|
||||
}
|
||||
long position = 12;
|
||||
long remaining = length - (12 + 20);
|
||||
boolean packHeadSkipped = false;
|
||||
|
@ -531,7 +510,6 @@ private long copyPackBypassCache(PackOutputStream out, DfsReader ctx)
|
|||
}
|
||||
return position;
|
||||
}
|
||||
}
|
||||
|
||||
private ByteBuffer newCopyBuffer(PackOutputStream out, ReadableChannel rc) {
|
||||
int bs = blockSize(rc);
|
||||
|
|
|
@ -128,7 +128,9 @@ public ByteBuffer read(long pos, int cnt) throws IOException {
|
|||
open().setReadAheadBytes(readAhead);
|
||||
}
|
||||
|
||||
DfsBlock block = cache.getOrLoad(file, pos, ctx, ch);
|
||||
DfsBlock block = cache.getOrLoad(file, pos, ctx, () -> {
|
||||
return open();
|
||||
});
|
||||
if (block.start == pos && block.size() >= cnt) {
|
||||
return block.zeroCopyByteBuffer(cnt);
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue