def __init__(self, config=Config()):
<EOL>self.cleanup_called = False<EOL>if isinstance(config, dict):<EOL><INDENT>raise ConfigurationError(<EOL>'<STR_LIT>'<EOL>'<STR_LIT>')<EOL><DEDENT>self._config = config<EOL>self.run_dir = make_rundir(config.run_dir)<EOL>parsl.set_file_logger("<STR_LIT>".format(self.run_dir), level=logging.DEBUG)<EOL>logger.debug("<STR_LIT>".format(config))<EOL>logger.info("<STR_LIT>".format(get_version()))<EOL>self.checkpoint_lock = threading.Lock()<EOL>self.usage_tracker = UsageTracker(self)<EOL>self.usage_tracker.send_message()<EOL>self.tasks_completed_count = <NUM_LIT:0><EOL>self.tasks_failed_count = <NUM_LIT:0><EOL>self.monitoring = config.monitoring<EOL>if self.monitoring:<EOL><INDENT>if self.monitoring.logdir is None:<EOL><INDENT>self.monitoring.logdir = self.run_dir<EOL><DEDENT>self.monitoring.start()<EOL><DEDENT>self.time_began = datetime.datetime.now()<EOL>self.time_completed = None<EOL>self.run_id = str(uuid4())<EOL>logger.info("<STR_LIT>" + self.run_id)<EOL>self.workflow_name = None<EOL>if self.monitoring is not None and self.monitoring.workflow_name is not None:<EOL><INDENT>self.workflow_name = self.monitoring.workflow_name<EOL><DEDENT>else:<EOL><INDENT>for frame in inspect.stack():<EOL><INDENT>fname = os.path.basename(str(frame.filename))<EOL>parsl_file_names = ['<STR_LIT>']<EOL>if fname not in parsl_file_names:<EOL><INDENT>self.workflow_name = fname<EOL>break<EOL><DEDENT><DEDENT><DEDENT>self.workflow_version = str(self.time_began)<EOL>if self.monitoring is not None and self.monitoring.workflow_version is not None:<EOL><INDENT>self.workflow_version = self.monitoring.workflow_version<EOL><DEDENT>workflow_info = {<EOL>'<STR_LIT>': "<STR_LIT>".format(sys.version_info.major,<EOL>sys.version_info.minor,<EOL>sys.version_info.micro),<EOL>'<STR_LIT>': get_version(),<EOL>"<STR_LIT>": self.time_began,<EOL>'<STR_LIT>': None,<EOL>'<STR_LIT>': None,<EOL>'<STR_LIT>': self.run_id,<EOL>'<STR_LIT>': self.workflow_name,<EOL>'<STR_LIT>': self.workflow_version,<EOL>'<STR_LIT>': self.run_dir,<EOL>'<STR_LIT>': self.tasks_completed_count,<EOL>'<STR_LIT>': self.tasks_failed_count,<EOL>'<STR_LIT:user>': getuser(),<EOL>'<STR_LIT:host>': gethostname(),<EOL>}<EOL>if self.monitoring:<EOL><INDENT>self.monitoring.send(MessageType.WORKFLOW_INFO,<EOL>workflow_info)<EOL><DEDENT>checkpoints = self.load_checkpoints(config.checkpoint_files)<EOL>self.memoizer = Memoizer(self, memoize=config.app_cache, checkpoint=checkpoints)<EOL>self.checkpointed_tasks = <NUM_LIT:0><EOL>self._checkpoint_timer = None<EOL>self.checkpoint_mode = config.checkpoint_mode<EOL>self.data_manager = DataManager(self, max_threads=config.data_management_max_threads)<EOL>self.executors = {}<EOL>self.add_executors(config.executors + [self.data_manager])<EOL>if self.checkpoint_mode == "<STR_LIT>":<EOL><INDENT>try:<EOL><INDENT>h, m, s = map(int, config.checkpoint_period.split('<STR_LIT::>'))<EOL>checkpoint_period = (h * <NUM_LIT>) + (m * <NUM_LIT>) + s<EOL>self._checkpoint_timer = Timer(self.checkpoint, interval=checkpoint_period)<EOL><DEDENT>except Exception:<EOL><INDENT>logger.error("<STR_LIT>".format(config.checkpoint_period))<EOL>self._checkpoint_timer = Timer(self.checkpoint, interval=(<NUM_LIT:30> * <NUM_LIT>))<EOL><DEDENT><DEDENT>if any([x.managed for x in config.executors]):<EOL><INDENT>self.flowcontrol = FlowControl(self)<EOL><DEDENT>else:<EOL><INDENT>self.flowcontrol = FlowNoControl(self)<EOL><DEDENT>self.task_count = <NUM_LIT:0><EOL>self.tasks = {}<EOL>self.submitter_lock = threading.Lock()<EOL>atexit.register(self.atexit_cleanup)<EOL>
Initialize the DataFlowKernel.

        Parameters
        ----------
        config : Config
            A specification of all configuration options. For more details see the
            :class:~`parsl.config.Config` documentation.