code: def abort(self, abort_message=''):
      
      # TODO: Use thread-local variable to enforce that this is not called
      # while a pipeline is executing in the current thread.
      if (self.async and self._root_pipeline_key == self._pipeline_key and
          not self.try_cancel()):
        # Handle the special case where the root pipeline is async and thus
        # cannot be aborted outright.
        return False
      else:
        return self._context.begin_abort(
            self._root_pipeline_key, abort_message=abort_message)
docs: Mark the entire pipeline up to the root as aborted.
  
      Note this should only be called from *outside* the context of a running
      pipeline. Synchronous and generator pipelines should raise the 'Abort'
      exception to cause this behavior during execution.
  
      Args:
        abort_message: Optional message explaining why the abort happened.
  
      Returns:
        True if the abort signal was sent successfully; False if the pipeline
        could not be aborted for any reason.