<commit_before>package eu.qualimaster.coordination;

import static eu.qualimaster.coordination.CoordinationUtils.getNamespace;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.FileUtils;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import eu.qualimaster.common.signal.AbstractTopologyExecutorSignal;
import eu.qualimaster.common.signal.AlgorithmChangeSignal;
import eu.qualimaster.common.signal.LoadSheddingSignal;
import eu.qualimaster.common.signal.MonitoringChangeSignal;
import eu.qualimaster.common.signal.ParameterChange;
import eu.qualimaster.common.signal.ParameterChangeSignal;
import eu.qualimaster.common.signal.ReplaySignal;
import eu.qualimaster.common.signal.ShutdownSignal;
import eu.qualimaster.common.signal.SignalException;
import eu.qualimaster.common.signal.SignalMechanism;
import eu.qualimaster.coordination.INameMapping.Algorithm;
import eu.qualimaster.coordination.INameMapping.Component;
import eu.qualimaster.coordination.INameMapping.ISubPipeline;
import eu.qualimaster.coordination.PipelineCache.PipelineElementCache;
import eu.qualimaster.coordination.RepositoryConnector.Models;
import eu.qualimaster.coordination.RepositoryConnector.Phase;
import eu.qualimaster.coordination.StormUtils.TopologyTestInfo;
import eu.qualimaster.coordination.commands.AlgorithmChangeCommand;
import eu.qualimaster.coordination.commands.CommandSequence;
import eu.qualimaster.coordination.commands.CommandSet;
import eu.qualimaster.coordination.commands.CoordinationCommand;
import eu.qualimaster.coordination.commands.CoordinationExecutionResult;
import eu.qualimaster.coordination.commands.LoadSheddingCommand;
import eu.qualimaster.coordination.commands.MonitoringChangeCommand;
import eu.qualimaster.coordination.commands.ParallelismChangeCommand;
import eu.qualimaster.coordination.commands.ParameterChangeCommand;
import eu.qualimaster.coordination.commands.PipelineCommand;
import eu.qualimaster.coordination.commands.PipelineCommand.Status;
import eu.qualimaster.coordination.commands.ProfileAlgorithmCommand;
import eu.qualimaster.coordination.commands.ReplayCommand;
import eu.qualimaster.coordination.commands.ScheduleWavefrontAdaptationCommand;
import eu.qualimaster.coordination.commands.ShutdownCommand;
import eu.qualimaster.coordination.commands.UpdateCommand;
import eu.qualimaster.coordination.events.CoordinationCommandExecutionEvent;
import eu.qualimaster.coordination.shutdown.Shutdown;
import eu.qualimaster.dataManagement.DataManager;
import eu.qualimaster.easy.extension.internal.AlgorithmProfileHelper;
import eu.qualimaster.easy.extension.internal.AlgorithmProfileHelper.ProfileData;
import eu.qualimaster.easy.extension.internal.QmProjectDescriptor;
import eu.qualimaster.events.EventManager;
import eu.qualimaster.file.Utils;
import eu.qualimaster.infrastructure.PipelineLifecycleEvent;
import eu.qualimaster.infrastructure.PipelineOptions;
import eu.qualimaster.monitoring.events.ChangeMonitoringEvent;
import net.ssehub.easy.basics.modelManagement.ModelInitializer;
import net.ssehub.easy.basics.modelManagement.ModelManagementException;
import net.ssehub.easy.basics.progress.ProgressObserver;
import net.ssehub.easy.instantiation.core.model.common.VilException;
import net.ssehub.easy.varModel.confModel.Configuration;
import net.ssehub.easy.varModel.management.VarModel;
import net.ssehub.easy.varModel.model.Project;

/**
 * Visits the commands for execution. Please note that when starting the execution of a single command,
 * the command must be placed on the {@link #commandStack} and at the end of the execution (no exception must
 * be thrown) {@link #writeCoordinationLog(CoordinationCommand, CoordinationExecutionResult)} must be
 * called which pops from the stack and performs the logging.
 * 
 * @author <NAME>
 */
class CoordinationCommandExecutionVisitor extends AbstractCoordinationCommandExecutionVisitor {

