incubator-connectors-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kwri...@apache.org
Subject svn commit: r1059695 - in /incubator/lcf/trunk: framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/ framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/ framework/pull-agent/src/main/java/org/apache/manifoldcf...
Date Sun, 16 Jan 2011 22:30:45 GMT
Author: kwright
Date: Sun Jan 16 22:30:45 2011
New Revision: 1059695

URL: http://svn.apache.org/viewvc?rev=1059695&view=rev
Log:
More of CONNECTORS-145.  Reconsider the delete model, because it turns out that the checktime field cannot be used for document states that are in use for crawling.

Added:
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java   (with props)
Modified:
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/DocumentDeleteThread.java
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
    incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
    incubator/lcf/trunk/site/src/documentation/content/xdocs/how-to-build-and-deploy.xml

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java Sun Jan 16 22:30:45 2011
@@ -163,6 +163,11 @@ public interface IJobManager
   public void resetDocCleanupWorkerStatus()
     throws ManifoldCFException;
 
+  /** Reset as part of restoring delete startup threads.
+  */
+  public void resetDeleteStartupWorkerStatus()
+    throws ManifoldCFException;
+
   /** Reset as part of restoring startup threads.
   */
   public void resetStartupWorkerStatus()
@@ -643,6 +648,12 @@ public interface IJobManager
   public void resetSeedJob(Long jobID)
     throws ManifoldCFException;
 
+  /** Get the list of jobs that are ready for deletion.
+  *@return jobs that were in the "readyfordelete" state.
+  */
+  public JobStartRecord[] getJobsReadyForDelete()
+    throws ManifoldCFException;
+    
   /** Get the list of jobs that are ready for startup.
   *@return jobs that were in the "readyforstartup" state.  These will be marked as being in the "starting up" state.
   */
@@ -660,15 +671,29 @@ public interface IJobManager
   */
   public void inactivateJob(Long jobID)
     throws ManifoldCFException;
-    
+
+  /** Reset a job starting for delete back to "ready for delete"
+  * state.
+  *@param jobID is the job id.
+  */
+  public void resetStartDeleteJob(Long jobID)
+    throws ManifoldCFException;
+
   /** Reset a starting job back to "ready for startup" state.
   *@param jobID is the job id.
   */
   public void resetStartupJob(Long jobID)
     throws ManifoldCFException;
 
+  /** Prepare for a delete scan.
+  *@param jobID is the job id.
+  */
+  public void prepareDeleteScan(Long jobID)
+    throws ManifoldCFException;
+
   /** Prepare for a full scan.
   *@param jobID is the job id.
+  *@param legalLinkTypes are the link types allowed for the job.
   *@param hopcountMethod describes how to handle deletions for hopcount purposes.
   */
   public void prepareFullScan(Long jobID, String[] legalLinkTypes, int hopcountMethod)
@@ -676,11 +701,19 @@ public interface IJobManager
 
   /** Prepare for an incremental scan.
   *@param jobID is the job id.
+  *@param legalLinkTypes are the link types allowed for the job.
   *@param hopcountMethod describes how to handle deletions for hopcount purposes.
   */
   public void prepareIncrementalScan(Long jobID, String[] legalLinkTypes, int hopcountMethod)
     throws ManifoldCFException;
 
+  /** Note job delete started.
+  *@param jobID is the job id.
+  *@param startTime is the job start time.
+  */
+  public void noteJobDeleteStarted(Long jobID, long startTime)
+    throws ManifoldCFException;
+
   /** Note job started.
   *@param jobID is the job id.
   *@param startTime is the job start time.

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobManager.java Sun Jan 16 22:30:45 2011
@@ -520,12 +520,11 @@ public class JobManager implements IJobM
       // If the job is running, throw an error
       ArrayList list = new ArrayList();
       list.add(id);
-      IResultSet set = database.performQuery("SELECT "+jobs.statusField+","+jobs.outputNameField+" FROM "+
+      IResultSet set = database.performQuery("SELECT "+jobs.statusField+" FROM "+
         jobs.getTableName()+" WHERE "+jobs.idField+"=? FOR UPDATE",list,null,null);
       if (set.getRowCount() == 0)
         throw new ManifoldCFException("Attempting to delete a job that doesn't exist: "+id);
       IResultRow row = set.getRow(0);
-      String outputName = (String)row.getValue(jobs.outputNameField);
       int status = jobs.stringToStatus(row.getValue(jobs.statusField).toString());
       if (status == jobs.STATUS_ACTIVE || status == jobs.STATUS_ACTIVESEEDING ||
         status == jobs.STATUS_ACTIVE_UNINSTALLED || status == jobs.STATUS_ACTIVESEEDING_UNINSTALLED ||
@@ -534,10 +533,7 @@ public class JobManager implements IJobM
       throw new ManifoldCFException("Job "+id+" is active; you must shut it down before deleting it");
       if (status != jobs.STATUS_INACTIVE)
         throw new ManifoldCFException("Job "+id+" is busy; you must wait and/or shut it down before deleting it");
-      if (outputMgr.checkConnectorExists(outputName))
-        jobs.writeStatus(id,jobs.STATUS_READYFORDELETE);
-      else
-        jobs.writeStatus(id,jobs.STATUS_READYFORDELETE_NOOUTPUT);
+      jobs.writeStatus(id,jobs.STATUS_READYFORDELETE);
       if (Logging.jobs.isDebugEnabled())
         Logging.jobs.debug("Job "+id+" marked for deletion");
     }
@@ -750,6 +746,16 @@ public class JobManager implements IJobM
     Logging.jobs.debug("Reset complete");
   }
 
+  /** Reset as part of restoring delete startup threads.
+  */
+  public void resetDeleteStartupWorkerStatus()
+    throws ManifoldCFException
+  {
+    Logging.jobs.debug("Resetting job delete starting up status");
+    jobs.resetDeleteStartupWorkerStatus();
+    Logging.jobs.debug("Reset complete");
+  }
+
   /** Reset as part of restoring startup threads.
   */
   public void resetStartupWorkerStatus()
@@ -1061,13 +1067,11 @@ public class JobManager implements IJobM
         // Note: This query does not do "FOR UPDATE", because it is running under the only thread that can possibly change the document's state to "being deleted".
         // If FOR UPDATE was included, deadlock happened a lot.
         ArrayList list = new ArrayList();
-        list.add(jobQueue.statusToString(jobQueue.STATUS_COMPLETE));
-        list.add(jobQueue.statusToString(jobQueue.STATUS_PURGATORY));
-        list.add(jobQueue.statusToString(jobQueue.STATUS_PENDINGPURGATORY));
+        list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
         
         list.add(new Long(currentTime));
         
