Introduce codec mapping caches to improve encoding and decoding performances.

* Switched the codec list to a thread safe variant to avoid the synchronized blocks. Even though `CopyOnWriteArrayList` is not super performant it should work fine in this context where the list should not be frequently updated.

* Switched `mockito-core` to `mockito-junit-jupiter` for Junit 5 support
* Refactored the codec registry to use a CodecFinder (default to SPI definition in the classpath)
* Provided 2 implementations of the codec finder, one without cache and another with cache
* Added a build cache method that will attempt to fill the cache when the codecs are updated. This cannot covers all the cases like the nested arrays, therefore for those type the cache will be filled dynamically on per-request basis
* Added microbenchmarks for codec encode and decode using the cache based implementation or not
* Disabled unixDomainSocketTest IT when running on Windows

[#444][resolves #409]

Author: vai-frd

pom.xml

@@ -210,7 +210,7 @@
         </dependency>
         <dependency>
             <groupId>org.mockito</groupId>
-            <artifactId>mockito-core</artifactId>
+            <artifactId>mockito-junit-jupiter</artifactId>
             <version>${mockito.version}</version>
             <scope>test</scope>
         </dependency>

src/jmh/java/io/r2dbc/postgresql/CodecRegistryBenchmarks.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.r2dbc.postgresql;
+
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import io.r2dbc.postgresql.codec.CodecFinderCacheImpl;
+import io.r2dbc.postgresql.codec.CodecFinderDefaultImpl;
+import io.r2dbc.postgresql.codec.Codecs;
+import io.r2dbc.postgresql.codec.DefaultCodecs;
+import io.r2dbc.postgresql.util.ByteBufUtils;
+import org.junit.platform.commons.annotation.Testable;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.infra.Blackhole;
+
+import java.time.LocalDateTime;
+import java.util.concurrent.TimeUnit;
+
+import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
+import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.FLOAT4;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.FLOAT8;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.FLOAT8_ARRAY;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.INT2;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.INT2_ARRAY;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.INT4;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.INT4_ARRAY;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.TIMESTAMP;
+import static io.r2dbc.postgresql.type.PostgresqlObjectId.VARCHAR;
+import static io.r2dbc.postgresql.util.TestByteBufAllocator.TEST;
+
+/**
+ * Benchmarks for codec encoding and decoding using cached enabled or disabled registries.
+ */
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+@Testable
+public class CodecRegistryBenchmarks extends BenchmarkSettings {
+
+    @State(Scope.Benchmark)
+    public static class CodecRegistryHolder {
+
+        @Param({"10", "100", "1000"})
+        public int iterations;
+
+        final ByteBufAllocator byteBufAllocator = new UnpooledByteBufAllocator(false, true);
+
+        DefaultCodecs cacheEnabledRegistry = new DefaultCodecs(byteBufAllocator, false, new CodecFinderCacheImpl());
+
+        DefaultCodecs cacheDisabledRegistry = new DefaultCodecs(byteBufAllocator, false, new CodecFinderDefaultImpl());
+
+    }
+
+    private void decode(Codecs codecs, int iterations, Blackhole voodoo) {
+        for (int i = 0; i < iterations; i++) {
+            voodoo.consume(codecs.decode(
+                TEST.buffer(4).writeInt(200), INT4.getObjectId(), FORMAT_BINARY, Integer.class));
+            voodoo.consume(codecs.decode(
+                ByteBufUtils.encode(TEST, "100"), INT2.getObjectId(), FORMAT_TEXT, Short.class));
+            voodoo.consume(codecs.decode(
+                ByteBufUtils.encode(TEST, "-125.369"), FLOAT8.getObjectId(), FORMAT_TEXT, Double.class));
+            voodoo.consume(codecs.decode(
+                TEST.buffer(4).writeFloat(-65.369f), FLOAT4.getObjectId(), FORMAT_BINARY, Float.class));
+            voodoo.consume(
+                codecs.decode(
+                    ByteBufUtils.encode(TEST, "test"),
+                    VARCHAR.getObjectId(),
+                    FORMAT_TEXT,
+                    String.class));
+            voodoo.consume(
+                codecs.decode(
+                    ByteBufUtils.encode(TEST, "2018-11-04 15:35:00.847108"),
+                    TIMESTAMP.getObjectId(),
+                    FORMAT_TEXT,
+                    LocalDateTime.class));
+            voodoo.consume(codecs.decode(ByteBufUtils.encode(TEST, "{100,200}"), INT2_ARRAY.getObjectId(), FORMAT_TEXT, Object.class));
+            voodoo.consume(codecs.decode(ByteBufUtils.encode(TEST, "{100,200}"), INT4_ARRAY.getObjectId(), FORMAT_TEXT, Object.class));
+            voodoo.consume(codecs.decode(ByteBufUtils.encode(TEST, "{100.5,200.8}"), FLOAT8_ARRAY.getObjectId(), FORMAT_TEXT, Object.class));
+        }
+    }
+
+    @Benchmark
+    public void decodeWithCacheEnabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        decode(holder.cacheEnabledRegistry, holder.iterations, voodoo);
+    }
+
+    @Benchmark
+    public void decodeWithCacheDisabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        decode(holder.cacheDisabledRegistry, holder.iterations, voodoo);
+    }
+
+    private void encode(Codecs codecs, int iterations, Blackhole voodoo) {
+        for (int i = 0; i < iterations; i++) {
+            voodoo.consume(codecs.encode((short) 12));
+            voodoo.consume(codecs.encode(35698));
+            voodoo.consume(codecs.encode(-256.3698));
+            voodoo.consume(codecs.encode(85.7458f));
+            voodoo.consume(codecs.encode("A text value"));
+            voodoo.consume(codecs.encode(LocalDateTime.now()));
+            voodoo.consume(codecs.encode(new Long[]{100L, 200L}));
+            voodoo.consume(codecs.encode(new Double[]{100.5, 200.25}));
+            voodoo.consume(codecs.encodeNull(Integer.class));
+            voodoo.consume(codecs.encodeNull(String.class));
+            voodoo.consume(codecs.encodeNull(Double[].class));
+        }
+    }
+
+    @Benchmark
+    public void encodeWithCacheEnabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        encode(holder.cacheEnabledRegistry, holder.iterations, voodoo);
+    }
+
+    @Benchmark
+    public void encodeWithCacheDisabledRegistry(CodecRegistryHolder holder, Blackhole voodoo) {
+        encode(holder.cacheDisabledRegistry, holder.iterations, voodoo);
+    }
+
+}

