':param processor_manager: manager to use
:type processor_manager: DagFileProcessorManager
:return: None'

def _execute_helper(self, processor_manager):

    self.executor.start()
    session = settings.Session()
    self.logger.info(u'Resetting    orphaned    tasks    for    active    dag    runs')
    self.reset_state_for_orphaned_tasks(session=session)
    session.close()
    execute_start_time = datetime.now()
    last_stat_print_time = datetime(2000, 1, 1)
    last_self_heartbeat_time = datetime.now()
    last_dag_dir_refresh_time = datetime.now()
    known_file_paths = processor_manager.file_paths
    while (((datetime.now() - execute_start_time).total_seconds() < self.run_duration) or (self.run_duration < 0)):
        self.logger.debug(u'Starting    Loop...')
        loop_start_time = time.time()
        elapsed_time_since_refresh = (datetime.now() - last_dag_dir_refresh_time).total_seconds()
        if (elapsed_time_since_refresh > self.dag_dir_list_interval):
            self.logger.info(u'Searching    for    files    in    {}'.format(self.subdir))
            known_file_paths = list_py_file_paths(self.subdir)
            last_dag_dir_refresh_time = datetime.now()
            self.logger.info(u'There    are    {}    files    in    {}'.format(len(known_file_paths), self.subdir))
            processor_manager.set_file_paths(known_file_paths)
            self.logger.debug(u'Removing    old    import    errors')
            self.clear_nonexistent_import_errors(known_file_paths=known_file_paths)
        self.logger.info(u'Heartbeating    the    process    manager')
        simple_dags = processor_manager.heartbeat()
        if self.using_sqlite:
            self.logger.debug(u"Waiting    for    processors    to    finish    since    we're    using    sqlite")
            processor_manager.wait_until_finished()
        if (len(simple_dags) > 0):
            simple_dag_bag = SimpleDagBag(simple_dags)
            self._change_state_for_tis_without_dagrun(simple_dag_bag, [State.UP_FOR_RETRY], State.FAILED)
            self._change_state_for_tis_without_dagrun(simple_dag_bag, [State.QUEUED, State.SCHEDULED], State.NONE)
            self._execute_task_instances(simple_dag_bag, (State.SCHEDULED,))
        self.logger.info(u'Heartbeating    the    executor')
        self.executor.heartbeat()
        self._process_executor_events()
        time_since_last_heartbeat = (datetime.now() - last_self_heartbeat_time).total_seconds()
        if (time_since_last_heartbeat > self.heartrate):
            self.logger.info(u'Heartbeating    the    scheduler')
            self.heartbeat()
            last_self_heartbeat_time = datetime.now()
        if ((datetime.now() - last_stat_print_time).total_seconds() > self.print_stats_interval):
            if (len(known_file_paths) > 0):
                self._log_file_processing_stats(known_file_paths, processor_manager)
            last_stat_print_time = datetime.now()
        loop_end_time = time.time()
        self.logger.debug(u'Ran    scheduling    loop    in    {:.2f}s'.format((loop_end_time - loop_start_time)))
        self.logger.debug(u'Sleeping    for    {:.2f}s'.format(self._processor_poll_interval))
        time.sleep(self._processor_poll_interval)
        if processor_manager.max_runs_reached():
            self.logger.info(u'Exiting    loop    as    all    files    have    been    processed    {}    times'.format(self.num_runs))
            break
    processor_manager.terminate()
    all_files_processed = True
    for file_path in known_file_paths:
        if (processor_manager.get_last_finish_time(file_path) is None):
            all_files_processed = False
            break
    if all_files_processed:
        self.logger.info(u"Deactivating    DAGs    that    haven't    been    touched    since    {}".format(execute_start_time.isoformat()))
        models.DAG.deactivate_stale_dags(execute_start_time)
    self.executor.end()
    settings.Session.remove()