-        list.add(jobs.statusToString(jobs.STATUS_READYFORDELETE));
+        list.add(jobs.statusToString(jobs.STATUS_DELETING));
         
         list.add(jobQueue.statusToString(jobQueue.STATUS_ACTIVE));
         list.add(jobQueue.statusToString(jobQueue.STATUS_ACTIVEPURGATORY));
@@ -1079,8 +1083,8 @@ public class JobManager implements IJobM
         // The checktime is null field check is for backwards compatibility
         IResultSet set = database.performQuery("SELECT "+jobQueue.idField+","+jobQueue.jobIDField+","+jobQueue.docHashField+","+jobQueue.docIDField+","+
           jobQueue.failTimeField+","+jobQueue.failCountField+" FROM "+
-          jobQueue.getTableName()+" t0 WHERE t0."+jobQueue.statusField+" IN (?,?,?) "+
-          " AND (t0."+jobQueue.checkTimeField+" IS NULL OR t0."+jobQueue.checkTimeField+"<=?) "+
+          jobQueue.getTableName()+" t0 WHERE t0."+jobQueue.statusField+"=? "+
+          " AND t0."+jobQueue.checkTimeField+"<=? "+
           " AND EXISTS(SELECT 'x' FROM "+jobs.getTableName()+" t1 WHERE t0."+jobQueue.jobIDField+"=t1."+jobs.idField+
           " AND t1."+jobs.statusField+"=?"+
           ") AND NOT EXISTS(SELECT 'x' FROM "+jobQueue.getTableName()+" t2 WHERE t0."+jobQueue.docHashField+"=t2."+
@@ -1312,11 +1316,12 @@ public class JobManager implements IJobM
     list.add(jobQueue.statusToString(jobQueue.STATUS_PURGATORY));
     list.add(jobQueue.statusToString(jobQueue.STATUS_PENDINGPURGATORY));
     list.add(jobQueue.statusToString(jobQueue.STATUS_COMPLETE));
+    list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
     
     list.add(connectionName);
     list.add(outputConnectionName);
     
-    sb.append(") AND t0.").append(jobQueue.statusField).append(" IN (?,?,?) AND EXISTS(SELECT 'x' FROM ")
+    sb.append(") AND t0.").append(jobQueue.statusField).append(" IN (?,?,?,?) AND EXISTS(SELECT 'x' FROM ")
       .append(jobs.getTableName()).append(" t1 WHERE t0.")
       .append(jobQueue.jobIDField).append("=t1.").append(jobs.idField).append(" AND t1.")
       .append(jobs.connectionNameField).append("=? AND t1.")
@@ -4903,6 +4908,18 @@ public class JobManager implements IJobM
     }
   }
 
+  /** Note job delete started.
+  *@param jobID is the job id.
+  *@param startTime is the job delete start time.
+  */
+  public void noteJobDeleteStarted(Long jobID, long startTime)
+    throws ManifoldCFException
+  {
+    jobs.noteJobDeleteStarted(jobID,startTime);
+    if (Logging.jobs.isDebugEnabled())
+      Logging.jobs.debug("Job "+jobID+" delete is now started");
+  }
+
   /** Note job started.
   *@param jobID is the job id.
   *@param startTime is the job start time.
@@ -4927,8 +4944,20 @@ public class JobManager implements IJobM
       Logging.jobs.debug("Job "+jobID+" has been successfully reseeded");
   }
 
+  /** Prepare for a delete scan.
+  *@param jobID is the job id.
+  */
+  public void prepareDeleteScan(Long jobID)
+    throws ManifoldCFException
+  {
+    // No special treatment needed for hopcount or carrydown, since these all get deleted at once
+    // at the end of the job delete process.
+    jobQueue.prepareDeleteScan(jobID);
+  }
+  
   /** Prepare for a full scan.
   *@param jobID is the job id.
+  *@param legalLinkTypes are the link types allowed for the job.
   *@param hopcountMethod describes how to handle deletions for hopcount purposes.
   */
   public void prepareFullScan(Long jobID, String[] legalLinkTypes, int hopcountMethod)
@@ -4982,6 +5011,7 @@ public class JobManager implements IJobM
 
   /** Prepare for an incremental scan.
   *@param jobID is the job id.
+  *@param legalLinkTypes are the link types allowed for the job.
   *@param hopcountMethod describes how to handle deletions for hopcount purposes.
   */
   public void prepareIncrementalScan(Long jobID, String[] legalLinkTypes, int hopcountMethod)
@@ -5189,6 +5219,68 @@ public class JobManager implements IJobM
     }
   }
 
+  /** Get the list of jobs that are ready for deletion.
+  *@return jobs that were in the "readyfordelete" state.
+  */
+  public JobStartRecord[] getJobsReadyForDelete()
+    throws ManifoldCFException
+  {
+    while (true)
+    {
+      long sleepAmt = 0L;
+      database.beginTransaction();
+      try
+      {
+        // Do the query
+        ArrayList list = new ArrayList();
+        list.add(jobs.statusToString(jobs.STATUS_READYFORDELETE));
+        IResultSet set = database.performQuery("SELECT "+jobs.idField+" FROM "+
+          jobs.getTableName()+" WHERE "+jobs.statusField+"=? FOR UPDATE",list,null,null);
+        // Update them all
+        JobStartRecord[] rval = new JobStartRecord[set.getRowCount()];
+        int i = 0;
+        while (i < rval.length)
+        {
+          IResultRow row = set.getRow(i);
+          Long jobID = (Long)row.getValue(jobs.idField);
+
+          // Mark status of job as "starting delete"
+          jobs.writeStatus(jobID,jobs.STATUS_DELETESTARTINGUP);
+          if (Logging.jobs.isDebugEnabled())
+          {
+            Logging.jobs.debug("Marked job "+jobID+" for delete startup");
+          }
+
+          rval[i] = new JobStartRecord(jobID,0L);
+          i++;
+        }
+        return rval;
+      }
+      catch (ManifoldCFException e)
+      {
+        database.signalRollback();
+        if (e.getErrorCode() == e.DATABASE_TRANSACTION_ABORT)
+        {
+          if (Logging.perf.isDebugEnabled())
+            Logging.perf.debug("Aborted getting jobs ready for startup: "+e.getMessage());
+          sleepAmt = getRandomAmount();
+          continue;
+        }
+        throw e;
+      }
+      catch (Error e)
+      {
+        database.signalRollback();
+        throw e;
+      }
+      finally
+      {
+        database.endTransaction();
+        sleepFor(sleepAmt);
+      }
+    }
+  }
+
   /** Get the list of jobs that are ready for startup.
   *@return jobs that were in the "readyforstartup" state.  These will be marked as being in the "starting up" state.
   */
