<commit_before>/*
 * The contents of this file are subject to the license and copyright
 * detailed in the LICENSE and NOTICE files at the root of the source
 * tree and available online at
 *
 *     http://duracloud.org/license/
 */
package org.duracloud.mill.ltp;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;

import org.duracloud.common.queue.TaskQueue;
import org.duracloud.mill.common.storageprovider.StorageProviderFactory;
import org.duracloud.mill.credentials.CredentialsRepo;
import org.duracloud.mill.credentials.CredentialsRepoException;
import org.duracloud.mill.credentials.StorageProviderCredentials;
import org.duracloud.mill.notification.NotificationManager;
import org.duracloud.storage.provider.StorageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class is responsible for filling the duplication <code>TaskQueue</code>
 * by looping through all duplication policies for all accounts and spaces and
 * blindly creates duplication tasks. It will create tasks for all items in the
 * the source destination as well as all items in the destination provider but
 * not in the source provider. A notable feature of this task producer is that
 * it attempts to respect a designated maximum task queue size. Once the limit
 * has been reached, the producer will stop. On subsequent runs, the producer
 * will pick up where it left off, starting with the next account,space,set of
 * content items, and duplication store policy. If all content items are visited
 * within a single run before the task queue limit has been reached, the
 * producer will exit.
 * 
 * For more information about how this process fits into the whole system of
 * collaborating components, see
 * https://wiki.duraspace.org/display/DSPINT/DuraCloud+Duplication+-+System+Overview
 * 
 * @author <NAME> Date: Nov 5, 2013
 */
public abstract class LoopingTaskProducer<T extends Morsel> implements Runnable {
    private static Logger log = LoggerFactory.getLogger(LoopingTaskProducer.class);
    private TaskQueue taskQueue;
    private CredentialsRepo credentialsRepo;
    private StateManager<T> stateManager;
    private int maxTaskQueueSize;
    private StorageProviderFactory storageProviderFactory;
    private List<T> morselsToReload = new LinkedList<>();
    private Frequency frequency;
    private RunStats cumulativeTotals;
    private NotificationManager notificationManager;
    private LoopingTaskProducerConfigurationManager config;
    private Map<String,RunStats> runstats = new HashMap<>();

    public LoopingTaskProducer(CredentialsRepo credentialsRepo,
                               StorageProviderFactory storageProviderFactory,
                               TaskQueue taskQueue,
                               StateManager<T> state,
                               int maxTaskQueueSize, 
                               Frequency frequency,
                               NotificationManager notificationManager, 
                               LoopingTaskProducerConfigurationManager config) {
        
        this.credentialsRepo = credentialsRepo;
        this.storageProviderFactory = storageProviderFactory;
        this.taskQueue = taskQueue;
        this.stateManager = state;
        this.credentialsRepo = credentialsRepo;
        this.maxTaskQueueSize = maxTaskQueueSize;
        this.frequency = frequency;
        this.cumulativeTotals = createRunStats();
        this.notificationManager = notificationManager;
        this.config = config;
    }
    
    protected Frequency getFrequency(){
        return this.frequency;
    }
    


    protected CredentialsRepo getCredentialsRepo() {
        return credentialsRepo;
    }


    protected TaskQueue getTaskQueue() {
        return taskQueue;
    }

    protected int getMaxTaskQueueSize() {
        return maxTaskQueueSize;
    }
    
