<commit_before>/*
 * Copyright (c) Murdoch Childrens Research Institute and Contributers
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 * 
 * Redistributions in binary form must reproduce the above copyright notice, this
 * list of conditions and the following disclaimer in the documentation and/or
 * other materials provided with the distribution.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */  
package bpipe

import groovy.util.logging.Log;
import groovyx.gpars.GParsPool
import groovy.time.TimeCategory;
import groovy.transform.CompileStatic;

/**
 * A node in the dependency graph representing a set of outputs
 * sharing a common set of child nodes (who depend on the outputs) and
 * inputs (parents nodes) upon which the children depend.  
 * 
 * @author ssadedin
 */
class GraphEntry {
    
    List<Properties> values
    List<GraphEntry> parents = []
    List<GraphEntry> children = []
    
    @CompileStatic
    GraphEntry findBy(Closure c) {
        if(this.values != null && values.any { c(it) })
            return this
            
        for(GraphEntry child in children) {
           def result = child.findBy(c)
           if(result)
               return result;
        }
    }
    
    List<Properties> findAllOutputsBy(Closure c) {
        
        
        List<Properties> results = []
        
        if(this.values != null)
            results.addAll(this.values.grep { c(it)})
            
        for(GraphEntry child in children) {
           results.addAll(child.findAllOutputsBy(c))
        }
        
        return results
    }
  
    
    /**
     * Search the graph for entry with given outputfile
     */
    @CompileStatic
    GraphEntry entryFor(File outputFile) {
        // In case of non-default output directory, the outputFile itself may be in a directory
        final String outputFilePath = outputFile.canonicalPath
        return entryForCanonicalPath(outputFilePath)
    }
    
    @CompileStatic
    GraphEntry entryForCanonicalPath(String canonicalPath) {
        // In case of non-default output directory, the outputFile itself may be in a directory
        findBy { Properties p -> canonicalPathFor(p) == canonicalPath  }
    }
    
     /**
     * getCanonicalPath can be very slow on some systems, so 
     * this method caches them in the property file itself.
     * 
     * @param p
     * @return
     */
    @CompileStatic
    String canonicalPathFor(Properties p) {
        synchronized(p) {
            if(p.containsKey("canonicalPath"))
                return p["canonicalPath"]        
                
            p["canonicalPath"] = ((File)p["outputFile"]).canonicalPath
        }
    }
    
    /**
     * Search the graph for properties for the given output file
     */
    Properties propertiesFor(String outputFile) { 
       // In case of non-default output directory, the outputFile itself may be in a directory
       String outputFilePath = new File(outputFile).canonicalPath
       def values = entryForCanonicalPath(outputFilePath)?.values
       if(!values)
           return null
           
       for(def o in values) {
           if(canonicalPathFor(o) == outputFilePath) {
               return o
           }
       }
       return null
    }
    
    /**
     * Create a copy of the subtree for this GraphEntry and its children (but NOT parents!)
     */
    GraphEntry clone() {
        new GraphEntry(values: values.clone(), parents: parents.clone(), children: children.collect { it.clone() })
    }
    
    void depthFirst(Closure c) {
        
        c(this)
        
        this.children*.depthFirst(c)
    }
    
    /**
     * Filter the graph so that only paths containing the specified output remain
     */
    GraphEntry filter(String output) {
        GraphEntry outputEntry = this.entryFor(new File(output))
        
        if(!outputEntry)
            return null
        
        GraphEntry result = outputEntry.clone()
        
        GraphEntry root = null
        
        def trimParent 
        trimParent = { GraphEntry p, GraphEntry c, int index ->
            // Remove all parents that don't contain the child
            
            p = new GraphEntry(values: p.values, parents: p.parents, children: [c]) 
            c.parents[index] = p
            
            if(p in p.parents)
                throw new RuntimeException("Internal error: an output appears to be a dependency of itself")
            
            // Recurse upwards
            if(p.parents)
                p.parents.eachWithIndex { gp, pindex -> trimParent(gp, p, pindex) }
            else
                root = p
        }
        
        if(outputEntry.parents)
            outputEntry.parents.eachWithIndex { p,index -> trimParent(p, result, index) }
        else
            root = outputEntry
        
        return root
    }
    