@@ -5314,6 +5406,68 @@ public class JobManager implements IJobM
     }
   }
 
+  /** Reset a job starting for delete back to "ready for delete"
+  * state.
+  *@param jobID is the job id.
+  */
+  public void resetStartDeleteJob(Long jobID)
+    throws ManifoldCFException
+  {
+    while (true)
+    {
+      long sleepAmt = 0L;
+      database.beginTransaction();
+      try
+      {
+        // Check job status
+        ArrayList list = new ArrayList();
+        list.add(jobID);
+        IResultSet set = database.performQuery("SELECT "+jobs.statusField+" FROM "+jobs.getTableName()+
+          " WHERE "+jobs.idField+"=? FOR UPDATE",list,null,null);
+        if (set.getRowCount() == 0)
+          throw new ManifoldCFException("No such job: "+jobID);
+        IResultRow row = set.getRow(0);
+        int status = jobs.stringToStatus((String)row.getValue(jobs.statusField));
+
+        switch (status)
+        {
+        case Jobs.STATUS_DELETESTARTINGUP:
+          if (Logging.jobs.isDebugEnabled())
+            Logging.jobs.debug("Setting job "+jobID+" back to 'ReadyForDelete' state");
+
+          // Set the state of the job back to "ReadyForStartup"
+          jobs.writeStatus(jobID,jobs.STATUS_READYFORDELETE);
+          break;
+        default:
+          throw new ManifoldCFException("Unexpected job status: "+Integer.toString(status));
+        }
+        return;
+      }
+      catch (ManifoldCFException e)
+      {
+        database.signalRollback();
+        if (e.getErrorCode() == e.DATABASE_TRANSACTION_ABORT)
+        {
+          if (Logging.perf.isDebugEnabled())
+            Logging.perf.debug("Aborted resetting start delete job: "+e.getMessage());
+          sleepAmt = getRandomAmount();
+          continue;
+        }
+        throw e;
+      }
+      catch (Error e)
+      {
+        database.signalRollback();
+        throw e;
+      }
+      finally
+      {
+        database.endTransaction();
+        sleepFor(sleepAmt);
+      }
+    }
+  }
+
   /** Reset a starting job back to "ready for startup" state.
   *@param jobID is the job id.
   */
@@ -5551,10 +5705,9 @@ public class JobManager implements IJobM
 
         // Do the first query, getting the candidate jobs to be considered
         ArrayList list = new ArrayList();
-        list.add(jobs.statusToString(jobs.STATUS_READYFORDELETE));
-        list.add(jobs.statusToString(jobs.STATUS_READYFORDELETE_NOOUTPUT));
+        list.add(jobs.statusToString(jobs.STATUS_DELETING));
         IResultSet set = database.performQuery("SELECT "+jobs.idField+" FROM "+
-          jobs.getTableName()+" WHERE "+jobs.statusField+" IN(?,?) FOR UPDATE",list,null,null);
+          jobs.getTableName()+" WHERE "+jobs.statusField+"=? FOR UPDATE",list,null,null);
 
         // Now, loop through this list.  For each one, verify that it's okay to delete it
         int i = 0;
@@ -5565,23 +5718,12 @@ public class JobManager implements IJobM
 
           list.clear();
           list.add(jobID);
-          list.add(jobQueue.statusToString(jobQueue.STATUS_COMPLETE));
-          list.add(jobID);
-          list.add(jobQueue.statusToString(jobQueue.STATUS_PURGATORY));
-          list.add(jobID);
-          list.add(jobQueue.statusToString(jobQueue.STATUS_PENDINGPURGATORY));
-          list.add(jobID);
+          list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
           list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGDELETED));
-          list.add(jobID);
-          list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED));
 
           IResultSet confirmSet = database.performQuery("SELECT "+jobQueue.idField+" FROM "+
             jobQueue.getTableName()+" WHERE "+
-            "("+jobQueue.jobIDField+"=? AND "+jobQueue.statusField+"=?) OR "+
-            "("+jobQueue.jobIDField+"=? AND "+jobQueue.statusField+"=?) OR "+
-            "("+jobQueue.jobIDField+"=? AND "+jobQueue.statusField+"=?) OR "+
-            "("+jobQueue.jobIDField+"=? AND "+jobQueue.statusField+"=?) OR "+
-            "("+jobQueue.jobIDField+"=? AND "+jobQueue.statusField+"=?) "+database.constructOffsetLimitClause(0,1),list,null,null,1,null);
+            jobQueue.jobIDField+"=? AND "+jobQueue.statusField+" IN (?,?) "+database.constructOffsetLimitClause(0,1),list,null,null,1,null);
 
           if (confirmSet.getRowCount() > 0)
             continue;
@@ -6209,7 +6351,8 @@ public class JobManager implements IJobM
         rstatus = JobStatus.JOBSTATUS_STARTING;
         break;
       case Jobs.STATUS_READYFORDELETE:
-      case Jobs.STATUS_READYFORDELETE_NOOUTPUT:
+      case Jobs.STATUS_DELETING:
+      case Jobs.STATUS_DELETING_NOOUTPUT:
         rstatus = JobStatus.JOBSTATUS_DESTRUCTING;
         break;
       default:
@@ -6258,6 +6401,7 @@ public class JobManager implements IJobM
     list.add(jobQueue.statusToString(jobQueue.STATUS_PURGATORY));
     list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGDELETED));
     list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED));
+    list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
     
     list.add(jobQueue.statusToString(jobQueue.STATUS_COMPLETE));
     list.add(jobQueue.statusToString(jobQueue.STATUS_PURGATORY));
@@ -6313,6 +6457,7 @@ public class JobManager implements IJobM
       .append(" WHEN ").append("t0.").append(jobQueue.statusField).append("=? THEN 'Processed'")
       .append(" WHEN ").append("t0.").append(jobQueue.statusField).append("=? THEN 'Being removed'")
       .append(" WHEN ").append("t0.").append(jobQueue.statusField).append("=? THEN 'Being removed'")
+      .append(" WHEN ").append("t0.").append(jobQueue.statusField).append("=? THEN 'Being removed'")
       .append(" ELSE 'Unknown'")
       .append(" END AS state,")
       .append("CASE")
@@ -6441,6 +6586,7 @@ public class JobManager implements IJobM
     
     list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGDELETED));
     list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED));
+    list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
     
     list.add(jobQueue.actionToString(jobQueue.ACTION_RESCAN));
     list.add(jobQueue.statusToString(jobQueue.STATUS_PENDING));
@@ -6495,6 +6641,7 @@ public class JobManager implements IJobM
       .append(" WHEN ")
       .append(jobQueue.statusField).append("=?")
       .append(" OR ").append(jobQueue.statusField).append("=?")
