<commit_before>package nextflow.processor

import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.Condition
import java.util.concurrent.locks.Lock
import java.util.concurrent.locks.ReentrantLock

import groovy.util.logging.Slf4j
import nextflow.Session
import nextflow.util.Duration
/**
 *
 * Monitors the queued tasks waiting for their termination
 *
 * @author <NAME> <<EMAIL>>
 */

@Slf4j
class TaskPollingMonitor implements TaskMonitor {

    /**
     * The current session object
     */
    final Session session

    /**
     * The tasks dispatcher
     */
    final TaskDispatcher dispatcher

    /**
     * The time interval (in milliseconds) elapsed which execute a new poll
     */
    final long pollIntervalMillis

    /**
     * Determines how often the executor status is written in the application log file (default: 5min)
     */
    final Duration dumpInterval

    /**
     * The name of the executor associated to this monitor
     */
    final String name

    /**
     * A lock object used to access in a synchronous manner the polling queue
     */
    private Lock mutex

    /**
     * A condition that signal when at least a complete task is available
     */
    private Condition taskComplete

    private Condition notFull

    private Queue<TaskHandler> pollingQueue

    private int capacity

    /**
     * Initialise the monitor, creating the queue which will hold the tasks
     *
     * @param session
     * @param execName
     * @param defPollInterval
     */
    TaskPollingMonitor( Session session, int capacity, Duration pollInterval, Duration dumpInterval = Duration.of('5min') ) {
        assert session, "Session object cannot be null"

        this.session = session
        this.dispatcher = session.dispatcher
        this.pollIntervalMillis = pollInterval.toMillis()
        this.dumpInterval = dumpInterval
        this.capacity = capacity

        this.pollingQueue = new ConcurrentLinkedQueue<>()

    }


    static TaskPollingMonitor create( Session session, String name, int defQueueSize, Duration defPollInterval ) {
        assert session
        assert name
        final capacity = session.getQueueSize(name, defQueueSize)
        final pollInterval = session.getPollInterval(name, defPollInterval)
        final dumpInterval = session.getMonitorDumpInterval(name)

        log.debug "Creating task monitor for executor '$name' > capacity: $capacity; pollInterval: $pollInterval; dumpInterval: $dumpInterval "
        new TaskPollingMonitor(session, capacity, pollInterval, dumpInterval)
    }

    static TaskPollingMonitor create( Session session, String name, Duration defPollInterval ) {
        assert session
        assert name

        final pollInterval = session.getPollInterval(name, defPollInterval)
        final dumpInterval = session.getMonitorDumpInterval(name)

        log.debug "Creating task monitor for executor '$name' > pollInterval: $pollInterval; dumpInterval: $dumpInterval "
        new TaskPollingMonitor(session, 0, pollInterval, dumpInterval)
    }

    protected Queue<TaskHandler> getPollingQueue() { pollingQueue }

    /**
     * Set the number of tasks the monitor will handle in parallel.
     * This value must be changed by using a mutex lock
     *
     * @param value
     */
    protected void setCapacity( int value ) {
        this.capacity = value
    }

    /**
     * @return the current capacity value by the number of slots specified
     */
    public int getCapacity() { capacity }

    /**
     * Increment the monitor capacity b
     *
     * @param slots The number of slots to add
     * @return The new capacity value
     */
    protected int capacityInc( int slots = 1 ) {

        mutex.withLock {
            def result = capacity += slots
            notFull.signal()
            return result
        }

    }

    /**
     * Decrement the monitor capacity by the number of slots specified
     *
     * @param slots
     * @return The new capacity value
     */
    protected int capacityDec( int slots = 1 ) {

        mutex.withLock {
            return capacity -= slots
        }
    }


    /**
     * Add a new task handler to queue tasks queue
     *
     * @param handler
     */
    @Override
    void put(TaskHandler handler) {

        //
        // This guarantee that the 'pollingQueue' does not contain
        // more entries than the specified 'capacity'
        //
        mutex.withLock(true) {

            while ( pollingQueue.size() >= capacity )
                notFull.await();

            if( !session.isTerminated())
                handler.submit()

            pollingQueue.add(handler)
        }

        dispatcher.notifySubmitted(handler)
    }

