<commit_before>/* File:    $Id$
 * Revision: $Revision$
 * Author:   $Author$
 * Date:     $Date$
 *
 * The Netarchive Suite - Software to harvest and preserve websites
 * Copyright 2004-2012 The Royal Danish Library, the Danish State and 
 * University Library, the National Library of France and the Austrian 
 * National Library.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation,dk.netarkivet.harvester.schedulerFloor, Boston, MA  02110-1301  USA
 */
package dk.netarkivet.harvester.scheduler;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import dk.netarkivet.common.distribute.JMSConnection;
import dk.netarkivet.common.exceptions.ArgumentNotValid;
import dk.netarkivet.common.exceptions.IOFailure;
import dk.netarkivet.common.utils.ExceptionUtils;
import dk.netarkivet.harvester.datamodel.HarvestDefinitionDAO;
import dk.netarkivet.harvester.datamodel.Job;
import dk.netarkivet.harvester.datamodel.JobDAO;
import dk.netarkivet.harvester.datamodel.JobPriority;
import dk.netarkivet.harvester.datamodel.JobStatus;
import dk.netarkivet.harvester.datamodel.SparseFullHarvest;
import dk.netarkivet.harvester.datamodel.SparsePartialHarvest;
import dk.netarkivet.harvester.harvesting.HeritrixLauncher;
import dk.netarkivet.harvester.harvesting.distribute.DoOneCrawlMessage;
import dk.netarkivet.harvester.harvesting.distribute.JobChannelUtil;
import dk.netarkivet.harvester.harvesting.distribute.MetadataEntry;
import dk.netarkivet.harvester.harvesting.distribute.PersistentJobData.HarvestDefinitionInfo;

/**
 * This class handles dispatching of Harvest jobs to the Harvesters.
 */
public class JobDispatcher {
    /** The logger to use.    */
    private final Log log = LogFactory.getLog(getClass());
    /** Connection to JMS provider. */
    private JMSConnection jmsConnection;
    
    /**
     * @param jmsConnection The JMS connection to use.
     */
    public JobDispatcher(JMSConnection jmsConnection) {
        log.info("Creating JobDispatcher");
        ArgumentNotValid.checkNotNull(jmsConnection, "jmsConnection");
        this.jmsConnection = jmsConnection;
    }