    private static final Set<Class<? extends CoordinationCommand>> DEFER_SUCCESSFUL_EXECUTION 
        = new HashSet<Class<? extends CoordinationCommand>>();
    private long timestamp = System.currentTimeMillis();
    private ActiveCommands activeCommands = null;
    
    static {
        // later, all command shall be deferred and report about the actual completion
        DEFER_SUCCESSFUL_EXECUTION.add(PipelineCommand.class);
    }
    
    /**
     * Creates a command execution visitor.
     * 
     * @param tracer a tracer for obtaining information about the execution
     */
    CoordinationCommandExecutionVisitor(IExecutionTracer tracer) {
        super(tracer);
    }
    
    /**
     * Sets the top-level command of this execution.
     * 
     * @param command the command
     * @return the active commands object
     */
    ActiveCommands setTopLevelCommand(CoordinationCommand command) {
        activeCommands = new ActiveCommands(command);
        return activeCommands;
    }
    
    /**
     * Returns the actual message id.
     * 
     * @return the actual message id
     */
    private String getCauseMessageId() {
        return null == activeCommands ? "" : activeCommands.getCauseMessageId();
    }
    
    /**
     * Handles sending a (storm commons) signal.
     * 
     * @param command the command causing the signal
     * @param signal the signal
     * @return whether the execution failed (instance) or not (<b>null</b>)
     */
    private static CoordinationExecutionResult send(CoordinationCommand command, 
        AbstractTopologyExecutorSignal signal) {
        CoordinationExecutionResult failing = null;
        try {
            if (!CoordinationManager.isTestingMode()) {
                getLogger().info("Sending signal " + signal);
                signal.sendSignal();
            }
        } catch (SignalException e) {
            failing = new CoordinationExecutionResult(command, e.getMessage(), 
                CoordinationExecutionCode.SIGNAL_SENDING_ERROR);
        }
        return failing;
    }

    @Override
    public CoordinationExecutionResult visitAlgorithmChangeCommand(AlgorithmChangeCommand command) {
        return handleAlgorithmChange(command, null);
    }
    
    @Override
    protected CoordinationExecutionResult handleAlgorithmChange(AlgorithmChangeCommand command, 
        List<ParameterChange> parameters) {
        CoordinationExecutionResult result;
        String pipelineName = command.getPipeline();
        INameMapping mapping = CoordinationManager.getNameMapping(pipelineName);
        ISubPipeline subPip = mapping.getSubPipelineByAlgorithmName(command.getAlgorithm());
        if (null != subPip) {
            CoordinationManager.deferCommand(subPip.getName(), PipelineLifecycleEvent.Status.STARTED, 
                new AlgorithmChangeAction(command, parameters, getTracer()));
            PipelineCommand cmd = new PipelineCommand(subPip.getName(), PipelineCommand.Status.START, 
                getSubPipelineOptions(pipelineName, command.getOptions()));
            result = handlePipelineStart(cmd);
        } else {
            result = handleAlgorithmChangeImpl(command, parameters);
        }
        return result;
    }

