From 0b9d73fb7c213d466fd0aef4615709c34a949bed Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 17 Jul 2020 15:44:00 -0700 Subject: async copy --- .../java/seaweed/hdfs/SeaweedOutputStream.java | 35 ++++++++++++++++++---- 1 file changed, 29 insertions(+), 6 deletions(-) (limited to 'other/java') diff --git a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java index 1138ecca2..c1e7b593d 100644 --- a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java +++ b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java @@ -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 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; } -- cgit v1.2.3 From 3abd74b1d7d1685ca4ddcd741ddf3ac0585b848a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 17 Jul 2020 15:46:16 -0700 Subject: 1.3.8 --- other/java/client/pom.xml | 2 +- other/java/client/pom.xml.deploy | 2 +- other/java/client/pom_debug.xml | 2 +- other/java/hdfs2/dependency-reduced-pom.xml | 6 ++-- other/java/hdfs2/pom.xml | 6 ++-- other/java/hdfs3/dependency-reduced-pom.xml | 2 +- other/java/hdfs3/pom.xml | 6 ++-- .../java/seaweed/hdfs/SeaweedOutputStream.java | 35 ++++++++++++++++++---- 8 files changed, 42 insertions(+), 19 deletions(-) (limited to 'other/java') diff --git a/other/java/client/pom.xml b/other/java/client/pom.xml index 62134715f..c5b954070 100644 --- a/other/java/client/pom.xml +++ b/other/java/client/pom.xml @@ -5,7 +5,7 @@ com.github.chrislusf seaweedfs-client - 1.3.6 + 1.3.8 org.sonatype.oss diff --git a/other/java/client/pom.xml.deploy b/other/java/client/pom.xml.deploy index 62134715f..c5b954070 100644 --- a/other/java/client/pom.xml.deploy +++ b/other/java/client/pom.xml.deploy @@ -5,7 +5,7 @@ com.github.chrislusf seaweedfs-client - 1.3.6 + 1.3.8 org.sonatype.oss diff --git a/other/java/client/pom_debug.xml b/other/java/client/pom_debug.xml index dcedc2aa6..ebdf2e3af 100644 --- a/other/java/client/pom_debug.xml +++ b/other/java/client/pom_debug.xml @@ -5,7 +5,7 @@ com.github.chrislusf seaweedfs-client - 1.3.6 + 1.3.8 org.sonatype.oss diff --git a/other/java/hdfs2/dependency-reduced-pom.xml b/other/java/hdfs2/dependency-reduced-pom.xml index 218021a58..287200fd1 100644 --- a/other/java/hdfs2/dependency-reduced-pom.xml +++ b/other/java/hdfs2/dependency-reduced-pom.xml @@ -15,8 +15,8 @@ maven-compiler-plugin - 7 - 7 + 8 + 8 @@ -127,7 +127,7 @@ - 1.3.6 + 1.3.8 2.9.2 diff --git a/other/java/hdfs2/pom.xml b/other/java/hdfs2/pom.xml index 94f80a114..150e37e41 100644 --- a/other/java/hdfs2/pom.xml +++ b/other/java/hdfs2/pom.xml @@ -5,7 +5,7 @@ 4.0.0 - 1.3.6 + 1.3.8 2.9.2 @@ -31,8 +31,8 @@ org.apache.maven.plugins maven-compiler-plugin - 7 - 7 + 8 + 8 diff --git a/other/java/hdfs3/dependency-reduced-pom.xml b/other/java/hdfs3/dependency-reduced-pom.xml index 00e236aa2..51c4a4473 100644 --- a/other/java/hdfs3/dependency-reduced-pom.xml +++ b/other/java/hdfs3/dependency-reduced-pom.xml @@ -127,7 +127,7 @@ - 1.3.6 + 1.3.8 3.1.1 diff --git a/other/java/hdfs3/pom.xml b/other/java/hdfs3/pom.xml index a03068a48..cf7a6a632 100644 --- a/other/java/hdfs3/pom.xml +++ b/other/java/hdfs3/pom.xml @@ -5,7 +5,7 @@ 4.0.0 - 1.3.6 + 1.3.8 3.1.1 @@ -31,8 +31,8 @@ org.apache.maven.plugins maven-compiler-plugin - 7 - 7 + 8 + 8 diff --git a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java index 9ea26776b..d5c8ae521 100644 --- a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java +++ b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java @@ -216,12 +216,35 @@ public class SeaweedOutputStream extends OutputStream implements Syncable, Strea 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 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; } -- cgit v1.2.3 From f90d2c93c9e34997a8e76aeefb438ec06b1cd093 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 17 Jul 2020 16:11:57 -0700 Subject: 1.3.9 remove logs --- other/java/client/pom.xml | 2 +- other/java/client/pom.xml.deploy | 2 +- other/java/client/pom_debug.xml | 2 +- other/java/hdfs2/dependency-reduced-pom.xml | 2 +- other/java/hdfs2/pom.xml | 2 +- .../java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java | 4 ++-- other/java/hdfs3/dependency-reduced-pom.xml | 6 +++--- other/java/hdfs3/pom.xml | 2 +- .../java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java | 4 ++-- 9 files changed, 13 insertions(+), 13 deletions(-) (limited to 'other/java') diff --git a/other/java/client/pom.xml b/other/java/client/pom.xml index c5b954070..d60351671 100644 --- a/other/java/client/pom.xml +++ b/other/java/client/pom.xml @@ -5,7 +5,7 @@ com.github.chrislusf seaweedfs-client - 1.3.8 + 1.3.9 org.sonatype.oss diff --git a/other/java/client/pom.xml.deploy b/other/java/client/pom.xml.deploy index c5b954070..d60351671 100644 --- a/other/java/client/pom.xml.deploy +++ b/other/java/client/pom.xml.deploy @@ -5,7 +5,7 @@ com.github.chrislusf seaweedfs-client - 1.3.8 + 1.3.9 org.sonatype.oss diff --git a/other/java/client/pom_debug.xml b/other/java/client/pom_debug.xml index ebdf2e3af..c9fb256d4 100644 --- a/other/java/client/pom_debug.xml +++ b/other/java/client/pom_debug.xml @@ -5,7 +5,7 @@ com.github.chrislusf seaweedfs-client - 1.3.8 + 1.3.9 org.sonatype.oss diff --git a/other/java/hdfs2/dependency-reduced-pom.xml b/other/java/hdfs2/dependency-reduced-pom.xml index 287200fd1..62cd798b4 100644 --- a/other/java/hdfs2/dependency-reduced-pom.xml +++ b/other/java/hdfs2/dependency-reduced-pom.xml @@ -127,7 +127,7 @@ - 1.3.8 + 1.3.9 2.9.2 diff --git a/other/java/hdfs2/pom.xml b/other/java/hdfs2/pom.xml index 150e37e41..1f4e7aebc 100644 --- a/other/java/hdfs2/pom.xml +++ b/other/java/hdfs2/pom.xml @@ -5,7 +5,7 @@ 4.0.0 - 1.3.8 + 1.3.9 2.9.2 diff --git a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java index c1e7b593d..46de0c443 100644 --- a/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java +++ b/other/java/hdfs2/src/main/java/seaweed/hdfs/SeaweedOutputStream.java @@ -184,9 +184,9 @@ public class SeaweedOutputStream extends OutputStream { waitForTaskToComplete(); } final Future job = completionService.submit(() -> { - System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")"); + // 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) + ")"); + // System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")"); bufferToWrite.clear(); ByteBufferPool.release(bufferToWrite); return null; diff --git a/other/java/hdfs3/dependency-reduced-pom.xml b/other/java/hdfs3/dependency-reduced-pom.xml index 51c4a4473..5a077cb31 100644 --- a/other/java/hdfs3/dependency-reduced-pom.xml +++ b/other/java/hdfs3/dependency-reduced-pom.xml @@ -15,8 +15,8 @@ maven-compiler-plugin - 7 - 7 + 8 + 8 @@ -127,7 +127,7 @@ - 1.3.8 + 1.3.9 3.1.1 diff --git a/other/java/hdfs3/pom.xml b/other/java/hdfs3/pom.xml index cf7a6a632..e289a1855 100644 --- a/other/java/hdfs3/pom.xml +++ b/other/java/hdfs3/pom.xml @@ -5,7 +5,7 @@ 4.0.0 - 1.3.8 + 1.3.9 3.1.1 diff --git a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java index d5c8ae521..c602a0d81 100644 --- a/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java +++ b/other/java/hdfs3/src/main/java/seaweed/hdfs/SeaweedOutputStream.java @@ -231,9 +231,9 @@ public class SeaweedOutputStream extends OutputStream implements Syncable, Strea waitForTaskToComplete(); } final Future job = completionService.submit(() -> { - System.out.println(path + " is going to save [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")"); + // 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) + ")"); + // System.out.println(path + " saved [" + (writePosition) + "," + ((writePosition) + bytesLength) + ")"); bufferToWrite.clear(); ByteBufferPool.release(bufferToWrite); return null; -- cgit v1.2.3