Skywalking implements cross thread trace delivery

Preface

When asynchronous multithreading is used in the process , If nothing is done ,SkyWalking Trace the execution link trace Information is bound to be interrupted . Generally speaking, it is a rigid requirement to ensure the integrity of the execution link information , At this time, in order to realize trace Cross thread transmission of information , You need to use SkyWalking Asynchronous task wrapper class

1. SkyWalking Client asynchronous task wrapper class

1.1 Use of cross thread wrapper classes

SkyWalking Of Java The client provides an asynchronous task wrapper class for multithreading trace Cross thread delivery of , At present, there are several implementations as follows ：

• RunnableWrapper：Ruannable Interface The wrapper class
• CallableWrapper：Callable Interface The wrapper class
• ConsumerWrapper： Functional interface Consumer The wrapper class
• SupplierWrapper： Functional interface Supplier The wrapper class
• FunctionWrapper： Functional interface Function The wrapper class

With SupplierWrapper For example , Examples of its use are as follows ：

CompletableFuture.supplyAsync(new SupplierWrapper<>(() -> {
    
	LoggerFactory.getLogger(this.getClass()).info("SupplierWrapper");
    return "nathan";
}));

1.2 The principle of wrapping classes across threads

1.2.1 @TraceCrossThread annotation

The following is a SupplierWrapper Source code , Looking at the source code of the wrapper class mentioned in the previous section, you will find that they are annotated @TraceCrossThread modification . actually SkyWalking Will pass SPI Mechanism in skywalking-plugin.def Specify bytecode enhanced configuration class in the file , among CallableOrRunnableActivation It's specifically for @TraceCrossThread Annotate the configuration class for scanning

@TraceCrossThread
public class SupplierWrapper<V> implements Supplier<V> {
    
    final Supplier<V> supplier;

    public static <V> SupplierWrapper<V> of(Supplier<V> r) {
    
        return new SupplierWrapper<>(r);
    }

    public SupplierWrapper(Supplier<V> supplier) {
    
        this.supplier = supplier;
    }

    @Override
    public V get() {
    
        return supplier.get();
    }
}

1.2.2 CallableOrRunnableActivation Enhanced configuration for

CallableOrRunnableActivation Will @TraceCrossThread Annotations as the entry point for enhanced classes , At the same time, configure aspects for the methods in the class that need to be enhanced , So that bytecode enhancement can be implemented across threads trace Transfer function

The following is a CallableOrRunnableActivation Source code , You can see that the configuration mainly specifies two enhancements for the target class ：

1. Constructor enhancements
   Constructor enhanced configuration to match any constructor , The enhanced section is CallableOrRunnableConstructInterceptor
2. Specify method enhancements
   Specifies that method enhancements only match Above, 1.1 section The specific methods of the wrapper class mentioned , The enhanced section is CallableOrRunnableInvokeInterceptor

---

This article is only for SkyWalking Make a preliminary introduction to the enhanced configuration of , If readers are interested, they can make an in-depth analysis here

public class CallableOrRunnableActivation extends ClassInstanceMethodsEnhancePluginDefine {
    

    public static final String ANNOTATION_NAME = "org.apache.skywalking.apm.toolkit.trace.TraceCrossThread";
    private static final String INIT_METHOD_INTERCEPTOR = "org.apache.skywalking.apm.toolkit.activation.trace.CallableOrRunnableConstructInterceptor";
    private static final String CALL_METHOD_INTERCEPTOR = "org.apache.skywalking.apm.toolkit.activation.trace.CallableOrRunnableInvokeInterceptor";
    private static final String CALL_METHOD_NAME = "call";
    private static final String RUN_METHOD_NAME = "run";
    private static final String GET_METHOD_NAME = "get";
    private static final String APPLY_METHOD_NAME = "apply";
    private static final String ACCEPT_METHOD_NAME = "accept";

    @Override
    public ConstructorInterceptPoint[] getConstructorsInterceptPoints() {
    
        return new ConstructorInterceptPoint[] {
    
            new ConstructorInterceptPoint() {
    
                @Override
                public ElementMatcher<MethodDescription> getConstructorMatcher() {
    
                    return any();
                }

                @Override
                public String getConstructorInterceptor() {
    
                    return INIT_METHOD_INTERCEPTOR;
                }
            }
        };
    }

    @Override
    public InstanceMethodsInterceptPoint[] getInstanceMethodsInterceptPoints() {
    
        return new InstanceMethodsInterceptPoint[] {
    
            new InstanceMethodsInterceptPoint() {
    
                @Override
                public ElementMatcher<MethodDescription> getMethodsMatcher() {
    
                    return named(CALL_METHOD_NAME)
                        .and(takesArguments(0))
                        .or(named(RUN_METHOD_NAME).and(takesArguments(0)))
                        .or(named(GET_METHOD_NAME).and(takesArguments(0)))
                        .or(named(APPLY_METHOD_NAME).and(takesArguments(1)))
                        .or(named(ACCEPT_METHOD_NAME).and(takesArguments(1)));
                }

                @Override
                public String getMethodsInterceptor() {
    
                    return CALL_METHOD_INTERCEPTOR;
                }

                @Override
                public boolean isOverrideArgs() {
    
                    return false;
                }
            }
        };
    }

    @Override
    protected ClassMatch enhanceClass() {
    
        return byClassAnnotationMatch(new String[] {
    ANNOTATION_NAME});
    }

}

2. Custom implementation SkyWalking Cross thread wrapper classes

After the introduction in the previous section , We know @TraceCrossThread Annotations can only be enhanced CallableOrRunnableActivation The method specified in the configuration , use @TraceCrossThread It's hard to achieve free customization . But copy CallableOrRunnableActivation Construction method facet and specified method facet configured in , Realize cross thread delivery trace The asynchronous task wrapper class is not that difficult , One BiConsumerWrapper The code example of is as follows ：

This wrapper class implements trace There are only two steps in the core processing of cross thread information transmission ：

1. When creating a wrapper class asynchronous task , adopt ContextManager In the context of the thread that will submit the task trace The information is cached inside the wrapper class object
2. The worker thread first passes through before executing an asynchronous task ContextManager Create your own context , Then cache the asynchronous task trace Information is loaded into its own context , To achieve trace Cross thread delivery of

public class BiConsumerWrapper<T, U> implements BiConsumer<T, U> {
    
    private final BiConsumer<T, U> function;
    private ContextSnapshot snapshot;

    public BiConsumerWrapper(BiConsumer<T, U> function) {
    
        this.function = function;
        if (ContextManager.isActive()) {
    
            snapshot = ContextManager.capture();
        }
    }

    @Override
    public void accept(T t, U u) {
    
        Optional.ofNullable(snapshot).ifPresent(snapshot -> {
    
            ContextManager.createLocalSpan("BiConsumerWrapper");
            ContextManager.continued(snapshot);
        });
        function.accept(t, u);
        Optional.ofNullable(snapshot).ifPresent(value -> ContextManager.stopSpan());
    }
}