    public void run(){
        
        
        Timer timer = new Timer();
        try {
            
            timer.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    logSessionStats();
                }

            }, 5 * 60 * 1000, 5 * 60 * 1000);  
            
            if(runLater()){
                return;
            }

            log.info("Starting run...");
            Queue<T> morselQueue = loadMorselQueue();
            
            while(!morselQueue.isEmpty() && this.taskQueue.size() < maxTaskQueueSize){
                T morsel = morselQueue.peek();
                nibble(morselQueue);
                persistMorsels(morselQueue, morselsToReload);
               
                if(morselQueue.isEmpty()){
                    morselQueue = reloadMorselQueue();
                }else{
                    //break if nothing was removed from the queue
                    //if nothing was removed from the queue we can assume
                    //that the for whatever reason the morsel could not be processed
                    //at this time, so the process should wait for the next run.
                    if(morsel.equals(morselQueue.peek())){
                        break;
                    }
                }
                
               
            }
    
            if(morselQueue.isEmpty()){
                scheduleNextRun();
                writeCompletionFile();
            }
            
            logSessionStats();
            log.info("Session ended.");
        }finally {
            timer.cancel();
        }
    }

    /**
     * Writes zero length file to the work directory to mark the completion of a run.
     */
    private void writeCompletionFile() {
        File completionFile = getCompletionFile();
        try {
            if(completionFile.createNewFile()){
                log.info("successfully created completion marker file: {}",
                         completionFile.getAbsolutePath());<commit_msg>Adds notification for looping producer failures.<commit_after>/*
 * The contents of this file are subject to the license and copyright
 * detailed in the LICENSE and NOTICE files at the root of the source
 * tree and available online at
 *
 *     http://duracloud.org/license/
 */
package org.duracloud.mill.ltp;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;

import org.duracloud.common.queue.TaskQueue;
import org.duracloud.mill.common.storageprovider.StorageProviderFactory;
import org.duracloud.mill.credentials.CredentialsRepo;
import org.duracloud.mill.credentials.CredentialsRepoException;
import org.duracloud.mill.credentials.StorageProviderCredentials;
import org.duracloud.mill.notification.NotificationManager;
import org.duracloud.storage.provider.StorageProvider;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This class is responsible for filling the duplication <code>TaskQueue</code>
 * by looping through all duplication policies for all accounts and spaces and
 * blindly creates duplication tasks. It will create tasks for all items in the
 * the source destination as well as all items in the destination provider but
 * not in the source provider. A notable feature of this task producer is that
 * it attempts to respect a designated maximum task queue size. Once the limit
 * has been reached, the producer will stop. On subsequent runs, the producer
 * will pick up where it left off, starting with the next account,space,set of
 * content items, and duplication store policy. If all content items are visited
 * within a single run before the task queue limit has been reached, the
 * producer will exit.
 * 
 * For more information about how this process fits into the whole system of
 * collaborating components, see
 * https://wiki.duraspace.org/display/DSPINT/DuraCloud+Duplication+-+System+Overview
 * 
 * @author <NAME> Date: Nov 5, 2013
 */
public abstract class LoopingTaskProducer<T extends Morsel> implements Runnable {
    private static Logger log = LoggerFactory.getLogger(LoopingTaskProducer.class);
    private TaskQueue taskQueue;
    private CredentialsRepo credentialsRepo;
    private StateManager<T> stateManager;
    private int maxTaskQueueSize;
    private StorageProviderFactory storageProviderFactory;
    private List<T> morselsToReload = new LinkedList<>();
    private Frequency frequency;
    private RunStats cumulativeTotals;
    private NotificationManager notificationManager;
    private LoopingTaskProducerConfigurationManager config;
    private Map<String,RunStats> runstats = new HashMap<>();

    public LoopingTaskProducer(CredentialsRepo credentialsRepo,
                               StorageProviderFactory storageProviderFactory,
                               TaskQueue taskQueue,
                               StateManager<T> state,
                               int maxTaskQueueSize, 
                               Frequency frequency,
                               NotificationManager notificationManager, 
                               LoopingTaskProducerConfigurationManager config) {
        
        this.credentialsRepo = credentialsRepo;
        this.storageProviderFactory = storageProviderFactory;
        this.taskQueue = taskQueue;
        this.stateManager = state;
        this.credentialsRepo = credentialsRepo;
        this.maxTaskQueueSize = maxTaskQueueSize;
        this.frequency = frequency;
        this.cumulativeTotals = createRunStats();
        this.notificationManager = notificationManager;
        this.config = config;
    }
    
    protected Frequency getFrequency(){
        return this.frequency;
    }
    


    protected CredentialsRepo getCredentialsRepo() {
        return credentialsRepo;
    }


    protected TaskQueue getTaskQueue() {
        return taskQueue;
    }

    protected int getMaxTaskQueueSize() {
        return maxTaskQueueSize;
    }
    
    public void run(){
        
        
        Timer timer = new Timer();
        try {
            
            timer.scheduleAtFixedRate(new TimerTask() {
                @Override
                public void run() {
                    logSessionStats();
                }

            }, 5 * 60 * 1000, 5 * 60 * 1000);  
            
            if(runLater()){
                return;
            }

            log.info("Starting run...");
            Queue<T> morselQueue = loadMorselQueue();
            
            while(!morselQueue.isEmpty() && this.taskQueue.size() < maxTaskQueueSize){
                T morsel = morselQueue.peek();
                nibble(morselQueue);
                persistMorsels(morselQueue, morselsToReload);
               
                if(morselQueue.isEmpty()){
                    morselQueue = reloadMorselQueue();
                }else{
                    //break if nothing was removed from the queue
                    //if nothing was removed from the queue we can assume
                    //that the for whatever reason the morsel could not be processed
                    //at this time, so the process should wait for the next run.
                    if(morsel.equals(morselQueue.peek())){
                        break;
                    }
                }
                
               
            }
    
            if(morselQueue.isEmpty()){
                scheduleNextRun();
                writeCompletionFile();
            }
            
            logSessionStats();
            log.info("Session ended.");
        }catch(Exception ex){
            log.error("failed to complete run: " + ex.getMessage(), ex);
            this.notificationManager.sendEmail("failed to complete bit producer run" , ex.getMessage());
        }finally {
            timer.cancel();
        }
    }

    /**
     * Writes zero length file to the work directory to mark the completion of a run.
     */
    private void writeCompletionFile() {
        File completionFile = getCompletionFile();
        try {
            if(completionFile.createNewFile()){
                log.info("successfully created completion marker file: {}",
                         completionFile.getAbsolutePath());
{'fraction_non_alphanumeric': 0.07269016406311328, 'fraction_numerical': 0.003296962084936023, 'mean_word_length': 2.2302231237322516, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 25, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '833449', 'n_tokens_mistral': 3205, 'n_tokens_neox': 3080, 'n_words': 939}