<commit_before>############################################################################
##
## Copyright (C) 2006-2007 University of Utah. All rights reserved.
##
## This file is part of VisTrails.
##
## This file may be used under the terms of the GNU General Public
## License version 2.0 as published by the Free Software Foundation
## and appearing in the file LICENSE.GPL included in the packaging of
## this file.  Please review the following to ensure GNU General Public
## Licensing requirements will be met:
## http://www.opensource.org/licenses/gpl-license.php
##
## If you are unsure which license is appropriate for your use (for
## instance, you are interested in developing a commercial derivative
## of VisTrails), please contact us at <EMAIL>.
##
## This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
## WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
##
############################################################################

from core import modules
from core.data_structures import Bidict
from core.modules.module_utils import FilePool
from core.modules.vistrails_module import ModuleConnector, ModuleError
from core.utils import withIndex, InstanceObject, lock_method
import copy
import core.interpreter.base
import core.interpreter.utils
import core.vistrail.pipeline
import gc

##############################################################################

class CachedInterpreter(core.interpreter.base.BaseInterpreter):

    def __init__(self):
        core.interpreter.base.BaseInterpreter.__init__(self)
        self.create()

    def create(self):
        self._file_pool = FilePool()
        self._persistent_pipeline = core.vistrail.pipeline.Pipeline()
        self._objects = {}
        self._executed = {}
        self.filePool = self._file_pool
        
    def clear(self):
        self._file_pool.cleanup()
        self._persistent_pipeline.clear()
        for obj in self._objects.itervalues():
            obj.clear()
        self._objects = {}
        self._executed = {}

    def __del__(self):
        self.clear()

    def clean_non_cacheable_modules(self):
        """clean_non_cacheable_modules() -> None

Removes all modules that are not cacheable from the persistent pipeline,
and the modules that depend on them."""
        non_cacheable_modules = [i for
                                 (i, mod) in self._objects.iteritems()
                                 if not mod.is_cacheable()]
        if non_cacheable_modules == []:
            return
        g = self._persistent_pipeline.graph
        dependencies = g.vertices_topological_sort(non_cacheable_modules)
        for v in dependencies:
            self._persistent_pipeline.deleteModule(v)
            del self._objects[v]

    def unlocked_execute(self, pipeline, vistrailName, currentVersion,
                view, logger):
        """unlocked_execute(pipeline, vistrailName, currentVersion, view, logger):
Executes a pipeline using caching. Caching works by reusing pipelines directly.
This means that there exists one global pipeline whose parts get executed over
and over again. This allows nested execution."""

        (module_map,
         conn_map,
         module_added_set,
         conn_added_set) = self.add_to_persistent_pipeline(pipeline)

        self.resolveAliases(self._persistent_pipeline)

        # the executed dict works on persistent ids
        def addToExecuted(obj):
            executed[obj.id] = True
        # views work on local ids
        def beginCompute(obj):
            i = module_map.inverse[obj.id]
            if view:
                view.setModuleComputing(i)
        # views and loggers work on local ids
        def beginUpdate(obj):
            i = module_map.inverse[obj.id]
            if view:
                view.setModuleActive(i)
            reg = modules.module_registry.registry
            name = reg.getDescriptor(obj.__class__).name
            if logger:
                logger.startModuleExecution(vistrailName, 
                                            currentVersion, i, name)
        # views and loggers work on local ids
        def endUpdate(obj, error=''):
            i = module_map.inverse[obj.id]
            if view:
                if not error:
                    view.setModuleSuccess(i)
                else:
                    view.setModuleError(i, error)
            if logger:
                logger.finishModuleExecution(vistrailName, 
                                             currentVersion, i)
        # views and loggers work on local ids
        def annotate(obj, d):
            i = module_map.inverse[obj.id]
            if logger:
                logger.insertAnnotationDB(vistrailName, 
                                          currentVersion, i, d)

        def createNull():
            """Creates a Null value"""
            reg = modules.module_registry.registry
            return reg.getDescriptorByName('Null').module()
        
        def createConstant(param):
            """Creates a Constant from a parameter spec"""
            reg = modules.module_registry.registry
            constant = reg.getDescriptorByName(param.type).module()
            constant.setValue(p.evaluatedStrValue)
            return constant
                
        persistent_sinks = [module_map[sink]
                            for sink
                            in pipeline.graph.sinks()]
        logging_obj = InstanceObject(signalSuccess=addToExecuted,
                                     beginUpdate=beginUpdate,
                                     beginCompute=beginCompute,
                                     endUpdate=endUpdate,
                                     annotate=annotate)
        errors = {}
        executed = {}
        # Create the new objects
        for i in module_added_set:
            persistent_id = module_map[i]
            module = self._persistent_pipeline.modules[persistent_id]
            self._objects[persistent_id] = module.summon()
            obj = self._objects[persistent_id]
            obj.interpreter = self
            obj.id = persistent_id
            if view:
                obj.logging = logging_obj
            obj.vistrailName = vistrailName
            obj.currentVersion = currentVersion
            reg = modules.module_registry.registry
            for f in module.functions:
                if len(f.params) == 0:
                    connector = ModuleConnector(createNull(), 'value')
                elif len(f.params) == 1:
                    p = f.params[0]
                    connector = ModuleConnector(createConstant(p), 'value')
                else:
                    tupleModule = reg.getDescriptorByName('Tuple').module()
                    tupleModule.length = len(f.params)
                    for (i,p) in withIndex(f.params):
                        constant = createConstant(p)
                        connector = ModuleConnector(constant, 'value')
                        tupleModule.setInputPort(i, connector)
                    connector = ModuleConnector(tupleModule, 'value')
                obj.setInputPort(f.name, connector)

        # Create the new connections
        for i in conn_added_set:<commit_msg>CachedInterpreter.unlocked_execute(): uses InternalTuple for constant values.<commit_after>############################################################################
