'This should be periodically called by the scheduler. This method will
kick of new processes to process DAG definition files and read the
results from the finished processors.
:return: a list of SimpleDags that were produced by processors that
have finished since the last time this was called
:rtype: list[SimpleDag]'

def heartbeat(self):

    finished_processors = {}
    u':type    :    dict[unicode,    AbstractDagFileProcessor]'
    running_processors = {}
    u':type    :    dict[unicode,    AbstractDagFileProcessor]'
    for (file_path, processor) in self._processors.items():
        if processor.done:
            self.logger.info(u'Processor    for    {}    finished'.format(file_path))
            now = datetime.now()
            finished_processors[file_path] = processor
            self._last_runtime[file_path] = (now - processor.start_time).total_seconds()
            self._last_finish_time[file_path] = now
            self._run_count[file_path] += 1
        else:
            running_processors[file_path] = processor
    self._processors = running_processors
    simple_dags = []
    for (file_path, processor) in finished_processors.items():
        if (processor.result is None):
            self.logger.warning(u'Processor    for    {}    exited    with    return    code    {}.    See    {}    for    details.'.format(processor.file_path, processor.exit_code, processor.log_file))
        else:
            for simple_dag in processor.result:
                simple_dags.append(simple_dag)
    if (len(self._file_path_queue) == 0):
        file_paths_in_progress = self._processors.keys()
        now = datetime.now()
        file_paths_recently_processed = []
        for file_path in self._file_paths:
            last_finish_time = self.get_last_finish_time(file_path)
            if ((last_finish_time is not None) and ((now - last_finish_time).total_seconds() < self._process_file_interval)):
                file_paths_recently_processed.append(file_path)
        files_paths_at_run_limit = [file_path for (file_path, num_runs) in self._run_count.items() if (num_runs == self._max_runs)]
        files_paths_to_queue = list((((set(self._file_paths) - set(file_paths_in_progress)) - set(file_paths_recently_processed)) - set(files_paths_at_run_limit)))
        for (file_path, processor) in self._processors.items():
            self.logger.debug(u'File    path    {}    is    still    being    processed    (started:    {})'.format(processor.file_path, processor.start_time.isoformat()))
        self.logger.debug(u'Queuing    the    following    files    for    processing:\n DCTB {}'.format(u'\n DCTB '.join(files_paths_to_queue)))
        self._file_path_queue.extend(files_paths_to_queue)
    while (((self._parallelism - len(self._processors)) > 0) and (len(self._file_path_queue) > 0)):
        file_path = self._file_path_queue.pop(0)
        log_file_path = self._get_log_file_path(file_path)
        processor = self._processor_factory(file_path, log_file_path)
        processor.start()
        self.logger.info(u'Started    a    process    (PID:    {})    to    generate    tasks    for    {}    -    logging    into    {}'.format(processor.pid, file_path, log_file_path))
        self._processors[file_path] = processor
    self.symlink_latest_log_directory()
    self._run_count[self._heart_beat_key] += 1
    return simple_dags
