Commit 66933e58 authored by Brian Long's avatar Brian Long
Browse files

separate reconcile threads from index/reindex threads

parent 83efe3a6

shared/src/main/java/com/inteligr8/alfresco/asie/service/AbstractNodeActionService.java

+18 −6
Original line number Diff line number Diff line
@@ -43,7 +43,7 @@ import com.inteligr8.solr.model.Action; 
import com.inteligr8.solr.model.ActionResponse;

import com.inteligr8.solr.model.BaseResponse;



public abstract class AbstractNodeActionService {

public abstract class AbstractNodeActionService implements DisposableBean {



	private final Logger logger = LoggerFactory.getLogger(this.getClass());
@@ -81,6 +81,22 @@ public abstract class AbstractNodeActionService { 
	

	protected abstract String getActionName();

	

	@Override

	public void destroy() {

		ExecutorService executor = this.executorManager.get(this.getThreadNamePrefix());

		if (executor != null) {

			this.logger.info("Shutting down throttled thread pool executor: {}", this.getThreadNamePrefix());

			executor.shutdown();

		}

	}



	private ThrottledThreadPoolExecutor getExecutor() {

		return this.executorManager.createThrottled(

				this.getThreadNamePrefix(),

				this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),

				1L, TimeUnit.MINUTES);

	}

	

