<commit_before>/*
 * Copyright (c) 2013-2015, Centre for Genomic Regulation (CRG).
 * Copyright (c) 2013-2015, <NAME> and the respective authors.
 *
 *   This file is part of 'Nextflow'.
 *
 *   Nextflow is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   Nextflow is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Nextflow.  If not, see <http://www.gnu.org/licenses/>.
 */

package nextflow
import java.nio.file.Path
import java.nio.file.Paths
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors

import groovy.transform.CompileStatic
import groovy.transform.Memoized
import groovy.transform.PackageScope
import groovy.util.logging.Slf4j
import groovyx.gpars.GParsConfig
import groovyx.gpars.dataflow.operator.DataflowProcessor
import nextflow.exception.MissingLibraryException
import nextflow.processor.TaskDispatcher
import nextflow.processor.TaskProcessor
import nextflow.processor.TaskRun
import nextflow.script.ScriptBinding
import nextflow.trace.TimelineObserver
import nextflow.trace.TraceFileObserver
import nextflow.trace.TraceObserver
import nextflow.util.Barrier
import nextflow.util.ConfigHelper
import nextflow.util.Duration
/**
 * Holds the information on the current execution
 *
 * @author <NAME> <<EMAIL>>
 */
@Slf4j
@CompileStatic
class Session implements ISession {

    static class TaskFault {
        Throwable error
        String report
        TaskRun task
    }

    static final String EXTRAE_TRACE_CLASS = 'nextflow.extrae.ExtraeTraceObserver'

    /**
     * Keep a list of all processor created
     */
    final List<DataflowProcessor> allProcessors = []

    /**
     * Dispatch tasks for executions
     */
    TaskDispatcher dispatcher

    /**
     * Holds the configuration object
     */
    Map config

    /**
     * Enable / disable tasks result caching
     */
    boolean cacheable

    /**
     * whenever it has been launched in resume mode
     */
    boolean resumeMode

    /**
     * The folder where tasks temporary files are stored
     */
    Path workDir

    /**
     * The folder where the main script is contained
     */
    Path baseDir

    /**
     * The pipeline script name (without parent path)
     */
    String scriptName

    /**
     * Folder(s) containing libs and classes to be added to the classpath
     */
    List<Path> libDir

    private Path binDir

    /**
     * The unique identifier of this session
     */
    private UUID uniqueId

    private Barrier processesBarrier = new Barrier()

    private Barrier monitorsBarrier = new Barrier()

    private volatile boolean aborted

    private volatile boolean terminated

    private volatile ExecutorService execService

    private volatile TaskFault fault

    private ScriptBinding binding

    private ClassLoader classLoader

    private List<Closure<Void>> shutdownCallbacks = []

    private int poolSize

    private List<TraceObserver> observers = []

    private boolean statsEnabled

    boolean getStatsEnabled() { statsEnabled }

    TaskFault getFault() { fault }

    /**
     * Creates a new session with an 'empty' (default) configuration
     */
    Session() {
        create(new ScriptBinding([:]))
    }

    /**
     * Create a new session given the {@link ScriptBinding} object
     *
     * @param binding
     */
    Session(ScriptBinding binding) {
        create(binding)
    }

    /**
     * Create a new session given the configuration specified
     *
     * @param config
     */
    Session(Map cfg) {
        final config = cfg instanceof ConfigObject ? cfg.toMap() : cfg
        create(new ScriptBinding(config))
    }

    /**
     * @return The current session {@link UUID}
     */
    UUID getUniqueId() { uniqueId }

    /**
     * @return The session max number of thread allowed
     */
    int getPoolSize() { poolSize }

    /**
     * @return The session {@link TaskDispatcher}
     */
    TaskDispatcher getDispatcher() { dispatcher }