+      .append(" OR ").append(jobQueue.statusField).append("=?")
       .append(" THEN 1 ELSE 0")
       .append(" END")
       .append(" as deleting,")
@@ -6636,11 +6783,12 @@ public class JobManager implements IJobM
         list.add(jobQueue.statusToString(jobQueue.STATUS_PENDINGPURGATORY));
         list.add(jobQueue.statusToString(jobQueue.STATUS_ACTIVEPURGATORY));
         list.add(jobQueue.statusToString(jobQueue.STATUS_ACTIVENEEDRESCANPURGATORY));
+        list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
         list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGDELETED));
         list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED));
         list.add(jobQueue.statusToString(jobQueue.STATUS_COMPLETE));
         list.add(jobQueue.statusToString(jobQueue.STATUS_PURGATORY));
-        sb.append(fieldPrefix).append(jobQueue.statusField).append(" IN (?,?,?,?,?,?,?)");
+        sb.append(fieldPrefix).append(jobQueue.statusField).append(" IN (?,?,?,?,?,?,?,?)");
         break;
       }
       k++;
@@ -6683,7 +6831,8 @@ public class JobManager implements IJobM
       case DOCSTATUS_DELETING:
         list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGDELETED));
         list.add(jobQueue.statusToString(jobQueue.STATUS_BEINGCLEANED));
-        sb.append(fieldPrefix).append(jobQueue.statusField).append(" IN (?,?)");
+        list.add(jobQueue.statusToString(jobQueue.STATUS_ELIGIBLEFORDELETE));
+        sb.append(fieldPrefix).append(jobQueue.statusField).append(" IN (?,?,?)");
         break;
       case DOCSTATUS_READYFORPROCESSING:
         list.add(jobQueue.statusToString(jobQueue.STATUS_PENDING));

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/JobQueue.java Sun Jan 16 22:30:45 2011
@@ -46,6 +46,7 @@ public class JobQueue extends org.apache
   public final static int STATUS_ACTIVENEEDRESCAN = 7;
   public final static int STATUS_ACTIVENEEDRESCANPURGATORY = 8;
   public final static int STATUS_BEINGCLEANED = 9;
+  public final static int STATUS_ELIGIBLEFORDELETE = 10;
 
   // Action values
   public final static int ACTION_RESCAN = 0;
@@ -99,6 +100,7 @@ public class JobQueue extends org.apache
     statusMap.put("G",new Integer(STATUS_PENDINGPURGATORY));
     statusMap.put("F",new Integer(STATUS_ACTIVEPURGATORY));
     statusMap.put("Z",new Integer(STATUS_PURGATORY));
+    statusMap.put("E",new Integer(STATUS_ELIGIBLEFORDELETE));
     statusMap.put("D",new Integer(STATUS_BEINGDELETED));
     statusMap.put("a",new Integer(STATUS_ACTIVENEEDRESCAN));
     statusMap.put("f",new Integer(STATUS_ACTIVENEEDRESCANPURGATORY));
@@ -306,8 +308,8 @@ public class JobQueue extends org.apache
     list.add(statusToString(STATUS_ACTIVENEEDRESCANPURGATORY));
     performUpdate(map,"WHERE "+statusField+"=? OR "+statusField+"=?",list,null);
 
-    // Map BEINGDELETED to COMPLETE
-    map.put(statusField,statusToString(STATUS_COMPLETE));
+    // Map BEINGDELETED to ELIGIBLEFORDELETE
+    map.put(statusField,statusToString(STATUS_ELIGIBLEFORDELETE));
     map.put(checkTimeField,new Long(0L));
     list.clear();
     list.add(statusToString(STATUS_BEINGDELETED));
@@ -382,8 +384,8 @@ public class JobQueue extends org.apache
   {
     HashMap map = new HashMap();
     ArrayList list = new ArrayList();
-    // Map BEINGDELETED to COMPLETE
-    map.put(statusField,statusToString(STATUS_COMPLETE));
+    // Map BEINGDELETED to ELIGIBLEFORDELETE
+    map.put(statusField,statusToString(STATUS_ELIGIBLEFORDELETE));
     map.put(checkTimeField,new Long(0L));
     list.clear();
     list.add(statusToString(STATUS_BEINGDELETED));
@@ -405,6 +407,45 @@ public class JobQueue extends org.apache
     performUpdate(map,"WHERE "+statusField+"=?",list,null);
   }
 
