[561308] Optimize bulk read/write operations in buffer streams

https://bugs.eclipse.org/bugs/show_bug.cgi?id=561308

3 files changed, 160 insertions, 2 deletions

diff --git a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/AllTests.java b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/AllTests.java
index 5b20fc74b7..6fa852f166 100644
--- a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/AllTests.java
+++ b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/AllTests.java
@@ -54,6 +54,7 @@ public class AllTests
     suite.addTestSuite(SortedFileMapTest.class);
     suite.addTestSuite(SynchronizingCorrelatorTest.class);
     suite.addTestSuite(BufferPoolTest.class);
+    suite.addTestSuite(BufferStreamTest.class);
     suite.addTestSuite(ExtendedIOTest.class);
     suite.addTestSuite(StringCompressorTest.class);
     suite.addTestSuite(SecurityTest.class);
diff --git a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/BufferStreamTest.java b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/BufferStreamTest.java
new file mode 100644
index 0000000000..474523b4b4
--- /dev/null
+++ b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/BufferStreamTest.java
@@ -0,0 +1,141 @@
+/*
+ * Copyright (c) 2008, 2009, 2011, 2012, 2015, 2016, 2019 Eike Stepper (Loehne, Germany) and others.
+ * All rights reserved. This program and the accompanying materials
+ * are made available under the terms of the Eclipse Public License v1.0
+ * which accompanies this distribution, and is available at
+ * http://www.eclipse.org/legal/epl-v10.html
+ *
+ * Contributors:
+ *    Eike Stepper - initial API and implementation
+ *    Teerawat Chaiyakijpichet (No Magic Asia Ltd.) - SSL
+ */
+package org.eclipse.net4j.tests;
+
+import static org.junit.Assert.assertArrayEquals;
+
+import org.eclipse.net4j.Net4jUtil;
+import org.eclipse.net4j.buffer.BufferInputStream;
+import org.eclipse.net4j.buffer.BufferOutputStream;
+import org.eclipse.net4j.buffer.IBuffer;
+import org.eclipse.net4j.buffer.IBufferHandler;
+import org.eclipse.net4j.buffer.IBufferPool;
+import org.eclipse.net4j.tests.data.HugeData;
+import org.eclipse.net4j.util.io.IOUtil;
+import org.eclipse.net4j.util.tests.AbstractOMTest;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * @author Eike Stepper
+ */
+public class BufferStreamTest extends AbstractOMTest
+{
+  private static final short CHANNEL_ID = 0;
+
+  public void testReadArray() throws Exception
+  {
+    byte[] data = { 1, 2, 3, 4, 5, 6, 7, 8, 9 };
+
+    BufferInputStream in = new BufferInputStream();
+    byte[] result = new byte[2 * data.length];
+    AtomicInteger bytesRead = new AtomicInteger();
+
+    IBufferPool bufferPool = Net4jUtil.createBufferPool();
+    BufferOutputStream out = new BufferOutputStream(new BufferStreamPipe(in), bufferPool, CHANNEL_ID);
+
+    Thread consumer = new Thread("CONSUMER")
+    {
+      @Override
+      public void run()
+      {
+        try
+        {
+          bytesRead.set(in.read(result));
+        }
+        catch (IOException ex)
+        {
+          ex.printStackTrace();
+        }
+      }
+    };
+
+    consumer.setDaemon(true);
+    consumer.start();
+
+    out.write(data);
+    out.flushWithEOS();
+    out.close();
+
+    consumer.join(10000000 * DEFAULT_TIMEOUT);
+
+    assertEquals(data.length, bytesRead.get());
+    byte[] actual = new byte[bytesRead.get()];
+    System.arraycopy(result, 0, actual, 0, bytesRead.get());
+    assertArrayEquals(data, actual);
+  }
+
+  public void testCopyBinary() throws Exception
+  {
+    byte[] data = HugeData.getBytes();
+
+    BufferInputStream in = new BufferInputStream();
+    ByteArrayOutputStream result = new ByteArrayOutputStream();
+
+    IBufferPool bufferPool = Net4jUtil.createBufferPool();
+    BufferOutputStream out = new BufferOutputStream(new BufferStreamPipe(in), bufferPool, CHANNEL_ID);
+
+    Thread consumer = new Thread("CONSUMER")
+    {
+      @Override
+      public void run()
+      {
+        try
+        {
+          IOUtil.copyBinary(in, result);
+        }
+        catch (IOException ex)
+        {
+          ex.printStackTrace();
+        }
+      }
+    };
+
+    consumer.setDaemon(true);
+    consumer.start();
+
+    out.write(data);
+    out.flushWithEOS();
+    out.close();
+
+    consumer.join(10000000 * DEFAULT_TIMEOUT);
+    assertArrayEquals(data, result.toByteArray());
+  }
+
+  /**
+   * @author Eike Stepper
+   */
+  private static final class BufferStreamPipe implements IBufferHandler
+  {
+    private final BufferInputStream in;
+
+    public BufferStreamPipe(BufferInputStream in)
+    {
+      this.in = in;
+    }
+
+    @Override
+    public void handleBuffer(IBuffer buffer)
+    {
+      ByteBuffer byteBuffer = buffer.getByteBuffer();
+      short payload = (short)(byteBuffer.position() - IBuffer.HEADER_SIZE);
+
+      byteBuffer.flip();
+      byteBuffer.putShort(CHANNEL_ID);
+      byteBuffer.putShort(payload);
+      in.handleBuffer(buffer);
+    }
+  }
+}
diff --git a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/ChannelTest.java b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/ChannelTest.java
index e4766e8f72..427e44e345 100644
--- a/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/ChannelTest.java
+++ b/plugins/org.eclipse.net4j.tests/src/org/eclipse/net4j/tests/ChannelTest.java
@@ -11,6 +11,8 @@
  */
 package org.eclipse.net4j.tests;
 
+import static org.junit.Assert.assertArrayEquals;
+
 import org.eclipse.net4j.ITransportConfigAware;
 import org.eclipse.net4j.channel.IChannel;
 import org.eclipse.net4j.connector.IConnector;
@@ -167,12 +169,26 @@ public class ChannelTest extends AbstractConfigTest
     assertActive(protocol);
 
     byte[] data = HugeData.getBytes();
-    for (int i = 0; i < 10000; i++)
+
+    // Warm-up
+    for (int i = 0; i < 1000; i++)
     {
       byte[] result = new ArrayRequest(protocol, data).send();
       assertEquals(true, Arrays.equals(data, result));
     }
 
+    final int SIGNALS = 10000;
+
+    long start = System.currentTimeMillis();
+    for (int i = 0; i < SIGNALS; i++)
+    {
+      byte[] result = new ArrayRequest(protocol, data).send();
+      assertArrayEquals(data, result);
+    }
+
+    long duration = System.currentTimeMillis() - start;
+    log("Millis for " + SIGNALS + " signals: " + duration);
+
     InternalChannel channel = (InternalChannel)protocol.getChannel();
     log("Sent buffers: " + channel.getSentBuffers());
     log("Received buffers: " + channel.getReceivedBuffers());
@@ -301,7 +317,7 @@ public class ChannelTest extends AbstractConfigTest
     {
       if (protocols != null)
       {
-        for (TestSignalProtocol protocol : new ArrayList<TestSignalProtocol>(protocols))
+        for (TestSignalProtocol protocol : new ArrayList<>(protocols))
         {
           protocol.close();
         }