class_docstr: Execute workflow with a distribution engine
  
  Combinations of ``proc_done`` and ``proc_pending``:
  
  +------------+---------------+--------------------------------+
  | proc_done  | proc_pending  | outcome                        |
  +============+===============+================================+
  | True       | False         | Process is finished            |
  +------------+---------------+--------------------------------+
  | True       | True          | Process is currently being run |
  +------------+---------------+--------------------------------+
  | False      | False         | Process is queued              |
  +------------+---------------+--------------------------------+
  | False      | True          | INVALID COMBINATION            |
  +------------+---------------+--------------------------------+
  
  Attributes
  ----------
  procs : :obj:`list`
      list (N) of underlying interface elements to be processed
  proc_done : :obj:`numpy.ndarray`
      a boolean numpy array (N,) signifying whether a process has been
      submitted for execution
  proc_pending : :obj:`numpy.ndarray`
      a boolean numpy array (N,) signifying whether a
      process is currently running.
  depidx : :obj:`numpy.matrix`
      a boolean matrix (NxN) storing the dependency structure across
      processes. Process dependencies are derived from each column.
class_docstr_tok: ["Execute","workflow","with","a","distribution","engine","Combinations","of","``","proc_done","``","and","``","proc_pending","``",":","+","--","--","--","--","--","--","+","--","--","--","--","--","--","--","-+","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","+","|","proc_done","|","proc_pending","|","outcome","|","+============+===============+================================+","|","True","|","False","|","Process","is","finished","|","+","--","--","--","--","--","--","+","--","--","--","--","--","--","--","-+","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","+","|","True","|","True","|","Process","is","currently","being","run","|","+","--","--","--","--","--","--","+","--","--","--","--","--","--","--","-+","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","+","|","False","|","False","|","Process","is","queued","|","+","--","--","--","--","--","--","+","--","--","--","--","--","--","--","-+","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","+","|","False","|","True","|","INVALID","COMBINATION","|","+","--","--","--","--","--","--","+","--","--","--","--","--","--","--","-+","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","--","+","Attributes","--","--","--","--","--","procs",":",":","obj",":","`","list","`","list","(","N",")","of","underlying","interface","elements","to","be","processed","proc_done",":",":","obj",":","`","numpy.ndarray","`","a","boolean","numpy","array","(","N",",",")","signifying","whether","a","process","has","been","submitted","for","execution","proc_pending",":",":","obj",":","`","numpy.ndarray","`","a","boolean","numpy","array","(","N",",",")","signifying","whether","a","process","is","currently","running",".","depidx",":",":","obj",":","`","numpy.matrix","`","a","boolean","matrix","(","NxN",")","storing","the","dependency","structure","across","processes",".","Process","dependencies","are","derived","from","each","column","."]
docstr: Executes a pre-defined pipeline using distributed approaches
docstr_tok: ["Executes","a","pre-defined","pipeline","using","distributed","approaches"]
code: def run(self, graph, config, updatehash=False):
      """
      Executes a pre-defined pipeline using distributed approaches
      """
      logger.info("Running in parallel.")
      self._config = config
      poll_sleep_secs = float(
          config["execution"]["poll_sleep_duration"]
      )
  
      self._prerun_check(graph)
      # Generate appropriate structures for worker-manager model
      self._generate_dependency_list(graph)
      self.mapnodes = []
      self.mapnodesubids = {}
      # setup polling - TODO: change to threaded model
      notrun = []
      errors = []
  
      old_progress_stats = None
      old_presub_stats = None
      while not np.all(self.proc_done) or np.any(self.proc_pending):
          loop_start = time()
          # Check if a job is available (jobs with all dependencies run)
          # https://github.com/nipy/nipype/pull/2200#discussion_r141605722
          jobs_ready = np.nonzero(
              ~self.proc_done & (self.depidx.sum(0) == 0)
          )[1]
  
          progress_stats = (
              len(self.proc_done),
              np.sum(self.proc_done ^ self.proc_pending),
              np.sum(self.proc_done & self.proc_pending),
              len(jobs_ready),
              len(self.pending_tasks),
              np.sum(~self.proc_done & ~self.proc_pending),
          )
          display_stats = progress_stats != old_progress_stats
          if display_stats:
              logger.debug(
                  "Progress: %d jobs, %d/%d/%d "
                  "(done/running/ready), %d/%d "
                  "(pending_tasks/waiting).",
                  *progress_stats,
              )
              old_progress_stats = progress_stats
          toappend = []
          # trigger callbacks for any pending results
          while self.pending_tasks:
              taskid, jobid = self.pending_tasks.pop()
              try:
                  result = self._get_result(taskid)
              except Exception as exc:
                  notrun.append(self._clean_queue(jobid, graph))
                  errors.append(exc)
              else:
                  if result:
                      if result["traceback"]:
                          notrun.append(
                              self._clean_queue(
                                  jobid, graph, result=result
                              )
                          )
                          errors.append("".join(result["traceback"]))
                      else:
                          self._task_finished_cb(jobid)
                          self._remove_node_dirs()
                      self._clear_task(taskid)
                  else:
                      assert (
                          self.proc_done[jobid]
                          and self.proc_pending[jobid]
                      )
                      toappend.insert(0, (taskid, jobid))
  
          if toappend:
              self.pending_tasks.extend(toappend)
  
          num_jobs = len(self.pending_tasks)
          presub_stats = (
              num_jobs,
              np.sum(self.proc_done & self.proc_pending),
          )
          display_stats = (
              display_stats or presub_stats != old_presub_stats
          )
          if display_stats:
              logger.debug(
                  "Tasks currently running: %d. Pending: %d.",
                  *presub_stats,
              )
              old_presub_stats = presub_stats
          if num_jobs < self.max_jobs:
              self._send_procs_to_workers(
                  updatehash=updatehash, graph=graph
              )
          elif display_stats:
              logger.debug("Not submitting (max jobs reached)")
  
          sleep_til = loop_start + poll_sleep_secs
          sleep(max(0, sleep_til - time()))
  
      self._remove_node_dirs()
      report_nodes_not_run(notrun)
  
      # close any open resources
      self._postrun_check()
  
      if errors:
          # If one or more nodes failed, re-rise first of them
          error, cause = errors[0], None
          if isinstance(error, str):
              error = RuntimeError(error)
  
          if len(errors) > 1:
              error, cause = (
                  RuntimeError(
                      f"{len(errors)} raised. Re-raising first."
                  ),
                  error,
              )
  
          raise error from cause
  
imports: import sys
  from copy import deepcopy
  from glob import glob
  import os
  import shutil
  from time import sleep, time
  from traceback import format_exception
  import numpy
  from ...None import logging
  from ...utils.misc import str2bool
  from ..engine.utils import topological_sort, load_resultfile
  from ..engine import MapNode
  from .tools import report_crash, report_nodes_not_run, create_pyscript
label_desc: Use image node_id 4 for calling the DistributedPluginBase obj's underlying member method code with example usage: obj.run(graph, config, updatehash) without return types