    /**
     * Creates a new session using the configuration properties provided
     *
     * @param binding
     */
    private void create( ScriptBinding binding ) {
        assert binding != null

        this.binding = binding
        this.config = binding.config

        // poor man session object dependency injection
        Global.setSession(this)
        Global.setConfig(config)

        cacheable = config.cacheable

        // sets resumeMode and uniqueId
        if( config.resume ) {
            resumeMode = true
            uniqueId = UUID.fromString(config.resume as String)
        }
        else {
           uniqueId = UUID.randomUUID()
        }
        log.debug "Session uuid: $uniqueId"

        // normalize taskConfig object
        if( config.process == null ) config.process = [:]
        if( config.env == null ) config.env = [:]

        if( !config.poolSize ) {
            final cpus = Runtime.getRuntime().availableProcessors()
            config.poolSize = cpus +1
        }

        //set the thread pool size
        this.poolSize = config.poolSize as int
        log.debug "Executor pool size: ${poolSize}"

        // create the task dispatcher instance
        this.dispatcher = new TaskDispatcher(this)

    }

    /**
     * Initialize the session workDir, libDir, baseDir and scriptName variables
     */
    void init( Path scriptPath ) {

        this.workDir = ((config.workDir ?: 'work') as Path).complete()
        this.setLibDir( config.libDir as String )

        if( scriptPath ) {
            // the folder that contains the main script
            this.setBaseDir(scriptPath.parent)
            // set the script name attribute
            this.setScriptName(scriptPath.name)
        }

        this.observers = createObservers()
        this.statsEnabled = observers.size()>0
    }

    /**
     * Given the `run` command line options creates the required {@link TraceObserver}s
     *
     * @param runOpts The {@code CmdRun} object holding the run command options
     * @return A list of {@link TraceObserver} objects or an empty list
     */
    @PackageScope
    List createObservers() {
        def result = []

        createTraceFileObserver(result)
        createTimelineObserver(result)
        createExtraeObserver(result)

        return result
    }

    /**
     * create the Extrae trace observer
     */
    protected void createExtraeObserver(List result) {
        Boolean isEnabled = config.navigate('extrae.enabled') as Boolean
        if( isEnabled ) {
            try {
                result << (TraceObserver)Class.forName(EXTRAE_TRACE_CLASS).newInstance()
            }
            catch( Exception e ) {
                log.warn("Unable to load Extrae profiler",e)
            }
        }
    }

    /**
     * Create timeline report file observer
     */
    protected void createTimelineObserver(List result) {
        Boolean isEnabled = config.navigate('timeline.enabled') as Boolean
        if( isEnabled ) {
            String fileName = config.navigate('timeline.file')
            if( !fileName ) fileName = TimelineObserver.DEF_FILE_NAME
            def traceFile = (fileName as Path).complete()
            def observer = new TimelineObserver(traceFile)
            result << observer
        }
    }

    /*
     * create the execution trace observer
     */
    protected void createTraceFileObserver(List result) {
        Boolean isEnabled = config.navigate('trace.enabled') as Boolean
        if( isEnabled ) {
            String fileName = config.navigate('trace.file')
            if( !fileName ) fileName = TraceFileObserver.DEF_FILE_NAME
            def traceFile = (fileName as Path).complete()
            def observer = new TraceFileObserver(traceFile)
            config.navigate('trace.raw') { it -> observer.useRawNumbers(it == true) }
            config.navigate('trace.sep') { observer.separator = it }
            config.navigate('trace.fields') { observer.setFieldsAndFormats(it) }
            result << observer
        }
    }

    def Session start() {
        log.debug "Session start invoked"

        /*
         * - register all of them in the dispatcher class
         * - register the onComplete event
         */
        for( int i=0; i<observers.size(); i++ ) {
            def trace = observers.get(i)
            log.debug "Registering observer: ${trace.class.name}"
            dispatcher.register(trace)
            onShutdown { trace.onFlowComplete() }
        }

        // register shut-down cleanup hooks
        Global.onShutdown { cleanUp() }
        // create tasks executor
        execService = Executors.newFixedThreadPool(poolSize)
        // signal start to tasks dispatcher
        dispatcher.start()
        // signal start to trace observers
        observers.each { trace -> trace.onFlowStart(this) }

        return this
    }

    ScriptBinding getBinding() { binding }

    ClassLoader getClassLoader() { classLoader }

    Session setClassLoader( ClassLoader loader ) {
        this.classLoader = loader
        return this
    }

    @PackageScope
    Barrier getBarrier() { monitorsBarrier }

    /**
     * The folder where script binaries file are located, by default the folder 'bin'
     * in the script base directory
     */
    Path getBinDir() {
        binDir
    }

