def _data_scan(self, data_id=None, executor='resolwe.flow.executors.local', **kwargs):
        """Scan for new Data objects and execute them.

        :param data_id: Optional id of Data object which (+ its
            children) should be scanned. If it is not given, all
            resolving objects are processed.
        :param executor: The fully qualified name of the executor to use
            for all :class:`~resolwe.flow.models.Data` objects
            discovered in this pass.
        """
        def process_data_object(data):
            """Process a single data object."""
            # Lock for update. Note that we want this transaction to be as short as possible in
            # order to reduce contention and avoid deadlocks. This is why we do not lock all
            # resolving objects for update, but instead only lock one object at a time. This
            # allows managers running in parallel to process different objects.
            data = Data.objects.select_for_update().get(pk=data.pk)
            if data.status != Data.STATUS_RESOLVING:
                # The object might have already been processed while waiting for the lock to be
                # obtained. In this case, skip the object.
                return

            dep_status = dependency_status(data)

            if dep_status == Data.STATUS_ERROR:
                data.status = Data.STATUS_ERROR
                data.process_error.append("One or more inputs have status ERROR")
                data.process_rc = 1
                data.save()
                return

            elif dep_status != Data.STATUS_DONE:
                return

            if data.process.run:
                try:
                    execution_engine = data.process.run.get('language', None)
                    # Evaluation by the execution engine may spawn additional data objects and
                    # perform other queries on the database. Queries of all possible execution
                    # engines need to be audited for possibilities of deadlocks in case any
                    # additional locks are introduced. Currently, we only take an explicit lock on
                    # the currently processing object.
                    program = self.get_execution_engine(execution_engine).evaluate(data)
                except (ExecutionError, InvalidEngineError) as error:
                    data.status = Data.STATUS_ERROR
                    data.process_error.append("Error in process script: {}".format(error))
                    data.save()
                    return

                # Set allocated resources:
                resource_limits = data.process.get_resource_limits()
                data.process_memory = resource_limits['memory']
                data.process_cores = resource_limits['cores']
            else:
                # If there is no run section, then we should not try to run anything. But the
                # program must not be set to None as then the process will be stuck in waiting
                # state.
                program = ''

            if data.status != Data.STATUS_DONE:
                # The data object may already be marked as done by the execution engine. In this
                # case we must not revert the status to STATUS_WAITING.
                data.status = Data.STATUS_WAITING
            data.save(render_name=True)

            # Actually run the object only if there was nothing with the transaction.
            transaction.on_commit(
                # Make sure the closure gets the right values here, since they're
                # changed in the loop.
                lambda d=data, p=program: self._data_execute(d, p, executor)
            )

        logger.debug(__("Manager processing communicate command triggered by Data with id {}.", data_id))

        if is_testing():
            # NOTE: This is a work-around for Django issue #10827
            # (https://code.djangoproject.com/ticket/10827), same as in
            # TestCaseHelpers._pre_setup(). Because the worker is running
            # independently, it must clear the cache on its own.
            ContentType.objects.clear_cache()

            # Ensure settings overrides apply
            self.discover_engines(executor=executor)

        try:
            queryset = Data.objects.filter(status=Data.STATUS_RESOLVING)
            if data_id is not None:
                # Scan only given data object and its children.
                queryset = queryset.filter(Q(parents=data_id) | Q(id=data_id)).distinct()

            for data in queryset:
                try:
                    with transaction.atomic():
                        process_data_object(data)

                        # All data objects created by the execution engine are commited after this
                        # point and may be processed by other managers running in parallel. At the
                        # same time, the lock for the current data object is released.
                except Exception as error:  # pylint: disable=broad-except
                    logger.exception(__(
                        "Unhandled exception in _data_scan while processing data object {}.",
                        data.pk
                    ))

                    # Unhandled error while processing a data object. We must set its
                    # status to STATUS_ERROR to prevent the object from being retried
                    # on next _data_scan run. We must perform this operation without
                    # using the Django ORM as using the ORM may be the reason the error
                    # occurred in the first place.
                    error_msg = "Internal error: {}".format(error)
                    process_error_field = Data._meta.get_field('process_error')  # pylint: disable=protected-access
                    max_length = process_error_field.base_field.max_length
                    if len(error_msg) > max_length:
                        error_msg = error_msg[:max_length - 3] + '...'

                    try:
                        with connection.cursor() as cursor:
                            cursor.execute(
                                """
                                    UPDATE {table}
                                    SET
                                        status = %(status)s,
                                        process_error = process_error || (%(error)s)::varchar[]
                                    WHERE id = %(id)s
                                """.format(
                                    table=Data._meta.db_table  # pylint: disable=protected-access
                                ),
                                {
                                    'status': Data.STATUS_ERROR,
                                    'error': [error_msg],
                                    'id': data.pk
                                }
                            )
                    except Exception as error:  # pylint: disable=broad-except
                        # If object's state cannot be changed due to some database-related
                        # issue, at least skip the object for this run.
                        logger.exception(__(
                            "Unhandled exception in _data_scan while trying to emit error for {}.",
                            data.pk
                        ))

        except IntegrityError as exp:
            logger.error(__("IntegrityError in manager {}", exp))
            return