<commit_before>package nextflow.processor
import java.util.concurrent.ArrayBlockingQueue
import java.util.concurrent.BlockingQueue

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 submitted tasks queue. It is implemented using a blocking queue,
     * so that no more than a fixed amount of tasks can be submitted concurrently
     */
    final BlockingQueue<TaskHandler> queue

    /**
     * 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

    final Duration dumpInterval

    final int queueSize

    /**
     * Initialise the monitor, creating the queue which will hold the tasks
     *
     * @param session
     * @param defQueueSize
     * @param defPollInterval
     */
    TaskPollingMonitor( Session session, String name, int defQueueSize, Duration defPollInterval ) {
        assert name, "Executor name cannot be empty"
        assert session, "Session object cannot be null"

        this.session = session
        this.dispatcher = session.dispatcher
        this.pollIntervalMillis = session.getPollInterval(name, defPollInterval).toMillis()
        this.queueSize = session.getQueueSize(name, defQueueSize)
        this.dumpInterval = session.getMonitorDumpInterval(name)

        log.debug "Creating executor '$name' > queue size: $queueSize; poll-interval: $pollIntervalMillis; dump-interval: $dumpInterval"
        this.queue = new ArrayBlockingQueue<TaskHandler>(defQueueSize)

        killOnExit()
    }

    /**
     * Launch the monitoring thread
     *
     * @return The monitor object itself
     */
    def TaskPollingMonitor start() {
        log.debug ">>> phaser register (scheduler)"
        session.phaser.register()

        Thread.start {
            try {
                pollLoop()
            }
            finally {
                log.debug "<<< phaser de-register (scheduler)"
                session.phaser.arriveAndDeregister()
            }
        }

        return this
    }

    /**
     * Implements the polling strategy
     */
    protected void pollLoop() {

        while( true ) {
            long time = System.currentTimeMillis()
            log.trace "Scheduler queue size: ${queue.size()}"

            // check all running tasks for termination
            checkAll(queue)

            if( session.isTerminated() && queue.size() == 0 ) {
                break
            }

            def delta = this.pollIntervalMillis - (System.currentTimeMillis() - time)
            if( delta>0 ) {
                sleep(delta)
            }

            // dump this line every two minutes
            dumpInterval.throttle(true) {
                log.debug "!!Tasks to be completed: ${queue.size()} -- first: ${queue.peek()}"
            }
        }
    }


    /**
     * Add an entry to the queue of monitored tasks
     *
     * @param handler The {@code TaskHandler} instance for the task
     */
    @Override<commit_msg>Patching bug when setting the monitor queue-size<commit_after>package nextflow.processor
import java.util.concurrent.ArrayBlockingQueue
import java.util.concurrent.BlockingQueue

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 submitted tasks queue. It is implemented using a blocking queue,
     * so that no more than a fixed amount of tasks can be submitted concurrently
     */
    final BlockingQueue<TaskHandler> queue

    /**
     * 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

    final Duration dumpInterval

    final int queueSize

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

    /**
     * Initialise the monitor, creating the queue which will hold the tasks
     *
     * @param session
     * @param defQueueSize
     * @param defPollInterval
     */
    TaskPollingMonitor( Session session, String execName, int defQueueSize, Duration defPollInterval ) {
        assert execName, "Executor name cannot be empty"
        assert session, "Session object cannot be null"

        this.name = execName
        this.session = session
        this.dispatcher = session.dispatcher
        this.pollIntervalMillis = session.getPollInterval(name, defPollInterval).toMillis()
        this.queueSize = session.getQueueSize(name, defQueueSize)
        this.dumpInterval = session.getMonitorDumpInterval(name)

        log.debug "Creating executor '$name' > queue size: $queueSize; poll-interval: $pollIntervalMillis; dump-interval: $dumpInterval"
        this.queue = new ArrayBlockingQueue<TaskHandler>(queueSize,true)

        killOnExit()
    }

    /**
     * Launch the monitoring thread
     *
     * @return The monitor object itself
     */
    def TaskPollingMonitor start() {
        log.debug ">>> phaser register (scheduler)"
        session.phaser.register()

        Thread.start {
            try {
                pollLoop()
            }
            finally {
                log.debug "<<< phaser de-register (scheduler)"
                session.phaser.arriveAndDeregister()
            }
        }

        return this
    }

    /**
     * Implements the polling strategy
     */
    protected void pollLoop() {

        while( true ) {
            long time = System.currentTimeMillis()
            log.trace "Scheduler queue size: ${queue.size()}"

            // check all running tasks for termination
            checkAll(queue)

            if( session.isTerminated() && queue.size() == 0 ) {
                break
            }

            def delta = this.pollIntervalMillis - (System.currentTimeMillis() - time)
            if( delta>0 ) {
                sleep(delta)
            }

            // dump this line every two minutes
            dumpInterval.throttle(true) {
                log.debug "!! executor $name > tasks to be completed: ${queue.size()} -- first: ${queue.peek()}"
            }
        }
    }


    /**
     * Add an entry to the queue of monitored tasks
     *
     * @param handler The {@code TaskHandler} instance for the task
     */
    @Override
{'fraction_non_alphanumeric': 0.0873388042203986, 'fraction_numerical': 0.0011723329425556857, 'mean_word_length': 2.0994550408719346, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 24, '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': '6887457', 'n_tokens_mistral': 1827, 'n_tokens_neox': 1747, 'n_words': 535}