class_docstr: A set of magics useful when controlling a parallel IPython cluster.
docstr: implementation used by %px and %%parallel
docstr_tok: ["implementation","used","by","%","px","and","%","%","parallel"]
code: def parallel_execute(
      self,
      cell,
      block=None,
      groupby="type",
      save_name=None,
      stream_output=None,
      progress_after=None,
      signal_on_interrupt=None,
  ):
      """implementation used by %px and %%parallel"""
  
      # defaults:
      block = self.view.block if block is None else block
      stream_output = (
          self.stream_output if stream_output is None else stream_output
      )
      signal_on_interrupt = (
          self.signal_on_interrupt
          if signal_on_interrupt is None
          else signal_on_interrupt
      )
  
      base = "Parallel" if block else "Async parallel"
  
      targets = self.view.targets
      if isinstance(targets, list) and len(targets) > 10:
          str_targets = (
              str(targets[:4])[:-1] + ", ..., " + str(targets[-4:])[1:]
          )
      else:
          str_targets = str(targets)
      if self.verbose:
          print(base + " execution on engine(s): %s" % str_targets)
  
      result = self.view.execute(cell, silent=False, block=False)
      result._fname = "%px"
      self.last_result = result
  
      if save_name:
          self.shell.user_ns[save_name] = result
  
      if block:
          try:
              if progress_after is None:
                  progress_after = self.progress_after_seconds
  
              cm = (
                  result.stream_output()
                  if stream_output
                  else nullcontext()
              )
              with cm:
                  finished_waiting = False
                  if progress_after > 0:
                      # finite progress-after timeout
                      # wait for 'quick' results before showing progress
                      tic = time.perf_counter()
                      deadline = tic + progress_after
                      result.wait(timeout=progress_after)
                      remaining = max(deadline - time.perf_counter(), 0)
                      result.wait_for_output(timeout=remaining)
                      finished_waiting = result.done()
  
                  if not finished_waiting:
                      if progress_after >= 0:
                          # not an immediate result, start interactive progress
                          result.wait_interactive()
                          result.wait_for_output(1)
  
                  try:
                      result.get()
                  except error.CompositeError as e:
                      if stream_output and result._output_ready:
                          # already streamed, show an abbreviated result
                          raise error.AlreadyDisplayedError(e) from None
                      else:
                          raise
              # Skip redisplay if streaming output
          except KeyboardInterrupt:
              if signal_on_interrupt is not None:
                  print(
                      f"Received Keyboard Interrupt. Sending signal {signal_on_interrupt} to engines...",
                      file=sys.stderr,
                  )
                  self.view.client.send_signal(
                      signal_on_interrupt, targets=targets, block=True
                  )
              else:
                  raise
          finally:
              # always redisplay outputs if not streaming,
              # on both success and error
  
              if not stream_output:
                  # wait for at most 1 second for output to be complete
                  result.wait_for_output(1)
                  result.display_outputs(groupby)
      else:
          # return AsyncResult only on non-blocking submission
          return result
  
imports: import time
  from contextlib import contextmanager
  import inspect
  import re
  import sys
  from textwrap import dedent
  from IPython.core import magic_arguments
  from IPython.core.error import UsageError
  from IPython.core.magic import Magics, no_var_expand
  import ipyparallel
  from ..None import error
label_desc: Use image node_id 7 for calling the ParallelMagics obj's underlying member method code with example usage: obj.parallel_execute(cell, block, groupby, save_name, stream_output, progress_after, signal_on_interrupt) and returns: result