KVCAsync

KVCAsync.java

package strimillinn.xperimental.async;

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

/**
 * Mark a method in SMAsyncComponent with this annotation to have it added to keys that get processed simultaneously
 */

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface KVCAsync {}

SMAsyncComponent.java

package strimillinn.xperimental.async;

import java.lang.reflect.Method;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.webobjects.appserver.WOContext;

import er.extensions.components.ERXComponent;

/**
 * Components that inherit from this class can annotate methods with @KVCAsync.
 * Annotated methods will be run simultaneously upon the invocation of the first one.
 * Useful if you have a component that contains a lot of heavy methods where some concurrency could help.
 *
 * FIXME: Methods prefixed with "get" will not work with this
 */

public abstract class SMAsyncComponent extends ERXComponent {

	private static final Logger logger = LoggerFactory.getLogger( SMAsyncComponent.class );

	/**
	 * Stores a list of all keys annotated with @KVCAsync
	 */
	private Set<String> _asyncKeys;

	/**
	 * Stores the values of asyncKeys, once they've been calculated.
	 */
	private Map<String, Object> _asyncValues;

	public SMAsyncComponent( WOContext context ) {
		super( context );
	}

	/**
	 * Indicates if you want to enable async resolution of methods annotated with @KVCAsync
	 */
	protected boolean useKVCAsync() {
		return true;
	}

	/**
	 * Indicates if you want to use platform threads or virtual threads
	 */
	protected boolean useVirtualThreads() {
		return true;
	}

	/**
	 * @return The executor used to run the threads.
	 */
	protected Executor createExecutor() {
		if( useVirtualThreads() ) {
			return Executors.newVirtualThreadPerTaskExecutor();
		}

		return ForkJoinPool.commonPool();
	}

	/**
	 * Checks if the requested key is annotated with @KVCAsync. The first time an annotated method is accessed,
	 * all keys annotated with @KVCAsync will be resolved/processed simultaneously and their values cached.
	 */
	@Override
	public Object valueForKey( String key ) {
		if( useKVCAsync() ) {
			if( isAsync( key ) ) {
				if( _asyncValues == null ) {
					populateAsyncValues();
				}

				return _asyncValues.get( key );
			}
		}

		return super.valueForKey( key );
	}

	/**
	 * @return A list of method names/keys annotated with @KVCAsync
	 */
	private Set<String> asyncKeys() {
		if( _asyncKeys == null ) {
			_asyncKeys = new HashSet<>();

			for( Method method : getClass().getDeclaredMethods() ) {
				if( method.isAnnotationPresent( KVCAsync.class ) ) {
					_asyncKeys.add( method.getName() );
				}
			}
		}

		return _asyncKeys;
	}

	/**
	 * @return true if the given key references a method annotated with @KVCAsync
	 */
	private boolean isAsync( String key ) {
		return asyncKeys().contains( key );
	}

	/**
	 * Populates the map of cached asynchronous values
	 */
	private void populateAsyncValues() {
		logger.debug( "Entering populateAsyncValues()" );

		_asyncValues = Collections.synchronizedMap( new HashMap<>() );

		final Executor executor = createExecutor();
		final CompletableFuture<?>[] futures = new CompletableFuture[asyncKeys().size()];

		// Time it actually takes to invoke all the methods concurrently
		long asyncInvocationTime = System.currentTimeMillis();

		// Sum of the invocation time of each method
		final AtomicLong syncInvocationTime = new AtomicLong();

		int i = 0;

		for( String key : asyncKeys() ) {
			futures[i++] = CompletableFuture.runAsync( () -> {
				long currentTime = System.currentTimeMillis();
				_asyncValues.put( key, super.valueForKey( key ) );
				final long elapsedTime = System.currentTimeMillis() - currentTime;
				syncInvocationTime.addAndGet( elapsedTime );
				logger.info( "Populated KVCAsync value '{}' in {}ms: ", key, elapsedTime );
			}, executor );
		}

		try {
			CompletableFuture.allOf( futures ).get();
		}
		catch( InterruptedException | ExecutionException e ) {
			throw new RuntimeException( "Failed to populate async fields", e );
		}

		asyncInvocationTime = System.currentTimeMillis() - asyncInvocationTime;

		// Calculate the amount of time we saved by going the async route
		long savedTimeMS = syncInvocationTime.longValue() - asyncInvocationTime;

		logger.info( "Invoked {} methods in {}ms instead of {}ms. You saved {}ms", futures.length, asyncInvocationTime, syncInvocationTime, savedTimeMS );
	}
}