	/**

	 * This method executes an action on the specified node in Solr using its

	 * ACS unique database identifier.  The callback handles all the return
@@ -140,11 +156,7 @@ public abstract class AbstractNodeActionService { 
		this.logger.debug("Will attempt to {} ACS node against {} shard instances: {}", this.getActionName(), eligibleInstances.size(), nodeDbId);

		

		CompositeFuture<Void> future = new CompositeFuture<>();

		

		ThrottledThreadPoolExecutor executor = this.executorManager.createThrottled(

				this.getThreadNamePrefix(),

				this.getConcurrency(), this.getConcurrency(), this.getConcurrentQueueSize(),

				1L, TimeUnit.MINUTES);

		ThrottledThreadPoolExecutor executor = this.getExecutor();

		

		for (final com.inteligr8.alfresco.asie.model.ShardInstance instance : eligibleInstances) {

			this.logger.trace("Will attempt to {} ACS node against shard instance: {}: {}", this.getActionName(), nodeDbId, instance);

shared/src/main/java/com/inteligr8/alfresco/asie/service/AcsReconcileService.java

+63 −67
Original line number Diff line number Diff line 
package com.inteligr8.alfresco.asie.service;



import java.util.HashMap;

import java.util.HashSet;

import java.util.Map;

import java.util.Collections;

import java.util.Set;

import java.util.concurrent.Callable;

import java.util.concurrent.ExecutionException;

import java.util.concurrent.ExecutorService;

import java.util.concurrent.Future;

import java.util.concurrent.TimeUnit;

import java.util.concurrent.TimeoutException;
@@ -26,7 +25,6 @@ import org.apache.commons.collections4.SetUtils; 
import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.DisposableBean;

import org.springframework.beans.factory.InitializingBean;

import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.beans.factory.annotation.Value;

import org.springframework.stereotype.Component;
@@ -39,7 +37,7 @@ import com.inteligr8.alfresco.asie.util.CompositeFuture; 
import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor;



@Component

public class AcsReconcileService implements InitializingBean, DisposableBean {

public class AcsReconcileService implements DisposableBean {



	private final Logger logger = LoggerFactory.getLogger(this.getClass());

	private final Logger reconcileLogger = LoggerFactory.getLogger("inteligr8.asie.reconcile");
@@ -62,6 +60,9 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	@Autowired

	private ReindexService reindexService;

	

	@Autowired

	private ExecutorManager executorManager;

	

	@Value("${inteligr8.asie.reconciliation.nodesChunkSize}")

	private int nodesChunkSize;
@@ -74,17 +75,20 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	@Value("${inteligr8.asie.reconciliation.concurrency}")

	private int concurrency;

	

	private ThrottledThreadPoolExecutor executor;

	

	@Override

	public void afterPropertiesSet() {

		this.executor = new ThrottledThreadPoolExecutor(this.concurrency, this.concurrency, this.concurrentQueueSize, 1L, TimeUnit.MINUTES, "solr-reconcile");

		this.executor.prestartAllCoreThreads();

	public void destroy() {

		ExecutorService executor = this.executorManager.get("solr-reconcile");

		if (executor != null) {

			this.logger.info("Shutting down throttled thread pool executor: {}", "solr-reconcile");

			executor.shutdown();

		}

	}



	@Override

	public void destroy() {

		this.executor.shutdown();

	private ThrottledThreadPoolExecutor getExecutor() {

		return this.executorManager.createThrottled(

				"solr-reconcile",

				this.concurrency, this.concurrency, this.concurrentQueueSize,

				10L, TimeUnit.SECONDS);

	}

	

	/**
@@ -205,6 +209,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
		}

		

		CompositeFuture<Void> future = new CompositeFuture<>();

		ThrottledThreadPoolExecutor executor = this.getExecutor();



		for (long _nodeDbId = fromDbId; _nodeDbId < toDbId; _nodeDbId++) {

			final long nodeDbId = _nodeDbId;
@@ -222,7 +227,11 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
						callback.reconciled(nodeDbId);



						if (reindexReconciled)

							reindex(nodeDbId, nodeRefs[dbIdIndex], callback, execTimeout, execUnit);

							reindex(nodeDbId, nodeRefs[dbIdIndex], callback);

							// purposefully forgetting about the returned future

							// its results will be logged

							// the reconcile thread will continue independently

							// the callback will lag

						return null;

					}

				};
@@ -230,16 +239,16 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
				callable = new Callable<Void>() {

					@Override

					public Void call() throws InterruptedException, TimeoutException {

						reconcile(nodeDbId, indexUnreconciled, callback, execTimeout, execUnit);

						reconcile(nodeDbId, indexUnreconciled, callback);

						return null;

					}

				};

			}

			

			if (queueTimeout < 0L) {

				future.combine(this.executor.submit(callable, -1L, null));

				future.combine(executor.submit(callable, -1L, null));

			} else {

				future.combine(this.executor.submit(callable, queueTimeout, queueUnit));

				future.combine(executor.submit(callable, queueTimeout, queueUnit));

			}

		}
@@ -248,8 +257,7 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
	

	public void reconcile(long nodeDbId,

			boolean index,

			ReconcileCallback callback,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

			ReconcileCallback callback) throws InterruptedException, TimeoutException {

		NodeRef nodeRef = this.nodeService.getNodeRef(nodeDbId);

		if (nodeRef == null) {

			this.logger.trace("No such ACS node: {}; skipping ...", nodeDbId);
@@ -273,93 +281,81 @@ public class AcsReconcileService implements InitializingBean, DisposableBean { 
			this.reconcileLogger.info("UNRECONCILED: {} <=> {}", nodeDbId, nodeRef);

			callback.unreconciled(nodeDbId);

		} else {

			logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef);

			this.index(nodeDbId, nodeRef, callback, execTimeout, execUnit);

			this.logger.debug("A node in the DB is not indexed in Solr; attempt to index: {}: {}", nodeDbId, nodeRef);

			this.index(nodeDbId, nodeRef, callback);

			// purposefully forgetting about the returned future

			// its results will be logged

			// the reconcile thread will continue independently

			// the callback will lag

		}

	}

	

	public void index(long nodeDbId, NodeRef nodeRef,

			ReconcileCallback callback,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

		Set<ShardInstance> syncHosts = new HashSet<>();

		Set<ShardInstance> asyncHosts = new HashSet<>();

		Map<ShardInstance, String> errorHosts = new HashMap<>();

		

	public Future<Void> index(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException, TimeoutException {

		IndexCallback indexCallback = new IndexCallback() {

			

			@Override

			public void success(ShardInstance instance) {

				reconcileLogger.info("INDEXED: {} <=> {} in {}", nodeDbId, nodeRef, instance);

				syncHosts.add(instance);

				if (callback != null)

					callback.processed(nodeDbId,

						Collections.singleton(instance), Collections.emptySet(),

						Collections.emptyMap());

			}

			

			@Override

			public void scheduled(ShardInstance instance) {

				reconcileLogger.info("INDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance);

				asyncHosts.add(instance);

				if (callback != null)

					callback.processed(nodeDbId,

						Collections.emptySet(), Collections.singleton(instance),

						Collections.emptyMap());

			}

			

			@Override

			public void error(ShardInstance instance, String message) {

				reconcileLogger.info("FAILED INDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance);

				errorHosts.put(instance, message);

				if (callback != null)

					callback.processed(nodeDbId,

						Collections.emptySet(), Collections.emptySet(),

						Collections.singletonMap(instance, message));

			}

		};

		

		try {

			if (execTimeout < 0L) {

				this.indexService.index(nodeDbId, indexCallback).get();

			} else {

				this.indexService.index(nodeDbId, indexCallback).get(execTimeout, execUnit);

			}

		} catch (ExecutionException ee) {

			throw new RuntimeException("An unexpected exception occurred: " + ee.getMessage(), ee);

		return this.indexService.index(nodeDbId, indexCallback);

	}

	

		if (callback != null)

			callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);

	}

	

	public void reindex(long nodeDbId, NodeRef nodeRef,

			ReconcileCallback callback,

			long execTimeout, TimeUnit execUnit) throws InterruptedException, TimeoutException {

		Set<ShardInstance> syncHosts = new HashSet<>();

		Set<ShardInstance> asyncHosts = new HashSet<>();

		Map<ShardInstance, String> errorHosts = new HashMap<>();

		

	public Future<Void> reindex(long nodeDbId, NodeRef nodeRef, ReconcileCallback callback) throws InterruptedException {

		ReindexCallback reindexCallback = new ReindexCallback() {

			

			@Override

			public void success(ShardInstance instance) {

				reconcileLogger.info("REINDEXED: {} <=> {} in {}", nodeDbId, nodeRef, instance);

				syncHosts.add(instance);

				if (callback != null)

					callback.processed(nodeDbId,

						Collections.singleton(instance), Collections.emptySet(),

						Collections.emptyMap());

			}

			

			@Override

			public void scheduled(ShardInstance instance) {

				reconcileLogger.info("REINDEXING: {} <=> {} in {}", nodeDbId, nodeRef, instance);

				asyncHosts.add(instance);

				if (callback != null)

					callback.processed(nodeDbId,

						Collections.emptySet(), Collections.singleton(instance),

						Collections.emptyMap());

			}

			

			@Override

			public void error(ShardInstance instance, String message) {

				reconcileLogger.info("FAILED REINDEX: {} <=> {} in {}", nodeDbId, nodeRef, instance);

				errorHosts.put(instance, message);

				if (callback != null)

					callback.processed(nodeDbId,

						Collections.emptySet(), Collections.emptySet(),

						Collections.singletonMap(instance, message));

			}

		};



		try {

			if (execTimeout < 0L) {

				this.reindexService.reindex(nodeDbId, reindexCallback).get();

			} else {

				this.reindexService.reindex(nodeDbId, reindexCallback).get(execTimeout, execUnit);

			}

		} catch (ExecutionException ee) {

			throw new RuntimeException("An unexpected exception occurred: " + ee.getMessage(), ee);

		}

		

		if (callback != null)

			callback.processed(nodeDbId, syncHosts, asyncHosts, errorHosts);

		return this.reindexService.reindex(nodeDbId, reindexCallback);

	}

	

	private String formatForFts(QName qname) {

shared/src/main/java/com/inteligr8/alfresco/asie/service/ExecutorManager.java

+19 −39
Original line number Diff line number Diff line
@@ -6,6 +6,8 @@ import java.util.concurrent.ExecutorService; 
import java.util.concurrent.RejectedExecutionHandler;

import java.util.concurrent.TimeUnit;



import org.slf4j.Logger;

import org.slf4j.LoggerFactory;

import org.springframework.beans.factory.DisposableBean;

import org.springframework.beans.factory.InitializingBean;

import org.springframework.beans.factory.annotation.Value;
@@ -31,41 +33,35 @@ import com.inteligr8.alfresco.asie.util.ThrottledThreadPoolExecutor; 
@Component

public class ExecutorManager implements InitializingBean, DisposableBean, RemovalListener<String, ExecutorService> {

	

	private final Logger logger = LoggerFactory.getLogger(this.getClass());



	@Value("${inteligr8.asie.executors.expireTimeInMinutes}")

	private int expireTimeInMinutes;

	

	private Cache<String, ExecutorService> refCache;

	private Cache<String, ExecutorService> expiringCache;

	private Cache<String, ThrottledThreadPoolExecutor> cache;

	

	@Override

	public void afterPropertiesSet() throws Exception {

		// a weak value happens when the executor is no longer referenced

		// the possible references are by the caller temporarily using and the `expiringCache` (below; so it expired)

		// this keeps the pool from being shutdown after it expires if the caller is still referencing it

		// ultimately, if it is cached, it will be in this cache and MAY be in the `expiringCache`. 

		this.refCache = CacheBuilder.newBuilder()

				.initialCapacity(8)

				.weakValues()

				.removalListener(this)

				.build();

		

		this.expiringCache = CacheBuilder.newBuilder()

		this.cache = CacheBuilder.newBuilder()

				.initialCapacity(8)

				.expireAfterAccess(this.expireTimeInMinutes, TimeUnit.MINUTES)

				.removalListener(this)

				.build();

	}

	

	@Override

	public void destroy() throws Exception {

		this.refCache.invalidateAll();

		this.refCache.cleanUp();

		this.expiringCache.invalidateAll();

		this.expiringCache.cleanUp();

		this.cache.invalidateAll();

		this.cache.cleanUp();

	}

	

	@Override

	public void onRemoval(RemovalNotification<String, ExecutorService> notification) {

		this.logger.debug("Throttled thread pool removed/expired from cache: {}", notification.getKey());

		if (!notification.getValue().isShutdown()) {

			notification.getValue().shutdown();

			this.logger.info("Throttled thread pool shut down: {}", notification.getKey());

		}

	}

	

	public ThrottledThreadPoolExecutor createThrottled(
@@ -88,9 +84,10 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova 
            final RejectedExecutionHandler rejectedExecutionHandler) {

		try {

			// if it is already cached, reuse the cache; otherwise create one

			final ExecutorService executor = this.refCache.get(name, new Callable<ThrottledThreadPoolExecutor>() {

			return this.cache.get(name, new Callable<ThrottledThreadPoolExecutor>() {

				@Override

				public ThrottledThreadPoolExecutor call() {

					logger.info("Creating throttled thread pool: {}", name);

					ThrottledThreadPoolExecutor executor = null;

					if (rejectedExecutionHandler == null) {

						executor = new ThrottledThreadPoolExecutor(coreThreadPoolSize, maximumThreadPoolSize, maximumQueueSize,
@@ -103,14 +100,9 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova 
								rejectedExecutionHandler);

					}

					

					logger.debug("Created throttled thread pool: {}; threads: {}; queue: {}", name, maximumThreadPoolSize, maximumQueueSize);

					executor.prestartAllCoreThreads();

					return executor;

				}

			});

			

			return (ThrottledThreadPoolExecutor) this.expiringCache.get(name, new Callable<ExecutorService>() {

				@Override

				public ExecutorService call() throws Exception {

					logger.trace("Started {} core threads in thread pool: {}", coreThreadPoolSize, name);

					return executor;

				}

			});
@@ -120,19 +112,7 @@ public class ExecutorManager implements InitializingBean, DisposableBean, Remova 
	}

	

	public ExecutorService get(String name) {

		// grab from the expiring cache first, so we can 

		ExecutorService executor = this.expiringCache.getIfPresent(name);

		if (executor != null)

			return executor;

		

		executor = this.refCache.getIfPresent(name);

		if (executor == null)

			return null;

		

		// the executor expired, but it was still referenced by the caller

		// re-cache it

		this.expiringCache.put(name, executor);

		return executor;

		return this.cache.getIfPresent(name);

	}

	

}

shared/src/main/java/com/inteligr8/alfresco/asie/util/ThrottledThreadPoolExecutor.java

+0 −3
Original line number Diff line number Diff line
@@ -80,9 +80,6 @@ public class ThrottledThreadPoolExecutor extends ThreadPoolExecutor { 
	}

	

	private WaitableRunnable submit(WaitableRunnable runnable, long throttlingBlockTimeout, TimeUnit throttlingBlockUnit) throws InterruptedException, TimeoutException {

		// if no core threads are running, the queue won't be monitored for runnables

		this.prestartAllCoreThreads();



		if (throttlingBlockTimeout < 0L) {

			this.getQueue().put(runnable);

		} else {