src/main/java/io/r2dbc/postgresql/codec/AbstractBinaryCodec.java

@@ -26,6 +26,7 @@
 import io.r2dbc.postgresql.util.ByteBufUtils;
 
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
 import java.util.Objects;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -62,6 +63,11 @@ boolean doCanDecode(PostgresqlObjectId type, Format format) {
         return BYTEA == type;
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return Collections.singleton(BYTEA);
+    }
+
     byte[] decode(Format format, ByteBuf byteBuf) {
         if (format == FORMAT_TEXT) {
             return decodeFromHex(byteBuf);

src/main/java/io/r2dbc/postgresql/codec/AbstractCodec.java

@@ -26,9 +26,12 @@
 import reactor.core.publisher.Mono;
 import reactor.util.annotation.Nullable;
 
+import java.util.EnumSet;
 import java.util.function.Supplier;
 
 import static io.r2dbc.postgresql.client.Parameter.NULL_VALUE;
+import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
+import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
 
 /**
  * Abstract codec class that provides a basis for all concrete
@@ -95,6 +98,12 @@ public Class<?> type() {
         return this.type;
     }
 
+    @Override
+    public Iterable<Format> getFormats() {
+        // Unless overridden all codecs supports both text and binary format
+        return EnumSet.of(FORMAT_TEXT, FORMAT_BINARY);
+    }
+
     /**
      * Create a {@link Parameter}.
      *

src/main/java/io/r2dbc/postgresql/codec/AbstractGeometryCodec.java

@@ -25,6 +25,7 @@
 import io.r2dbc.postgresql.util.ByteBufUtils;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -103,6 +104,11 @@ public Parameter encodeNull() {
         return createNull(this.postgresqlObjectId, Format.FORMAT_BINARY);
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return Collections.singleton(this.postgresqlObjectId);
+    }
+
     /**
      * Create a {@link TokenStream} given {@code content}.
      *

src/main/java/io/r2dbc/postgresql/codec/AbstractJsonCodec.java

@@ -21,12 +21,17 @@
 import io.r2dbc.postgresql.type.PostgresqlObjectId;
 import io.r2dbc.postgresql.util.Assert;
 
+import java.util.EnumSet;
+import java.util.Set;
+
 import static io.r2dbc.postgresql.message.Format.FORMAT_BINARY;
 import static io.r2dbc.postgresql.type.PostgresqlObjectId.JSON;
 import static io.r2dbc.postgresql.type.PostgresqlObjectId.JSONB;
 
 abstract class AbstractJsonCodec<T> extends AbstractCodec<T> {
 
+    private static final Set<PostgresqlObjectId> SUPPORTED_TYPES = EnumSet.of(JSON, JSONB);
+
     AbstractJsonCodec(Class<T> type) {
         super(type);
     }
@@ -41,7 +46,12 @@ boolean doCanDecode(PostgresqlObjectId type, Format format) {
         Assert.requireNonNull(format, "format must not be null");
         Assert.requireNonNull(type, "type must not be null");
 
-        return JSONB == type || JSON == type;
+        return SUPPORTED_TYPES.contains(type);
+    }
+
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return SUPPORTED_TYPES;
     }
 
 }

src/main/java/io/r2dbc/postgresql/codec/AbstractNumericCodec.java

@@ -87,6 +87,10 @@ T decodeNumber(ByteBuf buffer, PostgresqlObjectId dataType, @Nullable Format for
         return potentiallyConvert(number, expectedType, converter);
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return SUPPORTED_TYPES;
+    }
     /**
      * Returns the {@link PostgresqlObjectId} for to identify whether this codec is the default codec.
      *

src/main/java/io/r2dbc/postgresql/codec/AbstractTemporalCodec.java

@@ -158,4 +158,9 @@ static <T> T potentiallyConvert(Temporal temporal, Class<T> expectedType, Functi
         return expectedType.isInstance(temporal) ? expectedType.cast(temporal) : converter.apply(temporal);
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return SUPPORTED_TYPES;
+    }
+
 }

src/main/java/io/r2dbc/postgresql/codec/BigDecimalArrayCodec.java

@@ -24,6 +24,7 @@
 import io.r2dbc.postgresql.util.Assert;
 
 import java.math.BigDecimal;
+import java.util.Collections;
 import java.util.function.Supplier;
 
 final class BigDecimalArrayCodec extends AbstractArrayCodec<BigDecimal> {
@@ -37,6 +38,11 @@ public Parameter encodeNull() {
         return createNull(PostgresqlObjectId.NUMERIC_ARRAY, Format.FORMAT_TEXT);
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return Collections.singleton(PostgresqlObjectId.NUMERIC_ARRAY);
+    }
+
     @Override
     boolean doCanDecode(PostgresqlObjectId type, Format format) {
         Assert.requireNonNull(type, "type must not be null");

src/main/java/io/r2dbc/postgresql/codec/BlobCodec.java

@@ -29,6 +29,7 @@
 import reactor.util.annotation.Nullable;
 
 import java.nio.ByteBuffer;
+import java.util.Collections;
 
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
 import static io.r2dbc.postgresql.type.PostgresqlObjectId.BYTEA;
@@ -75,6 +76,11 @@ Parameter doEncode(Blob value) {
         );
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return Collections.singleton(BYTEA);
+    }
+
     private static final class ByteABlob implements Blob {
 
         private final ByteBuf byteBuf;

src/main/java/io/r2dbc/postgresql/codec/BooleanArrayCodec.java

@@ -23,6 +23,7 @@
 import io.r2dbc.postgresql.type.PostgresqlObjectId;
 import io.r2dbc.postgresql.util.Assert;
 
+import java.util.Collections;
 import java.util.function.Supplier;
 
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
@@ -42,6 +43,11 @@ public Parameter encodeNull() {
         return createNull(BOOL_ARRAY, FORMAT_TEXT);
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return Collections.singleton(BOOL_ARRAY);
+    }
+
     @Override
     Boolean doDecodeBinary(ByteBuf byteBuffer) {
         return byteBuffer.readBoolean();

src/main/java/io/r2dbc/postgresql/codec/BooleanCodec.java

@@ -25,6 +25,8 @@
 import io.r2dbc.postgresql.util.ByteBufUtils;
 import reactor.util.annotation.Nullable;
 
+import java.util.Collections;
+
 import static io.r2dbc.postgresql.message.Format.FORMAT_TEXT;
 import static io.r2dbc.postgresql.type.PostgresqlObjectId.BOOL;
 
@@ -42,6 +44,11 @@ public Parameter encodeNull() {
         return createNull(BOOL, FORMAT_TEXT);
     }
 
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return Collections.singleton(BOOL);
+    }
+
     @Override
     boolean doCanDecode(PostgresqlObjectId type, Format format) {
         Assert.requireNonNull(format, "format must not be null");

src/main/java/io/r2dbc/postgresql/codec/ByteCodec.java

@@ -62,4 +62,14 @@ Parameter doEncode(Byte value) {
         return this.delegate.doEncode((short) value);
     }
 
+    @Override
+    public Iterable<Format> getFormats() {
+        return this.delegate.getFormats();
+    }
+
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return this.delegate.getDataTypes();
+    }
+
 }

src/main/java/io/r2dbc/postgresql/codec/CharacterCodec.java

@@ -62,4 +62,14 @@ Parameter doEncode(Character value) {
         return this.delegate.doEncode(value.toString());
     }
 
+    @Override
+    public Iterable<Format> getFormats() {
+        return this.delegate.getFormats();
+    }
+
+    @Override
+    public Iterable<PostgresqlObjectId> getDataTypes() {
+        return this.delegate.getDataTypes();
+    }
+
 }