<commit_before>/**
 *  Copyright 2007-2008 University Of Southern California
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */


package org.griphyn.cPlanner.transfer.implementation;

import edu.isi.pegasus.common.logging.LoggingKeys;
import edu.isi.pegasus.planner.catalog.site.classes.GridGateway;
import edu.isi.pegasus.planner.catalog.site.classes.SiteCatalogEntry;
import org.griphyn.cPlanner.classes.ADag;
import org.griphyn.cPlanner.classes.SubInfo;
import org.griphyn.cPlanner.classes.TransferJob;
import org.griphyn.cPlanner.classes.FileTransfer;
import org.griphyn.cPlanner.classes.PegasusBag;

import org.griphyn.cPlanner.common.LogManager;

import org.griphyn.cPlanner.transfer.MultipleFTPerXFERJob;

import org.griphyn.common.catalog.TransformationCatalogEntry;

import org.griphyn.common.classes.TCType;

import org.griphyn.common.util.Separator;

import org.griphyn.cPlanner.cluster.aggregator.JobAggregatorFactory;
import org.griphyn.cPlanner.cluster.JobAggregator;

import java.io.File;

import java.util.Collection;
import java.util.List;
import java.util.LinkedList;
import java.util.Iterator;
import org.griphyn.cPlanner.transfer.Refiner;
import org.griphyn.cPlanner.classes.Profile;
import java.util.ArrayList;


/**
 * A Windward implementation that uses the seqexec client to execute
 *
 * -DC Transfer client to fetch the raw data sources
 * -Pegasus transfer client to fetch the patterns from the pattern catalog.
 *
 * @author <NAME>
 * @version $Revision$
 */

