def execute_pipeline(self, pipeline, tmp_id_to_module_map, 
                         persistent_to_tmp_id_map, **kwargs):
        def fetch(name, default):
            return kwargs.pop(name, default)
        controller = fetch('controller', None)
        locator = fetch('locator', None)
        current_version = fetch('current_version', None)
        view = fetch('view', DummyView())
        vistrail_variables = fetch('vistrail_variables', None)
        aliases = fetch('aliases', None)
        params = fetch('params', None)
        extra_info = fetch('extra_info', None)
        logger = fetch('logger', DummyLogController)
        sinks = fetch('sinks', None)
        reason = fetch('reason', None)
        actions = fetch('actions', None)
        module_executed_hook = fetch('module_executed_hook', [])
        done_summon_hooks = fetch('done_summon_hooks', [])
        clean_pipeline = fetch('clean_pipeline', False)
        stop_on_error = fetch('stop_on_error', True)
        parent_exec = fetch('parent_exec', None)
        job_monitor = fetch('job_monitor', None)

        if len(kwargs) > 0:
            raise VistrailsInternalError('Wrong parameters passed '
                                         'to execute_pipeline: %s' % kwargs)

        # LOGGING SETUP
        def get_remapped_id(id):
            return persistent_to_tmp_id_map[id]

        logging_obj = ViewUpdatingLogController(
                logger=logger,
                view=view,
                remap_id=get_remapped_id,
                ids=pipeline.modules.keys(),
                module_executed_hook=module_executed_hook)

        # PARAMETER CHANGES SETUP
        parameter_changes = []
        def change_parameter(obj, name, value):
            parameter_changes.append((get_remapped_id(obj.id),
                                      name, value))
        def make_change_parameter(obj):
            return lambda *args: change_parameter(obj, *args)

        # Update **all** modules in the current pipeline
        for i, obj in tmp_id_to_module_map.iteritems():
            obj.in_pipeline = True # set flag to indicate in pipeline
            obj.logging = logging_obj
            obj.change_parameter = make_change_parameter(obj)
            
            # Update object pipeline information
            obj.moduleInfo['locator'] = locator
            obj.moduleInfo['version'] = current_version
            obj.moduleInfo['moduleId'] = i
            obj.moduleInfo['pipeline'] = pipeline
            obj.moduleInfo['controller'] = controller
            # extract job monitor from controller if this is the top level
            if controller:
                obj.moduleInfo['job_monitor'] = controller.jobMonitor
            else:
                obj.moduleInfo['job_monitor'] = job_monitor

            if extra_info is not None:
                obj.moduleInfo['extra_info'] = extra_info
            if reason is not None:
                obj.moduleInfo['reason'] = reason
            if actions is not None:
                obj.moduleInfo['actions'] = actions

        ## Checking 'sinks' from kwargs to resolve only requested sinks
        # Note that we accept any module in 'sinks', even if it's not actually
        # a sink in the graph
        if sinks is not None:
            persistent_sinks = [tmp_id_to_module_map[sink]
                                for sink in sinks
                                if sink in tmp_id_to_module_map]
        else:
            persistent_sinks = [tmp_id_to_module_map[sink]
                                for sink in pipeline.graph.sinks()]

        self._streams.append(Generator.generators)
        Generator.generators = []

        # Update new sinks
        for obj in persistent_sinks:
            abort = False
            try:
                obj.update()
                continue
            except ModuleWasSuspended:
                continue
            except ModuleHadError:
                pass
            except AbortExecution:
                break
            except ModuleSuspended, ms:
                ms.module.logging.end_update(ms.module, ms,
                                             was_suspended=True)
                continue
            except ModuleErrors, mes:
                for me in mes.module_errors:
                    me.module.logging.end_update(me.module, me)
                    logging_obj.signalError(me.module, me)
                    abort = abort or me.abort
            except ModuleError, me:
                me.module.logging.end_update(me.module, me, me.errorTrace)
                logging_obj.signalError(me.module, me)
                abort = me.abort
            except ModuleBreakpoint, mb:
                mb.module.logging.end_update(mb.module)
                logging_obj.signalError(mb.module, mb)
                abort = True
            if stop_on_error or abort:
                break

        # execute all generators until inputs are exhausted
        # this makes sure branching and multiple sinks are executed correctly
        if not logging_obj.errors and not logging_obj.suspended and \
                                                          Generator.generators:
            result = True
            abort = False
            while result is not None:
                try:
                    for m in Generator.generators:
                        result = m.generator.next()
                    continue
                except AbortExecution:
                    break
                except ModuleErrors, mes:
                    for me in mes.module_errors:
                        me.module.logging.end_update(me.module, me)
                        logging_obj.signalError(me.module, me)
                        abort = abort or me.abort
                except ModuleError, me:
                    me.module.logging.end_update(me.module, me, me.errorTrace)
                    logging_obj.signalError(me.module, me)
                    abort = me.abort
                except ModuleBreakpoint, mb:
                    mb.module.logging.end_update(mb.module)
                    logging_obj.signalError(mb.module, mb)
                    abort = True
                except Exception, e:
                    debug.unexpected_exception(e)
                    debug.critical("Exception running generators: %s" % e,
                                   debug.format_exc())
                    abort = True
                if stop_on_error or abort:
                    break

        Generator.generators = self._streams.pop()

        if self.done_update_hook:
            self.done_update_hook(self._persistent_pipeline, self._objects)
                
        # objs, errs, and execs are mappings that use the local ids as keys,
        # as opposed to the persistent ids.
        # They are thus ideal to external consumption.
        objs = {}
        # dict([(i, self._objects[tmp_to_persistent_module_map[i]])
        #              for i in tmp_to_persistent_module_map.keys()])
        errs = {}
        execs = {}
        suspends = {}
        caches = {}

        to_delete = []
        for (tmp_id, obj) in tmp_id_to_module_map.iteritems():
            if clean_pipeline:
                to_delete.append(obj.id)
            objs[tmp_id] = obj
            if obj.id in logging_obj.errors:
                errs[tmp_id] = logging_obj.errors[obj.id]
                if not clean_pipeline:
                    to_delete.append(obj.id)
            executed = False
            if obj.id in logging_obj.executed:
                execs[tmp_id] = logging_obj.executed[obj.id]
                executed = True
            if obj.id in logging_obj.suspended:
                suspends[tmp_id] = logging_obj.suspended[obj.id]
                if not clean_pipeline:
                    to_delete.append(obj.id)
                executed = True
            if obj.id in logging_obj.cached:
                caches[tmp_id] = logging_obj.cached[obj.id]
                executed = True
            if not executed:
                # these modules didn't execute
                execs[tmp_id] = False

        return (to_delete, objs, errs, execs, suspends, caches, parameter_changes)
dataset/ETHPy150Open VisTrails/VisTrails/vistrails/core/interpreter/cached.py/CachedInterpreter.execute_pipeline/original