[강운덕] [LUCYSUS-1744] arcus cancel부분 수정. 비동기 호출 추적부분 수정.

git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-tomcat-profiler/trunk@859 84d0f5b1-2673-498c-a247-62c4ff18d310

src/main/java/com/profiler/context/AsyncTrace.java

@@ -3,6 +3,9 @@ package com.profiler.context;
 import com.profiler.common.util.AnnotationTranscoder;
 import com.profiler.sender.DataSender;
 
+
+import java.util.TimerTask;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
@@ -16,8 +19,17 @@ public class AsyncTrace {
     //    private int id;
     // 비동기일 경우 traceenable의 경우 애매함. span을 보내는것으로 데이터를 생성하므로 약간 이상.
 //    private boolean tracingEnabled;
+
+    private static int COMPLATE_STATE_NONE = 0;
+    private static int COMPLATE_STATE_FIRE = 1;
+    private static int COMPLATE_STATE_TIMEOUT = 2;
+
+    private final AtomicInteger complate = new AtomicInteger(COMPLATE_STATE_NONE);
+
+    private int asyncId;
     private Span span;
     private DataSender dataSender;
+    private TimerTask timeoutTask;
 
     public AsyncTrace(Span span) {
         this.span = span;
@@ -27,6 +39,18 @@ public class AsyncTrace {
         this.dataSender = dataSender;
     }
 
+    public void setTimeoutTask(TimerTask timeoutTask) {
+        this.timeoutTask = timeoutTask;
+    }
+
+    public void setAsyncId(int asyncId) {
+        this.asyncId = asyncId;
+    }
+
+    public int getAsyncId() {
+        return asyncId;
+    }
+
     public Span getSpan() {
         return span;
     }
@@ -43,6 +67,9 @@ public class AsyncTrace {
 
 
     public void record(Annotation annotation) {
+        if (complate.get() == COMPLATE_STATE_FIRE) {
+
+        }
         annotate(annotation.getCode(), null);
     }
 
@@ -129,4 +156,21 @@ public class AsyncTrace {
             logger.log(Level.SEVERE, e.getMessage(), e);
         }
     }
+
+    public void timeout() {
+        if (complate.compareAndSet(0, COMPLATE_STATE_TIMEOUT)) {
+            // TODO timeout log 던지기.
+            // 뭘 어떤 내용을 던져야 되는지 아직 모르겠음????
+        }
+    }
+
+    public boolean cancelTimeout() {
+        if (complate.compareAndSet(0, COMPLATE_STATE_FIRE)) {
+            this.timeoutTask.cancel();
+            return true;
+        }
+        return false;
+    }
+
+
 }

src/main/java/com/profiler/context/GlobalCallTrace.java

@@ -3,6 +3,7 @@ package com.profiler.context;
 import com.profiler.sender.DataSender;
 
 import java.util.Timer;
+import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -11,6 +12,9 @@ import java.util.concurrent.atomic.AtomicInteger;
  *
  */
 public class GlobalCallTrace {
+
+    private static final long FLUSH_TIMEOUT = 120000L; // 2 minutes
+
     private static AtomicInteger timerId = new AtomicInteger(0);
 
     private ConcurrentMap<Integer, AsyncTrace> trace = new ConcurrentHashMap<Integer, AsyncTrace>(32);
@@ -19,19 +23,61 @@ public class GlobalCallTrace {
 
     private Timer timer = new Timer("GlobalCallTrace-Timer-" + timerId.getAndIncrement(), true);
 
-    public int registerTraceObject(AsyncTrace target) {
-        // datasender쪽 전달부분이 영 별로임.
-        target.setDataSender(this.dataSender);
-        int id = idGenerator.getAndIncrement();
-        trace.put(id, target);
+    public int registerTraceObject(AsyncTrace asyncTrace) {
+        // TODO 연관관계가 전달부분이 영 별로임.
+        asyncTrace.setDataSender(this.dataSender);
+
+        TimeoutTask timeoutTask = new TimeoutTask(trace, asyncTrace.getAsyncId());
+        asyncTrace.setTimeoutTask(timeoutTask);
+
+        int id = put(asyncTrace);
+        asyncTrace.setAsyncId(id);
+        timer.schedule(timeoutTask, FLUSH_TIMEOUT);
         return id;
     }
 
-    public AsyncTrace removeTraceObject(int id) {
-        return trace.get(id);
+    private int put(AsyncTrace asyncTrace) {
+        int id = idGenerator.getAndIncrement();
+        trace.put(id, asyncTrace);
+        return id;
+    }
+
+    public AsyncTrace getTraceObject(int asyncId) {
+        return trace.get(asyncId);
+    }
+
+    public AsyncTrace removeTraceObject(int asyncId) {
+        AsyncTrace asyncTrace = trace.remove(asyncId);
+        if (asyncTrace != null) {
+            boolean result = asyncTrace.cancelTimeout();
+            if (!result) {
+                // 이미 timeout된 asyncTrace임.
+                return null;
+            }
+        }
+        return asyncTrace;
     }
 
     public void setDataSender(DataSender dataSender) {
         this.dataSender = dataSender;
     }
+
+    private final class TimeoutTask extends TimerTask {
+        private ConcurrentMap<Integer, AsyncTrace> trace;
+        private int id;
+//        private final AsyncTrace asyncTrace;
+
+        public TimeoutTask(ConcurrentMap<Integer, AsyncTrace> trace, int id) {
+            this.trace = trace;
+            this.id = id;
+        }
+
+        @Override
+        public void run() {
+            AsyncTrace asyncTrace = trace.remove(id);
+            if (asyncTrace != null) {
+                asyncTrace.timeout();
+            }
+        }
+    }
 }

src/main/java/com/profiler/context/Span.java

@@ -24,7 +24,6 @@ public class Span {
     private boolean isTerminal = false;
 
     private final List<HippoAnnotation> annotations = new ArrayList<HippoAnnotation>(5);
-    private final Set<String> annotationKeys = new HashSet<String>(5);
 
     private long rpcStartTime;
     private long rpcEndTime;
@@ -37,7 +36,6 @@ public class Span {
     }
 
     public boolean addAnnotation(HippoAnnotation annotation) {
-        annotationKeys.add(annotation.getKey());
         if (annotation.getKey().equals(Annotation.ClientSend.getCode()) || annotation.getKey().equals(Annotation.ServerRecv.getCode())) {
             rpcStartTime = annotation.getTimestamp();
         }
@@ -51,15 +49,6 @@ public class Span {
         return annotations.size();
     }
 
-    /**
-     * this method only works for Trace.mutate()
-     *
-     * @param value
-     * @return
-     */
-    public boolean isExistsAnnotationKey(String key) {
-        return annotationKeys.contains(key);
-    }
 
     public String getEndPoint() {
         return this.endPoint;

src/main/java/com/profiler/context/Trace.java

@@ -60,8 +60,6 @@ public final class Trace {
             StackFrame stackFrame = createStackFrame(nextId, HANDLER_STACKID);
             callStack.setStackFrame(stackFrame);
             handler.handle(nextId);
-        } catch (Exception e) {
-            e.printStackTrace();
         } finally {
             // stackID check하면 좋을듯.
             callStack.pop();

src/main/java/com/profiler/modifier/arcus/ArcusClientModifier.java

@@ -49,66 +49,4 @@ public class ArcusClientModifier extends AbstractModifier {
             return null;
         }
     }
-
-    private String getCancelBeforeCode() {
-        StringBuilder code = new StringBuilder();
-
-        code.append("{");
-        code.append("	if (!cancelled) {");
-        code.append("		__setCancelledTime(System.nanoTime());");
-        code.append("	}");
-        code.append("}");
-
-        return code.toString();
-    }
-
-    private String getTransitionStateAfterCode() {
-        StringBuilder code = new StringBuilder();
-
-        code.append("{");
-//		code.append("com.profiler.context.Trace.traceBlockBegin();");
-
-        /**
-         * always override traceid
-         */
-        code.append("com.profiler.context.Trace.setTraceId(__nextTraceId);");
-
-        /**
-         * After sending command to the Arcus server. now waiting server
-         * response.
-         */
-        code.append("if (newState == net.spy.memcached.ops.OperationState.READING) {");
-
-        code.append("	java.net.SocketAddress socketAddress = handlingNode.getSocketAddress();");
-        code.append("	if (socketAddress instanceof java.net.InetSocketAddress) {");
-        code.append("		java.net.InetSocketAddress addr = (java.net.InetSocketAddress) handlingNode.getSocketAddress();");
-        code.append("		com.profiler.context.Trace.recordTerminalEndPoint(\"ARCUS:\" + addr.getHostName() + \":\" + addr.getPort());");
-        code.append("	}");
-        code.append("	com.profiler.context.Trace.recordRpcName(\"ARCUS\", this.getClass().getSimpleName());");
-        code.append("	com.profiler.context.Trace.recordAttribute(\"arcus.command\", ((cmd == null) ? \"UNKNOWN\" : new String(cmd.array())));");
-        code.append("	com.profiler.StopWatch.start(this.hashCode());");
-        code.append("	com.profiler.context.Trace.record(com.profiler.context.Annotation.ClientSend, System.nanoTime() - __commandCreatedTime);");
-
-        /**
-         * Received all response or timed out.
-         */
-        code.append("} else if (newState == net.spy.memcached.ops.OperationState.COMPLETE || newState == net.spy.memcached.ops.OperationState.TIMEDOUT) {");
-        code.append("	if (exception != null) { ");
-        code.append("		com.profiler.context.Trace.recordAttribute(\"exception\", com.profiler.util.InterceptorUtils.exceptionToString(exception));");
-        code.append("	}");
-
-        code.append("	if (!cancelled) {");
-        code.append("		com.profiler.context.Trace.record(com.profiler.context.Annotation.ClientRecv, com.profiler.StopWatch.stopAndGetElapsed(this.hashCode()));");
-        code.append("	} else {");
-        code.append("		com.profiler.context.Trace.recordAttribute(\"exception\", \"cancelled by user\");");
-        code.append("		com.profiler.context.Trace.record(com.profiler.context.Annotation.ClientRecv, System.nanoTime() - __cancelledTime);");
-        code.append("	}");
-
-        code.append("}");
-
-//		code.append("com.profiler.context.Trace.traceBlockEnd();");
-        code.append("}");
-
-        return code.toString();
-    }
 }

src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationCancelInterceptor.java

@@ -30,10 +30,14 @@ public class BaseOperationCancelInterceptor implements StaticBeforeInterceptor {
         GlobalCallTrace globalCallTrace = traceContext.getGlobalCallTrace();
         Object asyncId = asyncTraceId.invoke(target);
         if (asyncId == null) {
-            logger.info("asyncId not found");
+            logger.fine("asyncId not found");
+            return;
+        }
+        AsyncTrace asyncTrace = globalCallTrace.getTraceObject((Integer) asyncId);
+        if (asyncTrace == null) {
+            logger.fine("asyncTrace expired");
             return;
         }
-        AsyncTrace asyncTrace = globalCallTrace.removeTraceObject((Integer) asyncId);
         BaseOperationImpl baseOperation = (BaseOperationImpl) target;
         if (!baseOperation.isCancelled()) {
             TimeObject timeObject = (TimeObject) asyncTrace.getAttachObject();

src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationTransitionStateInterceptor.java

@@ -38,17 +38,21 @@ public class BaseOperationTransitionStateInterceptor implements StaticBeforeInte
 
         Object asyncId = asyncTraceId.invoke(target);
         if (asyncId == null) {
-            logger.info("asyncId not found");
+            logger.fine("asyncId not found");
             return;
         }
         // saynctrace를 제거한다.
         AsyncTrace asyncTrace = globalCallTrace.removeTraceObject((Integer) asyncId);
+        if (asyncTrace == null) {
+            logger.fine("AsyncTrace already timeout");
+            return;
+        }
         OperationState newState = (OperationState) args[0];
 
         BaseOperationImpl baseOperation = (BaseOperationImpl) target;
         if (newState == OperationState.READING) {
             if (logger.isLoggable(Level.FINE)) {
-                logger.log(Level.FINE, "event:" + newState);
+                logger.fine("event:" + newState + " asyncId:" + asyncId);
             }
             MemcachedNode handlingNode = baseOperation.getHandlingNode();
             SocketAddress socketAddress = handlingNode.getSocketAddress();
@@ -67,7 +71,7 @@ public class BaseOperationTransitionStateInterceptor implements StaticBeforeInte
             asyncTrace.record(Annotation.ClientSend, System.currentTimeMillis() - createTime);
         } else if (newState == OperationState.COMPLETE || newState == OperationState.TIMEDOUT) {
             if (logger.isLoggable(Level.FINE)) {
-                logger.log(Level.FINE, "event:" + newState);
+                logger.fine("event:" + newState + " asyncId:" + asyncId);
             }
             Exception exception = baseOperation.getException();
             if (exception != null) {

src/main/java/com/profiler/modifier/arcus/interceptors/ConstructInterceptor.java

@@ -15,7 +15,7 @@ import java.util.logging.Logger;
 public class ConstructInterceptor implements StaticAfterInterceptor {
 
     private final Logger logger = Logger.getLogger(ConstructInterceptor.class.getName());
-    private MetaObject asyncTraceId = new MetaObject<Integer>("__setAsyncTraceId", int.class);
+    private MetaObject<Integer> asyncTraceId = new MetaObject<Integer>("__setAsyncTraceId", int.class);
 
     @Override
     public void after(Object target, String className, String methodName, String parameterDescription, Object[] args, Object result) {