    void setBaseDir( Path baseDir ) {
        this.baseDir = baseDir

        def path = baseDir.resolve('bin')
        if( path.exists() && path.isDirectory() ) {
            this.binDir = path
        }
        else {
            log.debug "Script base path does not exist or is not a directory: ${path}"
        }
    }

    def void setLibDir( String str ) {

        if( !str ) return

        def files = str.split( File.pathSeparator ).collect { String it -> Paths.get(it) }
        if( !files ) return

        libDir = []
        for( Path file : files ) {
            if( !file.exists() )
                throw new MissingLibraryException("Cannot find specified library: ${file.complete()}")

            libDir << file
        }
    }

    def List<Path> getLibDir() {
        if( libDir )
            return libDir

        libDir = []
        def localLib = baseDir ? baseDir.resolve('lib') : Paths.get('lib')
        if( localLib.exists() ) {
            log.debug "Using default localLib path: $localLib"
            libDir << localLib
        }
        return libDir
    }

    /**
     * Await the termination of all processors
     */
    void await() {
        log.debug "Session await"
        processesBarrier.awaitCompletion()
        log.debug "Session await > processes completed"
        terminated = true
        monitorsBarrier.awaitCompletion()
        log.debug "Session await > done"
    }

    void destroy() {
        log.trace "Session > destroying"
        cleanUp()
        log.trace "Session > after cleanup"

        if( !aborted ) {
            allProcessors *. join()
            log.trace "Session > after processors join"
        }

        execService.shutdown()
        log.trace "Session > executor shutdown"
        execService = null
        log.debug "Session destroyed"
    }