##
## Copyright (C) 2006-2007 University of Utah. All rights reserved.
##
## This file is part of VisTrails.
##
## This file may be used under the terms of the GNU General Public
## License version 2.0 as published by the Free Software Foundation
## and appearing in the file LICENSE.GPL included in the packaging of
## this file.  Please review the following to ensure GNU General Public
## Licensing requirements will be met:
## http://www.opensource.org/licenses/gpl-license.php
##
## If you are unsure which license is appropriate for your use (for
## instance, you are interested in developing a commercial derivative
## of VisTrails), please contact us at <EMAIL>.
##
## This file is provided AS IS with NO WARRANTY OF ANY KIND, INCLUDING THE
## WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE.
##
############################################################################

from core import modules
from core.data_structures import Bidict
from core.modules.module_utils import FilePool
from core.modules.vistrails_module import ModuleConnector, ModuleError
from core.utils import withIndex, InstanceObject, lock_method
import copy
import core.interpreter.base
import core.interpreter.utils
import core.vistrail.pipeline
import gc

##############################################################################

class CachedInterpreter(core.interpreter.base.BaseInterpreter):

    def __init__(self):
        core.interpreter.base.BaseInterpreter.__init__(self)
        self.create()

    def create(self):
        self._file_pool = FilePool()
        self._persistent_pipeline = core.vistrail.pipeline.Pipeline()
        self._objects = {}
        self._executed = {}
        self.filePool = self._file_pool
        
    def clear(self):
        self._file_pool.cleanup()
        self._persistent_pipeline.clear()
        for obj in self._objects.itervalues():
            obj.clear()
        self._objects = {}
        self._executed = {}

    def __del__(self):
        self.clear()

    def clean_non_cacheable_modules(self):
        """clean_non_cacheable_modules() -> None

Removes all modules that are not cacheable from the persistent pipeline,
and the modules that depend on them."""
        non_cacheable_modules = [i for
                                 (i, mod) in self._objects.iteritems()
                                 if not mod.is_cacheable()]
        if non_cacheable_modules == []:
            return
        g = self._persistent_pipeline.graph
        dependencies = g.vertices_topological_sort(non_cacheable_modules)
        for v in dependencies:
            self._persistent_pipeline.deleteModule(v)
            del self._objects[v]

    def unlocked_execute(self, pipeline, vistrailName, currentVersion,
                view, logger):
        """unlocked_execute(pipeline, vistrailName, currentVersion, view, logger):
Executes a pipeline using caching. Caching works by reusing pipelines directly.
This means that there exists one global pipeline whose parts get executed over
and over again. This allows nested execution."""

        (module_map,
         conn_map,
         module_added_set,
         conn_added_set) = self.add_to_persistent_pipeline(pipeline)

        self.resolveAliases(self._persistent_pipeline)

        # the executed dict works on persistent ids
        def addToExecuted(obj):
            executed[obj.id] = True
        # views work on local ids
        def beginCompute(obj):
            i = module_map.inverse[obj.id]
            if view:
                view.setModuleComputing(i)
        # views and loggers work on local ids
        def beginUpdate(obj):
            i = module_map.inverse[obj.id]
            if view:
                view.setModuleActive(i)
            reg = modules.module_registry.registry
            name = reg.getDescriptor(obj.__class__).name
            if logger:
                logger.startModuleExecution(vistrailName, 
                                            currentVersion, i, name)
        # views and loggers work on local ids
        def endUpdate(obj, error=''):
            i = module_map.inverse[obj.id]
            if view:
                if not error:
                    view.setModuleSuccess(i)
                else:
                    view.setModuleError(i, error)
            if logger:
                logger.finishModuleExecution(vistrailName, 
                                             currentVersion, i)
        # views and loggers work on local ids
        def annotate(obj, d):
            i = module_map.inverse[obj.id]
            if logger:
                logger.insertAnnotationDB(vistrailName, 
                                          currentVersion, i, d)

        def createNull():
            """Creates a Null value"""
            reg = modules.module_registry.registry
            return reg.getDescriptorByName('Null').module()
        
        def createConstant(param):
            """Creates a Constant from a parameter spec"""
            reg = modules.module_registry.registry
            constant = reg.getDescriptorByName(param.type).module()
            constant.setValue(p.evaluatedStrValue)
            return constant
                
        persistent_sinks = [module_map[sink]
                            for sink
                            in pipeline.graph.sinks()]
        logging_obj = InstanceObject(signalSuccess=addToExecuted,
                                     beginUpdate=beginUpdate,
                                     beginCompute=beginCompute,
                                     endUpdate=endUpdate,
                                     annotate=annotate)
        errors = {}
        executed = {}
        # Create the new objects
        for i in module_added_set:
            persistent_id = module_map[i]
            module = self._persistent_pipeline.modules[persistent_id]
            self._objects[persistent_id] = module.summon()
            obj = self._objects[persistent_id]
            obj.interpreter = self
            obj.id = persistent_id
            if view:
                obj.logging = logging_obj
            obj.vistrailName = vistrailName
            obj.currentVersion = currentVersion
            reg = modules.module_registry.registry
            for f in module.functions:
                if len(f.params) == 0:
                    connector = ModuleConnector(createNull(), 'value')
                elif len(f.params) == 1:
                    p = f.params[0]
                    connector = ModuleConnector(createConstant(p), 'value')
                else:
                    tupleModule = core.interpreter.base.InternalTuple()
                    tupleModule.length = len(f.params)
                    for (i,p) in withIndex(f.params):
                        constant = createConstant(p)
                        constant.update()
                        connector = ModuleConnector(constant, 'value')
                        tupleModule.setInputPort(i, connector)
                    connector = ModuleConnector(tupleModule, 'value')
                obj.setInputPort(f.name, connector)

        # Create the new connections
        for i in conn_added_set:
{'fraction_non_alphanumeric': 0.10460337032470202, 'fraction_numerical': 0.0017810658994382791, 'mean_word_length': 1.981212987543394, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7374713', 'n_tokens_mistral': 3599, 'n_tokens_neox': 3490, 'n_words': 975}