Makes SpanBytesDecoder work on ByteBuffer (openzipkin#2589)

* Makes SpanBytesDecoder work on ByteBuffer Internally, this also covers InputStream. This was easier (though not easy) due to splitting how we handle encoding from decoding. Incidentally, the jar size of zipkin is smaller still vs the last release. Fixes openzipkin#2435 * whoops forgot the important parts * de-optimizes for thrift in order to increase clarity * bug * Adds json comparison benchmarks * just review feedback * mah endian * less code * bench
abesto · Sep 10, 2019 · 24fc499 · 24fc499
1 parent 672e7c4
commit 24fc499
Show file tree

Hide file tree

Showing 48 changed files with 1,833 additions and 1,085 deletions.
diff --git a/benchmarks/pom.xml b/benchmarks/pom.xml
@@ -94,6 +94,13 @@
       <groupId>org.apache.zipkin.proto3</groupId>
       <artifactId>zipkin-proto3</artifactId>
     </dependency>
+
+    <dependency>
+      <groupId>${project.groupId}.zipkin2</groupId>
+      <artifactId>zipkin-tests</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

diff --git a/benchmarks/src/main/java/zipkin2/codec/JsonCodecBenchmarks.java b/benchmarks/src/main/java/zipkin2/codec/JsonCodecBenchmarks.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 zipkin2.codec;
+
+import com.google.common.io.Resources;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.PooledByteBufAllocator;
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+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.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+import zipkin2.Span;
+
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(3)
+@BenchmarkMode(Mode.SampleTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Thread)
+@Threads(1)
+public class JsonCodecBenchmarks {
+  static final MoshiSpanDecoder MOSHI = MoshiSpanDecoder.create();
+
+  static final byte[] clientSpanJsonV2 = read("/zipkin2-client.json");
+  static final Span clientSpan = SpanBytesDecoder.JSON_V2.decodeOne(clientSpanJsonV2);
+
+  // Assume a message is 1000 spans (which is a high number for as this is per-node-second)
+  static final List<Span> spans = Collections.nCopies(1000, clientSpan);
+  static final byte[] encodedBytes = SpanBytesEncoder.JSON_V2.encodeList(spans);
+
+  private ByteBuf encodedBuf;
+
+  @Setup public void setup() {
+    encodedBuf = PooledByteBufAllocator.DEFAULT.buffer(encodedBytes.length);
+    encodedBuf.writeBytes(encodedBytes);
+  }
+
+  @TearDown public void tearDown() {
+    encodedBuf.release();
+  }
+
+  @Benchmark public List<Span> bytes_moshiDecoder() {
+    return MOSHI.decodeList(encodedBytes);
+  }
+
+  @Benchmark public List<Span> bytes_zipkinDecoder() {
+    return SpanBytesDecoder.JSON_V2.decodeList(encodedBytes);
+  }
+
+  @Benchmark public List<Span> bytebuffer_moshiDecoder() {
+    return MOSHI.decodeList(encodedBuf.nioBuffer());
+  }
+
+  @Benchmark public List<Span> bytebuffer_zipkinDecoder() {
+    return SpanBytesDecoder.JSON_V2.decodeList(encodedBuf.nioBuffer());
+  }
+
+  // Convenience main entry-point
+  public static void main(String[] args) throws Exception {
+    Options opt = new OptionsBuilder()
+      .include(".*" + JsonCodecBenchmarks.class.getSimpleName())
+      .addProfiler("gc")
+      .build();
+
+    new Runner(opt).run();
+  }
+
+  static byte[] read(String resource) {
+    try {
+      return Resources.toByteArray(Resources.getResource(CodecBenchmarks.class, resource));
+    } catch (IOException e) {
+      throw new IllegalStateException(e);
+    }
+  }
+}
diff --git a/benchmarks/src/main/java/zipkin2/codec/MoshiSpanDecoder.java b/benchmarks/src/main/java/zipkin2/codec/MoshiSpanDecoder.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 zipkin2.codec;
+
+import com.squareup.moshi.JsonAdapter;
+import com.squareup.moshi.JsonReader;
+import com.squareup.moshi.Moshi;
+import com.squareup.moshi.Types;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.List;
+import okio.Buffer;
+import okio.BufferedSource;
+import okio.Okio;
+import okio.Timeout;
+import zipkin2.Span;
+import zipkin2.elasticsearch.ElasticsearchAccess;
+
+/**
+ * Read-only json adapters resurrected from before we switched to Java 6 as storage components can
+ * be Java 7+
+ */
+public final class MoshiSpanDecoder {
+  final JsonAdapter<List<Span>> listSpansAdapter;
+
+  public static MoshiSpanDecoder create() {
+    return new MoshiSpanDecoder();
+  }
+
+  MoshiSpanDecoder() {
+    listSpansAdapter = new Moshi.Builder()
+      .add(Span.class, ElasticsearchAccess.jsonSpanAdapter())
+      .build().adapter(Types.newParameterizedType(List.class, Span.class));
+  }
+
+  public List<Span> decodeList(byte[] spans) {
+    BufferedSource source = new Buffer().write(spans);
+    try {
+      return listSpansAdapter.fromJson(source);
+    } catch (IOException e) {
+      throw new AssertionError(e); // no I/O
+    }
+  }
+
+  public List<Span> decodeList(ByteBuffer spans) {
+    try {
+      return listSpansAdapter.fromJson(JsonReader.of(Okio.buffer(new ByteBufferSource(spans))));
+    } catch (IOException e) {
+      throw new AssertionError(e); // no I/O
+    }
+  }
+
+  final class ByteBufferSource implements okio.Source {
+    final ByteBuffer source;
+
+    final Buffer.UnsafeCursor cursor = new Buffer.UnsafeCursor();
+
+    ByteBufferSource(ByteBuffer source) {
+      this.source = source;
+    }
+
+    @Override public long read(Buffer sink, long byteCount) {
+      try (Buffer.UnsafeCursor ignored = sink.readAndWriteUnsafe(cursor)) {
+        long oldSize = sink.size();
+        int length = (int) Math.min(source.remaining(), Math.min(8192, byteCount));
+        if (length == 0) return -1;
+        cursor.expandBuffer(length);
+        source.get(cursor.data, cursor.start, length);
+        cursor.resizeBuffer(oldSize + length);
+        return length;
+      }
+    }
+
+    @Override public Timeout timeout() {
+      return Timeout.NONE;
+    }
+
+    @Override public void close() {
+    }
+  }
+}
diff --git a/benchmarks/src/main/java/zipkin2/codec/ProtoCodecBenchmarks.java b/benchmarks/src/main/java/zipkin2/codec/ProtoCodecBenchmarks.java
@@ -87,7 +87,7 @@ public List<Span> bytes_wireDecoder() {
 
   @Benchmark
   public List<Span> bytebuffer_zipkinDecoder() {
-    return SpanBytesDecoder.PROTO3.decodeList(ByteBufUtil.getBytes(encodedBuf));
+    return SpanBytesDecoder.PROTO3.decodeList(encodedBuf.nioBuffer());
   }
 
   @Benchmark
@@ -103,7 +103,7 @@ public List<Span> bytebuffer_wireDecoder() {
   // Convenience main entry-point
   public static void main(String[] args) throws Exception {
     Options opt = new OptionsBuilder()
-      .include(".*" + ProtoCodecBenchmarks.class.getSimpleName() + ".*bytes.*")
+      .include(".*" + ProtoCodecBenchmarks.class.getSimpleName())
       .addProfiler("gc")
       .build();
 

diff --git a/benchmarks/src/main/java/zipkin2/elasticsearch/ElasticsearchAccess.java b/benchmarks/src/main/java/zipkin2/elasticsearch/ElasticsearchAccess.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 zipkin2.elasticsearch;
+
+import com.squareup.moshi.JsonAdapter;
+import zipkin2.Span;
+
+public class ElasticsearchAccess {
+  public static JsonAdapter<Span> jsonSpanAdapter() {
+    return JsonAdapters.SPAN_ADAPTER;
+  }
+}
diff --git a/benchmarks/src/main/java/zipkin2/internal/ReadBufferBenchmarks.java b/benchmarks/src/main/java/zipkin2/internal/ReadBufferBenchmarks.java
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 zipkin2.internal;
+
+import java.util.concurrent.TimeUnit;
+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.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+@Measurement(iterations = 5, time = 1)
+@Warmup(iterations = 10, time = 1)
+@Fork(3)
+@BenchmarkMode(Mode.SampleTime)
+@OutputTimeUnit(TimeUnit.MICROSECONDS)
+@State(Scope.Thread)
+@Threads(1)
+public class ReadBufferBenchmarks {
+  byte[] longBuff = {
+    (byte) 0x01, (byte) 0x02, (byte) 0x03, (byte) 0x04,
+    (byte) 0x05, (byte) 0x06, (byte) 0x07, (byte) 0x08,
+  };
+
+  @Benchmark public long readLong() {
+    int pos = 0;
+    return (longBuff[pos] & 0xffL) << 56
+      | (longBuff[pos + 1] & 0xffL) << 48
+      | (longBuff[pos + 2] & 0xffL) << 40
+      | (longBuff[pos + 3] & 0xffL) << 32
+      | (longBuff[pos + 4] & 0xffL) << 24
+      | (longBuff[pos + 5] & 0xffL) << 16
+      | (longBuff[pos + 6] & 0xffL) << 8
+      | (longBuff[pos + 7] & 0xffL);
+  }
+
+  @Benchmark public long readLong_localArray() {
+    int pos = 0;
+    byte[] longBuff = this.longBuff;
+    return (longBuff[pos] & 0xffL) << 56
+      | (longBuff[pos + 1] & 0xffL) << 48
+      | (longBuff[pos + 2] & 0xffL) << 40
+      | (longBuff[pos + 3] & 0xffL) << 32
+      | (longBuff[pos + 4] & 0xffL) << 24
+      | (longBuff[pos + 5] & 0xffL) << 16
+      | (longBuff[pos + 6] & 0xffL) << 8
+      | (longBuff[pos + 7] & 0xffL);
+  }
+
+  @Benchmark public long readLong_8arity_localArray() {
+    int pos = 0;
+    return readLong(
+      longBuff[pos] & 0xff,
+      longBuff[pos + 1] & 0xff,
+      longBuff[pos + 2] & 0xff,
+      longBuff[pos + 3] & 0xff,
+      longBuff[pos + 4] & 0xff,
+      longBuff[pos + 5] & 0xff,
+      longBuff[pos + 6] & 0xff,
+      longBuff[pos + 7] & 0xff
+    );
+  }
+
+  @Benchmark public long readLong_8arity() {
+    int pos = 0;
+    byte[] longBuff = this.longBuff;
+    return readLong(
+      longBuff[pos] & 0xff,
+      longBuff[pos + 1] & 0xff,
+      longBuff[pos + 2] & 0xff,
+      longBuff[pos + 3] & 0xff,
+      longBuff[pos + 4] & 0xff,
+      longBuff[pos + 5] & 0xff,
+      longBuff[pos + 6] & 0xff,
+      longBuff[pos + 7] & 0xff
+    );
+  }
+
+  static long readLong(int p0, int p1, int p2, int p3, int p4, int p5, int p6, int p7) {
+    return (p0 & 0xffL) << 56
+      | (p1 & 0xffL) << 48
+      | (p2 & 0xffL) << 40
+      | (p3 & 0xffL) << 32
+      | (p4 & 0xffL) << 24
+      | (p5 & 0xffL) << 16
+      | (p6 & 0xffL) << 8
+      | (p7 & 0xffL);
+  }
+
+  @Benchmark public long readLongReverseBytes() {
+    return Long.reverseBytes(readLong());
+  }
+
+  // Convenience main entry-point
+  public static void main(String[] args) throws RunnerException {
+    Options opt = new OptionsBuilder()
+      .include(".*" + ReadBufferBenchmarks.class.getSimpleName() + ".*")
+      .addProfiler("gc")
+      .build();
+
+    new Runner(opt).run();
+  }
+}