    /**
     * Divide the outputs into groups depending on their 
     * inputs.
     * @return
     */
    def groupOutputs(List<Properties> outputs = null) {
        def outputGroups = [:]
        if(outputs == null)
            outputs = this.values
            
        for(def o in outputs) {
            def key = o.inputs.sort().join(',')
            if(!outputGroups.containsKey(key)) {
              outputGroups[key] = [o]
            }
            else {
                outputGroups[key] << o
            }
        }
        return outputGroups
    }
    
    String dump() {
        
        def inputs = children*.values*.inputs.flatten()
        if(!inputs)
            inputs = ["<no inputs>"]
                  
        String inputValue = inputs.join('\n') + ' => \n'
        return inputValue + dumpChildren(inputs.collect {it.size()}.max())
        
//        return dumpChildren(0)
    }
   
    /**
     * Render this graph entry as a tree
     */
    String dumpChildren(int indent = 0, List<Properties> filter = null) {
        
       
        def valuesToDump = filter != null ? filter : values
        
        def names = values*.outputFile*.name
        
        String me = names?names.collect { " " * indent + it }.join("\n") + (children?" => \n":"") :""
        
        StringBuilder result = new StringBuilder()
        def filteredChildren = children
        if(filter) {
          def filterOutputs = filter*.outputFile*.name.flatten()
          
          filteredChildren = children.grep { child ->
              // Only output children who have one of the output files in the filter
              // as an input
              child.values*.inputs.flatten().find { childInput ->  
//                  if(filterOutputs.contains(childInput))
//                    println "Compare $filterOutputs with $childInput YES"
//                  else
//                    println "Compare $filterOutputs with $childInput NO"
//                    
                  return filterOutputs.contains(childInput)  
              }
          }
        }
        return me + filteredChildren*.dumpChildren(indent+(names.collect{it.size()}.max()?:0)).join('\n')
    }
}

/**
 * Manages dependency tracking functions
 * 
 * @author <EMAIL>
 */
@Singleton
@Log
class Dependencies {
    
    GraphEntry outputGraph
    
    /**
     * List of output files created in *this run*
     */
    List<String> outputFilesGenerated = []
    
    /**
     * Return true iff all the outputs and downstream outputs for which
     * they are dependencies are present and up to date.
     * 
     * @param outputs
     * @param inputs
     * @return  true iff file is newer than all inputs, but older than all 
     *          non-up-to-date outputs
     */
    boolean checkUpToDate(def outputs, def inputs) {
        
        inputs = Utils.box(inputs)
        
        // If there are no outputs then by definition, they are all up to date
        if(!outputs)
            return true
        
        // If the outputs are created from nothing (no inputs)
        // then they are up to date as long as they exist
        if(!inputs)  {
            return outputs.collect { new File(it) }.every { it.exists() }
        }
            
        // The most obvious case: all the outputs exist and are newer than 
        // the inputs. We can return straight away here
        List<File> older = Utils.findOlder(outputs,inputs)
        if(!older) {
            log.info "No missing / older files from inputs: $outputs are up to date"
            return true
        }
        else
        if(older.any { it.exists() }) { // If any of the older files exist then we have no choice but to rebuild them
            log.info "Not up to date because these files exist and are older than inputs: " + older.grep { it.exists() }
            return false
        }
        else {
            log.info "Found these missing / older files: " + older
        }
        
        GraphEntry graph = this.getOutputGraph()
        
        def outDated = older.collect { it.canonicalPath }.grep { out ->
             def p = graph.propertiesFor(out); 
             if(!p || !p.cleaned)  {
                 if(!p)
                     log.info "Output properties file is not available for $out: assume NOT cleaned up"
                 else
                     log.info "Output properties are available, indicating file was NOT cleaned up"
                     
                 return true 
             }
             else {
                 log.info "File $out has output properties available: upToDate=$p.upToDate"
                 return !p.upToDate
             }
        }
        
        if(!outDated) {
            log.info "All missing files are up to date"
            return true
        }
        else {
            log.info "Some files are outdated : $outDated"
            return false
        }
    }
    
