D-Net

package eu.dnetlib.msro.workflows.sarasvati.loader;

import java.io.File;

import java.util.Comparator;

import java.util.List;

import java.util.Map;

import java.util.concurrent.Executors;

import java.util.concurrent.PriorityBlockingQueue;

import java.util.concurrent.ScheduledExecutorService;

import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.lang.math.NumberUtils;

import org.apache.commons.logging.Log;

import org.apache.commons.logging.LogFactory;

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

import com.googlecode.sarasvati.Graph;

import com.googlecode.sarasvati.GraphProcess;

import com.googlecode.sarasvati.mem.MemEngine;

import com.googlecode.sarasvati.mem.MemGraphProcess;

import eu.dnetlib.enabling.common.Stoppable;

import eu.dnetlib.enabling.common.StoppableDetails;

import eu.dnetlib.enabling.common.StoppableDetails.StopStatus;

import eu.dnetlib.enabling.is.lookup.rmi.ISLookUpService;

import eu.dnetlib.enabling.tools.ServiceLocator;

import eu.dnetlib.miscutils.datetime.DateUtils;

import eu.dnetlib.msro.rmi.MSROException;

import eu.dnetlib.msro.workflows.sarasvati.registry.GraphProcessRegistry;

import eu.dnetlib.msro.workflows.util.WorkflowsConstants;

public class WorkflowExecutor implements Stoppable {

	private MemEngine engine;

	private GraphLoader graphLoader;

	private GraphProcessRegistry graphProcessRegistry;

	private ProfileToSarasvatiConverter profileToSarasvatiConverter;

	private ScheduledExecutorService queueConsumers;

	private boolean paused = false;

	@Resource(name="lookupLocator")

	private ServiceLocator<ISLookUpService> lookupLocator;

	public boolean isPaused() {

		return paused;

	}

	public void setPaused(boolean paused) {

		this.paused = paused;

	}

	private static final Log log = LogFactory.getLog(WorkflowExecutor.class);

