[#5156] AsyncId optimization

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/LocalAsyncIdBo.java

@@ -0,0 +1,64 @@
+/*
+ * Copyright 2019 NAVER Corp.
+ *
+ * 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 com.navercorp.pinpoint.common.server.bo;
+
+/**
+ * @author Woonduk Kang(emeroad)
+ */
+public class LocalAsyncIdBo {
+    private final int asyncId;
+    private final int sequence;
+
+    public LocalAsyncIdBo(int asyncId, int sequence) {
+        this.asyncId = asyncId;
+        this.sequence = sequence;
+    }
+
+    public int getAsyncId() {
+        return asyncId;
+    }
+
+    public int getSequence() {
+        return sequence;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (!(o instanceof LocalAsyncIdBo)) return false;
+
+        LocalAsyncIdBo that = (LocalAsyncIdBo) o;
+
+        if (asyncId != that.asyncId) return false;
+        return sequence == that.sequence;
+    }
+
+    @Override
+    public int hashCode() {
+        int result = asyncId;
+        result = 31 * result + sequence;
+        return result;
+    }
+
+    @Override
+    public String toString() {
+        return "LocalAsyncIdBo{" +
+                "asyncId=" + asyncId +
+                ", sequence=" + sequence +
+                '}';
+    }
+}

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/SpanBo.java

@@ -50,11 +50,12 @@ public class SpanBo implements Event, BasicSpan {
     private String endPoint;
     private int apiId;
 
-    private List<AnnotationBo> annotationBoList = new ArrayList<AnnotationBo>();
+    private List<AnnotationBo> annotationBoList = new ArrayList<>();
     private short flag; // optional
     private int errCode;
 
-    private List<SpanEventBo> spanEventBoList = new ArrayList<SpanEventBo>();
+    private List<SpanEventBo> spanEventBoList = new ArrayList<>();
+    private List<SpanChunkBo> asyncSpanChunkBoList;
 
     private long collectorAcceptTime;
 
@@ -223,6 +224,20 @@ public List<SpanEventBo> getSpanEventBoList() {
         return spanEventBoList;
     }
 
+    public List<SpanChunkBo> getAsyncSpanChunkBoList() {
+        if (asyncSpanChunkBoList == null) {
+            this.asyncSpanChunkBoList = new ArrayList<>();
+        }
+        return asyncSpanChunkBoList;
+    }
+
+    public void addAsyncSpanBo(SpanChunkBo asyncSpanBo) {
+        if (asyncSpanChunkBoList == null) {
+            this.asyncSpanChunkBoList = new ArrayList<>();
+        }
+        this.asyncSpanChunkBoList.add(asyncSpanBo);
+    }
+
     public short getServiceType() {
         return serviceType;
     }

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/SpanChunkBo.java

@@ -29,6 +29,7 @@ public class SpanChunkBo implements BasicSpan {
 
     private long collectorAcceptTime;
 
+    private LocalAsyncIdBo localAsyncId;
 
 
     public SpanChunkBo() {
@@ -136,6 +137,14 @@ public void addSpanEventBoList(List<SpanEventBo> spanEventBoList) {
         this.spanEventBoList.addAll(spanEventBoList);
     }
 
+    public LocalAsyncIdBo getLocalAsyncId() {
+        return localAsyncId;
+    }
+
+    public void setLocalAsyncId(LocalAsyncIdBo localAsyncId) {
+        this.localAsyncId = localAsyncId;
+    }
+
     @Override
     public String toString() {
         return "SpanChunkBo{" +
@@ -150,6 +159,7 @@ public String toString() {
                 ", applicationServiceType=" + applicationServiceType +
                 ", spanEventBoList=" + spanEventBoList +
                 ", collectorAcceptTime=" + collectorAcceptTime +
+                ", localAsyncId=" + localAsyncId +
                 '}';
     }
 }

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/SpanEventBo.java

@@ -54,8 +54,9 @@ public class SpanEventBo implements Event {
 
     private int nextAsyncId = -1;
 
-
+    @Deprecated
     private int asyncId = -1;
+    @Deprecated
     private short asyncSequence = -1;
 
     public SpanEventBo() {
@@ -196,26 +197,31 @@ public void setExceptionClass(String exceptionClass) {
         this.exceptionClass = exceptionClass;
     }
 
+    public int getNextAsyncId() {
+        return nextAsyncId;
+    }
+
+    public void setNextAsyncId(int nextAsyncId) {
+        this.nextAsyncId = nextAsyncId;
+    }
+
+
+    @Deprecated
     public int getAsyncId() {
         return asyncId;
     }
 
+    @Deprecated
     public void setAsyncId(int asyncId) {
         this.asyncId = asyncId;
     }
 
-    public int getNextAsyncId() {
-        return nextAsyncId;
-    }
-
-    public void setNextAsyncId(int nextAsyncId) {
-        this.nextAsyncId = nextAsyncId;
-    }
-
+    @Deprecated
     public short getAsyncSequence() {
         return asyncSequence;
     }
 
+    @Deprecated
     public void setAsyncSequence(short asyncSequence) {
         this.asyncSequence = asyncSequence;
     }

commons-server/src/main/java/com/navercorp/pinpoint/common/server/bo/SpanFactory.java

@@ -16,10 +16,13 @@
 import com.navercorp.pinpoint.thrift.dto.TSpanChunk;
 import com.navercorp.pinpoint.thrift.dto.TSpanEvent;
 import org.apache.commons.collections.CollectionUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
+import java.util.Iterator;
 import java.util.List;
 
 /**
@@ -28,14 +31,20 @@
 @Component
 public class SpanFactory {
 
+    private final Logger logger = LoggerFactory.getLogger(this.getClass());
 
     private SpanEventFilter spanEventFilter = new EmptySpanEventFilter();
 
     private AcceptedTimeService acceptedTimeService = new EmptyAcceptedTimeService();
 
     private static final AnnotationTranscoder transcoder = new AnnotationTranscoder();
 
+    // TODO
+    private final boolean fastAsyncIdGen;
+
     public SpanFactory() {
+        final String fastAsyncIdGen = System.getProperty("collector.spanfactory.fastasyncidgen", "true");
+        this.fastAsyncIdGen = Boolean.parseBoolean(fastAsyncIdGen);
     }
 
     @Autowired(required = false)
@@ -53,7 +62,7 @@ public SpanBo buildSpanBo(TSpan tSpan) {
         final SpanBo spanBo = newSpanBo(tSpan);
 
         List<TSpanEvent> spanEventList = tSpan.getSpanEventList();
-        List<SpanEventBo> spanEventBoList = buildSpanEventBoList(spanEventList, null);
+        List<SpanEventBo> spanEventBoList = buildSpanEventBoList(spanEventList);
         spanBo.addSpanEventBoList(spanEventBoList);
 
         long acceptedTime = acceptedTimeService.getAcceptedTime();
@@ -117,7 +126,7 @@ SpanBo newSpanBo(TSpan tSpan) {
     }
 
 
-    private void bind(SpanEventBo spanEvent, TSpanEvent tSpanEvent, TLocalAsyncId localAsyncId) {
+    private void bind(SpanEventBo spanEvent, TSpanEvent tSpanEvent) {
 
         spanEvent.setSequence(tSpanEvent.getSequence());
 
@@ -154,27 +163,30 @@ private void bind(SpanEventBo spanEvent, TSpanEvent tSpanEvent, TLocalAsyncId lo
         }
 
         // async id
-        if (localAsyncId == null) {
-            if (tSpanEvent.isSetAsyncId()) {
-                spanEvent.setAsyncId(tSpanEvent.getAsyncId());
-            }
-            if (tSpanEvent.isSetAsyncSequence()) {
-                spanEvent.setAsyncSequence(tSpanEvent.getAsyncSequence());
-            }
-        } else {
-            spanEvent.setAsyncId(localAsyncId.getAsyncId());
-            spanEvent.setAsyncSequence((short) localAsyncId.getSequence());
-        }
+//        if (localAsyncId == null) {
+//            if (tSpanEvent.isSetAsyncId()) {
+//                spanEvent.setAsyncId(tSpanEvent.getAsyncId());
+//            }
+//            if (tSpanEvent.isSetAsyncSequence()) {
+//                spanEvent.setAsyncSequence(tSpanEvent.getAsyncSequence());
+//            }
+//        } else {
+//            spanEvent.setAsyncId(localAsyncId.getAsyncId());
+//            spanEvent.setAsyncSequence((short) localAsyncId.getSequence());
+//        }
     }
 
 
 
     public SpanChunkBo buildSpanChunkBo(TSpanChunk tSpanChunk) {
         final SpanChunkBo spanChunkBo = newSpanChunkBo(tSpanChunk);
+        final LocalAsyncIdBo localAsyncIdBo = getLocalAsyncId(tSpanChunk, spanChunkBo);
+        if (localAsyncIdBo != null) {
+            spanChunkBo.setLocalAsyncId(localAsyncIdBo);
+        }
 
         List<TSpanEvent> spanEventList = tSpanChunk.getSpanEventList();
-        TLocalAsyncId localAsyncId = tSpanChunk.getLocalAsyncId();
-        List<SpanEventBo> spanEventBoList = buildSpanEventBoList(spanEventList, localAsyncId);
+        List<SpanEventBo> spanEventBoList = buildSpanEventBoList(spanEventList);
         spanChunkBo.addSpanEventBoList(spanEventBoList);
 
 
@@ -184,6 +196,63 @@ public SpanChunkBo buildSpanChunkBo(TSpanChunk tSpanChunk) {
         return spanChunkBo;
     }
 
+    private LocalAsyncIdBo getLocalAsyncId(TSpanChunk tSpanChunk, SpanChunkBo spanChunkBo) {
+        final TLocalAsyncId localAsyncId = tSpanChunk.getLocalAsyncId();
+        if (localAsyncId != null) {
+            return new LocalAsyncIdBo(localAsyncId.getAsyncId(), localAsyncId.getSequence());
+        } else {
+            return extractLocalAsyncId(tSpanChunk);
+        }
+    }
+
+    // for compatibility
+    // https://github.com/naver/pinpoint/issues/5156
+    private LocalAsyncIdBo extractLocalAsyncId(TSpanChunk tSpanChunk) {
+        List<TSpanEvent> tSpanEventList = tSpanChunk.getSpanEventList();
+        if (CollectionUtils.isEmpty(tSpanEventList)) {
+            return null;
+        }
+        if (fastAsyncIdGen) {
+            final TSpanEvent first = tSpanEventList.get(0);
+            final int asyncId = first.getAsyncId();
+            if (asyncId == -1) {
+                return null;
+            } else {
+                return new LocalAsyncIdBo(asyncId, first.getAsyncSequence());
+            }
+        } else {
+            int asyncId = -1;
+            int asyncSequence = -1;
+            boolean first = true;
+            boolean asyncIdNotSame = false;
+            for (TSpanEvent tSpanEvent : tSpanEventList) {
+                if (first) {
+                    first = false;
+                    asyncId = tSpanEvent.getAsyncId();
+                    asyncSequence = tSpanEvent.getAsyncSequence();
+                } else {
+                    if (asyncId != tSpanEvent.getAsyncId()) {
+                        asyncIdNotSame = true;
+                        break;
+                    }
+                    if (asyncSequence != tSpanEvent.getAsyncSequence()) {
+                        asyncIdNotSame = true;
+                        break;
+                    }
+                }
+            }
+            if (asyncIdNotSame) {
+                logger.warn("AsyncId consistency is broken. SpanChunk:{}", tSpanChunk);
+                return null;
+            }
+            if (asyncId != -1 && asyncSequence != -1) {
+                return new LocalAsyncIdBo(asyncId, asyncSequence);
+            }
+            // non async
+            return null;
+        }
+    }
+
     // for test
     SpanChunkBo newSpanChunkBo(TSpanChunk tSpanChunk) {
         final SpanChunkBo spanChunkBo = new SpanChunkBo();
@@ -217,13 +286,13 @@ private TransactionId newTransactionId(byte[] transactionIdBytes, BasicSpan basi
     }
 
 
-    private List<SpanEventBo> buildSpanEventBoList(List<TSpanEvent> spanEventList, TLocalAsyncId localAsyncId) {
+    private List<SpanEventBo> buildSpanEventBoList(List<TSpanEvent> spanEventList) {
         if (CollectionUtils.isEmpty(spanEventList)) {
             return new ArrayList<>();
         }
         List<SpanEventBo> spanEventBoList = new ArrayList<>(spanEventList.size());
         for (TSpanEvent tSpanEvent : spanEventList) {
-            final SpanEventBo spanEventBo = buildSpanEventBo(tSpanEvent, localAsyncId);
+            final SpanEventBo spanEventBo = buildSpanEventBo(tSpanEvent);
             if (!spanEventFilter.filter(spanEventBo)) {
                 continue;
             }
@@ -249,13 +318,13 @@ private List<AnnotationBo> buildAnnotationList(List<TAnnotation> tAnnotationList
     }
 
     // for test
-    public SpanEventBo buildSpanEventBo(TSpanEvent tSpanEvent, TLocalAsyncId parentAsyncId) {
+    public SpanEventBo buildSpanEventBo(TSpanEvent tSpanEvent) {
         if (tSpanEvent == null) {
             throw new NullPointerException("tSpanEvent must not be null");
         }
 
         final SpanEventBo spanEvent = new SpanEventBo();
-        bind(spanEvent, tSpanEvent, parentAsyncId);
+        bind(spanEvent, tSpanEvent);
         return spanEvent;
     }