From bd65b0d937db12624da685e4936bccbd40f446f4 Mon Sep 17 00:00:00 2001 From: Woonduk Kang Date: Tue, 13 Nov 2012 09:49:42 +0000 Subject: [PATCH] =?UTF-8?q?[=EA=B0=95=EC=9A=B4=EB=8D=95]=20[LUCYSUS-1744]?= =?UTF-8?q?=20arcus=20cancel=EB=B6=80=EB=B6=84=20=EC=88=98=EC=A0=95.=20?= =?UTF-8?q?=EB=B9=84=EB=8F=99=EA=B8=B0=20=ED=98=B8=EC=B6=9C=20=EC=B6=94?= =?UTF-8?q?=EC=A0=81=EB=B6=80=EB=B6=84=20=EC=88=98=EC=A0=95.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-tomcat-profiler/trunk@859 84d0f5b1-2673-498c-a247-62c4ff18d310 --- .../java/com/profiler/context/AsyncTrace.java | 44 +++++++++++++ .../com/profiler/context/GlobalCallTrace.java | 60 +++++++++++++++--- src/main/java/com/profiler/context/Span.java | 11 ---- src/main/java/com/profiler/context/Trace.java | 2 - .../modifier/arcus/ArcusClientModifier.java | 62 ------------------- .../BaseOperationCancelInterceptor.java | 8 ++- ...seOperationTransitionStateInterceptor.java | 10 ++- .../interceptors/ConstructInterceptor.java | 2 +- 8 files changed, 111 insertions(+), 88 deletions(-) diff --git a/src/main/java/com/profiler/context/AsyncTrace.java b/src/main/java/com/profiler/context/AsyncTrace.java index d2965b305..9410c4bc1 100644 --- a/src/main/java/com/profiler/context/AsyncTrace.java +++ b/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; + } + + } diff --git a/src/main/java/com/profiler/context/GlobalCallTrace.java b/src/main/java/com/profiler/context/GlobalCallTrace.java index 247a8d63c..091845576 100644 --- a/src/main/java/com/profiler/context/GlobalCallTrace.java +++ b/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 trace = new ConcurrentHashMap(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 trace; + private int id; +// private final AsyncTrace asyncTrace; + + public TimeoutTask(ConcurrentMap trace, int id) { + this.trace = trace; + this.id = id; + } + + @Override + public void run() { + AsyncTrace asyncTrace = trace.remove(id); + if (asyncTrace != null) { + asyncTrace.timeout(); + } + } + } } diff --git a/src/main/java/com/profiler/context/Span.java b/src/main/java/com/profiler/context/Span.java index 5ff2fa14d..120e6c7a4 100644 --- a/src/main/java/com/profiler/context/Span.java +++ b/src/main/java/com/profiler/context/Span.java @@ -24,7 +24,6 @@ public class Span { private boolean isTerminal = false; private final List annotations = new ArrayList(5); - private final Set annotationKeys = new HashSet(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; diff --git a/src/main/java/com/profiler/context/Trace.java b/src/main/java/com/profiler/context/Trace.java index be4c29ce6..d075e3e33 100644 --- a/src/main/java/com/profiler/context/Trace.java +++ b/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(); diff --git a/src/main/java/com/profiler/modifier/arcus/ArcusClientModifier.java b/src/main/java/com/profiler/modifier/arcus/ArcusClientModifier.java index 08d58491e..96a5283b8 100644 --- a/src/main/java/com/profiler/modifier/arcus/ArcusClientModifier.java +++ b/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(); - } } \ No newline at end of file diff --git a/src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationCancelInterceptor.java b/src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationCancelInterceptor.java index 3cd6d74e2..804bea10a 100644 --- a/src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationCancelInterceptor.java +++ b/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(); diff --git a/src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationTransitionStateInterceptor.java b/src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationTransitionStateInterceptor.java index 936a72091..20e855851 100644 --- a/src/main/java/com/profiler/modifier/arcus/interceptors/BaseOperationTransitionStateInterceptor.java +++ b/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) { diff --git a/src/main/java/com/profiler/modifier/arcus/interceptors/ConstructInterceptor.java b/src/main/java/com/profiler/modifier/arcus/interceptors/ConstructInterceptor.java index 8b78d394c..ba096f5d6 100644 --- a/src/main/java/com/profiler/modifier/arcus/interceptors/ConstructInterceptor.java +++ b/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("__setAsyncTraceId", int.class); + private MetaObject asyncTraceId = new MetaObject("__setAsyncTraceId", int.class); @Override public void after(Object target, String className, String methodName, String parameterDescription, Object[] args, Object result) {