public class Windward extends Abstract
                      implements MultipleFTPerXFERJob  {




    /**
     * The prefix to identify the raw data sources.
     */
    public static final String DATA_SOURCE_PREFIX = "DS";

    /**
     * A short description of the transfer implementation.
     */
    public static final String DESCRIPTION = "Seqexec Transfer Wrapper around Pegasus Transfer and DC Transfer Client";

    /**
     * The transformation namespace for for the transfer job.
     */
    public static final String TRANSFORMATION_NAMESPACE = "windward";


    /**
     * The name of the underlying transformation that is queried for in the
     * Transformation Catalog.
     */
    public static final String TRANSFORMATION_NAME = "dc-transfer";

    /**
     * The version number for the transfer job.
     */
    public static final String TRANSFORMATION_VERSION = null;

    /**
     * The derivation namespace for for the transfer job.
     */
    public static final String DERIVATION_NAMESPACE = "windward";

    /**
     * The name of the underlying derivation.
     */
    public static final String DERIVATION_NAME = "dc-transfer";

    /**
     * The derivation version number for the transfer job.
     */
    public static final String DERIVATION_VERSION = null;


    /**
     * The handle to the transfer implementation.
     */
    private Transfer mPegasusTransfer;

    
    /**
     * The handle to the BAE transfer implementation.
     */
    private BAE mBAETransfer;
    
    /**
     * The seqexec job aggregator.
     */
    private JobAggregator mSeqExecAggregator;
    
    /**
     * The refiner beig used.
     */
    private Refiner mRefiner;

    /**
     * The overloaded constructor, that is called by the Factory to load the
     * class.
     *
     * @param bag   the bag of initialization objects.
     */
    public Windward( PegasusBag bag ) {
        super( bag );

        //should probably go through the factory
        mPegasusTransfer = new Transfer( bag );
        mBAETransfer     = new BAE( bag );

        //just to pass the label have to send an empty ADag.
        //should be fixed
        ADag dag = new ADag();
        dag.dagInfo.setLabel( "windward" );

        mSeqExecAggregator = JobAggregatorFactory.loadInstance( JobAggregatorFactory.SEQ_EXEC_CLASS,
                                                                dag,
                                                                bag  );
    }

    /**
     * Sets the callback to the refiner, that has loaded this implementation.
     *
     * @param refiner  the transfer refiner that loaded the implementation.
     */
    public void setRefiner(Refiner refiner){
        super.setRefiner( refiner );
        //also set the refiner for hte internal pegasus transfer
        mPegasusTransfer.setRefiner( refiner );
        mBAETransfer.setRefiner(refiner);
        mRefiner = refiner;
    }


    /**
     *
     *
     * @param job         the SubInfo object for the job, in relation to which
     *                    the transfer node is being added. Either the transfer
     *                    node can be transferring this jobs input files to
     *                    the execution pool, or transferring this job's output
     *                    files to the output pool.
     * @param files       collection of <code>FileTransfer</code> objects
     *                    representing the data files and staged executables to be
     *                    transferred.
     * @param execFiles   subset collection of the files parameter, that identifies
     *                    the executable files that are being transferred.
     * @param txJobName   the name of transfer node.
     * @param jobClass    the job Class for the newly added job. Can be one of the
     *                    following:
     *                              stage-in
     *                              stage-out
     *                              inter-pool transfer
     *
     * @return  the created TransferJob.
     */
    public TransferJob createTransferJob( SubInfo job,
                                          Collection files,
                                          Collection execFiles,
                                          String txJobName,
                                          int jobClass) {


        //iterate through all the files and identify the patterns
        //and the other data sources
        Collection rawDataSources = new LinkedList();
        Collection patterns       = new LinkedList();

        for( Iterator it = files.iterator(); it.hasNext(); ){
            FileTransfer ft = ( FileTransfer )it.next();
            //no way to distinguish a pattern right now
            /*
            if( ft.getLFN().startsWith( DATA_SOURCE_PREFIX ) && !ft.getSourceURL().getValue().endsWith( ".zip" )){
                //it a raw data source that will have to be ingested
                rawDataSources.add( ft );
            }
            else{
                //everything else is a pattern
                patterns.add( ft );
            }*/
            rawDataSources.add( ft );
        }

        List txJobs = new LinkedList();

        //use the Pegasus Transfer to handle the patterns
        TransferJob patternTXJob = null;
        String patternTXJobStdin = null;
        List<String> txJobIDs = new LinkedList<String>();
        if( !patterns.isEmpty() ){
            patternTXJob = mPegasusTransfer.createTransferJob( job,
                                                               patterns,
                                                               null,
                                                               txJobName,
                                                               jobClass );

            //get the stdin and set it as lof in the arguments
            patternTXJobStdin = patternTXJob.getStdIn();
            StringBuffer patternArgs = new StringBuffer();
            patternArgs.append( patternTXJob.getArguments() ).append( " " ).
                append( patternTXJobStdin );
            patternTXJob.setArguments( patternArgs.toString() );
            patternTXJob.setStdIn( "" );
            txJobs.add( patternTXJob );
            txJobIDs.add( patternTXJob.getID() );
        }


      

        //this should in fact only be set
        // for non third party pools
        //we first check if there entry for transfer universe,
        //if no then go for globus
//        SiteInfo ePool = mSCHandle.getTXPoolEntry( job.getSiteHandle() );
//        JobManager jobmanager = ePool.selectJobManager(this.TRANSFER_UNIVERSE,true);
        SiteCatalogEntry ePool = mSiteStore.lookup( job.getSiteHandle() );
        GridGateway jobmanager = ePool.selectGridGateway( GridGateway.JOB_TYPE.transfer );


        //use the DC transfer client to handle the data sources        
        for( Iterator it = rawDataSources.iterator(); it.hasNext(); ){
            FileTransfer ft = (FileTransfer)it.next();            
            List<FileTransfer> l = new LinkedList<FileTransfer>();
            l.add( ft );
            TransferJob dcTXJob = mBAETransfer.createTransferJob( job,
                                                                  l,
                                                                  null,
                                                                  txJobName, 
                                                                  jobClass );
            txJobs.add( dcTXJob );
            txJobIDs.add( patternTXJob.getID() );
        }

        //only merging if more than only one data set being staged
        TransferJob txJob = null;
        if( txJobs.size() > 1 ){        
            //now lets merge all these jobs
            SubInfo merged = mSeqExecAggregator.construct( txJobs, "transfer", txJobName  );
            txJob = new TransferJob( merged );


            //set the name of the merged job back to the name of
            //transfer job passed in the function call
            txJob.setName( txJobName );
            txJob.setJobType( jobClass );
        }else{
            txJob = (TransferJob) txJobs.get( 0 );
        }

        //if a pattern job was constructed add the pattern stdin
        //as an input file for condor to transfer
        if( patternTXJobStdin != null ){
            txJob.condorVariables.addIPFileForTransfer( patternTXJobStdin );
        }
        //take care of transfer of proxies
        this.checkAndTransferProxy( txJob );

        //apply the priority to the transfer job
        this.applyPriority( txJob );
<commit_msg>Fixed a NPE that was occuring due to faulty logging<commit_after>/**
 *  Copyright 2007-2008 University Of Southern California
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *  http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */


package org.griphyn.cPlanner.transfer.implementation;

import edu.isi.pegasus.common.logging.LoggingKeys;
import edu.isi.pegasus.planner.catalog.site.classes.GridGateway;
import edu.isi.pegasus.planner.catalog.site.classes.SiteCatalogEntry;
import org.griphyn.cPlanner.classes.ADag;
import org.griphyn.cPlanner.classes.SubInfo;
import org.griphyn.cPlanner.classes.TransferJob;
import org.griphyn.cPlanner.classes.FileTransfer;
import org.griphyn.cPlanner.classes.PegasusBag;

import org.griphyn.cPlanner.common.LogManager;

import org.griphyn.cPlanner.transfer.MultipleFTPerXFERJob;

import org.griphyn.common.catalog.TransformationCatalogEntry;

import org.griphyn.common.classes.TCType;

import org.griphyn.common.util.Separator;

import org.griphyn.cPlanner.cluster.aggregator.JobAggregatorFactory;
import org.griphyn.cPlanner.cluster.JobAggregator;

import java.io.File;

import java.util.Collection;
import java.util.List;
import java.util.LinkedList;
import java.util.Iterator;
import org.griphyn.cPlanner.transfer.Refiner;
import org.griphyn.cPlanner.classes.Profile;
import java.util.ArrayList;


/**
 * A Windward implementation that uses the seqexec client to execute
 *
 * -DC Transfer client to fetch the raw data sources
 * -Pegasus transfer client to fetch the patterns from the pattern catalog.
 *
 * @author <NAME>
 * @version $Revision$
 */

public class Windward extends Abstract
                      implements MultipleFTPerXFERJob  {




    /**
     * The prefix to identify the raw data sources.
     */
    public static final String DATA_SOURCE_PREFIX = "DS";

    /**
     * A short description of the transfer implementation.
     */
    public static final String DESCRIPTION = "Seqexec Transfer Wrapper around Pegasus Transfer and DC Transfer Client";

    /**
     * The transformation namespace for for the transfer job.
     */
    public static final String TRANSFORMATION_NAMESPACE = "windward";


    /**
     * The name of the underlying transformation that is queried for in the
     * Transformation Catalog.
     */
    public static final String TRANSFORMATION_NAME = "dc-transfer";

    /**
     * The version number for the transfer job.
     */
    public static final String TRANSFORMATION_VERSION = null;

    /**
     * The derivation namespace for for the transfer job.
     */
    public static final String DERIVATION_NAMESPACE = "windward";

    /**
     * The name of the underlying derivation.
     */
    public static final String DERIVATION_NAME = "dc-transfer";

    /**
     * The derivation version number for the transfer job.
     */
    public static final String DERIVATION_VERSION = null;


    /**
     * The handle to the transfer implementation.
     */
    private Transfer mPegasusTransfer;

    
    /**
     * The handle to the BAE transfer implementation.
     */
    private BAE mBAETransfer;
    
    /**
     * The seqexec job aggregator.
     */
    private JobAggregator mSeqExecAggregator;
    
    /**
     * The refiner beig used.
     */
    private Refiner mRefiner;

    /**
     * The overloaded constructor, that is called by the Factory to load the
     * class.
     *
     * @param bag   the bag of initialization objects.
     */
    public Windward( PegasusBag bag ) {
        super( bag );

        //should probably go through the factory
        mPegasusTransfer = new Transfer( bag );
        mBAETransfer     = new BAE( bag );

        //just to pass the label have to send an empty ADag.
        //should be fixed
        ADag dag = new ADag();
        dag.dagInfo.setLabel( "windward" );

        mSeqExecAggregator = JobAggregatorFactory.loadInstance( JobAggregatorFactory.SEQ_EXEC_CLASS,
                                                                dag,
                                                                bag  );
    }

    /**
     * Sets the callback to the refiner, that has loaded this implementation.
     *
     * @param refiner  the transfer refiner that loaded the implementation.
     */
    public void setRefiner(Refiner refiner){
        super.setRefiner( refiner );
        //also set the refiner for hte internal pegasus transfer
        mPegasusTransfer.setRefiner( refiner );
        mBAETransfer.setRefiner(refiner);
        mRefiner = refiner;
    }


    /**
     *
     *
     * @param job         the SubInfo object for the job, in relation to which
     *                    the transfer node is being added. Either the transfer
     *                    node can be transferring this jobs input files to
     *                    the execution pool, or transferring this job's output
     *                    files to the output pool.
     * @param files       collection of <code>FileTransfer</code> objects
     *                    representing the data files and staged executables to be
     *                    transferred.
     * @param execFiles   subset collection of the files parameter, that identifies
     *                    the executable files that are being transferred.
     * @param txJobName   the name of transfer node.
     * @param jobClass    the job Class for the newly added job. Can be one of the
     *                    following:
     *                              stage-in
     *                              stage-out
     *                              inter-pool transfer
     *
     * @return  the created TransferJob.
     */
    public TransferJob createTransferJob( SubInfo job,
                                          Collection files,
                                          Collection execFiles,
                                          String txJobName,
                                          int jobClass) {


        //iterate through all the files and identify the patterns
        //and the other data sources
        Collection rawDataSources = new LinkedList();
        Collection patterns       = new LinkedList();

        for( Iterator it = files.iterator(); it.hasNext(); ){
            FileTransfer ft = ( FileTransfer )it.next();
            //no way to distinguish a pattern right now
            /*
            if( ft.getLFN().startsWith( DATA_SOURCE_PREFIX ) && !ft.getSourceURL().getValue().endsWith( ".zip" )){
                //it a raw data source that will have to be ingested
                rawDataSources.add( ft );
            }
            else{
                //everything else is a pattern
                patterns.add( ft );
            }*/
            rawDataSources.add( ft );
        }

        List txJobs = new LinkedList();

        //use the Pegasus Transfer to handle the patterns
        TransferJob patternTXJob = null;
        String patternTXJobStdin = null;
        List<String> txJobIDs = new LinkedList<String>();
        if( !patterns.isEmpty() ){
            patternTXJob = mPegasusTransfer.createTransferJob( job,
                                                               patterns,
                                                               null,
                                                               txJobName,
                                                               jobClass );

            //get the stdin and set it as lof in the arguments
            patternTXJobStdin = patternTXJob.getStdIn();
            StringBuffer patternArgs = new StringBuffer();
            patternArgs.append( patternTXJob.getArguments() ).append( " " ).
                append( patternTXJobStdin );
            patternTXJob.setArguments( patternArgs.toString() );
            patternTXJob.setStdIn( "" );
            txJobs.add( patternTXJob );
            txJobIDs.add( patternTXJob.getID() );
        }


      

        //this should in fact only be set
        // for non third party pools
        //we first check if there entry for transfer universe,
        //if no then go for globus
//        SiteInfo ePool = mSCHandle.getTXPoolEntry( job.getSiteHandle() );
//        JobManager jobmanager = ePool.selectJobManager(this.TRANSFER_UNIVERSE,true);
        SiteCatalogEntry ePool = mSiteStore.lookup( job.getSiteHandle() );
        GridGateway jobmanager = ePool.selectGridGateway( GridGateway.JOB_TYPE.transfer );


        //use the DC transfer client to handle the data sources        
        for( Iterator it = rawDataSources.iterator(); it.hasNext(); ){
            FileTransfer ft = (FileTransfer)it.next();            
            List<FileTransfer> l = new LinkedList<FileTransfer>();
            l.add( ft );
            TransferJob dcTXJob = mBAETransfer.createTransferJob( job,
                                                                  l,
                                                                  null,
                                                                  txJobName, 
                                                                  jobClass );
            txJobs.add( dcTXJob );
            txJobIDs.add( dcTXJob.getID() );
        }

        //only merging if more than only one data set being staged
        TransferJob txJob = null;
        if( txJobs.size() > 1 ){        
            //now lets merge all these jobs
            SubInfo merged = mSeqExecAggregator.construct( txJobs, "transfer", txJobName  );
            txJob = new TransferJob( merged );


            //set the name of the merged job back to the name of
            //transfer job passed in the function call
            txJob.setName( txJobName );
            txJob.setJobType( jobClass );
        }else{
            txJob = (TransferJob) txJobs.get( 0 );
        }

        //if a pattern job was constructed add the pattern stdin
        //as an input file for condor to transfer
        if( patternTXJobStdin != null ){
            txJob.condorVariables.addIPFileForTransfer( patternTXJobStdin );
        }
        //take care of transfer of proxies
        this.checkAndTransferProxy( txJob );

        //apply the priority to the transfer job
        this.applyPriority( txJob );

{'fraction_non_alphanumeric': 0.07380941187532321, 'fraction_numerical': 0.0013163461990503503, 'mean_word_length': 1.7547267547267547, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 62, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '980314', 'n_tokens_mistral': 5385, 'n_tokens_neox': 5233, 'n_words': 1731}