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
            ))
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.