fix oom; do benchmark

all/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>catty-parent</artifactId>
     <groupId>pink.catty</groupId>
-    <version>0.2.2</version>
+    <version>0.2.3</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
   <artifactId>catty-all</artifactId>
@@ -30,27 +30,27 @@
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-core</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-invokers</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-config</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-extension</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-spring</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
   </dependencies>
 

benchmark/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>catty-parent</artifactId>
     <groupId>pink.catty</groupId>
-    <version>0.2.2</version>
+    <version>0.2.3</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
@@ -30,7 +30,7 @@
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-all</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
     <dependency>
       <groupId>org.apache.dubbo</groupId>
@@ -40,18 +40,50 @@
     <dependency>
       <groupId>com.weibo</groupId>
       <artifactId>motan-core</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-api</artifactId>
+        </exclusion>
+      </exclusions>
       <version>1.0.0</version>
     </dependency>
     <dependency>
       <groupId>com.weibo</groupId>
       <artifactId>motan-transport-netty</artifactId>
+      <exclusions>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-log4j12</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.slf4j</groupId>
+          <artifactId>slf4j-api</artifactId>
+        </exclusion>
+      </exclusions>
       <version>1.0.0</version>
     </dependency>
     <dependency>
       <groupId>com.google.protobuf</groupId>
       <artifactId>protobuf-java</artifactId>
       <version>3.8.0</version>
     </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+      <version>2.11.2</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-core</artifactId>
+      <version>2.11.2</version>
+      <scope>compile</scope>
+    </dependency>
   </dependencies>
 
   <build>

benchmark/src/main/java/pink/catty/benchmark/common/PojoWrkGateway.java

@@ -14,6 +14,7 @@
  */
 package pink.catty.benchmark.common;
 
+import io.netty.buffer.PooledByteBufAllocator;
 import pink.catty.benchmark.service.PojoService;
 import io.netty.bootstrap.ServerBootstrap;
 import io.netty.channel.Channel;
@@ -42,6 +43,7 @@ public void start(PojoService service) {
     workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);
     ChannelHandler handler = new PojoWrkHandler(service);
 