	private PriorityBlockingQueue<GraphProcess> pendingProcs = new PriorityBlockingQueue<GraphProcess>(20, new Comparator<GraphProcess>() {

		@Override

		public int compare(GraphProcess p1, GraphProcess p2) {

			int n1 = NumberUtils.toInt(p1.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PRIORITY), WorkflowsConstants.DEFAULT_WF_PRIORITY);

			int n2 = NumberUtils.toInt(p2.getEnv().getAttribute(WorkflowsConstants.SYSTEM_WF_PRIORITY), WorkflowsConstants.DEFAULT_WF_PRIORITY);

			return NumberUtils.compare(n1, n2);

		}

	});

	public void init() {

		this.queueConsumers = Executors.newScheduledThreadPool(WorkflowsConstants.MAX_WF_THREADS);

		final int period = 60;

		final int step   = period / WorkflowsConstants.MAX_WF_THREADS;

		for (int i=0; i<WorkflowsConstants.MAX_WF_THREADS; i++) {

			this.queueConsumers.scheduleAtFixedRate(new Runnable() {

				@Override

				public void run() {

					if (isPaused()) {

						return;

					}

					final GraphProcess process = pendingProcs.poll();

					if (process != null) {

						log.info("Starting workflow: " + process);

						final long now = DateUtils.now();

						process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_START_DATE, now);

						process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_START_HUMAN_DATE, DateUtils.calculate_ISO8601(now));

						engine.startProcess(process);

					} else {

						log.debug("Process queue is empty");

					}

				}

			}, i * step, period, TimeUnit.SECONDS);

		}

	}

	public String startProcess(final String profileId) throws Exception {

		return startProcess(profileId, null);

	}

	public String startProcess(final String profileId, final Map<String, Object> params) throws Exception {

		final WfProfileDescriptor desc = profileToSarasvatiConverter.getSarasvatiWorkflow(profileId);

		if (isPaused()) {

			log.warn("Wf " + profileId + " not launched, because WorkflowExecutor is preparing for shutdown");

			throw new MSROException("WorkflowExecutor is preparing for shutdown");

		}

		if (!desc.isReady()) {

			log.warn("Wf " + profileId + " not launched, because it is not ready to start");

			throw new MSROException("Workflow " + profileId + " is not ready to start");

		}

		if (pendingProcs.size() > WorkflowsConstants.MAX_PENDING_PROCS_SIZE) {

			log.warn("Wf " + profileId + " not launched, Max number of pending procs reached: " + WorkflowsConstants.MAX_PENDING_PROCS_SIZE);

        	throw new MSROException("Max number of pending procs reached: " + WorkflowsConstants.MAX_PENDING_PROCS_SIZE);

        }

		final File tmpFile = File.createTempFile("wftfs", null);

		try {

			final Graph graph = graphLoader.loadGraph(desc.getWorkflowXml());

			final GraphProcess process = new MemGraphProcess(graph);

	        final String procId = graphProcessRegistry.registerProcess(process);

	        graphProcessRegistry.associateProcessWithResource(process, profileId);

	        process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROCESS_ID, procId);

	        process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_ID, profileId);

	        process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_NAME, desc.getName());

	        process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_NAME, graph.getName());

	        process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PROFILE_FAMILY, desc.getType());

	        process.getEnv().setAttribute(WorkflowsConstants.SYSTEM_WF_PRIORITY, desc.getPriority());

	        if (params != null) {

	        	for (Map.Entry<String, Object> e : params.entrySet()) {

	        		process.getEnv().setAttribute(e.getKey(), e.getValue());

	        	} 

	        }

	        log.info("Process " + process + " in queue, priority=" + desc.getPriority());

	        pendingProcs.put(process);

			return procId;

		} catch (Exception e) {

			log.error("Error parsing workflow xml: " + desc.getWorkflowXml(), e);

			throw new IllegalArgumentException("Error parsing workflow");

		} finally {

			tmpFile.delete();

		}

	}

	public void startMetaWorkflow(final String id, final boolean manual) throws Exception {

		final String query = "/*[.//RESOURCE_IDENTIFIER/@value='" + id + "']//CONFIGURATION[@status='EXECUTABLE']/WORKFLOW/@id/string()";

		final List<String> list = lookupLocator.getService().quickSearchProfile(query);	

		if (list == null || list.isEmpty()) {

			throw new MSROException("Metaworkflow " + id + " not launched");

		}

		for (String wfId : list) {

			final String q = "/*[.//RESOURCE_IDENTIFIER/@value='" + wfId + "']//CONFIGURATION/@start/string()";

			if (manual || lookupLocator.getService().getResourceProfileByQuery(q).equals("auto")) {

				startProcess(wfId);

			} else {

				log.warn("Worflow " + wfId + " can not be launched AUTOMATICALLY");

			}

		}

	}

	public GraphLoader getGraphLoader() {

		return graphLoader;

	}

	@Required

	public void setGraphLoader(GraphLoader graphLoader) {

		this.graphLoader = graphLoader;

	}

	public MemEngine getEngine() {

		return engine;

	}

	@Required

	public void setEngine(MemEngine engine) {

		this.engine = engine;

	}

	public GraphProcessRegistry getGraphProcessRegistry() {

		return graphProcessRegistry;

	}

	@Required

	public void setGraphProcessRegistry(GraphProcessRegistry graphProcessRegistry) {

		this.graphProcessRegistry = graphProcessRegistry;

	}

	public ProfileToSarasvatiConverter getProfileToSarasvatiConverter() {

		return profileToSarasvatiConverter;

	}

	@Required

	public void setProfileToSarasvatiConverter(ProfileToSarasvatiConverter profileToSarasvatiConverter) {

		this.profileToSarasvatiConverter = profileToSarasvatiConverter;

	}

	@Override

	public void stop() {

		this.paused = true;

	}

	@Override

	public void resume() {

		this.paused = false;

	}

	@Override

	public StoppableDetails getStopDetails() {

		final int count = graphProcessRegistry.countRunningWfs();

		StoppableDetails.StopStatus status = StopStatus.RUNNING;

		if (isPaused()) {

			if (count == 0) {

				status = StopStatus.STOPPED;

			} else {

				status = StopStatus.STOPPING;

			}

		}

		graphProcessRegistry.listIdentifiers();

		return new StoppableDetails("D-NET workflow manager", "Running workflows: " + count, status);

	}

}