+  /** Prepare for a job delete pass.  This will not be called
+  * unless the job is in an INACTIVE state.
+  * Does the following:
+  * (1) Delete PENDING entries
+  * (2) Maps PENDINGPURGATORY, PURGATORY, and COMPLETED entries to ELIGIBLEFORDELETE
+  *@param jobID is the job identifier.
+  */
+  public void prepareDeleteScan(Long jobID)
+    throws ManifoldCFException
+  {
+    // Delete PENDING entries
+    ArrayList list = new ArrayList();
+    list.add(jobID);
+    list.add(statusToString(STATUS_PENDING));
+    // Clean out prereqevents table first
+    prereqEventManager.deleteRows(getTableName()+" t0","t0."+idField,"t0."+jobIDField+"=? AND t0."+statusField+"=?",list);
+    performDelete("WHERE "+jobIDField+"=? AND "+statusField+"=?",list,null);
+
+    // Turn PENDINGPURGATORY, PURGATORY, COMPLETED into ELIGIBLEFORDELETE.
+    HashMap map = new HashMap();
+    map.put(statusField,statusToString(STATUS_ELIGIBLEFORDELETE));
+    map.put(checkTimeField,new Long(0L));
+    map.put(checkActionField,null);
+    map.put(failTimeField,null);
+    map.put(failCountField,null);
+    list.clear();
+    list.add(jobID);
+    list.add(statusToString(STATUS_PENDINGPURGATORY));
+    list.add(statusToString(STATUS_COMPLETE));
+    list.add(statusToString(STATUS_PURGATORY));
+    performUpdate(map,"WHERE "+jobIDField+"=? AND "+statusField+" IN (?,?,?)",list,null);
+
+    // Not accurate, but best we can do without overhead
+    noteModifications(0,2,0);
+    // Do an analyze, otherwise our plans are going to be crap right off the bat
+    unconditionallyAnalyzeTables();
+
+  }
+  
   /** Prepare for a "full scan" job.  This will not be called
   * unless the job is in the "INACTIVE" state.
   * This does the following:
@@ -416,7 +457,7 @@ public class JobQueue extends org.apache
   public void prepareFullScan(Long jobID)
     throws ManifoldCFException
   {
-    // Delete PENDING and ACTIVE entries
+    // Delete PENDING entries
     ArrayList list = new ArrayList();
     list.add(jobID);
     list.add(statusToString(STATUS_PENDING));
@@ -559,7 +600,7 @@ public class JobQueue extends org.apache
     case STATUS_ACTIVEPURGATORY:
       newStatus = STATUS_COMPLETE;
       actionFieldValue = null;
-      checkTimeValue = new Long(0L);
+      checkTimeValue = null;
       break;
     case STATUS_ACTIVENEEDRESCAN:
     case STATUS_ACTIVENEEDRESCANPURGATORY:
@@ -661,7 +702,7 @@ public class JobQueue extends org.apache
     throws ManifoldCFException
   {
     HashMap map = new HashMap();
-    map.put(statusField,statusToString(STATUS_COMPLETE));
+    map.put(statusField,statusToString(STATUS_ELIGIBLEFORDELETE));
     map.put(checkTimeField,new Long(checkTime));
     map.put(checkActionField,null);
     map.put(failTimeField,null);
@@ -1276,6 +1317,8 @@ public class JobQueue extends org.apache
       return "F";
     case STATUS_PURGATORY:
       return "Z";
+    case STATUS_ELIGIBLEFORDELETE:
+      return "E";
     case STATUS_BEINGDELETED:
       return "D";
     case STATUS_ACTIVENEEDRESCAN:

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Jobs.java Sun Jan 16 22:30:45 2011
@@ -51,6 +51,8 @@ public class Jobs extends org.apache.man
   public static final int STATUS_ABORTINGFORRESTARTSEEDING = 17;  // Seeding version of aborting for restart
   public static final int STATUS_ABORTINGSTARTINGUPFORRESTART = 18; // Starting up version of aborting for restart
   public static final int STATUS_NOTIFYINGOFCOMPLETION = 19;    // Notifying connector of terminating job (either aborted, or finished)
+  public static final int STATUS_DELETING = 20;                         // The job is deleting.
+  public static final int STATUS_DELETESTARTINGUP = 21;         // The delete is starting up.
   
   // These statuses have to do with whether a job has an installed underlying connector or not.
   // There are two reasons to have a special state here: (1) if the behavior of the crawler differs, or (2) if the
@@ -61,13 +63,13 @@ public class Jobs extends org.apache.man
   // But, since there is no indication in the jobs table of an uninstalled connector for such jobs, the code which starts
   // jobs up (or otherwise would enter any state that has a corresponding special state) must check to see if the underlying
   // connector exists before deciding what state to put the job into.
-  public static final int STATUS_ACTIVE_UNINSTALLED = 21;               // Active, but repository connector not installed
-  public static final int STATUS_ACTIVESEEDING_UNINSTALLED = 22;   // Active and seeding, but repository connector not installed
-  public static final int STATUS_ACTIVE_NOOUTPUT = 23;                  // Active, but output connector not installed
-  public static final int STATUS_ACTIVESEEDING_NOOUTPUT = 24;       // Active and seeding, but output connector not installed
-  public static final int STATUS_ACTIVE_NEITHER = 25;                     // Active, but neither repository connector nor output connector installed
-  public static final int STATUS_ACTIVESEEDING_NEITHER = 26;          // Active and seeding, but neither repository connector nor output connector installed
-  public static final int STATUS_READYFORDELETE_NOOUTPUT = 27;    // Job is being deleted but there's no output connector installed
+  public static final int STATUS_ACTIVE_UNINSTALLED = 22;               // Active, but repository connector not installed
+  public static final int STATUS_ACTIVESEEDING_UNINSTALLED = 23;   // Active and seeding, but repository connector not installed
+  public static final int STATUS_ACTIVE_NOOUTPUT = 24;                  // Active, but output connector not installed
+  public static final int STATUS_ACTIVESEEDING_NOOUTPUT = 25;       // Active and seeding, but output connector not installed
+  public static final int STATUS_ACTIVE_NEITHER = 26;                     // Active, but neither repository connector nor output connector installed
+  public static final int STATUS_ACTIVESEEDING_NEITHER = 27;          // Active and seeding, but neither repository connector nor output connector installed
+  public static final int STATUS_DELETING_NOOUTPUT = 28;                // Job is being deleted but there's no output connector installed
 
   // Type field values
   public static final int TYPE_CONTINUOUS = IJobDescription.TYPE_CONTINUOUS;
@@ -144,6 +146,8 @@ public class Jobs extends org.apache.man
     statusMap.put("Q",new Integer(STATUS_ABORTINGSTARTINGUP));
     statusMap.put("C",new Integer(STATUS_READYFORSTARTUP));
     statusMap.put("E",new Integer(STATUS_READYFORDELETE));
+    statusMap.put("V",new Integer(STATUS_DELETESTARTINGUP));
+    statusMap.put("e",new Integer(STATUS_DELETING));
     statusMap.put("Y",new Integer(STATUS_ABORTINGFORRESTART));
     statusMap.put("T",new Integer(STATUS_ABORTINGSTARTINGUPFORRESTART));
 
@@ -161,7 +165,7 @@ public class Jobs extends org.apache.man
     statusMap.put("o",new Integer(STATUS_ACTIVESEEDING_NOOUTPUT));
     statusMap.put("U",new Integer(STATUS_ACTIVE_NEITHER));
     statusMap.put("u",new Integer(STATUS_ACTIVESEEDING_NEITHER));
-    statusMap.put("D",new Integer(STATUS_READYFORDELETE_NOOUTPUT));
+    statusMap.put("D",new Integer(STATUS_DELETING_NOOUTPUT));
     
     typeMap = new HashMap();
     typeMap.put("C",new Integer(TYPE_CONTINUOUS));
@@ -419,9 +423,11 @@ public class Jobs extends org.apache.man
 
       ArrayList list = new ArrayList();
       list.add(statusToString(STATUS_READYFORDELETE));
-      list.add(statusToString(STATUS_READYFORDELETE_NOOUTPUT));
+      list.add(statusToString(STATUS_DELETESTARTINGUP));
+      list.add(statusToString(STATUS_DELETING));
+      list.add(statusToString(STATUS_DELETING_NOOUTPUT));
       IResultSet set = performQuery("SELECT "+idField+","+descriptionField+" FROM "+
-        getTableName()+" WHERE "+statusField+"!=? AND "+statusField+"!=?"+
+        getTableName()+" WHERE "+statusField+"!=? AND "+statusField+"!=? AND "+statusField+"!=? AND "+statusField+"!=?"+
         " ORDER BY "+descriptionField+" ASC",list,cacheKeys,null);
       // Convert to an array of id's, and then load them
       Long[] ids = new Long[set.getRowCount()];
@@ -711,10 +717,16 @@ public class Jobs extends org.apache.man
       StringSet invKey = new StringSet(getJobStatusKey());
       ArrayList list = new ArrayList();
 
+      // Starting up delete goes back to just being ready for delete
+      list.add(statusToString(STATUS_DELETESTARTINGUP));
+      HashMap map = new HashMap();
+      map.put(statusField,statusToString(STATUS_READYFORDELETE));
+      performUpdate(map,"WHERE "+statusField+"=?",list,invKey);
+
       // Starting up or aborting starting up goes back to just being ready
+      list.clear();
       list.add(statusToString(STATUS_STARTINGUP));
       list.add(statusToString(STATUS_ABORTINGSTARTINGUP));
-      HashMap map = new HashMap();
       map.put(statusField,statusToString(STATUS_READYFORSTARTUP));
       performUpdate(map,"WHERE "+statusField+"=? OR "+statusField+"=?",list,invKey);
 
@@ -809,8 +821,8 @@ public class Jobs extends org.apache.man
     case STATUS_ACTIVESEEDING_UNINSTALLED:
       newStatusValue = STATUS_ACTIVESEEDING_NEITHER;
       break;
-    case STATUS_READYFORDELETE:
-      newStatusValue = STATUS_READYFORDELETE_NOOUTPUT;
+    case STATUS_DELETING:
+      newStatusValue = STATUS_DELETING_NOOUTPUT;
       break;
     default:
       newStatusValue = oldStatusValue;
@@ -852,8 +864,8 @@ public class Jobs extends org.apache.man
     case STATUS_ACTIVESEEDING_NEITHER:
       newStatusValue = STATUS_ACTIVESEEDING_UNINSTALLED;
       break;
-    case STATUS_READYFORDELETE_NOOUTPUT:
-      newStatusValue = STATUS_READYFORDELETE;
+    case STATUS_DELETING_NOOUTPUT:
+      newStatusValue = STATUS_DELETING;
       break;
     default:
       newStatusValue = oldStatusValue;
@@ -1004,6 +1016,21 @@ public class Jobs extends org.apache.man
     return (status == STATUS_ACTIVE || status == STATUS_ACTIVESEEDING || status == STATUS_STARTINGUP);
   }
 
+  /** Reset delete startup worker thread status.
+  */
+  public void resetDeleteStartupWorkerStatus()
+    throws ManifoldCFException
+  {
+    // This handles everything that the delete startup thread would resolve.
+
+    ArrayList list = new ArrayList();
+    list.add(statusToString(STATUS_DELETESTARTINGUP));
+    HashMap map = new HashMap();
+    map.put(statusField,statusToString(STATUS_READYFORDELETE));
+    performUpdate(map,"WHERE "+statusField+"=?",list,new StringSet(getJobStatusKey()));
+
+  }
+  
   /** Reset startup worker thread status.
   */
   public void resetStartupWorkerStatus()
