gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

Hadoop: fix reading file tail

Browse source
This commit is contained in:
Chris Lu 2020-10-10 00:57:54 -07:00
parent e1a3ffcdbf
commit 00a75d7c99
1 changed files with 14 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
other/java/client/src/main/java/seaweedfs/client/SeaweedRead.java
View file

@ -40,7 +40,7 @@ public class SeaweedRead {
//TODO parallel this //TODO parallel this
long readCount = 0; long readCount = 0;
int startOffset = bufferOffset; long startOffset = position;
for (ChunkView chunkView : chunkViews) { for (ChunkView chunkView : chunkViews) {
if (startOffset < chunkView.logicOffset) { if (startOffset < chunkView.logicOffset) {
@ -57,7 +57,7 @@ public class SeaweedRead {
return 0; return 0;
} }
int len = readChunkView(position, buffer, startOffset, chunkView, locations); int len = readChunkView(startOffset, buffer, bufferOffset + readCount, chunkView, locations);
LOG.debug("read [{},{}) {} size {}", startOffset, startOffset + len, chunkView.fileId, chunkView.size); LOG.debug("read [{},{}) {} size {}", startOffset, startOffset + len, chunkView.fileId, chunkView.size);
@ -66,7 +66,7 @@ public class SeaweedRead {
} }
long limit = Math.min(bufferLength, fileSize); long limit = Math.min(bufferOffset + bufferLength, fileSize);
if (startOffset < limit) { if (startOffset < limit) {
long gap = limit - startOffset; long gap = limit - startOffset;
@ -78,7 +78,7 @@ public class SeaweedRead {
return readCount; return readCount;
} }
private static int readChunkView(long position, byte[] buffer, int startOffset, ChunkView chunkView, FilerProto.Locations locations) throws IOException { private static int readChunkView(long startOffset, byte[] buffer, long bufOffset, ChunkView chunkView, FilerProto.Locations locations) throws IOException {
byte[] chunkData = chunkCache.getChunk(chunkView.fileId); byte[] chunkData = chunkCache.getChunk(chunkView.fileId);
@ -88,9 +88,9 @@ public class SeaweedRead {
} }
int len = (int) chunkView.size; int len = (int) chunkView.size;
LOG.debug("readChunkView fid:{} chunkData.length:{} chunkView.offset:{} buffer.length:{} startOffset:{} len:{}", LOG.debug("readChunkView fid:{} chunkData.length:{} chunkView[{};{}) buf[{},{})/{} startOffset:{}",
chunkView.fileId, chunkData.length, chunkView.offset, buffer.length, startOffset, len); chunkView.fileId, chunkData.length, chunkView.offset, chunkView.offset+chunkView.size, bufOffset, bufOffset+len, buffer.length, startOffset);
System.arraycopy(chunkData, startOffset - (int) (chunkView.logicOffset - chunkView.offset), buffer, startOffset, len); System.arraycopy(chunkData, (int) (startOffset - chunkView.logicOffset + chunkView.offset), buffer, (int)bufOffset, len);
return len; return len;
} }
@ -98,7 +98,8 @@ public class SeaweedRead {
public static byte[] doFetchFullChunkData(ChunkView chunkView, FilerProto.Locations locations) throws IOException { public static byte[] doFetchFullChunkData(ChunkView chunkView, FilerProto.Locations locations) throws IOException {
byte[] data = null; byte[] data = null;
for (long waitTime = 230L; waitTime < 20 * 1000; waitTime += waitTime / 2) { IOException lastException = null;
for (long waitTime = 1000L; waitTime < 10 * 1000; waitTime += waitTime / 2) {
for (FilerProto.Location location : locations.getLocationsList()) { for (FilerProto.Location location : locations.getLocationsList()) {
String url = String.format("http://%s/%s", location.getUrl(), chunkView.fileId); String url = String.format("http://%s/%s", location.getUrl(), chunkView.fileId);
try { try {
@ -106,6 +107,7 @@ public class SeaweedRead {
break; break;
} catch (IOException ioe) { } catch (IOException ioe) {
LOG.debug("doFetchFullChunkData {} :{}", url, ioe); LOG.debug("doFetchFullChunkData {} :{}", url, ioe);
lastException = ioe;
} }
} }
if (data != null) { if (data != null) {
@ -117,6 +119,10 @@ public class SeaweedRead {
} }
} }
if (data == null) {
throw lastException;
}
LOG.debug("doFetchFullChunkData fid:{} chunkData.length:{}", chunkView.fileId, data.length); LOG.debug("doFetchFullChunkData fid:{} chunkData.length:{}", chunkView.fileId, data.length);
return data; return data;