<commit_before>'''
DataFlowKernel
==============

The DataFlowKernel adds dependency awareness to an existing executor.
It is responsible for managing futures, such that when dependencies are resolved, pending tasks
move to the runnable state.

Here's a simplified diagram of what happens internally::

    User             |        DFK         |    Executor
    ----------------------------------------------------------
                     |                    |
          Task-------+> +Submit           |
        App_Fu<------+--|                 |
                     |  Dependencies met  |
                     |         task-------+--> +Submit
                     |        Ex_Fu<------+----|
'''

import copy
import uuid
import logging
from inspect import signature
from concurrent.futures import Future
from functools import partial

from parsl.dataflow.error import *
from parsl.dataflow.states import States
from parsl.dataflow.futures import AppFuture
from parsl.app.futures import DataFuture

# Exceptions

logger = logging.getLogger(__name__)

class DataFlowKernel(object):
    """ DataFlowKernel
    """

    def __init__(self, executor, lazy_fail=True, fail_retries=2):
        """ Initialize the DataFlowKernel
        Args:
            executor (Executor): An executor object.

        KWargs:
            lazy_fail(Bool) : Default=True, determine failure behavior
            fail_retries(int): Default=2, Set the number of retry attempts in case of failure

        Returns:
            DataFlowKernel object
        """

        #self.pending         = {}
        #self.runnable        = {}
        #self.done            = {}
        self.task_count      = 0
        self.lazy_fail       = lazy_fail
        self.fail_retries    = 2
        self.fut_task_lookup = {}
        self.tasks           = {}
        self.executor       = executor
        self.scalable        = self.executor.scaling_enabled

    @staticmethod
    def _count_deps(depends, task_id):
        ''' Internal. Count the number of unresolved futures in the list depends'''

        count = 0
        for dep in depends:
            if isinstance(dep, Future) or issubclass(type(dep), Future):
                logger.debug("Task:%s dep:%s done:%s", task_id, dep, dep.done())
                if not dep.done():
                    count += 1

        return count

    def handle_update(self, task_id, future):
        ''' This function is called only as a callback from a task being done
        Move done task from runnable -> done
        Move newly doable tasks from pending -> runnable , and launch

        Args:
             task_id (string) : Task id which is a uuid string
             future (Future) : The future object corresponding to the task which makes this callback
        '''
        if future.done():
<commit_msg>DataFlowKernel initialization now can use configs to create appropriate structures<commit_after>'''
DataFlowKernel
==============

The DataFlowKernel adds dependency awareness to an existing executor.
It is responsible for managing futures, such that when dependencies are resolved, pending tasks
move to the runnable state.

Here's a simplified diagram of what happens internally::

    User             |        DFK         |    Executor
    ----------------------------------------------------------
                     |                    |
          Task-------+> +Submit           |
        App_Fu<------+--|                 |
                     |  Dependencies met  |
                     |         task-------+--> +Submit
                     |        Ex_Fu<------+----|
'''

import copy
import uuid
import logging
from inspect import signature
from concurrent.futures import Future
from functools import partial

from parsl.dataflow.error import *
from parsl.dataflow.states import States
from parsl.dataflow.futures import AppFuture
from parsl.app.futures import DataFuture
from parsl.execution_provider.provider_factory import ExecProviderFactory as EPF
# Exceptions

logger = logging.getLogger(__name__)

class DataFlowKernel(object):
    """ DataFlowKernel
    """

    def __init__(self, config=None, executors=None, lazy_fail=True, fail_retries=2):
        """ Initialize the DataFlowKernel

        Please note that keyword args passed to the DFK here will always override
        options passed in via the config.

        KWargs:
            config (Dict) : A single data object encapsulating all config attributes
            executors (list of Executor objs): Optional, kept for (somewhat) backward compatibility with 0.2.0
            lazy_fail(Bool) : Default=True, determine failure behavior
            fail_retries(int): Default=2, Set the number of retry attempts in case of failure

        Returns:
            DataFlowKernel object
        """

        self.config          = config
        if self.config :
            # Create the executors
            epf = EPF()
            self.executors = epf.make(self.config)

            # set global vars from config
            self.lazy_fail = self.config["globals"].get("lazyFail", lazy_fail)
            self.fail_retires = self.config["globals"].get("fail_retries", fail_retries)

        else:
            self.fail_retries = fail_retries
            self.lazy_fail    = lazy_fail
            self.executors    = executors

        self.task_count      = 0
        self.fut_task_lookup = {}
        self.tasks           = {}

        self.executor = self.executors[0]
        logger.warn("Using only executor: {0}".format(self.executor))


    @staticmethod
    def _count_deps(depends, task_id):
        ''' Internal. Count the number of unresolved futures in the list depends'''

        count = 0
        for dep in depends:
            if isinstance(dep, Future) or issubclass(type(dep), Future):
                logger.debug("Task:%s dep:%s done:%s", task_id, dep, dep.done())
                if not dep.done():
                    count += 1

        return count

    def handle_update(self, task_id, future):
        ''' This function is called only as a callback from a task being done
        Move done task from runnable -> done
        Move newly doable tasks from pending -> runnable , and launch

        Args:
             task_id (string) : Task id which is a uuid string
             future (Future) : The future object corresponding to the task which makes this callback
        '''
        if future.done():

{'fraction_non_alphanumeric': 0.10351258936897731, 'fraction_numerical': 0.0024867889337892445, 'mean_word_length': 1.827328646748682, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5796080', 'n_tokens_mistral': 1660, 'n_tokens_neox': 1581, 'n_words': 546}