You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@manifoldcf.apache.org by kw...@apache.org on 2013/11/26 12:45:55 UTC
svn commit: r1545624 [5/5] - in /manifoldcf/trunk: ./
framework/agents/src/main/java/org/apache/manifoldcf/agents/
framework/agents/src/main/java/org/apache/manifoldcf/agents/agentmanager/
framework/agents/src/main/java/org/apache/manifoldcf/agents/int...
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireStufferThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireStufferThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireStufferThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireStufferThread.java Tue Nov 26 11:45:53 2013
@@ -33,25 +33,28 @@ public class ExpireStufferThread extends
public static final String _rcsid = "@(#)$Id: ExpireStufferThread.java 988245 2010-08-23 18:39:35Z kwright $";
// Local data
- // This is a reference to the static main document expiration queue
- protected DocumentCleanupQueue documentQueue;
+ /** This is a reference to the static main document expiration queue */
+ protected final DocumentCleanupQueue documentQueue;
/** Worker thread pool reset manager */
- protected WorkerResetManager resetManager;
- // This is the number of entries we want to stuff at any one time.
- protected int n;
-
+ protected final WorkerResetManager resetManager;
+ /** This is the number of entries we want to stuff at any one time. */
+ protected final int n;
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*@param documentQueue is the document queue we'll be stuffing.
*@param n represents the number of threads that will be processing queued stuff, NOT the
* number of documents to be done at once!
*/
- public ExpireStufferThread(DocumentCleanupQueue documentQueue, int n, WorkerResetManager resetManager)
+ public ExpireStufferThread(DocumentCleanupQueue documentQueue, int n, WorkerResetManager resetManager, String processID)
throws ManifoldCFException
{
super();
this.documentQueue = documentQueue;
this.n = n;
this.resetManager = resetManager;
+ this.processID = processID;
setName("Expire stuffer thread");
setDaemon(true);
// The priority of this thread is higher than most others. We want stuffing to proceed even if the machine
@@ -115,7 +118,7 @@ public class ExpireStufferThread extends
// The number n passed in here thus cannot be used in a query to limit the number of returned
// results. Instead, it must be factored into the limit portion of the query.
long currentTime = System.currentTimeMillis();
- DocumentSetAndFlags docsAndFlags = jobManager.getExpiredDocuments(deleteChunkSize,currentTime);
+ DocumentSetAndFlags docsAndFlags = jobManager.getExpiredDocuments(processID,deleteChunkSize,currentTime);
DocumentDescription[] descs = docsAndFlags.getDocumentSet();
boolean[] deleteFromIndex = docsAndFlags.getFlags();
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ExpireThread.java Tue Nov 26 11:45:53 2013
@@ -33,18 +33,21 @@ public class ExpireThread extends Thread
// Local data
- protected String id;
- // This is a reference to the static main document queue
- protected DocumentCleanupQueue documentQueue;
+ /** Thread id */
+ protected final String id;
+ /** This is a reference to the static main document queue */
+ protected final DocumentCleanupQueue documentQueue;
/** Worker thread pool reset manager */
- protected WorkerResetManager resetManager;
+ protected final WorkerResetManager resetManager;
/** Queue tracker */
- protected QueueTracker queueTracker;
-
+ protected final QueueTracker queueTracker;
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*@param id is the expire thread id.
*/
- public ExpireThread(String id, DocumentCleanupQueue documentQueue, QueueTracker queueTracker, WorkerResetManager resetManager)
+ public ExpireThread(String id, DocumentCleanupQueue documentQueue, QueueTracker queueTracker, WorkerResetManager resetManager, String processID)
throws ManifoldCFException
{
super();
@@ -52,6 +55,7 @@ public class ExpireThread extends Thread
this.documentQueue = documentQueue;
this.resetManager = resetManager;
this.queueTracker = queueTracker;
+ this.processID = processID;
setName("Expiration thread '"+id+"'");
setDaemon(true);
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/FinisherThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/FinisherThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/FinisherThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/FinisherThread.java Tue Nov 26 11:45:53 2013
@@ -32,13 +32,16 @@ public class FinisherThread extends Thre
public static final String _rcsid = "@(#)$Id: FinisherThread.java 991295 2010-08-31 19:12:14Z kwright $";
// Local data
+ /** Process ID */
+ protected final String processID;
/** Constructor.
*/
- public FinisherThread()
+ public FinisherThread(String processID)
throws ManifoldCFException
{
super();
+ this.processID = processID;
setName("Finisher thread");
setDaemon(true);
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/HistoryCleanupThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/HistoryCleanupThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/HistoryCleanupThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/HistoryCleanupThread.java Tue Nov 26 11:45:53 2013
@@ -32,12 +32,17 @@ public class HistoryCleanupThread extend
protected static final String historyCleanupIntervalProperty = "org.apache.manifoldcf.crawler.historycleanupinterval";
+ // Local data
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*/
- public HistoryCleanupThread()
+ public HistoryCleanupThread(String processID)
throws ManifoldCFException
{
super();
+ this.processID = processID;
setName("History cleanup thread");
setDaemon(true);
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/IdleCleanupThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/IdleCleanupThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/IdleCleanupThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/IdleCleanupThread.java Tue Nov 26 11:45:53 2013
@@ -33,14 +33,16 @@ public class IdleCleanupThread extends T
public static final String _rcsid = "@(#)$Id: IdleCleanupThread.java 988245 2010-08-23 18:39:35Z kwright $";
// Local data
-
+ /** Process ID */
+ protected final String processID;
/** Constructor.
*/
- public IdleCleanupThread()
+ public IdleCleanupThread(String processID)
throws ManifoldCFException
{
super();
+ this.processID = processID;
setName("Idle cleanup thread");
setDaemon(true);
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobDeleteThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobDeleteThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobDeleteThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobDeleteThread.java Tue Nov 26 11:45:53 2013
@@ -35,13 +35,16 @@ public class JobDeleteThread extends Thr
public static final String _rcsid = "@(#)$Id: JobDeleteThread.java 988245 2010-08-23 18:39:35Z kwright $";
// Local data
-
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*/
- public JobDeleteThread()
+ public JobDeleteThread(String processID)
throws ManifoldCFException
{
super();
+ this.processID = processID;
setName("Job delete thread");
setDaemon(true);
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobNotificationThread.java Tue Nov 26 11:45:53 2013
@@ -33,14 +33,18 @@ public class JobNotificationThread exten
public static final String _rcsid = "@(#)$Id: JobNotificationThread.java 998081 2010-09-17 11:33:15Z kwright $";
/** Notification reset manager */
- protected static NotificationResetManager resetManager = new NotificationResetManager();
-
+ protected final NotificationResetManager resetManager;
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*/
- public JobNotificationThread()
+ public JobNotificationThread(NotificationResetManager resetManager, String processID)
throws ManifoldCFException
{
super();
+ this.resetManager = resetManager;
+ this.processID = processID;
setName("Job notification thread");
setDaemon(true);
}
@@ -66,7 +70,7 @@ public class JobNotificationThread exten
// Before we begin, conditionally reset
resetManager.waitForReset(threadContext);
- JobNotifyRecord[] jobsNeedingNotification = jobManager.getJobsReadyForInactivity();
+ JobNotifyRecord[] jobsNeedingNotification = jobManager.getJobsReadyForInactivity(processID);
try
{
HashMap connectionNames = new HashMap();
@@ -329,33 +333,4 @@ public class JobNotificationThread exten
}
- /** 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 NotificationResetManager extends ResetManager
- {
-
- /** Constructor. */
- public NotificationResetManager()
- {
- super();
- }
-
- /** Reset */
- protected void performResetLogic(IThreadContext tc)
- throws ManifoldCFException
- {
- IJobManager jobManager = JobManagerFactory.make(tc);
- jobManager.resetNotificationWorkerStatus();
- }
-
- /** Do the wakeup logic.
- */
- protected void performWakeupLogic()
- {
- }
-
- }
-
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobResetThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobResetThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobResetThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobResetThread.java Tue Nov 26 11:45:53 2013
@@ -33,17 +33,21 @@ public class JobResetThread extends Thre
public static final String _rcsid = "@(#)$Id: JobResetThread.java 991295 2010-08-31 19:12:14Z kwright $";
// Local data
- protected QueueTracker queueTracker;
+ /** Queue tracker */
+ protected final QueueTracker queueTracker;
+ /** Process ID */
+ protected final String processID;
/** Constructor.
*/
- public JobResetThread(QueueTracker queueTracker)
+ public JobResetThread(QueueTracker queueTracker, String processID)
throws ManifoldCFException
{
super();
setName("Job reset thread");
setDaemon(true);
this.queueTracker = queueTracker;
+ this.processID = processID;
}
public void run()
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobStartThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobStartThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobStartThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/JobStartThread.java Tue Nov 26 11:45:53 2013
@@ -30,12 +30,16 @@ public class JobStartThread extends Thre
{
public static final String _rcsid = "@(#)$Id: JobStartThread.java 988245 2010-08-23 18:39:35Z kwright $";
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*/
- public JobStartThread()
+ public JobStartThread(String processID)
throws ManifoldCFException
{
super();
+ this.processID = processID;
setName("Job start thread");
setDaemon(true);
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java Tue Nov 26 11:45:53 2013
@@ -51,50 +51,6 @@ public class ManifoldCF extends org.apac
// Initialization flag.
protected static boolean crawlerInitialized = false;
-
- // Thread objects.
- // These get filled in as threads are created.
- protected static InitializationThread initializationThread = null;
- protected static JobStartThread jobStartThread = null;
- protected static StufferThread stufferThread = null;
- 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;
- protected static ExpireThread[] expireThreads = null;
- protected static DocumentDeleteStufferThread deleteStufferThread = null;
- protected static DocumentDeleteThread[] deleteThreads = null;
- protected static DocumentCleanupStufferThread cleanupStufferThread = null;
- protected static DocumentCleanupThread[] cleanupThreads = null;
- protected static JobResetThread jobResetThread = null;
- protected static SeedingThread seedingThread = null;
- protected static IdleCleanupThread idleCleanupThread = null;
- protected static SetPriorityThread setPriorityThread = null;
- protected static HistoryCleanupThread historyCleanupThread = null;
-
- // Reset managers
- /** Worker thread pool reset manager */
- protected static WorkerResetManager workerResetManager = null;
- /** Delete thread pool reset manager */
- protected static DocDeleteResetManager docDeleteResetManager = null;
- /** Cleanup thread pool reset manager */
- protected static DocCleanupResetManager docCleanupResetManager = null;
-
- // Number of worker threads
- protected static int numWorkerThreads = 0;
- // Number of delete threads
- protected static int numDeleteThreads = 0;
- // Number of cleanup threads
- protected static int numCleanupThreads = 0;
- // Number of expiration threads
- protected static int numExpireThreads = 0;
- // Factor for low water level in queueing
- protected static float lowWaterFactor = 5.0f;
- // Factor in amount to stuff
- protected static float stuffAmtFactor = 0.5f;
// Properties
protected static final String workerThreadCountProperty = "org.apache.manifoldcf.crawler.threads";
@@ -108,8 +64,6 @@ public class ManifoldCF extends org.apac
protected static final String databaseSuperuserPasswordProperty = "org.apache.manifoldcf.dbsuperuserpassword";
protected static final String saltProperty = "org.apache.manifoldcf.salt";
- /** This object is used to make sure the initialization sequence is atomic. Shutdown cannot occur until the system is in a known state. */
- protected static Integer startupLock = new Integer(0);
/** Initialize environment.
*/
@@ -586,535 +540,6 @@ public class ManifoldCF extends org.apac
org.apache.manifoldcf.authorities.system.ManifoldCF.deinstallSystemTables(threadcontext);
}
-
- /** Start everything.
- */
- public static void startSystem(IThreadContext threadContext)
- throws ManifoldCFException
- {
- Logging.root.info("Starting up pull-agent...");
- synchronized (startupLock)
- {
- // Now, start all the threads
- numWorkerThreads = LockManagerFactory.getIntProperty(threadContext,workerThreadCountProperty,100);
- if (numWorkerThreads < 1 || numWorkerThreads > 300)
- throw new ManifoldCFException("Illegal value for the number of worker threads");
- numDeleteThreads = LockManagerFactory.getIntProperty(threadContext,deleteThreadCountProperty,10);
- numCleanupThreads = LockManagerFactory.getIntProperty(threadContext,cleanupThreadCountProperty,10);
- numExpireThreads = LockManagerFactory.getIntProperty(threadContext,expireThreadCountProperty,10);
- if (numDeleteThreads < 1 || numDeleteThreads > 300)
- throw new ManifoldCFException("Illegal value for the number of delete threads");
- if (numCleanupThreads < 1 || numCleanupThreads > 300)
- throw new ManifoldCFException("Illegal value for the number of cleanup threads");
- if (numExpireThreads < 1 || numExpireThreads > 300)
- throw new ManifoldCFException("Illegal value for the number of expire threads");
- lowWaterFactor = (float)LockManagerFactory.getDoubleProperty(threadContext,lowWaterFactorProperty,5.0);
- if (lowWaterFactor < 1.0 || lowWaterFactor > 1000.0)
- throw new ManifoldCFException("Illegal value for the low water factor");
- stuffAmtFactor = (float)LockManagerFactory.getDoubleProperty(threadContext,stuffAmtFactorProperty,2.0);
- if (stuffAmtFactor < 0.1 || stuffAmtFactor > 1000.0)
- throw new ManifoldCFException("Illegal value for the stuffing amount factor");
-
-
- // Create the threads and objects. This MUST be completed before there is any chance of "shutdownSystem" getting called.
-
- QueueTracker queueTracker = new QueueTracker();
-
-
- DocumentQueue documentQueue = new DocumentQueue();
- DocumentDeleteQueue documentDeleteQueue = new DocumentDeleteQueue();
- DocumentCleanupQueue documentCleanupQueue = new DocumentCleanupQueue();
- DocumentCleanupQueue expireQueue = new DocumentCleanupQueue();
-
- BlockingDocuments blockingDocuments = new BlockingDocuments();
-
- workerResetManager = new WorkerResetManager(documentQueue,expireQueue);
- docDeleteResetManager = new DocDeleteResetManager(documentDeleteQueue);
- docCleanupResetManager = new DocCleanupResetManager(documentCleanupQueue);
-
- jobStartThread = new JobStartThread();
- startupThread = new StartupThread(queueTracker);
- startDeleteThread = new StartDeleteThread();
- finisherThread = new FinisherThread();
- notificationThread = new JobNotificationThread();
- jobDeleteThread = new JobDeleteThread();
- stufferThread = new StufferThread(documentQueue,numWorkerThreads,workerResetManager,queueTracker,blockingDocuments,lowWaterFactor,stuffAmtFactor);
- expireStufferThread = new ExpireStufferThread(expireQueue,numExpireThreads,workerResetManager);
- setPriorityThread = new SetPriorityThread(queueTracker,numWorkerThreads,blockingDocuments);
- historyCleanupThread = new HistoryCleanupThread();
-
- workerThreads = new WorkerThread[numWorkerThreads];
- int i = 0;
- while (i < numWorkerThreads)
- {
- workerThreads[i] = new WorkerThread(Integer.toString(i),documentQueue,workerResetManager,queueTracker);
- i++;
- }
-
- expireThreads = new ExpireThread[numExpireThreads];
- i = 0;
- while (i < numExpireThreads)
- {
- expireThreads[i] = new ExpireThread(Integer.toString(i),expireQueue,queueTracker,workerResetManager);
- i++;
- }
-
- deleteStufferThread = new DocumentDeleteStufferThread(documentDeleteQueue,numDeleteThreads,docDeleteResetManager);
- deleteThreads = new DocumentDeleteThread[numDeleteThreads];
- i = 0;
- while (i < numDeleteThreads)
- {
- deleteThreads[i] = new DocumentDeleteThread(Integer.toString(i),documentDeleteQueue,docDeleteResetManager);
- i++;
- }
-
- cleanupStufferThread = new DocumentCleanupStufferThread(documentCleanupQueue,numCleanupThreads,docCleanupResetManager);
- cleanupThreads = new DocumentCleanupThread[numCleanupThreads];
- i = 0;
- while (i < numCleanupThreads)
- {
- cleanupThreads[i] = new DocumentCleanupThread(Integer.toString(i),documentCleanupQueue,queueTracker,docCleanupResetManager);
- i++;
- }
-
- jobResetThread = new JobResetThread(queueTracker);
- seedingThread = new SeedingThread(queueTracker);
- idleCleanupThread = new IdleCleanupThread();
-
- initializationThread = new InitializationThread(queueTracker);
- // Start the initialization thread. This does the initialization work and starts all the other threads when that's done. It then exits.
- initializationThread.start();
- }
- Logging.root.info("Pull-agent started");
- }
-
- protected static class InitializationThread extends Thread
- {
-
- protected final QueueTracker queueTracker;
-
- public InitializationThread(QueueTracker queueTracker)
- {
- super();
- this.queueTracker = queueTracker;
- setName("Initialization thread");
- setDaemon(true);
- }
-
- public void run()
- {
- int i;
-
- // Initialize the database
- try
- {
- IThreadContext threadContext = ThreadContextFactory.make();
-
- // First, get a job manager
- IJobManager jobManager = JobManagerFactory.make(threadContext);
- IRepositoryConnectionManager mgr = RepositoryConnectionManagerFactory.make(threadContext);
-
- Logging.threads.debug("Agents process starting initialization...");
-
- // Call the database to get it ready
- jobManager.prepareForStart();
-
- Logging.threads.debug("Agents process reprioritizing documents...");
-
- Map<String,IRepositoryConnection> connectionMap = new HashMap<String,IRepositoryConnection>();
- Map<Long,IJobDescription> jobDescriptionMap = new HashMap<Long,IJobDescription>();
- // Reprioritize all documents in the jobqueue, 1000 at a time
- long currentTime = System.currentTimeMillis();
-
- // Do the 'not yet processed' documents only. Documents that are queued for reprocessing will be assigned
- // new priorities. Already processed documents won't. This guarantees that our bins are appropriate for current thread
- // activity.
- // In order for this to be the correct functionality, ALL reseeding and requeuing operations MUST reset the associated document
- // priorities.
- while (true)
- {
- long startTime = System.currentTimeMillis();
-
- DocumentDescription[] docs = jobManager.getNextNotYetProcessedReprioritizationDocuments(currentTime, 10000);
- if (docs.length == 0)
- break;
-
- // Calculate new priorities for all these documents
- writeDocumentPriorities(threadContext,mgr,jobManager,docs,connectionMap,jobDescriptionMap,
- queueTracker,currentTime);
-
- Logging.threads.debug("Reprioritized "+Integer.toString(docs.length)+" not-yet-processed documents in "+new Long(System.currentTimeMillis()-startTime)+" ms");
- }
-
- Logging.threads.debug("Agents process initialization complete!");
-
- // Start all the threads
- jobStartThread.start();
- startupThread.start();
- startDeleteThread.start();
- finisherThread.start();
- notificationThread.start();
- jobDeleteThread.start();
- stufferThread.start();
- expireStufferThread.start();
- setPriorityThread.start();
- historyCleanupThread.start();
-
- i = 0;
- while (i < numWorkerThreads)
- {
- workerThreads[i].start();
- i++;
- }
-
- i = 0;
- while (i < numExpireThreads)
- {
- expireThreads[i].start();
- i++;
- }
-
- cleanupStufferThread.start();
- i = 0;
- while (i < numCleanupThreads)
- {
- cleanupThreads[i].start();
- i++;
- }
-
- deleteStufferThread.start();
- i = 0;
- while (i < numDeleteThreads)
- {
- deleteThreads[i].start();
- i++;
- }
-
- jobResetThread.start();
- seedingThread.start();
- idleCleanupThread.start();
- // exit!
- }
- catch (Throwable e)
- {
- // Severe error on initialization
- if (e instanceof ManifoldCFException)
- {
- // Deal with interrupted exception gracefully, because it means somebody is trying to shut us down before we got started.
- if (((ManifoldCFException)e).getErrorCode() == ManifoldCFException.INTERRUPTED)
- return;
- }
- System.err.println("agents process could not start - shutting down");
- Logging.threads.fatal("Startup initialization error tossed: "+e.getMessage(),e);
- System.exit(-300);
- }
- }
- }
-
- /** Stop the system.
- */
- public static void stopSystem(IThreadContext threadContext)
- throws ManifoldCFException
- {
- Logging.root.info("Shutting down pull-agent...");
- synchronized (startupLock)
- {
- 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 || historyCleanupThread != null)
- {
- // Send an interrupt to all threads that are still there.
- // In theory, this only needs to be done once. In practice, I have seen cases where the thread loses track of the fact that it has been
- // interrupted (which may be a JVM bug - who knows?), but in any case there's no harm in doing it again.
- if (initializationThread != null)
- {
- initializationThread.interrupt();
- }
- if (historyCleanupThread != null)
- {
- historyCleanupThread.interrupt();
- }
- if (setPriorityThread != null)
- {
- setPriorityThread.interrupt();
- }
- if (jobStartThread != null)
- {
- jobStartThread.interrupt();
- }
- if (jobDeleteThread != null)
- {
- jobDeleteThread.interrupt();
- }
- if (startupThread != null)
- {
- startupThread.interrupt();
- }
- if (startDeleteThread != null)
- {
- startDeleteThread.interrupt();
- }
- if (stufferThread != null)
- {
- stufferThread.interrupt();
- }
- if (expireStufferThread != null)
- {
- expireStufferThread.interrupt();
- }
- if (finisherThread != null)
- {
- finisherThread.interrupt();
- }
- if (notificationThread != null)
- {
- notificationThread.interrupt();
- }
- if (workerThreads != null)
- {
- int i = 0;
- while (i < workerThreads.length)
- {
- Thread workerThread = workerThreads[i++];
- if (workerThread != null)
- workerThread.interrupt();
- }
- }
- if (expireThreads != null)
- {
- int i = 0;
- while (i < expireThreads.length)
- {
- Thread expireThread = expireThreads[i++];
- if (expireThread != null)
- expireThread.interrupt();
- }
- }
- if (cleanupStufferThread != null)
- {
- cleanupStufferThread.interrupt();
- }
- if (cleanupThreads != null)
- {
- int i = 0;
- while (i < cleanupThreads.length)
- {
- Thread cleanupThread = cleanupThreads[i++];
- if (cleanupThread != null)
- cleanupThread.interrupt();
- }
- }
- if (deleteStufferThread != null)
- {
- deleteStufferThread.interrupt();
- }
- if (deleteThreads != null)
- {
- int i = 0;
- while (i < deleteThreads.length)
- {
- Thread deleteThread = deleteThreads[i++];
- if (deleteThread != null)
- deleteThread.interrupt();
- }
- }
- if (jobResetThread != null)
- {
- jobResetThread.interrupt();
- }
- if (seedingThread != null)
- {
- seedingThread.interrupt();
- }
- if (idleCleanupThread != null)
- {
- idleCleanupThread.interrupt();
- }
-
- // Now, wait for all threads to die.
- try
- {
- ManifoldCF.sleep(1000L);
- }
- catch (InterruptedException e)
- {
- }
-
- // Check to see which died.
- if (initializationThread != null)
- {
- if (!initializationThread.isAlive())
- initializationThread = null;
- }
- if (historyCleanupThread != null)
- {
- if (!historyCleanupThread.isAlive())
- historyCleanupThread = null;
- }
- if (setPriorityThread != null)
- {
- if (!setPriorityThread.isAlive())
- setPriorityThread = null;
- }
- if (jobDeleteThread != null)
- {
- if (!jobDeleteThread.isAlive())
- jobDeleteThread = null;
- }
- if (startupThread != null)
- {
- if (!startupThread.isAlive())
- startupThread = null;
- }
- if (startDeleteThread != null)
- {
- if (!startDeleteThread.isAlive())
- startDeleteThread = null;
- }
- if (jobStartThread != null)
- {
- if (!jobStartThread.isAlive())
- jobStartThread = null;
- }
- if (stufferThread != null)
- {
- if (!stufferThread.isAlive())
- stufferThread = null;
- }
- if (expireStufferThread != null)
- {
- if (!expireStufferThread.isAlive())
- expireStufferThread = null;
- }
- if (finisherThread != null)
- {
- if (!finisherThread.isAlive())
- finisherThread = null;
- }
- if (notificationThread != null)
- {
- if (!notificationThread.isAlive())
- notificationThread = null;
- }
- if (workerThreads != null)
- {
- int i = 0;
- boolean isAlive = false;
- while (i < workerThreads.length)
- {
- Thread workerThread = workerThreads[i];
- if (workerThread != null)
- {
- if (!workerThread.isAlive())
- workerThreads[i] = null;
- else
- isAlive = true;
- }
- i++;
- }
- if (!isAlive)
- workerThreads = null;
- }
-
- if (expireThreads != null)
- {
- int i = 0;
- boolean isAlive = false;
- while (i < expireThreads.length)
- {
- Thread expireThread = expireThreads[i];
- if (expireThread != null)
- {
- if (!expireThread.isAlive())
- expireThreads[i] = null;
- else
- isAlive = true;
- }
- i++;
- }
- if (!isAlive)
- expireThreads = null;
- }
-
- if (cleanupStufferThread != null)
- {
- if (!cleanupStufferThread.isAlive())
- cleanupStufferThread = null;
- }
- if (cleanupThreads != null)
- {
- int i = 0;
- boolean isAlive = false;
- while (i < cleanupThreads.length)
- {
- Thread cleanupThread = cleanupThreads[i];
- if (cleanupThread != null)
- {
- if (!cleanupThread.isAlive())
- cleanupThreads[i] = null;
- else
- isAlive = true;
- }
- i++;
- }
- if (!isAlive)
- cleanupThreads = null;
- }
-
- if (deleteStufferThread != null)
- {
- if (!deleteStufferThread.isAlive())
- deleteStufferThread = null;
- }
- if (deleteThreads != null)
- {
- int i = 0;
- boolean isAlive = false;
- while (i < deleteThreads.length)
- {
- Thread deleteThread = deleteThreads[i];
- if (deleteThread != null)
- {
- if (!deleteThread.isAlive())
- deleteThreads[i] = null;
- else
- isAlive = true;
- }
- i++;
- }
- if (!isAlive)
- deleteThreads = null;
- }
- if (jobResetThread != null)
- {
- if (!jobResetThread.isAlive())
- jobResetThread = null;
- }
- if (seedingThread != null)
- {
- if (!seedingThread.isAlive())
- seedingThread = null;
- }
- if (idleCleanupThread != null)
- {
- if (!idleCleanupThread.isAlive())
- idleCleanupThread = null;
- }
- }
-
- // Threads are down; release connectors
- RepositoryConnectorFactory.closeAllConnectors(threadContext);
- numWorkerThreads = 0;
- numDeleteThreads = 0;
- numExpireThreads = 0;
- }
- Logging.root.info("Pull-agent successfully shut down");
- }
-
-
/** Atomically export the crawler configuration */
public static void exportConfiguration(IThreadContext threadContext, String exportFilename, String passCode)
throws ManifoldCFException
@@ -1382,25 +807,36 @@ public class ManifoldCF extends org.apac
/** Get the maximum number of worker threads.
*/
- public static int getMaxWorkerThreads()
+ public static int getMaxWorkerThreads(IThreadContext threadContext)
+ throws ManifoldCFException
{
- return numWorkerThreads;
+ return LockManagerFactory.getIntProperty(threadContext,workerThreadCountProperty,100);
}
/** Get the maximum number of delete threads.
*/
- public static int getMaxDeleteThreads()
+ public static int getMaxDeleteThreads(IThreadContext threadContext)
+ throws ManifoldCFException
{
- return numDeleteThreads;
+ return LockManagerFactory.getIntProperty(threadContext,deleteThreadCountProperty,10);
}
/** Get the maximum number of expire threads.
*/
- public static int getMaxExpireThreads()
+ public static int getMaxExpireThreads(IThreadContext threadContext)
+ throws ManifoldCFException
{
- return numExpireThreads;
+ return LockManagerFactory.getIntProperty(threadContext,expireThreadCountProperty,10);
}
+ /** Get the maximum number of cleanup threads.
+ */
+ public static int getMaxCleanupThreads(IThreadContext threadContext)
+ throws ManifoldCFException
+ {
+ return LockManagerFactory.getIntProperty(threadContext,cleanupThreadCountProperty,10);
+ }
+
/** Requeue documents due to carrydown.
*/
public static void requeueDocumentsDueToCarrydown(IJobManager jobManager, DocumentDescription[] requeueCandidates,
@@ -1606,55 +1042,6 @@ public class ManifoldCF extends org.apac
}
- /** Request permission from agent to delete an output connection.
- *@param threadContext is the thread context.
- *@param connName is the name of the output connection.
- *@return true if the connection is in use, false otherwise.
- */
- public static boolean isOutputConnectionInUse(IThreadContext threadContext, String connName)
- throws ManifoldCFException
- {
- // Check with job manager.
- IJobManager jobManager = JobManagerFactory.make(threadContext);
- return jobManager.checkIfOutputReference(connName);
- }
-
- /** Note the deregistration of a set of output connections.
- *@param threadContext is the thread context.
- *@param connectionNames are the names of the connections being deregistered.
- */
- public static void noteOutputConnectorDeregistration(IThreadContext threadContext, String[] connectionNames)
- throws ManifoldCFException
- {
- // Notify job manager
- IJobManager jobManager = JobManagerFactory.make(threadContext);
- jobManager.noteOutputConnectorDeregistration(connectionNames);
- }
-
- /** Note the registration of a set of output connections.
- *@param threadContext is the thread context.
- *@param connectionNames are the names of the connections being registered.
- */
- public static void noteOutputConnectorRegistration(IThreadContext threadContext, String[] connectionNames)
- throws ManifoldCFException
- {
- // Notify job manager
- IJobManager jobManager = JobManagerFactory.make(threadContext);
- jobManager.noteOutputConnectorRegistration(connectionNames);
- }
-
- /** Note the change in configuration of an output connection.
- *@param threadContext is the thread context.
- *@param connectionName is the output connection name.
- */
- public static void noteOutputConnectionChange(IThreadContext threadContext, String connectionName)
- throws ManifoldCFException
- {
- // Notify job manager
- IJobManager jobManager = JobManagerFactory.make(threadContext);
- jobManager.noteOutputConnectionChange(connectionName);
- }
-
/** Qualify output activity name.
*@param outputActivityName is the name of the output activity.
*@param outputConnectionName is the corresponding name of the output connection.
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ResetManager.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ResetManager.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ResetManager.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ResetManager.java Tue Nov 26 11:45:53 2013
@@ -37,6 +37,9 @@ public abstract class ResetManager
{
public static final String _rcsid = "@(#)$Id: ResetManager.java 988245 2010-08-23 18:39:35Z kwright $";
+ /** Process ID */
+ protected final String processID;
+
/** Boolean which describes whether an event requiring reset has occurred. */
protected boolean resetRequired = false;
/** This is the count of the threads that care about this resource. */
@@ -46,8 +49,9 @@ public abstract class ResetManager
/** Constructor.
*/
- public ResetManager()
+ public ResetManager(String processID)
{
+ this.processID = processID;
}
/** Register a thread with this reset manager.
@@ -90,7 +94,7 @@ public abstract class ResetManager
// is tracked.
try
{
- performResetLogic(tc);
+ performResetLogic(tc, processID);
}
finally
{
@@ -112,7 +116,7 @@ public abstract class ResetManager
/** Do the reset logic.
*/
- protected abstract void performResetLogic(IThreadContext tc)
+ protected abstract void performResetLogic(IThreadContext tc, String processID)
throws ManifoldCFException;
/** Do the wakeup logic.
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingActivity.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingActivity.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingActivity.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingActivity.java Tue Nov 26 11:45:53 2013
@@ -36,29 +36,32 @@ public class SeedingActivity implements
protected static final int MAX_COUNT = 100;
// Variables
- protected String connectionName;
- protected IRepositoryConnectionManager connManager;
- protected IJobManager jobManager;
- protected QueueTracker queueTracker;
- protected IRepositoryConnection connection;
- protected IRepositoryConnector connector;
- protected Long jobID;
- protected String[] legalLinkTypes;
- protected boolean overrideSchedule;
- protected int hopcountMethod;
- protected String[] documentHashList = new String[MAX_COUNT];
- protected String[] documentList = new String[MAX_COUNT];
- protected String[][] documentPrereqList = new String[MAX_COUNT][];
+ protected final String processID;
+ protected final String connectionName;
+ protected final IRepositoryConnectionManager connManager;
+ protected final IJobManager jobManager;
+ protected final QueueTracker queueTracker;
+ protected final IRepositoryConnection connection;
+ protected final IRepositoryConnector connector;
+ protected final Long jobID;
+ protected final String[] legalLinkTypes;
+ protected final boolean overrideSchedule;
+ protected final int hopcountMethod;
+
+ protected final String[] documentHashList = new String[MAX_COUNT];
+ protected final String[] documentList = new String[MAX_COUNT];
+ protected final String[][] documentPrereqList = new String[MAX_COUNT][];
protected int documentCount = 0;
- protected String[] remainingDocumentHashList = new String[MAX_COUNT];
+ protected final String[] remainingDocumentHashList = new String[MAX_COUNT];
protected int remainingDocumentCount = 0;
/** Constructor.
*/
public SeedingActivity(String connectionName, IRepositoryConnectionManager connManager, IJobManager jobManager,
QueueTracker queueTracker, IRepositoryConnection connection, IRepositoryConnector connector,
- Long jobID, String[] legalLinkTypes, boolean overrideSchedule, int hopcountMethod)
+ Long jobID, String[] legalLinkTypes, boolean overrideSchedule, int hopcountMethod, String processID)
{
+ this.processID = processID;
this.connectionName = connectionName;
this.connManager = connManager;
this.jobManager = jobManager;
@@ -139,7 +142,7 @@ public class SeedingActivity implements
if (remainingDocumentCount == MAX_COUNT)
{
// Flush the remaining documents
- jobManager.addRemainingDocumentsInitial(jobID,legalLinkTypes,remainingDocumentHashList,hopcountMethod);
+ jobManager.addRemainingDocumentsInitial(processID,jobID,legalLinkTypes,remainingDocumentHashList,hopcountMethod);
remainingDocumentCount = 0;
}
remainingDocumentHashList[remainingDocumentCount++] = ManifoldCF.hash(documentIdentifier);
@@ -174,7 +177,7 @@ public class SeedingActivity implements
documents[i] = remainingDocumentHashList[i];
i++;
}
- jobManager.addRemainingDocumentsInitial(jobID,legalLinkTypes,documents,hopcountMethod);
+ jobManager.addRemainingDocumentsInitial(processID,jobID,legalLinkTypes,documents,hopcountMethod);
remainingDocumentCount = 0;
}
@@ -229,7 +232,8 @@ public class SeedingActivity implements
i++;
}
- boolean[] trackerNote = jobManager.addDocumentsInitial(jobID,legalLinkTypes,docIDHashes,docIDs,overrideSchedule,hopcountMethod,
+ boolean[] trackerNote = jobManager.addDocumentsInitial(processID,
+ jobID,legalLinkTypes,docIDHashes,docIDs,overrideSchedule,hopcountMethod,
prioritizationTime,docPriorities,prereqEventNames);
// Inform queuetracker about what we used and what we didn't
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SeedingThread.java Tue Nov 26 11:45:53 2013
@@ -35,24 +35,28 @@ public class SeedingThread extends Threa
{
public static final String _rcsid = "@(#)$Id: SeedingThread.java 988245 2010-08-23 18:39:35Z kwright $";
- /** Worker thread pool reset manager */
- protected static SeedingResetManager resetManager = new SeedingResetManager();
-
// Local data
- protected QueueTracker queueTracker;
+ /** Seeding reset manager */
+ protected final SeedingResetManager resetManager;
+ /** Queue tracker */
+ protected final QueueTracker queueTracker;
+ /** Process ID */
+ protected final String processID;
/** The number of documents that are added to the queue per transaction */
protected final static int MAX_COUNT = 100;
/** Constructor.
*/
- public SeedingThread(QueueTracker queueTracker)
+ public SeedingThread(QueueTracker queueTracker, SeedingResetManager resetManager, String processID)
throws ManifoldCFException
{
super();
setName("Seeding thread");
setDaemon(true);
+ this.resetManager = resetManager;
this.queueTracker = queueTracker;
+ this.processID = processID;
}
public void run()
@@ -91,7 +95,7 @@ public class SeedingThread extends Threa
Logging.threads.debug("Seeding thread woke up");
// Grab active, adaptive jobs (and set their state to xxxSEEDING as a side effect)
- JobSeedingRecord[] seedJobs = jobManager.getJobsReadyForSeeding(currentTime);
+ JobSeedingRecord[] seedJobs = jobManager.getJobsReadyForSeeding(processID,currentTime);
// Process these jobs, and do the seeding. The seeding is based on what came back
// in the job start record for sync time. If there's an interruption, we just go on
@@ -144,7 +148,7 @@ public class SeedingThread extends Threa
{
SeedingActivity activity = new SeedingActivity(connection.getName(),connectionMgr,jobManager,queueTracker,
- connection,connector,jobID,legalLinkTypes,false,hopcountMethod);
+ connection,connector,jobID,legalLinkTypes,false,hopcountMethod,processID);
if (Logging.threads.isDebugEnabled())
Logging.threads.debug("Seeding thread: Getting seeds for job "+jobID.toString());
@@ -287,33 +291,4 @@ public class SeedingThread extends Threa
}
}
- /** 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 SeedingResetManager extends ResetManager
- {
-
- /** Constructor. */
- public SeedingResetManager()
- {
- super();
- }
-
- /** Reset */
- protected void performResetLogic(IThreadContext tc)
- throws ManifoldCFException
- {
- IJobManager jobManager = JobManagerFactory.make(tc);
- jobManager.resetSeedingWorkerStatus();
- }
-
- /** Do the wakeup logic.
- */
- protected void performWakeupLogic()
- {
- }
-
- }
-
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/SetPriorityThread.java Tue Nov 26 11:45:53 2013
@@ -38,22 +38,25 @@ public class SetPriorityThread extends T
public static final String _rcsid = "@(#)$Id: SetPriorityThread.java 988245 2010-08-23 18:39:35Z kwright $";
// Local data
- // This is the queue tracker object.
- protected QueueTracker queueTracker;
- // This is the number of documents per cycle
- protected int cycleCount;
- // The blocking documents object
- protected BlockingDocuments blockingDocuments;
+ /** This is the queue tracker object. */
+ protected final QueueTracker queueTracker;
+ /** This is the number of documents per cycle */
+ protected final int cycleCount;
+ /** The blocking documents object */
+ protected final BlockingDocuments blockingDocuments;
+ /** Process ID */
+ protected final String processID;
/** Constructor.
*@param qt is the queue tracker object.
*/
- public SetPriorityThread(QueueTracker qt, int workerThreadCount, BlockingDocuments blockingDocuments)
+ public SetPriorityThread(QueueTracker qt, int workerThreadCount, BlockingDocuments blockingDocuments, String processID)
throws ManifoldCFException
{
super();
this.queueTracker = qt;
this.blockingDocuments = blockingDocuments;
+ this.processID = processID;
cycleCount = workerThreadCount * 10;
setName("Set priority thread");
setDaemon(true);
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartDeleteThread.java Tue Nov 26 11:45:53 2013
@@ -33,14 +33,18 @@ public class StartDeleteThread extends T
public static final String _rcsid = "@(#)$Id$";
/** Delete startup reset manager */
- protected static DeleteStartupResetManager resetManager = new DeleteStartupResetManager();
-
+ protected final DeleteStartupResetManager resetManager;
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*/
- public StartDeleteThread()
+ public StartDeleteThread(DeleteStartupResetManager resetManager, String processID)
throws ManifoldCFException
{
super();
+ this.resetManager = resetManager;
+ this.processID = processID;
setName("Delete startup thread");
setDaemon(true);
}
@@ -80,7 +84,7 @@ public class StartDeleteThread extends T
// 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.
- JobDeleteRecord[] deleteJobs = jobManager.getJobsReadyForDelete();
+ JobDeleteRecord[] deleteJobs = jobManager.getJobsReadyForDelete(processID);
try
{
@@ -208,33 +212,4 @@ public class StartDeleteThread extends T
}
}
- /** 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();
- }
-
- /** Do the wakeup logic.
- */
- protected void performWakeupLogic()
- {
- }
-
- }
-
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StartupThread.java Tue Nov 26 11:45:53 2013
@@ -32,21 +32,25 @@ public class StartupThread extends Threa
{
public static final String _rcsid = "@(#)$Id: StartupThread.java 988245 2010-08-23 18:39:35Z kwright $";
- /** Worker thread pool reset manager */
- protected static StartupResetManager resetManager = new StartupResetManager();
-
// Local data
- protected QueueTracker queueTracker;
-
+ /** Queue tracker */
+ protected final QueueTracker queueTracker;
+ /** Process ID */
+ protected final String processID;
+ /** Reset manager */
+ protected final StartupResetManager resetManager;
+
/** Constructor.
*/
- public StartupThread(QueueTracker queueTracker)
+ public StartupThread(QueueTracker queueTracker, StartupResetManager resetManager, String processID)
throws ManifoldCFException
{
super();
setName("Startup thread");
setDaemon(true);
this.queueTracker = queueTracker;
+ this.resetManager = resetManager;
+ this.processID = processID;
}
public void run()
@@ -84,7 +88,7 @@ public class StartupThread extends Threa
// 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[] startupJobs = jobManager.getJobsReadyForStartup();
+ JobStartRecord[] startupJobs = jobManager.getJobsReadyForStartup(processID);
try
{
@@ -148,7 +152,7 @@ public class StartupThread extends Threa
try
{
SeedingActivity activity = new SeedingActivity(connection.getName(),connectionMgr,jobManager,queueTracker,
- connection,connector,jobID,legalLinkTypes,true,hopcountMethod);
+ connection,connector,jobID,legalLinkTypes,true,hopcountMethod,processID);
if (Logging.threads.isDebugEnabled())
Logging.threads.debug("Adding initial seed documents for job "+jobID.toString()+"...");
@@ -279,33 +283,4 @@ public class StartupThread extends Threa
}
}
- /** 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 StartupResetManager extends ResetManager
- {
-
- /** Constructor. */
- public StartupResetManager()
- {
- super();
- }
-
- /** Reset */
- protected void performResetLogic(IThreadContext tc)
- throws ManifoldCFException
- {
- IJobManager jobManager = JobManagerFactory.make(tc);
- jobManager.resetStartupWorkerStatus();
- }
-
- /** Do the wakeup logic.
- */
- protected void performWakeupLogic()
- {
- }
-
- }
-
}
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StufferThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StufferThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StufferThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/StufferThread.java Tue Nov 26 11:45:53 2013
@@ -33,28 +33,31 @@ public class StufferThread extends Threa
public static final String _rcsid = "@(#)$Id: StufferThread.java 988245 2010-08-23 18:39:35Z kwright $";
// Local data
- // This is a reference to the static main document queue
- protected DocumentQueue documentQueue;
+
+ /** This is a reference to the static main document queue */
+ protected final DocumentQueue documentQueue;
/** Worker thread pool reset manager */
- protected WorkerResetManager resetManager;
- // This is the lowest number of entries we want ot stuff at any one time
- protected int lowestStuffAmt;
- // This is the number of entries we want to stuff at any one time.
+ protected final WorkerResetManager resetManager;
+ /** This is the lowest number of entries we want ot stuff at any one time */
+ protected final int lowestStuffAmt;
+ /** This is the number of entries we want to stuff at any one time. */
protected int stuffAmt;
- // This is the low water mark for attempting to restuff
- protected int lowWaterMark;
- // This is the queue tracker object.
- protected QueueTracker queueTracker;
- // Blocking documents object.
- protected BlockingDocuments blockingDocuments;
-
+ /** This is the low water mark for attempting to restuff */
+ protected final int lowWaterMark;
+ /** This is the queue tracker object. */
+ protected final QueueTracker queueTracker;
+ /** Blocking documents object. */
+ protected final BlockingDocuments blockingDocuments;
+ /** Process ID */
+ protected final String processID;
+
/** Constructor.
*@param documentQueue is the document queue we'll be stuffing.
*@param n represents the number of threads that will be processing queued stuff, NOT the
* number of documents to be done at once!
*/
public StufferThread(DocumentQueue documentQueue, int n, WorkerResetManager resetManager, QueueTracker qt,
- BlockingDocuments blockingDocuments, float lowWaterFactor, float stuffSizeFactor)
+ BlockingDocuments blockingDocuments, float lowWaterFactor, float stuffSizeFactor, String processID)
throws ManifoldCFException
{
super();
@@ -65,6 +68,7 @@ public class StufferThread extends Threa
this.resetManager = resetManager;
this.queueTracker = qt;
this.blockingDocuments = blockingDocuments;
+ this.processID = processID;
setName("Stuffer thread");
setDaemon(true);
// The priority of this thread is higher than most others. We want stuffing to proceed even if the machine
@@ -154,7 +158,7 @@ public class StufferThread extends Threa
DepthStatistics depthStatistics = new DepthStatistics();
long currentTime = System.currentTimeMillis();
lastQueueStart = currentTime;
- DocumentDescription[] descs = jobManager.getNextDocuments(stuffAmt,currentTime,currentTime-lastTime,
+ DocumentDescription[] descs = jobManager.getNextDocuments(processID,stuffAmt,currentTime,currentTime-lastTime,
blockingDocuments,queueTracker.getCurrentStatistics(),depthStatistics);
lastQueueEnd = System.currentTimeMillis();
lastQueueFullResults = (descs.length == stuffAmt);
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerResetManager.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerResetManager.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerResetManager.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerResetManager.java Tue Nov 26 11:45:53 2013
@@ -31,30 +31,32 @@ public class WorkerResetManager extends
public static final String _rcsid = "@(#)$Id: WorkerResetManager.java 988245 2010-08-23 18:39:35Z kwright $";
/** The document queue */
- protected DocumentQueue dq;
+ protected final DocumentQueue dq;
/** The expiration queue */
- protected DocumentCleanupQueue eq;
+ protected final DocumentCleanupQueue eq;
/** Constructor. */
- public WorkerResetManager(DocumentQueue dq, DocumentCleanupQueue eq)
+ public WorkerResetManager(DocumentQueue dq, DocumentCleanupQueue eq, String processID)
{
- super();
+ super(processID);
this.dq = dq;
this.eq = eq;
}
/** Reset */
- protected void performResetLogic(IThreadContext tc)
+ @Override
+ protected void performResetLogic(IThreadContext tc, String processID)
throws ManifoldCFException
{
IJobManager jobManager = JobManagerFactory.make(tc);
- jobManager.resetDocumentWorkerStatus();
+ jobManager.resetDocumentWorkerStatus(processID);
dq.clear();
eq.clear();
}
/** Do the wakeup logic.
*/
+ @Override
protected void performWakeupLogic()
{
// Wake up all sleeping worker threads
Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerThread.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerThread.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerThread.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/WorkerThread.java Tue Nov 26 11:45:53 2013
@@ -35,18 +35,21 @@ public class WorkerThread extends Thread
// Local data
- protected String id;
- // This is a reference to the static main document queue
- protected DocumentQueue documentQueue;
+ /** Thread id */
+ protected final String id;
+ /** This is a reference to the static main document queue */
+ protected final DocumentQueue documentQueue;
/** Worker thread pool reset manager */
- protected WorkerResetManager resetManager;
+ protected final WorkerResetManager resetManager;
/** Queue tracker */
- protected QueueTracker queueTracker;
+ protected final QueueTracker queueTracker;
+ /** Process ID */
+ protected final String processID;
/** Constructor.
*@param id is the worker thread id.
*/
- public WorkerThread(String id, DocumentQueue documentQueue, WorkerResetManager resetManager, QueueTracker queueTracker)
+ public WorkerThread(String id, DocumentQueue documentQueue, WorkerResetManager resetManager, QueueTracker queueTracker, String processID)
throws ManifoldCFException
{
super();
@@ -54,6 +57,7 @@ public class WorkerThread extends Thread
this.documentQueue = documentQueue;
this.resetManager = resetManager;
this.queueTracker = queueTracker;
+ this.processID = processID;
setName("Worker thread '"+id+"'");
setDaemon(true);
@@ -306,7 +310,7 @@ public class WorkerThread extends Thread
String outputVersion = ingester.getOutputDescription(outputName,outputSpec);
HashMap abortSet = new HashMap();
- VersionActivity versionActivity = new VersionActivity(connectionName,connMgr,jobManager,job,ingester,abortSet,outputVersion);
+ VersionActivity versionActivity = new VersionActivity(processID,connectionName,connMgr,jobManager,job,ingester,abortSet,outputVersion);
String aclAuthority = connection.getACLAuthority();
boolean isDefaultAuthority = (aclAuthority == null || aclAuthority.length() == 0);
@@ -522,7 +526,8 @@ public class WorkerThread extends Thread
}
// First, make the things we will need for all subsequent steps.
- ProcessActivity activity = new ProcessActivity(threadContext,queueTracker,jobManager,ingester,
+ ProcessActivity activity = new ProcessActivity(processID,
+ threadContext,queueTracker,jobManager,ingester,
currentTime,job,connection,connector,connMgr,legalLinkTypes,ingestLogger,abortSet,outputVersion,newParameterVersion);
try
{
@@ -1227,21 +1232,22 @@ public class WorkerThread extends Thread
*/
protected static class VersionActivity implements IVersionActivity
{
- protected String connectionName;
- protected IRepositoryConnectionManager connMgr;
- protected IJobManager jobManager;
- protected Long jobID;
- protected IJobDescription job;
- protected IIncrementalIngester ingester;
- protected HashMap abortSet;
- protected String outputVersion;
+ protected final String processID;
+ protected final String connectionName;
+ protected final IRepositoryConnectionManager connMgr;
+ protected final IJobManager jobManager;
+ protected final IJobDescription job;
+ protected final IIncrementalIngester ingester;
+ protected final HashMap abortSet;
+ protected final String outputVersion;
/** Constructor.
*/
- public VersionActivity(String connectionName, IRepositoryConnectionManager connMgr,
+ public VersionActivity(String processID, String connectionName, IRepositoryConnectionManager connMgr,
IJobManager jobManager, IJobDescription job, IIncrementalIngester ingester, HashMap abortSet,
String outputVersion)
{
+ this.processID = processID;
this.connectionName = connectionName;
this.connMgr = connMgr;
this.jobManager = jobManager;
@@ -1360,7 +1366,7 @@ public class WorkerThread extends Thread
public boolean beginEventSequence(String eventName)
throws ManifoldCFException
{
- return jobManager.beginEventSequence(eventName);
+ return jobManager.beginEventSequence(processID,eventName);
}
/** Complete an event sequence.
@@ -1413,7 +1419,7 @@ public class WorkerThread extends Thread
*/
public String createJobSpecificString(String simpleString)
{
- return ManifoldCF.createJobSpecificString(jobID,simpleString);
+ return ManifoldCF.createJobSpecificString(job.getID(),simpleString);
}
}
@@ -1423,6 +1429,7 @@ public class WorkerThread extends Thread
protected static class ProcessActivity implements IProcessActivity
{
// Member variables
+ protected final String processID;
protected final IThreadContext threadContext;
protected final IJobManager jobManager;
protected final IIncrementalIngester ingester;
@@ -1454,10 +1461,11 @@ public class WorkerThread extends Thread
*@param jobManager is the job manager
*@param ingester is the ingester
*/
- public ProcessActivity(IThreadContext threadContext, QueueTracker queueTracker, IJobManager jobManager, IIncrementalIngester ingester,
+ public ProcessActivity(String processID, IThreadContext threadContext, QueueTracker queueTracker, IJobManager jobManager, IIncrementalIngester ingester,
long currentTime, IJobDescription job, IRepositoryConnection connection, IRepositoryConnector connector, IRepositoryConnectionManager connMgr,
String[] legalLinkTypes, OutputActivity ingestLogger, HashMap abortSet, String outputVersion, String parameterVersion)
{
+ this.processID = processID;
this.threadContext = threadContext;
this.queueTracker = queueTracker;
this.jobManager = jobManager;
@@ -2014,7 +2022,8 @@ public class WorkerThread extends Thread
j++;
}
- boolean[] trackerNote = jobManager.addDocuments(job.getID(),legalLinkTypes,docidHashes,docids,db.getParentIdentifierHash(),db.getLinkType(),job.getHopcountMode(),
+ boolean[] trackerNote = jobManager.addDocuments(processID,
+ job.getID(),legalLinkTypes,docidHashes,docids,db.getParentIdentifierHash(),db.getLinkType(),job.getHopcountMode(),
dataNames,dataValues,currentTime,priorities,eventNames);
// Inform queuetracker about what we used and what we didn't
@@ -2058,7 +2067,7 @@ public class WorkerThread extends Thread
public boolean beginEventSequence(String eventName)
throws ManifoldCFException
{
- return jobManager.beginEventSequence(eventName);
+ return jobManager.beginEventSequence(processID,eventName);
}
/** Complete an event sequence.
Modified: manifoldcf/trunk/framework/pull-agent/src/test/java/org/apache/manifoldcf/crawler/tests/ManifoldCFInstance.java
URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/test/java/org/apache/manifoldcf/crawler/tests/ManifoldCFInstance.java?rev=1545624&r1=1545623&r2=1545624&view=diff
==============================================================================
--- manifoldcf/trunk/framework/pull-agent/src/test/java/org/apache/manifoldcf/crawler/tests/ManifoldCFInstance.java (original)
+++ manifoldcf/trunk/framework/pull-agent/src/test/java/org/apache/manifoldcf/crawler/tests/ManifoldCFInstance.java Tue Nov 26 11:45:53 2013
@@ -56,8 +56,6 @@ import org.apache.http.entity.ContentTyp
/** Tests that run the "agents daemon" should be derived from this */
public class ManifoldCFInstance
{
- public static final String agentShutdownSignal = "agent-process";
-
protected boolean singleWar = false;
protected int testPort = 8346;
@@ -548,8 +546,7 @@ public class ManifoldCFInstance
// If all worked, then we can start the daemon.
// Clear the agents shutdown signal.
IThreadContext tc = ThreadContextFactory.make();
- ILockManager lockManager = LockManagerFactory.make(tc);
- lockManager.clearGlobalFlag(agentShutdownSignal);
+ ManifoldCF.clearAgentsShutdownSignal(tc);
daemonThread = new DaemonThread();
daemonThread.start();
@@ -644,8 +641,7 @@ public class ManifoldCFInstance
if (!singleWar)
{
// Shut down daemon
- ILockManager lockManager = LockManagerFactory.make(tc);
- lockManager.setGlobalFlag(agentShutdownSignal);
+ ManifoldCF.assertAgentsShutdownSignal(tc);
// Wait for daemon thread to exit.
while (true)
@@ -692,30 +688,13 @@ public class ManifoldCFInstance
public void run()
{
+ String processID = ManifoldCF.getProcessID();
IThreadContext tc = ThreadContextFactory.make();
// Now, start the server, and then wait for the shutdown signal. On shutdown, we have to actually do the cleanup,
// because the JVM isn't going away.
try
{
- ILockManager lockManager = LockManagerFactory.make(tc);
- while (true)
- {
- // Any shutdown signal yet?
- if (lockManager.checkGlobalFlag(agentShutdownSignal))
- break;
-
- // Start whatever agents need to be started
- ManifoldCF.startAgents(tc);
-
- try
- {
- ManifoldCF.sleep(5000);
- }
- catch (InterruptedException e)
- {
- break;
- }
- }
+ ManifoldCF.runAgents(tc, processID);
}
catch (ManifoldCFException e)
{
@@ -725,7 +704,7 @@ public class ManifoldCFInstance
{
try
{
- ManifoldCF.stopAgents(tc);
+ ManifoldCF.stopAgents(tc, processID);
}
catch (ManifoldCFException e)
{