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

async copy

Browse source
This commit is contained in:
Chris Lu 2020-07-17 15:44:00 -07:00
parent 8a1baafcb3
commit 0b9d73fb7c
1 changed files with 29 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

35
other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java
View file

@ -169,12 +169,35 @@ public class SeaweedOutputStream extends OutputStream {
return;
}
buffer.flip();
int bytesLength = buffer.limit() - buffer.position();
SeaweedWrite.writeData(entry, replication, filerGrpcClient, position, buffer.array(), buffer.position(), buffer.limit());
// System.out.println(path + " saved [" + (position) + "," + ((position) + bytesLength) + ")");
position += bytesLength;
buffer.clear();
position += submitWriteBufferToService(buffer, position);
buffer = ByteBufferPool.request(bufferSize);
}
private synchronized int submitWriteBufferToService(final ByteBuffer bufferToWrite, final long writePosition) throws IOException {
bufferToWrite.flip();
int bytesLength = bufferToWrite.limit() - bufferToWrite.position();
if (threadExecutor.getQueue().size() >= maxConcurrentRequestCount * 2) {
waitForTaskToComplete();
}
final Future<Void> job = completionService.submit(() -> {
System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
SeaweedWrite.writeData(entry, replication, filerGrpcClient, writePosition, bufferToWrite.array(), bufferToWrite.position(), bufferToWrite.limit());
System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")");
bufferToWrite.clear();
ByteBufferPool.release(bufferToWrite);
return null;
});
writeOperations.add(new WriteOperation(job, writePosition, bytesLength));
// Try to shrink the queue
shrinkWriteOperationQueue();
return bytesLength;
}