From 0b9d73fb7c213d466fd0aef4615709c34a949bed Mon Sep 17 00:00:00 2001
From: Chris Lu <chris.lu@gmail.com>
Date: Fri, 17 Jul 2020 15:44:00 -0700
Subject: [PATCH] async copy

---
 .../seaweed/hdfs/SeaweedOutputStream.java     | 35 +++++++++++++++----
 1 file changed, 29 insertions(+), 6 deletions(-)

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<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;
 
     }