Avoid bytebufer copy

unknow-server-bench/src/main/java/unknow/server/bench/Main.java

@@ -17,13 +17,15 @@
 
 public class Main {
 	public static void main(String[] args) throws Exception {
-		Options o = new OptionsBuilder().forks(1).measurementIterations(10).verbosity(VerboseMode.SILENT).warmupIterations(5).build();
 
-		try (PrintStream w = args.length > 0 ? new PrintStream(Files.newOutputStream(Paths.get(args[0])), false, StandardCharsets.UTF_8) : System.out) {
-			for (Class<?> c : Arrays.asList(EncoderDecoder.class, BenchJaxb.class, BenchDocument.class, BenchProtostuff.class)) {
+		Options o = new OptionsBuilder().forks(1).measurementIterations(10).verbosity(VerboseMode.NORMAL).warmupIterations(5).build();
+
+		try (PrintStream w = args.length > 0 ? new PrintStream(Files.newOutputStream(Paths.get(args[0])), false, StandardCharsets.UTF_8) : System.err) {
+			for (Class<?> c : Arrays.asList(EncoderDecoder.class/*, BenchJaxb.class, BenchDocument.class, BenchProtostuff.class*/)) {
+				Collection<RunResult> result = new Runner(new OptionsBuilder().parent(o).include(c.getName()).build()).run();
+
 				w.println(c.getSimpleName());
 				w.println("```");
-				Collection<RunResult> result = new Runner(new OptionsBuilder().parent(o).include(c.getName()).build()).run();
 				ResultFormatFactory.getInstance(ResultFormatType.TEXT, w).writeOut(result);
 				w.println("```");
 			}

unknow-server-nio/src/main/java/unknow/server/nio/ConnectionStats.java

@@ -1,5 +1,8 @@
 package unknow.server.nio;
 
+/**
+ * statistics of a nio connection
+ */
 public class ConnectionStats {
 	private final boolean hasPengingWrite;
 	private final boolean closing;
@@ -13,18 +16,30 @@ public ConnectionStats(boolean hasPengingWrite, boolean closing, long lastCheck,
 		this.lastAction = lastAction;
 	}
 
+	/**
+	 * @return true if the connection has pending write
+	 */
 	public boolean hasPengingWrite() {
 		return hasPengingWrite;
 	}
 
+	/**
+	 * @return true if the connection is closing
+	 */
 	public boolean isClosing() {
 		return closing;
 	}
 
+	/**
+	 * @return nanotime of last connection check
+	 */
 	public long lastCheck() {
 		return lastCheck;
 	}
 
+	/**
+	 * @return nano time of last action  on the connection
+	 */
 	public long lastAction() {
 		return lastAction;
 	}

unknow-server-nio/src/main/java/unknow/server/nio/NIOConnection.java

@@ -10,9 +10,10 @@
 import java.nio.channels.SelectionKey;
 import java.nio.channels.SocketChannel;
 import java.util.Arrays;
-import java.util.concurrent.BlockingQueue;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Future;
-import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.net.ssl.SSLEngine;
 
@@ -36,10 +37,13 @@ public final class NIOConnection extends NIOHandlerDelegate {
 	protected final SelectionKey key;
 	protected final SocketChannel channel;
 
+	private final AtomicBoolean writeScheduled;
+	private final WriteCheck writeCheck;
+
 	private InetSocketAddress local;
 	private InetSocketAddress remote;
 
-	final BlockingQueue<ByteBuffer> pending;
+	final Queue<ByteBuffer> pending;
 	final ByteBuffers writes;
 
 	long lastCheck;
@@ -59,8 +63,10 @@ public NIOConnection(NIOWorker worker, SelectionKey key, NIOConnectionHandler ha
 		this.worker = worker;
 		this.key = key;
 		this.channel = (SocketChannel) key.channel();
+		this.writeScheduled = new AtomicBoolean(false);
+		this.writeCheck = new WriteCheck();
 		this.out = new Out(this);
-		this.pending = new LinkedBlockingDeque<>();
+		this.pending = new ConcurrentLinkedQueue<>();
 		this.writes = new ByteBuffers(16);
 	}
 
@@ -91,25 +97,14 @@ public void init(NIOConnection co, long now, SSLEngine sslEngine) throws IOExcep
 	 * add a buffers to the writing queue
 	 * 
 	 * @param buf buffer to be written
-	 * @throws InterruptedException in case of interruption
-	 * @throws IOException
+	 * @throws IOException in case of io error
 	 */
-	public final void write(ByteBuffer buf) throws InterruptedException, IOException {
+	public final void write(ByteBuffer buf) throws IOException {
 		if (!key.isValid())
 			throw new IOException("already closed");
-		pending.put(buf);
-		if (pending.size() > 10)
-			flush();
-		else
-			key.interestOpsOr(SelectionKey.OP_WRITE);
-	}
-
-	@SuppressWarnings("resource")
-	public final void flush() {
-		if (!hasPendingWrites())
-			return;
-		key.interestOpsOr(SelectionKey.OP_WRITE);
-		key.selector().wakeup();
+		pending.offer(buf);
+		if (writeScheduled.compareAndSet(false, true))
+			execute(writeCheck);
 	}
 
 	/**
@@ -161,17 +156,23 @@ public boolean isClosed() {
 	}
 
 	protected final void beforeWrite(long now) throws IOException {
-		while (writes.len < 16 && !pending.isEmpty())
-			handler.prepareWrite(pending.poll(), now, writes);
-		handler.beforeWrite(now, writes);
+		ByteBuffer b;
+		while ((b = pending.poll()) != null)
+			writes.accept(b);
+		handler.transformWrite(writes, now);
 	}
 
 	@Override
 	public final void onWrite(long now) throws IOException {
 		lastAction = now;
 		writes.compact();
-		if (!hasPendingWrites())
-			key.interestOpsAnd(~SelectionKey.OP_WRITE);
+		if (!hasPendingWrites()) {
+			writeScheduled.set(false);
+			if (!hasPendingWrites())
+				key.interestOpsAnd(~SelectionKey.OP_WRITE);
+			else if (writeScheduled.compareAndSet(false, true))
+				key.interestOpsOr(SelectionKey.OP_WRITE);
+		}
 		handler.onWrite(now);
 	}
 
@@ -203,18 +204,19 @@ public final void doneClosing() {
 
 	@Override
 	public String toString() {
-		return getClass() + "[local=" + getLocal() + " remote=" + getRemote() + "]";
+		return getClass() + "[local=" + getLocal() + " remote=" + getRemote() + "] writes: " + hasPendingWrites();
 	}
 
 	/** output stream for this connection */
 	public static final class Out extends OutputStream {
+		private static final int BUF_SIZE = 16 * 1024;
 		private NIOConnection h;
 
 		private ByteBuffer buf;
 
 		private Out(NIOConnection h) {
 			this.h = h;
-			this.buf = ByteBuffer.allocate(4096);
+			this.buf = ByteBuffer.allocate(BUF_SIZE);
 		}
 
 		@Override
@@ -246,15 +248,10 @@ public synchronized void write(byte[] b, int off, int len) throws IOException {
 				len -= r;
 				off += r;
 				writeBuffer();
-				if (len < 4096)
+				if (len < BUF_SIZE)
 					buf.put(b, off, len);
 				else
-					try {
-						h.write(ByteBuffer.wrap(Arrays.copyOfRange(b, off, off + len)));
-					} catch (InterruptedException e) {
-						Thread.currentThread().interrupt();
-						throw new IOException(e);
-					}
+					h.write(ByteBuffer.wrap(Arrays.copyOfRange(b, off, off + len)));
 			} else if (len == r) {
 				buf.put(b, off, len);
 				writeBuffer();
@@ -272,12 +269,7 @@ public synchronized void write(ByteBuffer b) throws IOException {
 			if (h == null)
 				throw new IOException("already closed");
 			writeBuffer();
-			try {
-				h.write(b);
-			} catch (InterruptedException e) {
-				Thread.currentThread().interrupt();
-				throw new IOException(e);
-			}
+			h.write(b);
 		}
 
 		@Override
@@ -302,19 +294,23 @@ public synchronized void flush() throws IOException {
 			if (h == null)
 				return;
 			writeBuffer();
-			h.flush();
 		}
 
 		private void writeBuffer() throws IOException {
 			if (h == null || buf.position() == 0)
 				return;
-			try {
-				h.write(buf.flip());
-				buf = ByteBuffer.allocate(4096);
-			} catch (InterruptedException e) {
-				Thread.currentThread().interrupt();
-				throw new IOException(e);
-			}
+			h.write(buf.flip());
+			buf = ByteBuffer.allocate(BUF_SIZE);
+		}
+	}
+
+	private class WriteCheck implements WorkerTask {
+		@Override
+		public void run(long now) {
+			if (hasPendingWrites())
+				key.interestOpsOr(SelectionKey.OP_WRITE);
+			else
+				writeScheduled.set(false);
 		}
 	}
 }

unknow-server-nio/src/main/java/unknow/server/nio/NIOConnectionHandler.java

@@ -6,6 +6,11 @@
 
 import javax.net.ssl.SSLEngine;
 
+import unknow.server.util.io.ByteBuffers;
+
+/**
+ * handle nio connection event
+ */
 @SuppressWarnings("unused")
 public interface NIOConnectionHandler {
 
@@ -40,33 +45,21 @@ default void onHandshakeDone(SSLEngine sslEngine, long now) throws IOException {
 	/**
 	 * called after some data has been read
 	 * 
-	 * @param b the read buffers
+	 * @param b the read buffers (should read all or copy content)
 	 * @param now nanoTime
 	 * @throws IOException on io exception
 	 */
 	default void onRead(ByteBuffer b, long now) throws IOException { // ok
 	}
 
 	/**
-	 * called before a buffer is written (allow to collect buffers)
-	 * 
-	 * @param b buffer to be written
-	 * @param now nanoTime
-	 * @param c consumer of generated buffers
-	 * @throws IOException on io exception
-	 */
-	default void prepareWrite(ByteBuffer b, long now, Consumer<ByteBuffer> c) throws IOException {
-		c.accept(b);
-	}
-
-	/**
-	 * called before a buffer is written
+	 * called before some buffers are written
 	 * 
+	 * @param buffers buffers to be written
 	 * @param now nanoTime
-	 * @param c consumer of generated buffers
 	 * @throws IOException on io exception
 	 */
-	default void beforeWrite(long now, Consumer<ByteBuffer> c) throws IOException { // ok
+	default void transformWrite(ByteBuffers buffers, long now) throws IOException { // ok
 	}
 
 	/**

unknow-server-nio/src/main/java/unknow/server/nio/NIOHandlerDelegate.java

@@ -2,10 +2,11 @@
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.util.function.Consumer;
 
 import javax.net.ssl.SSLEngine;
 
+import unknow.server.util.io.ByteBuffers;
+
 public class NIOHandlerDelegate implements NIOConnectionHandler {
 	protected final NIOConnectionHandler handler;
 
@@ -34,13 +35,8 @@ public void onRead(ByteBuffer b, long now) throws IOException {
 	}
 
 	@Override
-	public void prepareWrite(ByteBuffer b, long now, Consumer<ByteBuffer> c) throws IOException {
-		handler.prepareWrite(b, now, c);
-	}
-
-	@Override
-	public void beforeWrite(long now, Consumer<ByteBuffer> c) throws IOException {
-		handler.beforeWrite(now, c);
+	public void transformWrite(ByteBuffers buffers, long now) throws IOException {
+		handler.transformWrite(buffers, now);
 	}
 
 	@Override

unknow-server-nio/src/main/java/unknow/server/nio/NIOLoop.java

@@ -12,6 +12,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
+ * basic selector loop
  * @author unknow
  */
 public class NIOLoop implements Runnable {