    /**
     * Submit the next new job (the one with the lowest ID) with the given
     * priority, and updates the internal counter as needed. If no jobs are 
     * ready for the given priority, nothing is done
     * @param priority the job priority
     */
    protected void submitNextNewJob(JobPriority priority) {
        final JobDAO dao = JobDAO.getInstance();
        Iterator<Long> jobsToSubmit = dao.getAllJobIds(JobStatus.NEW, priority);
        if (!jobsToSubmit.hasNext()) {
            if (log.isTraceEnabled()) {
                log.trace("No " + priority + " jobs to be run at this time");
            }
        } else {
            if (log.isDebugEnabled()) {
                log.debug("Submitting new " + priority + " job");
            }
            final long jobID = jobsToSubmit.next();
            Job jobToSubmit = null;
            try {
                jobToSubmit = dao.read(jobID);

                jobToSubmit.setStatus(JobStatus.SUBMITTED);
                jobToSubmit.setSubmittedDate(new Date());
                dao.update(jobToSubmit);
                
                List<MetadataEntry> metadata = createMetadata(jobToSubmit);
                
                // Extract documentary information about the harvest
                HarvestDefinitionDAO hDao = HarvestDefinitionDAO.getInstance();
                String hName = hDao.getHarvestName(
                        jobToSubmit.getOrigHarvestDefinitionID());

                String schedule = "";
                String hdComments = "";
                SparseFullHarvest fh = hDao.getSparseFullHarvest(hName);
                if (fh != null) {
                    hdComments = fh.getComments();
                } else {
                    SparsePartialHarvest ph =
                            hDao.getSparsePartialHarvest(hName);

                    if (ph == null) {
                        throw new ArgumentNotValid("No harvest definition "
                                + "found for id '"
                                + jobToSubmit.getOrigHarvestDefinitionID()
                                + "', named '" + hName + "'");
                    }

                    // The schedule name can only be documented for
                    // focused crawls.
                    schedule = ph.getScheduleName();

                    hdComments = ph.getComments();
                }

                doOneCrawl(jobToSubmit, hName, hdComments, schedule, metadata);

                log.info("Submitting job: " + jobToSubmit);

            } catch (Throwable e) {
                String message = "Error while dispatching job " + jobID
                        + ". Job status changed to FAILED";
                log.warn(message, e);
                if (jobToSubmit != null) {
                    jobToSubmit.setStatus(JobStatus.FAILED);
                    jobToSubmit.appendHarvestErrors(message);
                    jobToSubmit.appendHarvestErrorDetails(
                            ExceptionUtils.getStackTrace(e));
                    dao.update(jobToSubmit);
                }
            }<commit_msg>NAS-2031: Harvest Jobs in state submitted can accumulate: Added synchronization to the test for newjobs and marking as submitted.<commit_after>/* File:    $Id$
 * Revision: $Revision$
 * Author:   $Author$
 * Date:     $Date$
 *
 * The Netarchive Suite - Software to harvest and preserve websites
 * Copyright 2004-2012 The Royal Danish Library, the Danish State and 
 * University Library, the National Library of France and the Austrian 
 * National Library.
 *
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or (at your option) any later version.
 *
 * This library is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 * Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation,dk.netarkivet.harvester.schedulerFloor, Boston, MA  02110-1301  USA
 */
package dk.netarkivet.harvester.scheduler;

import java.util.ArrayList;
import java.util.Date;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import dk.netarkivet.common.distribute.JMSConnection;
import dk.netarkivet.common.exceptions.ArgumentNotValid;
import dk.netarkivet.common.exceptions.IOFailure;
import dk.netarkivet.common.utils.ExceptionUtils;
import dk.netarkivet.harvester.datamodel.HarvestDefinitionDAO;
import dk.netarkivet.harvester.datamodel.Job;
import dk.netarkivet.harvester.datamodel.JobDAO;
import dk.netarkivet.harvester.datamodel.JobPriority;
import dk.netarkivet.harvester.datamodel.JobStatus;
import dk.netarkivet.harvester.datamodel.SparseFullHarvest;
import dk.netarkivet.harvester.datamodel.SparsePartialHarvest;
import dk.netarkivet.harvester.harvesting.HeritrixLauncher;
import dk.netarkivet.harvester.harvesting.distribute.DoOneCrawlMessage;
import dk.netarkivet.harvester.harvesting.distribute.JobChannelUtil;
import dk.netarkivet.harvester.harvesting.distribute.MetadataEntry;
import dk.netarkivet.harvester.harvesting.distribute.PersistentJobData.HarvestDefinitionInfo;

/**
 * This class handles dispatching of Harvest jobs to the Harvesters.
 */
public class JobDispatcher {
    /** The logger to use.    */
    private final Log log = LogFactory.getLog(getClass());
    /** Connection to JMS provider. */
    private JMSConnection jmsConnection;
    /** For jobDB access. */
    private JobDAO jobDao;
    
    /**
     * @param jmsConnection The JMS connection to use.
     */
    public JobDispatcher(JMSConnection jmsConnection, JobDAO dao) {
        log.info("Creating JobDispatcher");
        ArgumentNotValid.checkNotNull(jmsConnection, "jmsConnection");
        this.jmsConnection = jmsConnection;
    }

    /**
     * Submit the next new job (the one with the lowest ID) with the given
     * priority, and updates the internal counter as needed. If no jobs are 
     * ready for the given priority, nothing is done
     * @param priority the job priority
     */
    protected void submitNextNewJob(JobPriority priority) {
        final JobDAO dao = JobDAO.getInstance();
        Job jobToSubmit = prepareNextJobForSubmission(priority);
        if (jobToSubmit == null) {
            if (log.isTraceEnabled()) {
                log.trace("No " + priority + " jobs to be run at this time");
            }
        } else {
            if (log.isDebugEnabled()) {
                log.debug("Submitting new " + priority + " job");
            }
            try {
                List<MetadataEntry> metadata = createMetadata(jobToSubmit);
                
                // Extract documentary information about the harvest
                HarvestDefinitionDAO hDao = HarvestDefinitionDAO.getInstance();
                String hName = hDao.getHarvestName(
                        jobToSubmit.getOrigHarvestDefinitionID());

                String schedule = "";
                String hdComments = "";
                SparseFullHarvest fh = hDao.getSparseFullHarvest(hName);
                if (fh != null) {
                    hdComments = fh.getComments();
                } else {
                    SparsePartialHarvest ph =
                            hDao.getSparsePartialHarvest(hName);

                    if (ph == null) {
                        throw new ArgumentNotValid("No harvest definition "
                                + "found for id '"
                                + jobToSubmit.getOrigHarvestDefinitionID()
                                + "', named '" + hName + "'");
                    }

                    // The schedule name can only be documented for
                    // selective crawls.
                    schedule = ph.getScheduleName();

                    hdComments = ph.getComments();
                }

                doOneCrawl(jobToSubmit, hName, hdComments, schedule, metadata);

                log.info("Submitting job: " + jobToSubmit);

            } catch (Throwable e) {
                String message = "Error while dispatching job " + 
                        jobToSubmit.getJobID()
                        + ". Job status changed to FAILED";
                log.warn(message, e);
                if (jobToSubmit != null) {
                    jobToSubmit.setStatus(JobStatus.FAILED);
                    jobToSubmit.appendHarvestErrors(message);
                    jobToSubmit.appendHarvestErrorDetails(
                            ExceptionUtils.getStackTrace(e));
                    dao.update(jobToSubmit);
                }
            }
        }
    }
    
    /**
     * Will read the next job ready to run from the db and set the job to 
     * submitted. If no job are ready, null will be returned.
     * 
     * Note the operation is synchronized, so only one thread may start the 
     * submission of a job.
     * @param priority the job priority.
     * @return The job prepared for submission.
     */
    private synchronized Job prepareNextJobForSubmission(JobPriority priority) {
        final JobDAO dao = JobDAO.getInstance();
        Iterator<Long> jobsToSubmit = dao.getAllJobIds(JobStatus.NEW, priority);
        if (!jobsToSubmit.hasNext()) {
            return null;
        } else {
            if (log.isDebugEnabled()) {
                log.debug("Submitting new " + priority + " job");
            }
            final long jobID = jobsToSubmit.next();
            Job jobToSubmit = null;
            jobToSubmit = dao.read(jobID);

            jobToSubmit.setStatus(JobStatus.SUBMITTED);
            jobToSubmit.setSubmittedDate(new Date());
            dao.update(jobToSubmit);
            return jobToSubmit;
{'fraction_non_alphanumeric': 0.08244866330879504, 'fraction_numerical': 0.003254552499031383, 'mean_word_length': 2.2378324134470646, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '977265', 'n_tokens_mistral': 3600, 'n_tokens_neox': 3373, 'n_words': 943}