drop using double write buffer

Author: JavaSaBr

README.md

@@ -14,7 +14,7 @@ repositories {
 }
 
 ext {
- rlibVersion = "10.0.alpha3"
+ rlibVersion = "10.0.alpha4"
 }
 
 dependencies {

build.gradle

@@ -1,4 +1,4 @@
-rootProject.version = "10.0.alpha3"
+rootProject.version = "10.0.alpha4"
 group = 'javasabr.rlib'
 
 allprojects {

rlib-network/src/loadTest/java/javasabr/rlib/network/StringNetworkLoadTest.java

@@ -130,7 +130,7 @@ void testServerWithMultiplyClients() {
  var serverAllocator = new DefaultBufferAllocator(serverConfig);
  ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
 
- int clientCount = 200;
+ int clientCount = 500;
  int messagesPerIteration = 3_000;
  int expectedMessages = clientCount * messagesPerIteration * MAX_ITERATIONS;
 

rlib-network/src/main/java/javasabr/rlib/network/ServerNetworkConfig.java

@@ -77,6 +77,7 @@ default int scheduledThreadGroupSize() {
  return 1;
  }
 
+
  /**
  * Get a thread constructor which should be used to create network threads.
  */

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/AbstractNetworkPacketReader.java

@@ -121,6 +121,8 @@ protected void startReadImpl() {
  if (reading.compareAndSet(true, false)) {
  retryReadLater();
  }
+ } catch (Error error) {
+ throw error;
  }
  }
 

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/AbstractNetworkPacketWriter.java

@@ -53,14 +53,10 @@ public void failed(Throwable exc, @Nullable WritableNetworkPacket<C> packet) {
  final C connection;
  final AsynchronousSocketChannel socketChannel;
  final BufferAllocator bufferAllocator;
- final ByteBuffer firstWriteBuffer;
- final ByteBuffer secondWriteBuffer;
+ final ByteBuffer writeBuffer;
 
  @Nullable
- volatile ByteBuffer firstWriteTempBuffer;
- @Nullable
- volatile ByteBuffer secondWriteTempBuffer;
-
+ volatile ByteBuffer writeTempBuffer;
  @Getter(AccessLevel.PROTECTED)
  volatile ByteBuffer writingBuffer = EMPTY_BUFFER;
 
@@ -80,8 +76,7 @@ public AbstractNetworkPacketWriter(
  this.connection = connection;
  this.socketChannel = socketChannel;
  this.bufferAllocator = bufferAllocator;
- this.firstWriteBuffer = bufferAllocator.takeWriteBuffer();
- this.secondWriteBuffer = bufferAllocator.takeWriteBuffer();
+ this.writeBuffer = bufferAllocator.takeWriteBuffer();
  this.updateActivityFunction = updateActivityFunction;
  this.writablePacketProvider = packetProvider;
  this.serializedToChannelPacketHandler = serializedToChannelPacketHandler;
@@ -156,28 +151,23 @@ protected ByteBuffer serialize(WritableNetworkPacket<C> packet) {
  int totalSize = expectedLength == -1 ? -1 : totalSize(packet, expectedLength);
 
  // if the packet is too big to use a write buffer
- if (expectedLength != -1 && totalSize > firstWriteBuffer.capacity()) {
- ByteBuffer first = bufferAllocator.takeBuffer(totalSize);
- ByteBuffer second = bufferAllocator.takeBuffer(totalSize);
- firstWriteTempBuffer = first;
- secondWriteTempBuffer = second;
+ if (expectedLength != -1 && totalSize > writeBuffer.capacity()) {
+ ByteBuffer tempBuffer = bufferAllocator.takeBuffer(totalSize);
  try {
- return serialize(resultPacket, expectedLength, totalSize, first, second);
+ ByteBuffer serialized = serialize(resultPacket, expectedLength, totalSize, tempBuffer);
+ writeTempBuffer = tempBuffer;
+ return serialized;
  } catch (BufferOverflowException ex) {
  log.error(ex);
- bufferAllocator.putBuffer(first);
- bufferAllocator.putBuffer(second);
- firstWriteTempBuffer = null;
- secondWriteTempBuffer = null;
+ bufferAllocator.putBuffer(tempBuffer);
+ writeTempBuffer = null;
  throw new RuntimeException(ex);
  }
  } else {
  try {
- return serialize(resultPacket, expectedLength, totalSize, firstWriteBuffer, secondWriteBuffer);
+ return serialize(resultPacket, expectedLength, totalSize, writeBuffer);
  } catch (BufferOverflowException ex) {
  log.error(ex);
- firstWriteBuffer.clear();
- secondWriteBuffer.clear();
  throw new RuntimeException(ex);
  }
  }
@@ -196,84 +186,76 @@ protected ByteBuffer serialize(WritableNetworkPacket<C> packet) {
  * @param packet the network packet to serialize.
  * @param expectedLength the packet's expected size.
  * @param totalSize the packet's total size.
- * @param firstBuffer the first buffer.
- * @param secondBuffer the second buffer.
+ * @param writeBuffer the write buffer.
  * @return the final buffer to write to channel.
  */
  protected ByteBuffer serialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
-
- if (!onBeforeSerialize(packet, expectedLength, totalSize, firstBuffer, secondBuffer)) {
- return firstBuffer.clear().limit(0);
- } else if (!doSerialize(packet, expectedLength, totalSize, firstBuffer, secondBuffer)) {
- return firstBuffer.clear().limit(0);
- } else if (!onAfterSerialize(packet, expectedLength, totalSize, firstBuffer, secondBuffer)) {
- return firstBuffer.clear().limit(0);
+ ByteBuffer writeBuffer) {
+
+ if (!onBeforeSerialize(packet, expectedLength, totalSize, writeBuffer)) {
+ return writeBuffer.clear().limit(0);
+ } else if (!doSerialize(packet, expectedLength, totalSize, writeBuffer)) {
+ return writeBuffer.clear().limit(0);
+ } else if (!onAfterSerialize(packet, expectedLength, totalSize, writeBuffer)) {
+ return writeBuffer.clear().limit(0);
  }
 
- return onSerializeResult(packet, expectedLength, totalSize, firstBuffer, secondBuffer);
+ return onSerializeResult(packet, expectedLength, totalSize, writeBuffer);
  }
 
  /**
- * Handles the buffers before serializing packet's data.
+ * Handles the buffer before serializing packet's data.
  *
  * @param packet the network packet.
  * @param expectedLength the packet's expected size.
  * @param totalSize the packet's total size.
- * @param firstBuffer the first buffer.
- * @param secondBuffer the second buffer.
+ * @param writeBuffer the write buffer.
  * @return true if handling was successful.
  */
  protected boolean onBeforeSerialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- firstBuffer.clear();
+ ByteBuffer writeBuffer) {
+ writeBuffer.clear();
  return true;
  }
 
  /**
- * Serializes the network packet data to the buffers.
+ * Serializes the network packet data to the buffer.
  *
  * @param packet the network packet.
  * @param expectedLength the packet's expected size.
  * @param totalSize the packet's total size.
- * @param firstBuffer the first buffer.
- * @param secondBuffer the second buffer.
+ * @param writeBuffer the first buffer.
  * @return true if writing was successful.
  */
  protected boolean doSerialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- return packet.write(connection, firstBuffer);
+ ByteBuffer writeBuffer) {
+ return packet.write(connection, writeBuffer);
  }
 
  /**
- * Handles the buffers after serializing packet's data.
+ * Handles the buffer after serializing packet's data.
  *
  * @param packet the network packet.
  * @param expectedLength the packet's expected size.
  * @param totalSize the packet's total size.
- * @param firstBuffer the first buffer.
- * @param secondBuffer the second buffer.
+ * @param writeBuffer the write buffer.
  * @return true if handling was successful.
  */
  protected boolean onAfterSerialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- firstBuffer.flip();
+ ByteBuffer writeBuffer) {
+ writeBuffer.flip();
  return true;
  }
 
@@ -283,17 +265,15 @@ protected boolean onAfterSerialize(
  * @param packet the network packet.
  * @param expectedLength the packet's expected size.
  * @param totalSize the packet's total size.
- * @param firstBuffer the first buffer.
- * @param secondBuffer the second buffer.
+ * @param writeBuffer the write buffer.
  * @return the result buffer.
  */
  protected ByteBuffer onSerializeResult(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- return firstBuffer.position(0);
+ ByteBuffer writeBuffer) {
+ return writeBuffer;
  }
 
  protected ByteBuffer writeHeader(ByteBuffer buffer, int position, int value, int headerSize) {
@@ -405,27 +385,18 @@ protected void handleFailedWritingData(Throwable exception, @Nullable WritableNe
 
  @Override
  public void close() {
- bufferAllocator
- .putWriteBuffer(firstWriteBuffer)
- .putWriteBuffer(secondWriteBuffer);
+ bufferAllocator.putWriteBuffer(writeBuffer);
  clearTempBuffers();
  writingBuffer = EMPTY_BUFFER;
  writing.compareAndSet(true, false);
  }
 
  protected void clearTempBuffers() {
  this.writingBuffer = EMPTY_BUFFER;
-
- var firstWriteTempBuffer = this.firstWriteTempBuffer;
- if (firstWriteTempBuffer != null) {
- this.firstWriteTempBuffer = null;
- bufferAllocator.putBuffer(firstWriteTempBuffer);
- }
-
- var secondWriteTempBuffer = this.secondWriteTempBuffer;
- if (secondWriteTempBuffer != null) {
- this.secondWriteTempBuffer = null;
- bufferAllocator.putBuffer(secondWriteTempBuffer);
+ var writeTempBuffer = this.writeTempBuffer;
+ if (writeTempBuffer != null) {
+ this.writeTempBuffer = null;
+ bufferAllocator.putBuffer(writeTempBuffer);
  }
  }
 }

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/DefaultNetworkPacketWriter.java

@@ -52,9 +52,8 @@ protected boolean onBeforeSerialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- firstBuffer
+ ByteBuffer writeBuffer) {
+ writeBuffer
  .clear()
  .position(packetLengthHeaderSize);
  return true;
@@ -65,9 +64,8 @@ protected ByteBuffer onSerializeResult(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- return writePacketLength(firstBuffer, firstBuffer.limit())
+ ByteBuffer writeBuffer) {
+ return writePacketLength(writeBuffer, writeBuffer.limit())
  .position(0);
  }
 

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/DefaultSslNetworkPacketWriter.java

@@ -57,9 +57,8 @@ protected boolean onBeforeSerialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- firstBuffer
+ ByteBuffer writeBuffer) {
+ writeBuffer
  .clear()
  .position(packetLengthHeaderSize);
  return true;
@@ -70,9 +69,9 @@ protected ByteBuffer onSerializeResult(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
-  ByteBuffer secondBuffer) {
- return writePacketLength(firstBuffer, firstBuffer.limit()).position(0);
+ ByteBuffer writeBuffer) {
+ return writePacketLength(writeBuffer, writeBuffer.limit())
+  .position(0);
  }
 
  protected ByteBuffer writePacketLength(ByteBuffer buffer, int packetLength) {

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/IdBasedNetworkPacketWriter.java

@@ -50,9 +50,8 @@ protected boolean doSerialize(
  W packet,
  int expectedLength,
  int totalSize,
- ByteBuffer firstBuffer,
- ByteBuffer secondBuffer) {
- writeHeader(firstBuffer, packet.packetId(), packetIdHeaderSize);
- return super.doSerialize(packet, expectedLength, totalSize, firstBuffer, secondBuffer);
+ ByteBuffer writeBuffer) {
+ writeHeader(writeBuffer, packet.packetId(), packetIdHeaderSize);
+ return super.doSerialize(packet, expectedLength, totalSize, writeBuffer);
  }
 }

rlib-network/src/main/java/javasabr/rlib/network/server/impl/DefaultServerNetwork.java

@@ -11,6 +11,7 @@
 import java.nio.channels.CompletionHandler;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.SynchronousQueue;
@@ -180,9 +181,9 @@ protected ExecutorService buildExecutor(ServerNetworkConfig config) {
  config.threadGroupMaxSize(),
  120,
  TimeUnit.SECONDS,
- new SynchronousQueue<>(),
+ new LinkedBlockingQueue<>(),
  threadFactory,
- new ThreadPoolExecutor.CallerRunsPolicy());
+ new ThreadPoolExecutor.AbortPolicy());
  } else {
  executorService = Executors.newFixedThreadPool(config.threadGroupMinSize(), threadFactory);
  }