@@ -1176,6 +1203,63 @@ public class Jobs extends org.apache.man
     }
   }
   
+  /** Put job into "deleting" state, and set the start time field.
+  *@param jobID is the job identifier.
+  *@param startTime is the current time in milliseconds from start of epoch.
+  */
+  public void noteJobDeleteStarted(Long jobID, long startTime)
+    throws ManifoldCFException
+  {
+    beginTransaction();
+    try
+    {
+      ArrayList list = new ArrayList();
+      list.add(jobID);
+      IResultSet set = performQuery("SELECT "+statusField+","+connectionNameField+","+outputNameField+" FROM "+getTableName()+" WHERE "+
+        idField+"=? FOR UPDATE",list,null,null);
+      if (set.getRowCount() == 0)
+        throw new ManifoldCFException("Can't find job "+jobID.toString());
+      IResultRow row = set.getRow(0);
+      int status = stringToStatus((String)row.getValue(statusField));
+      int newStatus;
+      switch (status)
+      {
+      case STATUS_DELETESTARTINGUP:
+        if (outputMgr.checkConnectorExists((String)row.getValue(outputNameField)))
+          newStatus = STATUS_DELETING;
+        else
+          newStatus = STATUS_DELETING_NOOUTPUT;
+        break;
+      default:
+        // Complain!
+        throw new ManifoldCFException("Unexpected job status encountered: "+Integer.toString(status));
+      }
+
+      HashMap map = new HashMap();
+      map.put(statusField,statusToString(newStatus));
+      if (newStatus == STATUS_DELETING || newStatus == STATUS_DELETING_NOOUTPUT)
+      {
+        map.put(startTimeField,new Long(startTime));
+      }
+      map.put(lastCheckTimeField,new Long(startTime));
+      performUpdate(map,"WHERE "+idField+"=?",list,new StringSet(getJobStatusKey()));
+    }
+    catch (ManifoldCFException e)
+    {
+      signalRollback();
+      throw e;
+    }
+    catch (Error e)
+    {
+      signalRollback();
+      throw e;
+    }
+    finally
+    {
+      endTransaction();
+    }
+  }
+
   /** Make job active, and set the start time field.
   *@param jobID is the job identifier.
   *@param startTime is the current time in milliseconds from start of epoch.
@@ -1230,7 +1314,6 @@ public class Jobs extends org.apache.man
       {
         map.put(startTimeField,new Long(startTime));
       }
-      map.put(lastCheckTimeField,new Long(startTime));
       performUpdate(map,"WHERE "+idField+"=?",list,new StringSet(getJobStatusKey()));
     }
     catch (ManifoldCFException e)
@@ -1824,7 +1907,7 @@ public class Jobs extends org.apache.man
 
   }
 
-  /** Return true if there is a job in the READYFORDELETE state.  (This matches the
+  /** Return true if there is a job in the DELETING state.  (This matches the
   * conditions for values to be returned from
   * getNextDeletableDocuments).
   *@return true if such jobs exist.
@@ -1833,7 +1916,7 @@ public class Jobs extends org.apache.man
     throws ManifoldCFException
   {
     ArrayList list = new ArrayList();
-    list.add(statusToString(STATUS_READYFORDELETE));
+    list.add(statusToString(STATUS_DELETING));
     IResultSet set = performQuery("SELECT "+idField+" FROM "+getTableName()+" WHERE "+
       statusField+"=? "+constructOffsetLimitClause(0,1),
       list,new StringSet(getJobStatusKey()),null,1);
@@ -1926,6 +2009,10 @@ public class Jobs extends org.apache.man
       return "C";
     case STATUS_READYFORDELETE:
       return "E";
+    case STATUS_DELETESTARTINGUP:
+      return "V";
+    case STATUS_DELETING:
+      return "e";
     case STATUS_ACTIVESEEDING:
       return "a";
     case STATUS_ABORTINGSEEDING:
@@ -1950,7 +2037,7 @@ public class Jobs extends org.apache.man
       return "U";
     case STATUS_ACTIVESEEDING_NEITHER:
       return "u";
-    case STATUS_READYFORDELETE_NOOUTPUT:
+    case STATUS_DELETING_NOOUTPUT:
       return "D";
 
     default:

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/DocumentDeleteThread.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/DocumentDeleteThread.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/DocumentDeleteThread.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/DocumentDeleteThread.java Sun Jan 16 22:30:45 2011
@@ -91,6 +91,9 @@ public class DocumentDeleteThread extend
             // Reset
             continue;
 
+          if (Logging.threads.isDebugEnabled())
+            Logging.threads.debug("Document delete thread received "+Integer.toString(dds.getCount())+" documents to delete for job "+dds.getJobDescription().getID().toString());
+          
           IJobDescription job = dds.getJobDescription();
           String connectionName = job.getConnectionName();
           String outputConnectionName = job.getOutputConnectionName();

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java Sun Jan 16 22:30:45 2011
@@ -40,6 +40,7 @@ public class ManifoldCF extends org.apac
   protected static FinisherThread finisherThread = null;
   protected static JobNotificationThread notificationThread = null;
   protected static StartupThread startupThread = null;
+  protected static StartDeleteThread startDeleteThread = null;
   protected static JobDeleteThread jobDeleteThread = null;
   protected static WorkerThread[] workerThreads = null;
   protected static ExpireStufferThread expireStufferThread = null;
@@ -204,6 +205,7 @@ public class ManifoldCF extends org.apac
 
       jobStartThread = new JobStartThread();
       startupThread = new StartupThread(queueTracker);
+      startDeleteThread = new StartDeleteThread();
       finisherThread = new FinisherThread();
       notificationThread = new JobNotificationThread();
       jobDeleteThread = new JobDeleteThread();
@@ -315,6 +317,7 @@ public class ManifoldCF extends org.apac
         // Start all the threads
         jobStartThread.start();
         startupThread.start();
+        startDeleteThread.start();
         finisherThread.start();
         notificationThread.start();
         jobDeleteThread.start();
@@ -381,8 +384,9 @@ public class ManifoldCF extends org.apac
     Logging.root.info("Shutting down pull-agent...");
     synchronized (startupLock)
     {
-      while (initializationThread != null || jobDeleteThread != null || startupThread != null || jobStartThread != null || stufferThread != null ||
-        finisherThread != null || notificationThread != null || workerThreads != null || expireStufferThread != null | expireThreads != null ||
+      while (initializationThread != null || jobDeleteThread != null || startupThread != null || startDeleteThread != null ||
+        jobStartThread != null || stufferThread != null ||
+        finisherThread != null || notificationThread != null || workerThreads != null || expireStufferThread != null || expireThreads != null ||
         deleteStufferThread != null || deleteThreads != null ||
         cleanupStufferThread != null || cleanupThreads != null ||
         jobResetThread != null || seedingThread != null || idleCleanupThread != null || setPriorityThread != null)
@@ -410,6 +414,10 @@ public class ManifoldCF extends org.apac
         {
           startupThread.interrupt();
         }
+        if (startDeleteThread != null)
+        {
+          startDeleteThread.interrupt();
+        }
         if (stufferThread != null)
         {
           stufferThread.interrupt();
@@ -517,6 +525,11 @@ public class ManifoldCF extends org.apac
           if (!startupThread.isAlive())
             startupThread = null;
         }
+        if (startDeleteThread != null)
+        {
+          if (!startDeleteThread.isAlive())
+            startDeleteThread = null;
+        }
         if (jobStartThread != null)
         {
           if (!jobStartThread.isAlive())

Added: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java?rev=1059695&view=auto
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java (added)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java Sun Jan 16 22:30:45 2011
@@ -0,0 +1,236 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You 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.apache.manifoldcf.crawler.system;
+
+import org.apache.manifoldcf.core.interfaces.*;
+import org.apache.manifoldcf.agents.interfaces.*;
+import org.apache.manifoldcf.crawler.interfaces.*;
+import org.apache.manifoldcf.crawler.system.Logging;
+import java.util.*;
+import java.lang.reflect.*;
+
+/** This class represents the start delete thread.  This thread's job is to detect when a job is
+* ready for deletion, initialize it, and put it into the DELETING state.
+*/
+public class StartDeleteThread extends Thread
+{
+  public static final String _rcsid = "@(#)$Id$";
+
+  /** Worker thread pool reset manager */
+  protected static DeleteStartupResetManager resetManager = new DeleteStartupResetManager();
+
+  /** The number of documents that are added to the queue per transaction */
+  protected final static int MAX_COUNT = 100;
+
+  /** Constructor.
+  */
+  public StartDeleteThread()
+    throws ManifoldCFException
+  {
+    super();
+    setName("Delete startup thread");
+    setDaemon(true);
+  }
+
+  public void run()
+  {
+    resetManager.registerMe();
+
+    try
+    {
+      // Create a thread context object.
+      IThreadContext threadContext = ThreadContextFactory.make();
+      IJobManager jobManager = JobManagerFactory.make(threadContext);
+      IRepositoryConnectionManager connectionMgr = RepositoryConnectionManagerFactory.make(threadContext);
+
+      IDBInterface database = DBInterfaceFactory.make(threadContext,
+        ManifoldCF.getMasterDatabaseName(),
+        ManifoldCF.getMasterDatabaseUsername(),
+        ManifoldCF.getMasterDatabasePassword());
+
+      // Loop
+      while (true)
+      {
+        // Do another try/catch around everything in the loop
+        try
+        {
+          // Before we begin, conditionally reset
+          resetManager.waitForReset(threadContext);
+
+          // Accumulate the wait before doing the next check.
+          // We start with 10 seconds, which is the maximum.  If there's a service request
+          // that's faster than that, we'll adjust the time downward.
+          long waitTime = 10000L;
+
+          if (Logging.threads.isDebugEnabled())
+            Logging.threads.debug("Checking for deleting jobs");
+
+          // See if there are any starting jobs.
+          // Note: Since this following call changes the job state, we must be careful to reset it on any kind of failure.
+          JobStartRecord[] deleteJobs = jobManager.getJobsReadyForDelete();
+          try
+          {
+
+            if (deleteJobs.length == 0)
+            {
+              ManifoldCF.sleep(waitTime);
+              continue;
+            }
+
+            if (Logging.threads.isDebugEnabled())
+              Logging.threads.debug("Found "+Integer.toString(deleteJobs.length)+" jobs ready to be deleted");
+
+            long currentTime = System.currentTimeMillis();
+
+
+            // Loop through jobs
+            int i = 0;
+            while (i < deleteJobs.length)
+            {
+              JobStartRecord jsr = deleteJobs[i++];
+              Long jobID = jsr.getJobID();
+	      try
+	      {
+		jobManager.prepareDeleteScan(jobID);
+                // Start deleting this job!
+                jobManager.noteJobDeleteStarted(jobID,currentTime);
+                jsr.noteStarted();
+              }
+              catch (ManifoldCFException e)
+              {
+                if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
+                  throw new InterruptedException();
+                if (e.getErrorCode() == ManifoldCFException.DATABASE_CONNECTION_ERROR)
+                  throw e;
+		// We cannot abort the delete startup, but if we fall through, we'll put the job back into
+		// the state whence it came.  So, fall through.
+		Logging.threads.error("Exception tossed: "+e.getMessage(),e);
+              }
+            }
+          }
+          finally
+          {
+            // Clean up all jobs that did not start
+            ManifoldCFException exception = null;
+            int i = 0;
+            while (i < deleteJobs.length)
+            {
+              JobStartRecord jsr = deleteJobs[i++];
+              if (!jsr.wasStarted())
+              {
+                // Clean up from failed start.
+                try
+                {
+                  jobManager.resetStartDeleteJob(jsr.getJobID());
+                }
+                catch (ManifoldCFException e)
+                {
+                  exception = e;
+                }
+              }
+            }
+            if (exception != null)
+              throw exception;
+          }
+
+          // Sleep for the retry interval.
+          ManifoldCF.sleep(waitTime);
+        }
+        catch (ManifoldCFException e)
+        {
+          if (e.getErrorCode() == ManifoldCFException.INTERRUPTED)
+            break;
+
+          if (e.getErrorCode() == ManifoldCFException.DATABASE_CONNECTION_ERROR)
+          {
+            resetManager.noteEvent();
+
+            Logging.threads.error("Start delete thread aborting and restarting due to database connection reset: "+e.getMessage(),e);
+            try
+            {
+              // Give the database a chance to catch up/wake up
+              ManifoldCF.sleep(10000L);
+            }
+            catch (InterruptedException se)
+            {
+              break;
+            }
+            continue;
+          }
+
+          // Log it, but keep the thread alive
+          Logging.threads.error("Exception tossed: "+e.getMessage(),e);
+
+          if (e.getErrorCode() == ManifoldCFException.SETUP_ERROR)
+          {
+            // Shut the whole system down!
+            System.exit(1);
+          }
+
+        }
+        catch (InterruptedException e)
+        {
+          // We're supposed to quit
+          break;
+        }
+        catch (OutOfMemoryError e)
+        {
+          System.err.println("agents process ran out of memory - shutting down");
+          e.printStackTrace(System.err);
+          System.exit(-200);
+        }
+        catch (Throwable e)
+        {
+          // A more severe error - but stay alive
+          Logging.threads.fatal("Error tossed: "+e.getMessage(),e);
+        }
+      }
+    }
+    catch (Throwable e)
+    {
+      // Severe error on initialization
+      System.err.println("agents process could not start - shutting down");
+      Logging.threads.fatal("StartDeleteThread initialization error tossed: "+e.getMessage(),e);
+      System.exit(-300);
+    }
+  }
+
+  /** Class which handles reset for seeding thread pool (of which there's
+  * typically only one member).  The reset action here
+  * is to move the status of jobs back from "seeding" to normal.
+  */
+  protected static class DeleteStartupResetManager extends ResetManager
+  {
+
+    /** Constructor. */
+    public DeleteStartupResetManager()
+    {
+      super();
+    }
+
+    /** Reset */
+    protected void performResetLogic(IThreadContext tc)
+      throws ManifoldCFException
+    {
+      IJobManager jobManager = JobManagerFactory.make(tc);
+      jobManager.resetDeleteStartupWorkerStatus();
+    }
+  }
+
+}