    /**
     * Check either a single file passed as a string or a list
     * of files passed as a collection.  Throws an exception
     * if any of them are missing and cannot be accounted for
     * in an output property file.
     *
     * @param f     a String or collection of Strings representing file names
     */
    void checkFiles(def fileNames, type="input") {
        
        log.info "Checking $type (s) " + fileNames
        
        GraphEntry graph = this.getOutputGraph()
        List missing = Utils.box(fileNames).grep { String.valueOf(it) != "null" }.collect { new File(it.toString()) }.grep { File f ->
            
            log.info " Checking file $f"
            if(f.exists())
                return false
                
            Properties p = graph.propertiesFor(f.path)
            if(!p) {
                log.info "There are no properties for $f.path and file is missing"
                return true
            }
            
            if(p.cleaned) {
                log.info "File $f.path [$type] does not exist but has a properties file indicating it was cleaned up"
                return false
            }
            else {
                log.info "File $f.path [$type] does not exist, has a properties file indicating it is not cleaned up"
                return true
            }
        }
       
        if(missing)
            throw new PipelineError("Expected $type file ${missing[0]} could not be found")
    }

    
    synchronized GraphEntry getOutputGraph() {
        if(this.outputGraph == null) {
            List<Properties> propertiesFiles = scanOutputFolder()
//            List<String> outputPaths = 
            this.outputGraph = computeOutputGraph(propertiesFiles)
        }
        return this.outputGraph
    }
    
    void reset() {
        this.outputGraph = null
    }
    