    final protected void cleanUp() {

        log.trace "Shutdown: $shutdownCallbacks"
        List<Closure<Void>> all = new ArrayList<>(shutdownCallbacks)
        for( def hook : all ) {
            try {
                hook.call()
            }
            catch( Exception e ) {<commit_msg>Fixed Channel.subscribe is invoked after the Workflow.onComplete handler #105<commit_after>/*
 * Copyright (c) 2013-2015, Centre for Genomic Regulation (CRG).
 * Copyright (c) 2013-2015, <NAME> and the respective authors.
 *
 *   This file is part of 'Nextflow'.
 *
 *   Nextflow is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   Nextflow is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with Nextflow.  If not, see <http://www.gnu.org/licenses/>.
 */

package nextflow
import java.nio.file.Path
import java.nio.file.Paths
import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors

import groovy.transform.CompileStatic
import groovy.transform.Memoized
import groovy.transform.PackageScope
import groovy.util.logging.Slf4j
import groovyx.gpars.GParsConfig
import groovyx.gpars.dataflow.operator.DataflowProcessor
import nextflow.exception.MissingLibraryException
import nextflow.processor.TaskDispatcher
import nextflow.processor.TaskProcessor
import nextflow.processor.TaskRun
import nextflow.script.ScriptBinding
import nextflow.trace.TimelineObserver
import nextflow.trace.TraceFileObserver
import nextflow.trace.TraceObserver
import nextflow.util.Barrier
import nextflow.util.ConfigHelper
import nextflow.util.Duration
/**
 * Holds the information on the current execution
 *
 * @author <NAME> <<EMAIL>>
 */
@Slf4j
@CompileStatic
class Session implements ISession {

    static class TaskFault {
        Throwable error
        String report
        TaskRun task
    }

    static final String EXTRAE_TRACE_CLASS = 'nextflow.extrae.ExtraeTraceObserver'

    /**
     * Keep a list of all processor created
     */
    final List<DataflowProcessor> allProcessors = []

    /**
     * Dispatch tasks for executions
     */
    TaskDispatcher dispatcher

    /**
     * Holds the configuration object
     */
    Map config

    /**
     * Enable / disable tasks result caching
     */
    boolean cacheable

    /**
     * whenever it has been launched in resume mode
     */
    boolean resumeMode

    /**
     * The folder where tasks temporary files are stored
     */
    Path workDir

    /**
     * The folder where the main script is contained
     */
    Path baseDir

    /**
     * The pipeline script name (without parent path)
     */
    String scriptName

    /**
     * Folder(s) containing libs and classes to be added to the classpath
     */
    List<Path> libDir

    private Path binDir

    /**
     * The unique identifier of this session
     */
    private UUID uniqueId

    private Barrier processesBarrier = new Barrier()

    private Barrier monitorsBarrier = new Barrier()

    private volatile boolean aborted

    private volatile boolean terminated

    private volatile ExecutorService execService

    private volatile TaskFault fault

    private ScriptBinding binding

    private ClassLoader classLoader

    private List<Closure<Void>> shutdownCallbacks = []

    private int poolSize

    private List<TraceObserver> observers = []

    private boolean statsEnabled

    boolean getStatsEnabled() { statsEnabled }

    TaskFault getFault() { fault }

    /**
     * Creates a new session with an 'empty' (default) configuration
     */
    Session() {
        create(new ScriptBinding([:]))
    }

    /**
     * Create a new session given the {@link ScriptBinding} object
     *
     * @param binding
     */
    Session(ScriptBinding binding) {
        create(binding)
    }

    /**
     * Create a new session given the configuration specified
     *
     * @param config
     */
    Session(Map cfg) {
        final config = cfg instanceof ConfigObject ? cfg.toMap() : cfg
        create(new ScriptBinding(config))
    }

    /**
     * @return The current session {@link UUID}
     */
    UUID getUniqueId() { uniqueId }

    /**
     * @return The session max number of thread allowed
     */
    int getPoolSize() { poolSize }

    /**
     * @return The session {@link TaskDispatcher}
     */
    TaskDispatcher getDispatcher() { dispatcher }

    /**
     * Creates a new session using the configuration properties provided
     *
     * @param binding
     */
    private void create( ScriptBinding binding ) {
        assert binding != null

        this.binding = binding
        this.config = binding.config

        // poor man session object dependency injection
        Global.setSession(this)
        Global.setConfig(config)

        cacheable = config.cacheable

        // sets resumeMode and uniqueId
        if( config.resume ) {
            resumeMode = true
            uniqueId = UUID.fromString(config.resume as String)
        }
        else {
           uniqueId = UUID.randomUUID()
        }
        log.debug "Session uuid: $uniqueId"

        // normalize taskConfig object
        if( config.process == null ) config.process = [:]
        if( config.env == null ) config.env = [:]

        if( !config.poolSize ) {
            final cpus = Runtime.getRuntime().availableProcessors()
            config.poolSize = cpus +1
        }

        //set the thread pool size
        this.poolSize = config.poolSize as int
        log.debug "Executor pool size: ${poolSize}"

        // create the task dispatcher instance
        this.dispatcher = new TaskDispatcher(this)

    }

    /**
     * Initialize the session workDir, libDir, baseDir and scriptName variables
     */
    void init( Path scriptPath ) {

        this.workDir = ((config.workDir ?: 'work') as Path).complete()
        this.setLibDir( config.libDir as String )

        if( scriptPath ) {
            // the folder that contains the main script
            this.setBaseDir(scriptPath.parent)
            // set the script name attribute
            this.setScriptName(scriptPath.name)
        }

        this.observers = createObservers()
        this.statsEnabled = observers.size()>0
    }

    /**
     * Given the `run` command line options creates the required {@link TraceObserver}s
     *
     * @param runOpts The {@code CmdRun} object holding the run command options
     * @return A list of {@link TraceObserver} objects or an empty list
     */
    @PackageScope
    List createObservers() {
        def result = []

        createTraceFileObserver(result)
        createTimelineObserver(result)
        createExtraeObserver(result)

        return result
    }

    /**
     * create the Extrae trace observer
     */
    protected void createExtraeObserver(List result) {
        Boolean isEnabled = config.navigate('extrae.enabled') as Boolean
        if( isEnabled ) {
            try {
                result << (TraceObserver)Class.forName(EXTRAE_TRACE_CLASS).newInstance()
            }
            catch( Exception e ) {
                log.warn("Unable to load Extrae profiler",e)
            }
        }
    }

    /**
     * Create timeline report file observer
     */
    protected void createTimelineObserver(List result) {
        Boolean isEnabled = config.navigate('timeline.enabled') as Boolean
        if( isEnabled ) {
            String fileName = config.navigate('timeline.file')
            if( !fileName ) fileName = TimelineObserver.DEF_FILE_NAME
            def traceFile = (fileName as Path).complete()
            def observer = new TimelineObserver(traceFile)
            result << observer
        }
    }

    /*
     * create the execution trace observer
     */
    protected void createTraceFileObserver(List result) {
        Boolean isEnabled = config.navigate('trace.enabled') as Boolean
        if( isEnabled ) {
            String fileName = config.navigate('trace.file')
            if( !fileName ) fileName = TraceFileObserver.DEF_FILE_NAME
            def traceFile = (fileName as Path).complete()
            def observer = new TraceFileObserver(traceFile)
            config.navigate('trace.raw') { it -> observer.useRawNumbers(it == true) }
            config.navigate('trace.sep') { observer.separator = it }
            config.navigate('trace.fields') { observer.setFieldsAndFormats(it) }
            result << observer
        }
    }

    def Session start() {
        log.debug "Session start invoked"

        /*
         * - register all of them in the dispatcher class
         * - register the onComplete event
         */
        for( int i=0; i<observers.size(); i++ ) {
            def trace = observers.get(i)
            log.debug "Registering observer: ${trace.class.name}"
            dispatcher.register(trace)
            onShutdown { trace.onFlowComplete() }
        }

        // register shut-down cleanup hooks
        Global.onShutdown { cleanUp() }
        // create tasks executor
        execService = Executors.newFixedThreadPool(poolSize)
        // signal start to tasks dispatcher
        dispatcher.start()
        // signal start to trace observers
        observers.each { trace -> trace.onFlowStart(this) }

        return this
    }

    ScriptBinding getBinding() { binding }

    ClassLoader getClassLoader() { classLoader }

    Session setClassLoader( ClassLoader loader ) {
        this.classLoader = loader
        return this
    }

    @PackageScope
    Barrier getBarrier() { monitorsBarrier }

    /**
     * The folder where script binaries file are located, by default the folder 'bin'
     * in the script base directory
     */
    Path getBinDir() {
        binDir
    }

    void setBaseDir( Path baseDir ) {
        this.baseDir = baseDir

        def path = baseDir.resolve('bin')
        if( path.exists() && path.isDirectory() ) {
            this.binDir = path
        }
        else {
            log.debug "Script base path does not exist or is not a directory: ${path}"
        }
    }

    def void setLibDir( String str ) {

        if( !str ) return

        def files = str.split( File.pathSeparator ).collect { String it -> Paths.get(it) }
        if( !files ) return

        libDir = []
        for( Path file : files ) {
            if( !file.exists() )
                throw new MissingLibraryException("Cannot find specified library: ${file.complete()}")

            libDir << file
        }
    }

    def List<Path> getLibDir() {
        if( libDir )
            return libDir

        libDir = []
        def localLib = baseDir ? baseDir.resolve('lib') : Paths.get('lib')
        if( localLib.exists() ) {
            log.debug "Using default localLib path: $localLib"
            libDir << localLib
        }
        return libDir
    }

    /**
     * Await the termination of all processors
     */
    void await() {
        log.debug "Session await"
        processesBarrier.awaitCompletion()
        log.debug "Session await > processes completed"
        terminated = true
        monitorsBarrier.awaitCompletion()
        log.debug "Session await > done"
    }

    void destroy() {
        log.trace "Session > destroying"

        if( !aborted ) {
            allProcessors *. join()
            log.trace "Session > after processors join"
        }

        cleanUp()
        log.trace "Session > after cleanup"

        execService.shutdown()
        log.trace "Session > executor shutdown"
        execService = null
        log.debug "Session destroyed"
    }

    final protected void cleanUp() {

        log.trace "Shutdown: $shutdownCallbacks"
        List<Closure<Void>> all = new ArrayList<>(shutdownCallbacks)
        for( def hook : all ) {
            try {
                hook.call()
            }
            catch( Exception e ) {
{'fraction_non_alphanumeric': 0.08709365406451071, 'fraction_numerical': 0.0019765339164809284, 'mean_word_length': 2.165601703940362, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4809950', 'n_tokens_mistral': 6371, 'n_tokens_neox': 6151, 'n_words': 1916}