Propchange: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
------------------------------------------------------------------------------
    svn:eol-style = native

Propchange: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
------------------------------------------------------------------------------
    svn:keywords = Id

Modified: incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java (original)
+++ incubator/lcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java Sun Jan 16 22:30:45 2011
@@ -38,9 +38,6 @@ public class StartupThread extends Threa
   // Local data
   protected QueueTracker queueTracker;
 
-  /** The number of documents that are added to the queue per transaction */
-  protected final static int MAX_COUNT = 100;
-
   /** Constructor.
   */
   public StartupThread(QueueTracker queueTracker)
@@ -68,8 +65,6 @@ public class StartupThread extends Threa
         ManifoldCF.getMasterDatabaseUsername(),
         ManifoldCF.getMasterDatabasePassword());
 
-      String[] identifiers = new String[MAX_COUNT];
-
       // Loop
       while (true)
       {

Modified: incubator/lcf/trunk/site/src/documentation/content/xdocs/how-to-build-and-deploy.xml
URL: http://svn.apache.org/viewvc/incubator/lcf/trunk/site/src/documentation/content/xdocs/how-to-build-and-deploy.xml?rev=1059695&r1=1059694&r2=1059695&view=diff
==============================================================================
--- incubator/lcf/trunk/site/src/documentation/content/xdocs/how-to-build-and-deploy.xml (original)
+++ incubator/lcf/trunk/site/src/documentation/content/xdocs/how-to-build-and-deploy.xml Sun Jan 16 22:30:45 2011
@@ -410,6 +410,8 @@ cd dist/example
             <tr><td>org.apache.manifoldcf.database.username</td><td>No</td><td>Describes database user name for ManifoldCF; defaults to "manifoldcf" if not specified.</td></tr>
             <tr><td>org.apache.manifoldcf.database.password</td><td>No</td><td>Describes database user's password for ManifoldCF; defaults to "local_pg_password" if not specified.</td></tr>
             <tr><td>org.apache.manifoldcf.crawler.threads</td><td>No</td><td>Number of crawler worker threads created.  Suggest a value of 30.</td></tr>
+            <tr><td>org.apache.manifoldcf.crawler.expirethreads</td><td>No</td><td>Number of crawler expiration threads created.  Suggest a value of 10.</td></tr>
+            <tr><td>org.apache.manifoldcf.crawler.cleanupthreads</td><td>No</td><td>Number of crawler cleanup threads created.  Suggest a value of 10.</td></tr>
             <tr><td>org.apache.manifoldcf.crawler.deletethreads</td><td>No</td><td>Number of crawler delete threads created.  Suggest a value of 10.</td></tr>
             <tr><td>org.apache.manifoldcf.misc</td><td>No</td><td>Miscellaneous debugging output.  Legal values INFO, WARN, or DEBUG.</td></tr>
             <tr><td>org.apache.manifoldcf.db</td><td>No</td><td>Database debugging output.  Legal values INFO, WARN, or DEBUG.</td></tr>



Mime
View raw message