Add support for Jackson's new ByteBufferFeeder (apple#1711)

Motivation:

In 2.14.0, Jackson added support for feeding ByteBuffers directly
into the streaming parser engine (where previously only byte arrays
would be supported). This changeset incorporates this enhancement
into the JacksonStreamingSerializer infrastructure.

Modifications:

Before this changeset, the ByteArrayParser (and corresponding feeder)
were the only choice when using the streaming deserialization
infrastructure in Jackson, so the code was able to eaglerly initialize
the parser and feeder.

Now there is a choice that can be made at runtime which parser might
be the best one based on the incoming buffer in the stream. As such,
the code now checks the first Buffer and then decides if either the
array-backed or the bytebuffer-backed parser should be initalized.

This also has the advantage that if no items are emitted on an empty
stream, no parser needs to be initialized at all.

Note that the code still preserves the runtime "backed by type" checks
since (while likely not common but possible) buffers with different
backing types can arrive one after another and need to be handled. In
this case it is possible that a sub-optimal parser is chosen, but the
code optimizes for the likely scenario that all buffers on one stream
are backed by the same type.

Also a MethodHandle-based runtime check has been added which makes sure
that even if a user overrides the Jackson version to an older one on
the Classpath the code is still functional.

Result:

Added support for Jackson's new ByteBufferFeeder and choosing the
best strategy at runtime depending on the first buffer type that
arrives.

Author: daschl

servicetalk-benchmarks/build.gradle

