docs: def run(self, connection: Connection, master: Workflow):
          """Run a workflow.
  
          :param connection:
              A connection giving a sink to the job-queue and a source yielding
              results.
          :type connection: Connection
  
          :param master:
              The workflow.
          :type master: Workflow
          """
          # initiate worker slave army and take up reins ...
          source, sink = connection.setup()
  
          # schedule work
          self.add_workflow(master, master, master.root, sink)
          graceful_exit = False
          errors = []
  
          # process results
          for job_key, status, result, err_msg in source:
              wf, n = self.jobs[job_key]
              if status == 'error':
                  graceful_exit = True
                  errors.append(err_msg)
  
                  try:
                      sink.send(FlushQueue)
                  except StopIteration:
                      pass
  
                  print("Uncaught error running job: {}, {}".format(n, err_msg),
                        file=sys.stderr)
                  print("Flushing queue and waiting for threads to close.",
                        file=sys.stderr, flush=True)
  
              if status == 'aborted':
                  print("Job {} got aborted: {}".format(n, err_msg),
                        file=sys.stderr)
                  print("Flushing queue and waiting for threads to close.",
                        file=sys.stderr, flush=True)
                  graceful_exit = True
                  errors.append(err_msg)
                  try:
                      sink.send(FlushQueue)
                  except StopIteration:
                      pass
  
              if self.verbose:
                  print("sched result [{0}]: ".format(self.key_map[job_key]),
                        result,
                        file=sys.stderr, flush=True)
  
              del self.jobs[job_key]
              if len(self.jobs) == 0 and graceful_exit:
                  for error in errors:
                      print("Exception of type", type(error), ":")
                      print(error)
                  raise errors[0]
  
              # if this result is the root of a workflow, pop to parent
              # we do this before scheduling a child workflow, as to
              # achieve tail-call elimination.
              while n == wf.root and wf is not master:
                  child = id(wf)
                  _, wf, n = self.dynamic_links[child]
                  del self.dynamic_links[child]
  
              # if we retrieve a workflow, push a child
              if is_workflow(result) and not graceful_exit:
                  child_wf = get_workflow(result)
                  self.add_workflow(child_wf, wf, n, sink)
                  continue
  
              # insert the result in the nodes that need it
              wf.nodes[n].result = result
              for (tgt, address) in wf.links[n]:
                  insert_result(wf.nodes[tgt], address, result)
                  if is_node_ready(wf.nodes[tgt]) and not graceful_exit:
                      self.schedule(Job(workflow=wf, node_id=tgt), sink)
  
              # see if we're done
              if wf == master and n == master.root:
                  try:
                      sink.send(EndOfQueue)
                  except StopIteration:
                      pass
  
                  return result