+    bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
     bootstrap.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .childHandler(

benchmark/src/main/java/pink/catty/benchmark/common/PojoWrkHandler.java

@@ -59,9 +59,6 @@ protected void channelRead0(ChannelHandlerContext ctx, FullHttpRequest msg) {
       String response = service.service(content);
       if (MD5Utils.md5(content).equals(response)) {
         okResponse(ctx);
-        if (LOGGER.isInfoEnabled()) {
-          LOGGER.info("Request result:success cost:{} ms", System.currentTimeMillis() - start);
-        }
       } else {
         badReponse(ctx);
         if (LOGGER.isInfoEnabled()) {

benchmark/src/main/java/pink/catty/benchmark/dubbo/DubboBenchmark.java

@@ -50,7 +50,7 @@ private static void startProvider() {
 
     protocol.setName("dubbo");
     protocol.setPort(25500);
-    protocol.setThreads(256);
+    protocol.setThreads(400);
     protocol.setHost("0.0.0.0");
 
     ServiceConfig<PojoService> service = new ServiceConfig<>();

benchmark/src/main/resources/log4j2.xml

@@ -0,0 +1,17 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--   ALL < TRACE < DEBUG < INFO < WARN < ERROR < FATAL < OFF   -->
+<Configuration status="warn">
+
+  <Appenders>
+    <Console name="Console" target="SYSTEM_OUT">
+      <PatternLayout pattern="%d{HH:mm:ss.SSS} [%t] %-5level %logger{36} - %msg%n"/>
+    </Console>
+  </Appenders>
+
+  <Loggers>
+    <Root level="info">
+      <appender-ref ref="Console"/>
+    </Root>
+  </Loggers>
+</Configuration>

config/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>catty-parent</artifactId>
     <groupId>pink.catty</groupId>
-    <version>0.2.2</version>
+    <version>0.2.3</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 
@@ -30,12 +30,12 @@
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-invokers</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
     <dependency>
       <groupId>pink.catty</groupId>
       <artifactId>catty-extension</artifactId>
-      <version>0.2.2</version>
+      <version>0.2.3</version>
     </dependency>
   </dependencies>
 

core/pom.xml

@@ -20,7 +20,7 @@
   <parent>
     <artifactId>catty-parent</artifactId>
     <groupId>pink.catty</groupId>
-    <version>0.2.2</version>
+    <version>0.2.3</version>
   </parent>
   <modelVersion>4.0.0</modelVersion>
 

core/src/main/java/pink/catty/core/config/InnerClientConfig.java

@@ -73,4 +73,5 @@ public boolean equals(Object o) {
   public int hashCode() {
     return Objects.hash(ip, port, address, timeout, codecType);
   }
+
 }

core/src/main/java/pink/catty/core/extension/spi/AbstractEndpointFactory.java

@@ -15,36 +15,70 @@
 package pink.catty.core.extension.spi;
 
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import pink.catty.core.config.InnerClientConfig;
 import pink.catty.core.config.InnerServerConfig;
 import pink.catty.core.extension.ExtensionFactory;
 import pink.catty.core.invoker.Client;
 import pink.catty.core.invoker.Server;
+import pink.catty.core.support.ConcurrentHashSet;
 
 public abstract class AbstractEndpointFactory implements EndpointFactory {
 
+  protected Logger logger = LoggerFactory.getLogger(getClass());
+
   private Map<InnerClientConfig, Client> clientCache = new ConcurrentHashMap<>();
   private Map<InnerServerConfig, Server> serverCache = new ConcurrentHashMap<>();
+  private final Set<InnerClientConfig> inCreationClient = new ConcurrentHashSet<>();
+  private final Set<InnerServerConfig> inCreationServer = new ConcurrentHashSet<>();
 
   @Override
   public Client createClient(InnerClientConfig clientConfig) {
     Client client = clientCache.get(clientConfig);
-    if(client == null || !client.isAvailable()) {
-      Codec codec = ExtensionFactory.getCodec().getExtensionSingleton(clientConfig.getCodecType());
-      client = doCreateClient(clientConfig, codec);
-      clientCache.put(clientConfig, client);
+    if(client != null && !client.isAvailable()) {
+      clientCache.remove(clientConfig);
+      client = null;
+    }
+    if(client == null) {
+      if(!inCreationClient.contains(clientConfig)) {
+        synchronized (inCreationClient) {
+          if(!inCreationClient.contains(clientConfig)) {
+            inCreationClient.add(clientConfig);
+            Codec codec = ExtensionFactory.getCodec().getExtensionSingleton(clientConfig.getCodecType());
+            client = doCreateClient(clientConfig, codec);
+            clientCache.put(clientConfig, client);
+            inCreationClient.remove(clientConfig);
+            logger.info("EndpointFactory: a new client has bean created. ip: {}, port: {}.",
+                clientConfig.getServerIp(), clientConfig.getServerPort());
+          }
+        }
+      }
     }
     return client;
   }
 
   @Override
   public Server createServer(InnerServerConfig serverConfig) {
     Server server = serverCache.get(serverConfig);
-    if(server == null || !server.isAvailable()) {
-      Codec codec = ExtensionFactory.getCodec().getExtensionSingleton(serverConfig.getCodecType());
-      server = doCreateServer(serverConfig, codec);
-      serverCache.put(serverConfig, server);
+    if(server != null && !server.isAvailable()) {
+      serverCache.remove(serverConfig);
+      server = null;
+    }
+    if(server == null) {
+      if(!inCreationServer.contains(serverConfig)) {
+        synchronized (inCreationServer) {
+          if(!inCreationServer.contains(serverConfig)) {
+            Codec codec = ExtensionFactory.getCodec().getExtensionSingleton(serverConfig.getCodecType());
+            server = doCreateServer(serverConfig, codec);
+            serverCache.put(serverConfig, server);
+            logger.info("EndpointFactory: a new server has bean created. ip: {}, port: {}.",
+                serverConfig.getServerAddress().getIp(), serverConfig.getServerAddress().getPort());
+          }
+        }
+      }
     }
     return server;
   }

core/src/main/java/pink/catty/core/extension/spi/Codec.java

@@ -16,7 +16,7 @@
 
 import pink.catty.core.CodecException;
 
-public interface Codec extends PackageReader {
+public interface Codec<I, O> extends PackageReader<I, O> {
 
   byte[] encode(Object message, DataTypeEnum dataTypeEnum) throws CodecException;
 

core/src/main/java/pink/catty/core/extension/spi/PackageReader.java

@@ -14,16 +14,24 @@
  */
 package pink.catty.core.extension.spi;
 
-public interface PackageReader {
+import java.util.List;
+
+/**
+ * Distinguish integral data package from byte-streaming.
+ *
+ * @param <I> Different framework might have different encapsulation of byte-streaming such as
+ * ByteBuffer from jdk, ByteBuf from netty and byte[].
+ */
+public interface PackageReader<I, O> {
 
   /**
    * Reading data package received from tcp or other stream completely.
    *
-   * @param dataPackage data received from tcp.
-   * @return return an complete data.
+   * @param data data received from tcp.
+   * @param out integral data
+   * @return return the rest of data.
    * @throws BrokenDataPackageException If reading data package occurs some error.
-   * @throws PartialDataException If dataPackage does not contain an complete data.
    */
-  CompletePackage readPackage(byte[] dataPackage) throws PartialDataException, BrokenDataPackageException;
+  void readPackage(I data, List<O> out) throws BrokenDataPackageException;
 
 }