docs: def notify_barriers(self,
                        slot_key,
                        cursor,
                        use_barrier_indexes,
                        max_to_notify=_MAX_BARRIERS_TO_NOTIFY):
      """Searches for barriers affected by a slot and triggers completed ones.
  
      Args:
        slot_key: db.Key or stringified key of the _SlotRecord that was filled.
        cursor: Stringified Datastore cursor where the notification query
          should pick up.
        use_barrier_indexes: When True, use _BarrierIndex records to determine
          which _Barriers to trigger by having this _SlotRecord filled. When
          False, use the old method that queries for _BarrierRecords by
          the blocking_slots parameter.
        max_to_notify: Used for testing.
  
      Raises:
        PipelineStatusError: If any of the barriers are in a bad state.
      """
      if not isinstance(slot_key, db.Key):
        slot_key = db.Key(slot_key)
      logging.debug('Notifying slot %r', slot_key)
  
      if use_barrier_indexes:
        # Please see models.py:_BarrierIndex to understand how _BarrierIndex
        # entities relate to _BarrierRecord entities.
        query = (
            _BarrierIndex.all(cursor=cursor, keys_only=True)
            .ancestor(slot_key))
        barrier_index_list = query.fetch(max_to_notify)
        barrier_key_list = [
            _BarrierIndex.to_barrier_key(key) for key in barrier_index_list]
  
        # If there are task and pipeline kickoff retries it's possible for a
        # _BarrierIndex to exist for a _BarrierRecord that was not successfully
        # written. It's safe to ignore this because the original task that wrote
        # the _BarrierIndex and _BarrierRecord would not have made progress to
        # kick off a real pipeline or child pipeline unless all of the writes for
        # these dependent entities went through. We assume that the instigator
        # retried from scratch and somehwere there exists a good _BarrierIndex and
        # corresponding _BarrierRecord that tries to accomplish the same thing.
        barriers = db.get(barrier_key_list)
        results = []
        for barrier_key, barrier in zip(barrier_key_list, barriers):
          if barrier is None:
            logging.debug('Ignoring that Barrier "%r" is missing, '
                          'relies on Slot "%r"', barrier_key, slot_key)
          else:
            results.append(barrier)
      else:
        # TODO(user): Delete this backwards compatible codepath and
        # make use_barrier_indexes the assumed default in all cases.
        query = (
            _BarrierRecord.all(cursor=cursor)
            .filter('blocking_slots =', slot_key))
        results = query.fetch(max_to_notify)
  
      # Fetch all blocking _SlotRecords for any potentially triggered barriers.
      blocking_slot_keys = []
      for barrier in results:
        blocking_slot_keys.extend(barrier.blocking_slots)
  
      blocking_slot_dict = {}
      for slot_record in db.get(blocking_slot_keys):
        if slot_record is None:
          continue
        blocking_slot_dict[slot_record.key()] = slot_record
  
      task_list = []
      updated_barriers = []
      for barrier in results:
        ready_slots = []
        for blocking_slot_key in barrier.blocking_slots:
          slot_record = blocking_slot_dict.get(blocking_slot_key)
          if slot_record is None:
            raise UnexpectedPipelineError(
                'Barrier "%r" relies on Slot "%r" which is missing.' %
                (barrier.key(), blocking_slot_key))
          if slot_record.status == _SlotRecord.FILLED:
            ready_slots.append(blocking_slot_key)
  
        # When all of the blocking_slots have been filled, consider the barrier
        # ready to trigger. We'll trigger it regardless of the current
        # _BarrierRecord status, since there could be task queue failures at any
        # point in this flow; this rolls forward the state and de-dupes using
        # the task name tombstones.
        pending_slots = set(barrier.blocking_slots) - set(ready_slots)
        if not pending_slots:
          if barrier.status != _BarrierRecord.FIRED:
            barrier.status = _BarrierRecord.FIRED
            barrier.trigger_time = self._gettime()
            updated_barriers.append(barrier)
  
          purpose = barrier.key().name()
          if purpose == _BarrierRecord.START:
            path = self.pipeline_handler_path
            countdown = None
          else:
            path = self.finalized_handler_path
            # NOTE: Wait one second before finalization to prevent
            # contention on the _PipelineRecord entity.
            countdown = 1
          pipeline_key = _BarrierRecord.target.get_value_for_datastore(barrier)
          pipeline_record = db.get(pipeline_key)
          logging.debug('Firing barrier %r', barrier.key())
          task_list.append(taskqueue.Task(
              url=path,
              countdown=countdown,
              name='ae-barrier-fire-%s-%s' % (pipeline_key.name(), purpose),
              params=dict(pipeline_key=pipeline_key, purpose=purpose),
              headers={'X-Ae-Pipeline-Key': pipeline_key},
              target=pipeline_record.params['target']))
        else:
          logging.debug('Not firing barrier %r, Waiting for slots: %r',
                        barrier.key(), pending_slots)
  
      # Blindly overwrite _BarrierRecords that have an updated status. This is
      # acceptable because by this point all finalization barriers for
      # generator children should have already had their final outputs assigned.
      if updated_barriers:
        db.put(updated_barriers)
  
      # Task continuation with sequence number to prevent fork-bombs.
      if len(results) == max_to_notify:
        the_match = re.match('(.*)-ae-barrier-notify-([0-9]+)', self.task_name)
        if the_match:
          prefix = the_match.group(1)
          end = int(the_match.group(2)) + 1
        else:
          prefix = self.task_name
          end = 0
        task_list.append(taskqueue.Task(
            name='%s-ae-barrier-notify-%d' % (prefix, end),
            url=self.barrier_handler_path,
            params=dict(
                slot_key=slot_key,
                cursor=query.cursor(),
                use_barrier_indexes=use_barrier_indexes)))
  
      if task_list:
        try:
          taskqueue.Queue(self.queue_name).add(task_list)
        except (taskqueue.TombstonedTaskError, taskqueue.TaskAlreadyExistsError):
          pass