    /**
     * For each output file created in the context, save information
     * about it such that it can be reliably loaded by this same stage
     * if the pipeline is re-executed.
     */
    synchronized void saveOutputs(PipelineContext context, List<File> oldFiles, Map<File,Long> timestamps, List<String> inputs) {
        GraphEntry root = getOutputGraph()
        
        // Get the full branch path of this pipeline stage
        def pipeline = Pipeline.currentRuntimePipeline.get()
        String branchPath = pipeline.branchPath.join("/")
        
        context.trackedOutputs.each { String id, Command command ->
            
            String cmd = command.command<commit_msg>caching of DAG lookups by canonical path to speed up large, complex pipelines<commit_after>/*
 * Copyright (c) Murdoch Childrens Research Institute and Contributers
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 * 
 * Redistributions of source code must retain the above copyright notice, this
 * list of conditions and the following disclaimer.
 * 
 * Redistributions in binary form must reproduce the above copyright notice, this
 * list of conditions and the following disclaimer in the documentation and/or
 * other materials provided with the distribution.
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */  
package bpipe

import groovy.util.logging.Log;
import groovyx.gpars.GParsPool
import groovy.time.TimeCategory;
import groovy.transform.CompileStatic;

/**
 * A node in the dependency graph representing a set of outputs
 * sharing a common set of child nodes (who depend on the outputs) and
 * inputs (parents nodes) upon which the children depend.  
 * 
 * @author ssadedin
 */
class GraphEntry {
    
    List<Properties> values
    List<GraphEntry> parents = []
    List<GraphEntry> children = []
    
    /**
     * An optional index to speed up lookups by canonical path - not populated by default,
     * but can be populated by using index()
     */
    Map<String, GraphEntry> index = null
    
    @CompileStatic
    GraphEntry findBy(Closure c) {
        if(this.values != null && values.any { c(it) })
            return this
            
        for(GraphEntry child in children) {
           def result = child.findBy(c)
           if(result)
               return result;
        }
    }
    
    @CompileStatic
    void index(int sizeHint) {
        Utils.time("Index output graph") {
            Map<String, GraphEntry> indexTmp = new HashMap(sizeHint)
            depthFirst { GraphEntry e ->
                if(e.values) {
                    for(Properties p in e.values) {
                        indexTmp[(String)p.canonicalPath] = e
                    }
                }
            }
        }
    }
    
    List<Properties> findAllOutputsBy(Closure c) {
        
        
        List<Properties> results = []
        
        if(this.values != null)
            results.addAll(this.values.grep { c(it)})
            
        for(GraphEntry child in children) {
           results.addAll(child.findAllOutputsBy(c))
        }
        
        return results
    }
  
    
    /**
     * Search the graph for entry with given outputfile
     */
    @CompileStatic
    GraphEntry entryFor(File outputFile) {
        // In case of non-default output directory, the outputFile itself may be in a directory
        final String outputFilePath = outputFile.canonicalPath
        return entryForCanonicalPath(outputFilePath)
    }
    
    @CompileStatic
    GraphEntry entryForCanonicalPath(String canonicalPath) {
        GraphEntry entry = index?.get(canonicalPath)
        if(entry)
            return entry
        // In case of non-default output directory, the outputFile itself may be in a directory
        findBy { Properties p -> canonicalPathFor(p) == canonicalPath  }
    }
    
     /**
     * getCanonicalPath can be very slow on some systems, so 
     * this method caches them in the property file itself.
     * 
     * @param p
     * @return
     */
    @CompileStatic
    String canonicalPathFor(Properties p) {
        synchronized(p) {
            if(p.containsKey("canonicalPath"))
                return p["canonicalPath"]        
                
            p["canonicalPath"] = ((File)p["outputFile"]).canonicalPath
        }
    }
    
    /**
     * Search the graph for properties for the given output file
     */
    Properties propertiesFor(String outputFile) { 
       // In case of non-default output directory, the outputFile itself may be in a directory
       String outputFilePath = new File(outputFile).canonicalPath
       def values = entryForCanonicalPath(outputFilePath)?.values
       if(!values)
           return null
           
       for(def o in values) {
           if(canonicalPathFor(o) == outputFilePath) {
               return o
           }
       }
       return null
    }
    
    /**
     * Create a copy of the subtree for this GraphEntry and its children (but NOT parents!)
     */
    GraphEntry clone() {
        new GraphEntry(values: values.clone(), parents: parents.clone(), children: children.collect { it.clone() })
    }
    
    void depthFirst(Closure c) {
        
        c(this)
        
        this.children*.depthFirst(c)
    }
    
    /**
     * Filter the graph so that only paths containing the specified output remain
     */
    GraphEntry filter(String output) {
        GraphEntry outputEntry = this.entryFor(new File(output))
        
        if(!outputEntry)
            return null
        
        GraphEntry result = outputEntry.clone()
        
        GraphEntry root = null
        
        def trimParent 
        trimParent = { GraphEntry p, GraphEntry c, int index ->
            // Remove all parents that don't contain the child
            
            p = new GraphEntry(values: p.values, parents: p.parents, children: [c]) 
            c.parents[index] = p
            
            if(p in p.parents)
                throw new RuntimeException("Internal error: an output appears to be a dependency of itself")
            
            // Recurse upwards
            if(p.parents)
                p.parents.eachWithIndex { gp, pindex -> trimParent(gp, p, pindex) }
            else
                root = p
        }
        
        if(outputEntry.parents)
            outputEntry.parents.eachWithIndex { p,index -> trimParent(p, result, index) }
        else
            root = outputEntry
        
        return root
    }
    
    /**
     * Divide the outputs into groups depending on their 
     * inputs.
     * @return
     */
    def groupOutputs(List<Properties> outputs = null) {
        def outputGroups = [:]
        if(outputs == null)
            outputs = this.values
            
        for(def o in outputs) {
            def key = o.inputs.sort().join(',')
            if(!outputGroups.containsKey(key)) {
              outputGroups[key] = [o]
            }
            else {
                outputGroups[key] << o
            }
        }
        return outputGroups
    }
    
    String dump() {
        
        def inputs = children*.values*.inputs.flatten()
        if(!inputs)
            inputs = ["<no inputs>"]
                  
        String inputValue = inputs.join('\n') + ' => \n'
        return inputValue + dumpChildren(inputs.collect {it.size()}.max())
        
//        return dumpChildren(0)
    }
   
    /**
     * Render this graph entry as a tree
     */
    String dumpChildren(int indent = 0, List<Properties> filter = null) {
        
       
        def valuesToDump = filter != null ? filter : values
        
        def names = values*.outputFile*.name
        
        String me = names?names.collect { " " * indent + it }.join("\n") + (children?" => \n":"") :""
        
        StringBuilder result = new StringBuilder()
        def filteredChildren = children
        if(filter) {
          def filterOutputs = filter*.outputFile*.name.flatten()
          
          filteredChildren = children.grep { child ->
              // Only output children who have one of the output files in the filter
              // as an input
              child.values*.inputs.flatten().find { childInput ->  
//                  if(filterOutputs.contains(childInput))
//                    println "Compare $filterOutputs with $childInput YES"
//                  else
//                    println "Compare $filterOutputs with $childInput NO"
//                    
                  return filterOutputs.contains(childInput)  
              }
          }
        }
        return me + filteredChildren*.dumpChildren(indent+(names.collect{it.size()}.max()?:0)).join('\n')
    }
}

/**
 * Manages dependency tracking functions
 * 
 * @author <EMAIL>
 */
@Singleton
@Log
class Dependencies {
    
    GraphEntry outputGraph
    
    /**
     * List of output files created in *this run*
     */
    List<String> outputFilesGenerated = []
    
    /**
     * Return true iff all the outputs and downstream outputs for which
     * they are dependencies are present and up to date.
     * 
     * @param outputs
     * @param inputs
     * @return  true iff file is newer than all inputs, but older than all 
     *          non-up-to-date outputs
     */
    boolean checkUpToDate(def outputs, def inputs) {
        
        inputs = Utils.box(inputs)
        
        // If there are no outputs then by definition, they are all up to date
        if(!outputs)
            return true
        
        // If the outputs are created from nothing (no inputs)
        // then they are up to date as long as they exist
        if(!inputs)  {
            return outputs.collect { new File(it) }.every { it.exists() }
        }
            
        // The most obvious case: all the outputs exist and are newer than 
        // the inputs. We can return straight away here
        List<File> older = Utils.findOlder(outputs,inputs)
        if(!older) {
            log.info "No missing / older files from inputs: $outputs are up to date"
            return true
        }
        else
        if(older.any { it.exists() }) { // If any of the older files exist then we have no choice but to rebuild them
            log.info "Not up to date because these files exist and are older than inputs: " + older.grep { it.exists() }
            return false
        }
        else {
            log.info "Found these missing / older files: " + older
        }
        
        GraphEntry graph = this.getOutputGraph()
        
        def outDated = older.collect { it.canonicalPath }.grep { out ->
             def p = graph.propertiesFor(out); 
             if(!p || !p.cleaned)  {
                 if(!p)
                     log.info "Output properties file is not available for $out: assume NOT cleaned up"
                 else
                     log.info "Output properties are available, indicating file was NOT cleaned up"
                     
                 return true 
             }
             else {
                 log.info "File $out has output properties available: upToDate=$p.upToDate"
                 return !p.upToDate
             }
        }
        
        if(!outDated) {
            log.info "All missing files are up to date"
            return true
        }
        else {
            log.info "Some files are outdated : $outDated"
            return false
        }
    }
    
    /**
     * Check either a single file passed as a string or a list
     * of files passed as a collection.  Throws an exception
     * if any of them are missing and cannot be accounted for
     * in an output property file.
     *
     * @param f     a String or collection of Strings representing file names
     */
    void checkFiles(def fileNames, type="input") {
        
        log.info "Checking $type (s) " + fileNames
        
        GraphEntry graph = this.getOutputGraph()
        List missing = Utils.box(fileNames).grep { String.valueOf(it) != "null" }.collect { new File(it.toString()) }.grep { File f ->
            
            log.info " Checking file $f"
            if(f.exists())
                return false
                
            Properties p = graph.propertiesFor(f.path)
            if(!p) {
                log.info "There are no properties for $f.path and file is missing"
                return true
            }
            
            if(p.cleaned) {
                log.info "File $f.path [$type] does not exist but has a properties file indicating it was cleaned up"
                return false
            }
            else {
                log.info "File $f.path [$type] does not exist, has a properties file indicating it is not cleaned up"
                return true
            }
        }
       
        if(missing)
            throw new PipelineError("Expected $type file ${missing[0]} could not be found")
    }

    
    synchronized GraphEntry getOutputGraph() {
        if(this.outputGraph == null) {
            List<Properties> propertiesFiles = scanOutputFolder()
            this.outputGraph = computeOutputGraph(propertiesFiles)
            this.outputGraph.index(propertiesFiles.size()*2)
        }
        return this.outputGraph
    }
    
    void reset() {
        this.outputGraph = null
    }
    
    /**
     * For each output file created in the context, save information
     * about it such that it can be reliably loaded by this same stage
     * if the pipeline is re-executed.
     */
    synchronized void saveOutputs(PipelineContext context, List<File> oldFiles, Map<File,Long> timestamps, List<String> inputs) {
        GraphEntry root = getOutputGraph()
        
        // Get the full branch path of this pipeline stage
        def pipeline = Pipeline.currentRuntimePipeline.get()
        String branchPath = pipeline.branchPath.join("/")
        
        context.trackedOutputs.each { String id, Command command ->
            
            String cmd = command.command
{'fraction_non_alphanumeric': 0.08183807439824946, 'fraction_numerical': 0.00032822757111597374, 'mean_word_length': 1.770357648009699, 'pattern_counts': {'":': 2, '<': 40, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6015019', 'n_tokens_mistral': 7115, 'n_tokens_neox': 6866, 'n_words': 2409}