@@ -32,6 +32,7 @@ dependencies {
   implementation project(":servicetalk-http-netty")
   implementation project(":servicetalk-transport-netty-internal")
   implementation project(":servicetalk-loadbalancer")
+  implementation project(":servicetalk-data-jackson")
   implementation "com.google.code.findbugs:jsr305"
   implementation "io.netty:netty-codec-http"
   implementation "org.openjdk.jmh:jmh-core"

servicetalk-benchmarks/src/jmh/java/io/servicetalk/data/jackson/JacksonStreamingSerializerBenchmark.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright © 2022 Apple Inc. and the ServiceTalk project 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
+ *
+ *   http://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.servicetalk.data.jackson;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import io.servicetalk.buffer.api.Buffer;
+import io.servicetalk.buffer.netty.BufferAllocators;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+import java.util.Map;
+
+import static io.servicetalk.concurrent.api.Publisher.from;
+import static io.servicetalk.data.jackson.JacksonSerializerFactory.JACKSON;
+
+/**
+ * Performs (de)serialization benchmarks on the JacksonStreamingSerializer.
+ *
+ * Benchmark                                                            Mode  Cnt        Score        Error  Units
+ * JacksonStreamingSerializerBenchmark.deserializeLargeBackedByArray   thrpt    5   114568,190 ±   5174,037  ops/s
+ * JacksonStreamingSerializerBenchmark.deserializeLargeBackedByDirect  thrpt    5    96666,948 ±   4912,809  ops/s
+ * JacksonStreamingSerializerBenchmark.deserializeMidBackedByArray     thrpt    5   527170,664 ± 107330,448  ops/s
+ * JacksonStreamingSerializerBenchmark.deserializeMidBackedByDirect    thrpt    5   420306,213 ±  78607,263  ops/s
+ * JacksonStreamingSerializerBenchmark.deserializeSmallBackedByArray   thrpt    5  1307308,103 ± 128131,629  ops/s
+ * JacksonStreamingSerializerBenchmark.deserializeSmallBackedByDirect  thrpt    5   674688,550 ± 108705,165  ops/s
+ *
+ * NOTE/TODO: the different numbers on the small and mid-sized buffers are likely due to higher allocation overhead for
+ * direct buffers compared to array ones. The benchmark can be modified in the future to refactor the allocation out
+ * so that it is not accounted for.
+ */
+@Fork(value = 1)
+@State(Scope.Benchmark)
+@Warmup(iterations = 5, time = 3)
+@Measurement(iterations = 5, time = 3)
+@BenchmarkMode(Mode.Throughput)
+public class JacksonStreamingSerializerBenchmark {
+
+  /**
+   * Type reference used for deserialization mapping.
+   */
+  private static final TypeReference<Map<String, Object>> MAP_TYPE_REFERENCE =
+    new TypeReference<Map<String, Object>>() {};
+
+  /**
+   * Represents a 10-byte JSON object (encoded).
+   */
+  private static final String SMALL_JSON = "{\"a\":true}";
+
+  /**
+   * Represents a 100-byte JSON object (encoded).
+   */
+  private static final String MID_JSON = "{\"_id\":\"6371eb6af678bb79174\",\"a\":false,\"balance\":\"$2,087.12\"," +
+    "\"picture\":\"https://placehold.it/32x32\"}";
+
+  /**
+   * Represents a 1000-byte JSON object (encoded).
+   */
+  private static final String LARGE_JSON = "{\"_id\":\"6371eaaf236316f8be8b44cc\",\"index\":0,\"guid\":" +
+    "\"12cdb7e4-b133-4da7-94b3-b1c58c4b3ef6\",\"isActive\":true,\"balance\":\"$3,777.31\",\"picture\":" +
+    "\"http://placehold.it/32x32\",\"age\":24,\"eyeColor\":\"blue\",\"name\":\"Nikki Randall\"," +
+    "\"gender\":\"female\",\"company\":\"PROVIDCO\",\"email\":\"[email protected]\",\"phone\":" +
+    "\"+1 (981) 475-2422\",\"address\":\"914 Butler Street, Eggertsville, Maryland, 5970\",\"about\":" +
+    "\"Nulla proident velit culpa magna sit duis in in deserunt. Irure consectetur ea veniam quis amet " +
+    "nulla tempor in esse ipsum. Do aute magna in qui qui dolor adipisicing ipsum nulla deserunt. Labore " +
+    "voluptate sint laborum adipisicing dolor anim amet pariatur sint. Ad nulla commodo ipsum aliqua " +
+    "est.\\r\\n\",\"registered\":\"2022-01-01T03:44:39 -01:00\",\"latitude\":61.061004,\"longitude\":38.544479," +
+    "\"tags\":[\"qui\",\"nulla\",\"cillum\",\"dolor\",\"deserunt\",\"amet\",\"magna\"],\"friends\":[{\"id\":0," +
+    "\"name\":\"Gibson Benson\"},{\"id\":1,\"name\":\"Marshall Atkins\"},{\"id\":2,\"name\":\"Gabriel Clay\"}," +
+    "{\"id\":3,\"name\":\"Aguilar\"},{\"id\":4,\"name\":\"Bauer\"}]}";
+
+  @Benchmark
+  public Map<String, Object> deserializeSmallBackedByArray() throws Exception {
+    return deserialize(BufferAllocators.PREFER_HEAP_ALLOCATOR.fromUtf8(SMALL_JSON));
+  }
+
+  @Benchmark
+  public Map<String, Object> deserializeSmallBackedByDirect() throws Exception {
+    return deserialize(BufferAllocators.PREFER_DIRECT_ALLOCATOR.fromUtf8(SMALL_JSON));
+  }
+
+  @Benchmark
+  public Map<String, Object> deserializeMidBackedByArray() throws Exception {
+    return deserialize(BufferAllocators.PREFER_HEAP_ALLOCATOR.fromUtf8(MID_JSON));
+  }
+
+  @Benchmark
+  public Map<String, Object> deserializeMidBackedByDirect() throws Exception {
+    return deserialize(BufferAllocators.PREFER_DIRECT_ALLOCATOR.fromUtf8(MID_JSON));
+  }
+
+  @Benchmark
+  public Map<String, Object> deserializeLargeBackedByArray() throws Exception {
+    return deserialize(BufferAllocators.PREFER_HEAP_ALLOCATOR.fromUtf8(LARGE_JSON));
+  }
+
+  @Benchmark
+  public Map<String, Object> deserializeLargeBackedByDirect() throws Exception {
+    return deserialize(BufferAllocators.PREFER_DIRECT_ALLOCATOR.fromUtf8(LARGE_JSON));
+  }
+
+  private static Map<String, Object> deserialize(final Buffer fromBuffer) throws Exception {
+    return JACKSON
+      .streamingSerializerDeserializer(MAP_TYPE_REFERENCE)
+      .deserialize(from(fromBuffer), BufferAllocators.DEFAULT_ALLOCATOR)
+      .firstOrError()
+      .toFuture()
+      .get();
+  }
+
+}

servicetalk-data-jackson/build.gradle

@@ -29,6 +29,7 @@ dependencies {
   implementation project(":servicetalk-concurrent-internal")
   implementation project(":servicetalk-utils-internal")
   implementation "com.google.code.findbugs:jsr305"
+  implementation "org.slf4j:slf4j-api"
 
   testImplementation testFixtures(project(":servicetalk-concurrent-api"))
   testImplementation testFixtures(project(":servicetalk-concurrent-internal"))