    /**
     * Remove a task from the processing tasks queue
     *
     * @param handler A {@code HzTaskHandler} instance
     * @return
     */
    @Override
    boolean drop(TaskHandler handler) {
        if( !handler ) {
            log.debug "Unknown task handler to drop: $handler"
            return false
        }

        log.trace "Dropping task handler: $handler"
        mutex.withLock {
            if( pollingQueue.remove(handler) ) {<commit_msg>Adding task to queue before submitting for execution it<commit_after>package nextflow.processor

import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.locks.Condition
import java.util.concurrent.locks.Lock
import java.util.concurrent.locks.ReentrantLock

import groovy.util.logging.Slf4j
import nextflow.Session
import nextflow.util.Duration
/**
 *
 * Monitors the queued tasks waiting for their termination
 *
 * @author <NAME> <<EMAIL>>
 */

@Slf4j
class TaskPollingMonitor implements TaskMonitor {

    /**
     * The current session object
     */
    final Session session

    /**
     * The tasks dispatcher
     */
    final TaskDispatcher dispatcher

    /**
     * The time interval (in milliseconds) elapsed which execute a new poll
     */
    final long pollIntervalMillis

    /**
     * Determines how often the executor status is written in the application log file (default: 5min)
     */
    final Duration dumpInterval

    /**
     * The name of the executor associated to this monitor
     */
    final String name

    /**
     * A lock object used to access in a synchronous manner the polling queue
     */
    private Lock mutex

    /**
     * A condition that signal when at least a complete task is available
     */
    private Condition taskComplete

    private Condition notFull

    private Queue<TaskHandler> pollingQueue

    private int capacity

    /**
     * Initialise the monitor, creating the queue which will hold the tasks
     *
     * @param session
     * @param execName
     * @param defPollInterval
     */
    TaskPollingMonitor( Session session, int capacity, Duration pollInterval, Duration dumpInterval = Duration.of('5min') ) {
        assert session, "Session object cannot be null"

        this.session = session
        this.dispatcher = session.dispatcher
        this.pollIntervalMillis = pollInterval.toMillis()
        this.dumpInterval = dumpInterval
        this.capacity = capacity

        this.pollingQueue = new ConcurrentLinkedQueue<>()

    }


    static TaskPollingMonitor create( Session session, String name, int defQueueSize, Duration defPollInterval ) {
        assert session
        assert name
        final capacity = session.getQueueSize(name, defQueueSize)
        final pollInterval = session.getPollInterval(name, defPollInterval)
        final dumpInterval = session.getMonitorDumpInterval(name)

        log.debug "Creating task monitor for executor '$name' > capacity: $capacity; pollInterval: $pollInterval; dumpInterval: $dumpInterval "
        new TaskPollingMonitor(session, capacity, pollInterval, dumpInterval)
    }

    static TaskPollingMonitor create( Session session, String name, Duration defPollInterval ) {
        assert session
        assert name

        final pollInterval = session.getPollInterval(name, defPollInterval)
        final dumpInterval = session.getMonitorDumpInterval(name)

        log.debug "Creating task monitor for executor '$name' > pollInterval: $pollInterval; dumpInterval: $dumpInterval "
        new TaskPollingMonitor(session, 0, pollInterval, dumpInterval)
    }

    protected Queue<TaskHandler> getPollingQueue() { pollingQueue }

    /**
     * Set the number of tasks the monitor will handle in parallel.
     * This value must be changed by using a mutex lock
     *
     * @param value
     */
    protected void setCapacity( int value ) {
        this.capacity = value
    }

    /**
     * @return the current capacity value by the number of slots specified
     */
    public int getCapacity() { capacity }

    /**
     * Increment the monitor capacity b
     *
     * @param slots The number of slots to add
     * @return The new capacity value
     */
    protected int capacityInc( int slots = 1 ) {

        mutex.withLock {
            def result = capacity += slots
            notFull.signal()
            return result
        }

    }

    /**
     * Decrement the monitor capacity by the number of slots specified
     *
     * @param slots
     * @return The new capacity value
     */
    protected int capacityDec( int slots = 1 ) {

        mutex.withLock {
            return capacity -= slots
        }
    }


    /**
     * Add a new task handler to queue tasks queue
     *
     * @param handler
     */
    @Override
    void put(TaskHandler handler) {

        //
        // This guarantee that the 'pollingQueue' does not contain
        // more entries than the specified 'capacity'
        //
        boolean done = false
        mutex.withLock(true) {

            while ( pollingQueue.size() >= capacity )
                notFull.await();

            if( !session.isTerminated()) {
                pollingQueue.add(handler)
                handler.submit()
                done = true
            }

        }

        if( done )
            dispatcher.notifySubmitted(handler)
    }

    /**
     * Remove a task from the processing tasks queue
     *
     * @param handler A {@code HzTaskHandler} instance
     * @return
     */
    @Override
    boolean drop(TaskHandler handler) {
        if( !handler ) {
            log.debug "Unknown task handler to drop: $handler"
            return false
        }

        log.trace "Dropping task handler: $handler"
        mutex.withLock {
            if( pollingQueue.remove(handler) ) {
{'fraction_non_alphanumeric': 0.07535410764872522, 'fraction_numerical': 0.0013220018885741266, 'mean_word_length': 2.3211037942928816, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 21, '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': '1086087', 'n_tokens_mistral': 2794, 'n_tokens_neox': 2635, 'n_words': 903}