docs: async def communicate(self, data_id=None, run_sync=False, save_settings=True):
          """Scan database for resolving Data objects and process them.
  
          This is submitted as a task to the manager's channel workers.
  
          :param data_id: Optional id of Data object which (+ its
              children) should be processes. If it is not given, all
              resolving objects are processed.
          :param run_sync: If ``True``, wait until all processes spawned
              from this point on have finished processing. If no processes
              are spawned, this results in a deadlock, since counts are
              handled on process finish.
          :param save_settings: If ``True``, save the current Django
              settings context to the global state. This should never be
              ``True`` for "automatic" calls, such as from Django signals,
              which can be invoked from inappropriate contexts (such as in
              the listener). For user code, it should be left at the
              default value. The saved settings are in effect until the
              next such call.
          """
          executor = getattr(settings, 'FLOW_EXECUTOR', {}).get('NAME', 'resolwe.flow.executors.local')
          logger.debug(__(
              "Manager sending communicate command on '{}' triggered by Data with id {}.",
              state.MANAGER_CONTROL_CHANNEL,
              data_id,
          ))
  
          saved_settings = self.state.settings_override
          if save_settings:
              saved_settings = self._marshal_settings()
              self.state.settings_override = saved_settings
  
          if run_sync:
              self._ensure_counter()
          await self.sync_counter.inc('communicate')
          try:
              await consumer.send_event({
                  WorkerProtocol.COMMAND: WorkerProtocol.COMMUNICATE,
                  WorkerProtocol.COMMUNICATE_SETTINGS: saved_settings,
                  WorkerProtocol.COMMUNICATE_EXTRA: {
                      'data_id': data_id,
                      'executor': executor,
                  },
              })
          except ChannelFull:
              logger.exception("ChannelFull error occurred while sending communicate message.")
              await self.sync_counter.dec('communicate')
  
          if run_sync and not self.sync_counter.active:
              logger.debug(__(
                  "Manager on channel '{}' entering synchronization block.",
                  state.MANAGER_CONTROL_CHANNEL
              ))
              await self.execution_barrier()
              logger.debug(__(
                  "Manager on channel '{}' exiting synchronization block.",
                  state.MANAGER_CONTROL_CHANNEL
              ))