    /**
     * Returns the sub pipeline options.
     * 
     * @param mainPipeline the main pipeline
     * @param commandOptions the options introduced by the command
     * @return the options for the sub-pipeline marked as sub-pipeline
     */
    private PipelineOptions getSubPipelineOptions(String mainPipeline, PipelineOptions commandOptions) {
        PipelineOptions opts = CoordinationManager.getPipelineOptions(mainPipeline);
        if (null == opts) {
            opts = new PipelineOptions(); // shall not occur
        } else {
            opts = new PipelineOptions(opts); // clone
        }
        if (null != commandOptions) { // take into account decisions done by the adaptation for algorithm change
            opts.merge(commandOptions);
        }
        opts.markAsSubPipeline(mainPipeline);
        return opts;
    }<commit_msg>transporting the initial sub-pipeline status upon sub-pip switch<commit_after>package eu.qualimaster.coordination;

import static eu.qualimaster.coordination.CoordinationUtils.getNamespace;

import java.io.File;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.commons.io.FileUtils;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import eu.qualimaster.common.signal.AbstractTopologyExecutorSignal;
import eu.qualimaster.common.signal.AlgorithmChangeSignal;
import eu.qualimaster.common.signal.Constants;
import eu.qualimaster.common.signal.LoadSheddingSignal;
import eu.qualimaster.common.signal.MonitoringChangeSignal;
import eu.qualimaster.common.signal.ParameterChange;
import eu.qualimaster.common.signal.ParameterChangeSignal;
import eu.qualimaster.common.signal.ReplaySignal;
import eu.qualimaster.common.signal.ShutdownSignal;
import eu.qualimaster.common.signal.SignalException;
import eu.qualimaster.common.signal.SignalMechanism;
import eu.qualimaster.coordination.INameMapping.Algorithm;
import eu.qualimaster.coordination.INameMapping.Component;
import eu.qualimaster.coordination.INameMapping.ISubPipeline;
import eu.qualimaster.coordination.PipelineCache.PipelineElementCache;
import eu.qualimaster.coordination.RepositoryConnector.Models;
import eu.qualimaster.coordination.RepositoryConnector.Phase;
import eu.qualimaster.coordination.StormUtils.TopologyTestInfo;
import eu.qualimaster.coordination.commands.AlgorithmChangeCommand;
import eu.qualimaster.coordination.commands.CommandSequence;
import eu.qualimaster.coordination.commands.CommandSet;
import eu.qualimaster.coordination.commands.CoordinationCommand;
import eu.qualimaster.coordination.commands.CoordinationExecutionResult;
import eu.qualimaster.coordination.commands.LoadSheddingCommand;
import eu.qualimaster.coordination.commands.MonitoringChangeCommand;
import eu.qualimaster.coordination.commands.ParallelismChangeCommand;
import eu.qualimaster.coordination.commands.ParameterChangeCommand;
import eu.qualimaster.coordination.commands.PipelineCommand;
import eu.qualimaster.coordination.commands.PipelineCommand.Status;
import eu.qualimaster.coordination.commands.ProfileAlgorithmCommand;
import eu.qualimaster.coordination.commands.ReplayCommand;
import eu.qualimaster.coordination.commands.ScheduleWavefrontAdaptationCommand;
import eu.qualimaster.coordination.commands.ShutdownCommand;
import eu.qualimaster.coordination.commands.UpdateCommand;
import eu.qualimaster.coordination.events.CoordinationCommandExecutionEvent;
import eu.qualimaster.coordination.shutdown.Shutdown;
import eu.qualimaster.dataManagement.DataManager;
import eu.qualimaster.easy.extension.internal.AlgorithmProfileHelper;
import eu.qualimaster.easy.extension.internal.AlgorithmProfileHelper.ProfileData;
import eu.qualimaster.easy.extension.internal.QmProjectDescriptor;
import eu.qualimaster.events.EventManager;
import eu.qualimaster.file.Utils;
import eu.qualimaster.infrastructure.PipelineLifecycleEvent;
import eu.qualimaster.infrastructure.PipelineOptions;
import eu.qualimaster.monitoring.events.ChangeMonitoringEvent;
import net.ssehub.easy.basics.modelManagement.ModelInitializer;
import net.ssehub.easy.basics.modelManagement.ModelManagementException;
import net.ssehub.easy.basics.progress.ProgressObserver;
import net.ssehub.easy.instantiation.core.model.common.VilException;
import net.ssehub.easy.varModel.confModel.Configuration;
import net.ssehub.easy.varModel.management.VarModel;
import net.ssehub.easy.varModel.model.Project;

/**
 * Visits the commands for execution. Please note that when starting the execution of a single command,
 * the command must be placed on the {@link #commandStack} and at the end of the execution (no exception must
 * be thrown) {@link #writeCoordinationLog(CoordinationCommand, CoordinationExecutionResult)} must be
 * called which pops from the stack and performs the logging.
 * 
 * @author <NAME>
 */
class CoordinationCommandExecutionVisitor extends AbstractCoordinationCommandExecutionVisitor {

    private static final Set<Class<? extends CoordinationCommand>> DEFER_SUCCESSFUL_EXECUTION 
        = new HashSet<Class<? extends CoordinationCommand>>();
    private long timestamp = System.currentTimeMillis();
    private ActiveCommands activeCommands = null;
    
    static {
        // later, all command shall be deferred and report about the actual completion
        DEFER_SUCCESSFUL_EXECUTION.add(PipelineCommand.class);
    }
    
    /**
     * Creates a command execution visitor.
     * 
     * @param tracer a tracer for obtaining information about the execution
     */
    CoordinationCommandExecutionVisitor(IExecutionTracer tracer) {
        super(tracer);
    }
    
    /**
     * Sets the top-level command of this execution.
     * 
     * @param command the command
     * @return the active commands object
     */
    ActiveCommands setTopLevelCommand(CoordinationCommand command) {
        activeCommands = new ActiveCommands(command);
        return activeCommands;
    }
    
    /**
     * Returns the actual message id.
     * 
     * @return the actual message id
     */
    private String getCauseMessageId() {
        return null == activeCommands ? "" : activeCommands.getCauseMessageId();
    }
    
    /**
     * Handles sending a (storm commons) signal.
     * 
     * @param command the command causing the signal
     * @param signal the signal
     * @return whether the execution failed (instance) or not (<b>null</b>)
     */
    private static CoordinationExecutionResult send(CoordinationCommand command, 
        AbstractTopologyExecutorSignal signal) {
        CoordinationExecutionResult failing = null;
        try {
            if (!CoordinationManager.isTestingMode()) {
                getLogger().info("Sending signal " + signal);
                signal.sendSignal();
            }
        } catch (SignalException e) {
            failing = new CoordinationExecutionResult(command, e.getMessage(), 
                CoordinationExecutionCode.SIGNAL_SENDING_ERROR);
        }
        return failing;
    }

    @Override
    public CoordinationExecutionResult visitAlgorithmChangeCommand(AlgorithmChangeCommand command) {
        return handleAlgorithmChange(command, null);
    }
    
    @Override
    protected CoordinationExecutionResult handleAlgorithmChange(AlgorithmChangeCommand command, 
        List<ParameterChange> parameters) {
        CoordinationExecutionResult result;
        String pipelineName = command.getPipeline();
        INameMapping mapping = CoordinationManager.getNameMapping(pipelineName);
        ISubPipeline subPip = mapping.getSubPipelineByAlgorithmName(command.getAlgorithm());
        if (null != subPip) {
            CoordinationManager.deferCommand(subPip.getName(), PipelineLifecycleEvent.Status.STARTED, 
                new AlgorithmChangeAction(command, parameters, getTracer()));
            PipelineCommand cmd = new PipelineCommand(subPip.getName(), PipelineCommand.Status.START, 
                getSubPipelineOptions(pipelineName, command.getPipelineElement(), command.getOptions()));
            result = handlePipelineStart(cmd);
        } else {
            result = handleAlgorithmChangeImpl(command, parameters);
        }
        return result;
    }

    /**
     * Returns the sub pipeline options.
     * 
     * @param mainPipeline the main pipeline
     * @param mainPipelineElement the element in the main pipeline to start the sub-pipeline for
     * @param commandOptions the options introduced by the command
     * @return the options for the sub-pipeline marked as sub-pipeline
     */
    private PipelineOptions getSubPipelineOptions(String mainPipeline, String mainPipelineElement, 
        PipelineOptions commandOptions) {
        PipelineOptions opts = CoordinationManager.getPipelineOptions(mainPipeline);
        if (null == opts) {
            opts = new PipelineOptions(); // shall not occur
        } else {
            opts = new PipelineOptions(opts); // clone
        }
        if (null != commandOptions) { // take into account decisions done by the adaptation for algorithm change
            opts.merge(commandOptions);
        }
        opts.markAsSubPipeline(mainPipeline);
        PipelineInfo info = CoordinationManager.getPipelineInfo(mainPipeline);
        String algorithm = null != info ? info.getEnactedAlgorithm(mainPipelineElement) : null;
        opts.setOption(Constants.CONFIG_KEY_INITIAL_SUBPIPELINE, Boolean.toString(null == algorithm));
        return opts;
    }
{'fraction_non_alphanumeric': 0.0804652266793259, 'fraction_numerical': 0.00023736055067647758, 'mean_word_length': 4.169631901840491, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6004412', 'n_tokens_mistral': 4386, 'n_tokens_neox': 4332, 'n_words': 937}