Modified: manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java (original) +++ manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/lockmanager/ZooKeeperLockManager.java Tue Nov 26 11:45:53 2013 @@ -40,7 +40,10 @@ public class ZooKeeperLockManager extend private final static String CONFIGURATION_PATH = "/org.apache.manifoldcf.configuration"; private final static String RESOURCE_PATH_PREFIX = "/org.apache.manifoldcf.resources-"; private final static String FLAG_PATH_PREFIX = "/org.apache.manifoldcf.flags-"; - + private final static String SERVICETYPE_LOCK_PATH_PREFIX = "/org.apache.manifoldcf.servicelock-"; + private final static String SERVICETYPE_ACTIVE_PATH_PREFIX = "/org.apache.manifoldcf.serviceactive-"; + private final static String SERVICETYPE_REGISTER_PATH_PREFIX = "/org.apache.manifoldcf.service-"; + // ZooKeeper connection pool protected static Integer connectionPoolLock = new Integer(0); protected static ZooKeeperConnectionPool pool = null; @@ -73,6 +76,343 @@ public class ZooKeeperLockManager extend } } + // The node synchronization model involves keeping track of active agents entities, so that other entities + // can perform any necessary cleanup if one of the agents processes goes away unexpectedly. There is a + // registration primitive (which can fail if the same guid is used as is already registered and active), a + // shutdown primitive (which makes a process id go inactive), and various inspection primitives. + + // For the zookeeper implementation, we'll need the following: + // - a service-type-specific global write lock transient node + // - a service-type-specific permanent root node that has registered services as children + // - a service-type-specific transient root node that has active services as children + // + // This is not necessarily the best implementation that meets the constraints, but it is straightforward + // and will serve until we come up with a better one. + + /** Register a service and begin service activity. + * This atomic operation creates a permanent registration entry for a service. + * If the permanent registration entry already exists, this method will not create it or + * treat it as an error. This operation also enters the "active" zone for the service. The "active" zone will remain in force until it is + * canceled, or until the process is interrupted. Ideally, the corresponding endServiceActivity method will be + * called when the service shuts down. Some ILockManager implementations require that this take place for + * proper management. + * If the transient registration already exists, it is treated as an error and an exception will be thrown. + * If registration will succeed, then this method may call an appropriate IServiceCleanup method to clean up either the + * current service, or all services on the cluster. + *@param serviceType is the type of service. + *@param serviceName is the name of the service to register. + *@param cleanup is called to clean up either the current service, or all services of this type, if no other active service exists + */ + @Override + public void registerServiceBeginServiceActivity(String serviceType, String serviceName, IServiceCleanup cleanup) + throws ManifoldCFException + { + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + enterServiceRegistryLock(connection, serviceType); + try + { + String activePath = buildServiceTypeActivePath(serviceType, serviceName); + if (connection.checkNodeExists(activePath)) + throw new ManifoldCFException("Service '"+serviceName+"' of type '"+serviceType+"' is already active"); + // First, see where we stand. + // We need to find out whether (a) our service is already registered; (b) how many registered services there are; + // (c) whether there are other active services. But no changes will be made at this time. + String registrationNodePath = buildServiceTypeRegistrationPath(serviceType); + List<String> children = connection.getChildren(registrationNodePath); + boolean foundService = false; + boolean foundActiveService = false; + for (String registeredServiceName : children) + { + if (registeredServiceName.equals(serviceName)) + foundService = true; + if (connection.checkNodeExists(buildServiceTypeActivePath(serviceType, registeredServiceName))) + foundActiveService = true; + } + + // Call the appropriate cleanup. This will depend on what's actually registered, and what's active. + // If there were no services registered at all when we started, then no cleanup is needed, just cluster init. + // If this fails, we must revert to having our service not be registered and not be active. + boolean unregisterAll = false; + if (cleanup != null) + { + if (children.size() == 0) + { + // If we could count on locks never being cleaned up, clusterInit() + // would be sufficient here. But then there's no way to recover from + // a lock clean. + cleanup.cleanUpAllServices(); + cleanup.clusterInit(); + } + else if (foundService && foundActiveService) + cleanup.cleanUpService(serviceName); + else if (!foundActiveService) + { + cleanup.cleanUpAllServices(); + cleanup.clusterInit(); + unregisterAll = true; + } + } + + if (unregisterAll) + { + // Unregister all (since we did a global cleanup) + for (String registeredServiceName : children) + { + if (!registeredServiceName.equals(serviceName)) + connection.deleteChild(registrationNodePath, registeredServiceName); + } + } + + // Now, register (if needed) + if (!foundService) + { + connection.createChild(registrationNodePath, serviceName); + } + + // Last, set the appropriate active flag + connection.createNode(activePath); + } + finally + { + leaveServiceRegistryLock(connection); + } + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } + } + + /** Count all active services of a given type. + *@param serviceType is the service type. + *@return the count. + */ + @Override + public int countActiveServices(String serviceType) + throws ManifoldCFException + { + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + enterServiceRegistryLock(connection, serviceType); + try + { + String registrationNodePath = buildServiceTypeRegistrationPath(serviceType); + List<String> children = connection.getChildren(registrationNodePath); + int activeServiceCount = 0; + for (String registeredServiceName : children) + { + if (connection.checkNodeExists(buildServiceTypeActivePath(serviceType, registeredServiceName))) + activeServiceCount++; + } + return activeServiceCount; + } + finally + { + leaveServiceRegistryLock(connection); + } + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } + } + + /** Clean up any inactive services found. + * Calling this method will invoke cleanup of one inactive service at a time. + * If there are no inactive services around, then false will be returned. + * Note that this method will block whatever service it finds from starting up + * for the time the cleanup is proceeding. At the end of the cleanup, if + * successful, the service will be atomically unregistered. + *@param serviceType is the service type. + *@param cleanup is the object to call to clean up an inactive service. + *@return true if there were no cleanup operations necessary. + */ + @Override + public boolean cleanupInactiveService(String serviceType, IServiceCleanup cleanup) + throws ManifoldCFException + { + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + enterServiceRegistryLock(connection, serviceType); + try + { + // We find ONE service that is registered but inactive, and clean up after that one. + // Presumably the caller will lather, rinse, and repeat. + String registrationNodePath = buildServiceTypeRegistrationPath(serviceType); + List<String> children = connection.getChildren(registrationNodePath); + String serviceName = null; + for (String registeredServiceName : children) + { + if (!connection.checkNodeExists(buildServiceTypeActivePath(serviceType, registeredServiceName))) + { + serviceName = registeredServiceName; + break; + } + } + if (serviceName == null) + return true; + + // Found one, in serviceName, at position i + // Ideally, we should signal at this point that we're cleaning up after it, and then leave + // the exclusive lock, so that other activity can take place. MHL + cleanup.cleanUpService(serviceName); + + // Unregister the service. + connection.deleteChild(registrationNodePath, serviceName); + return false; + } + finally + { + leaveServiceRegistryLock(connection); + } + + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } + } + + /** End service activity. + * This operation exits the "active" zone for the service. This must take place using the same ILockManager + * object that was used to registerServiceBeginServiceActivity() - which implies that it is the same thread. + *@param serviceType is the type of service. + *@param serviceName is the name of the service to exit. + */ + @Override + public void endServiceActivity(String serviceType, String serviceName) + throws ManifoldCFException + { + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + enterServiceRegistryLock(connection, serviceType); + try + { + connection.deleteNode(buildServiceTypeActivePath(serviceType, serviceName)); + } + finally + { + leaveServiceRegistryLock(connection); + } + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } + } + + /** Check whether a service is active or not. + * This operation returns true if the specified service is considered active at the moment. Once a service + * is not active anymore, it can only return to activity by calling beginServiceActivity() once more. + *@param serviceType is the type of service. + *@param serviceName is the name of the service to check on. + *@return true if the service is considered active. + */ + @Override + public boolean checkServiceActive(String serviceType, String serviceName) + throws ManifoldCFException + { + try + { + ZooKeeperConnection connection = pool.grab(); + try + { + enterServiceRegistryLock(connection, serviceType); + try + { + return connection.checkNodeExists(buildServiceTypeActivePath(serviceType, serviceName)); + } + finally + { + leaveServiceRegistryLock(connection); + } + } + finally + { + pool.release(connection); + } + } + catch (InterruptedException e) + { + throw new ManifoldCFException(e.getMessage(),e,ManifoldCFException.INTERRUPTED); + } + } + + /** Enter service registry lock */ + protected void enterServiceRegistryLock(ZooKeeperConnection connection, String serviceType) + throws ManifoldCFException, InterruptedException + { + while (true) + { + if (connection.obtainWriteLockNoWait(buildServiceTypeLockPath(serviceType))) + return; + ManifoldCF.sleep(100L); + } + } + + /** Leave service registry lock */ + protected void leaveServiceRegistryLock(ZooKeeperConnection connection) + throws ManifoldCFException, InterruptedException + { + connection.releaseLock(); + } + + /** Build a zk path for the lock for a specific service type. + */ + protected static String buildServiceTypeLockPath(String serviceType) + { + return SERVICETYPE_LOCK_PATH_PREFIX + serviceType; + } + + /** Build a zk path for the active node for a specific service of a specific type. + */ + protected static String buildServiceTypeActivePath(String serviceType, String serviceName) + { + return SERVICETYPE_ACTIVE_PATH_PREFIX + serviceType + "-" + serviceName; + } + + /** Build a zk path for the registration node for a specific service type. + */ + protected static String buildServiceTypeRegistrationPath(String serviceType) + { + return SERVICETYPE_REGISTER_PATH_PREFIX + serviceType; + } + + // Shared configuration + /** Get the current shared configuration. This configuration is available in common among all nodes, * and thus must not be accessed through here for the purpose of finding configuration data that is specific to any one * specific node.
Modified: manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/system/ManifoldCF.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/system/ManifoldCF.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/system/ManifoldCF.java (original) +++ manifoldcf/trunk/framework/core/src/main/java/org/apache/manifoldcf/core/system/ManifoldCF.java Tue Nov 26 11:45:53 2013 @@ -31,6 +31,11 @@ public class ManifoldCF public static final String NODE_LIBDIR = "libdir"; public static final String ATTRIBUTE_PATH = "path"; + // This is the unique process identifier, which has to be unique and repeatable within a cluster + + /** Process ID (no more than 16 characters) */ + protected static String processID = null; + // "Working directory" /** This is the working directory file object. */ @@ -94,6 +99,10 @@ public class ManifoldCF // System property/config file property names + // Process ID property + /** Process ID - cannot exceed 16 characters */ + public static final String processIDProperty = "org.apache.manifoldcf.processid"; + // Admin properties /** UI login user name */ public static final String loginUserNameProperty = "org.apache.manifoldcf.login.name"; @@ -155,6 +164,7 @@ public class ManifoldCF { // Clean up the system doing the same thing the shutdown thread would have if the process was killed cleanUpEnvironment(threadContext); + processID = null; loginUserName = null; loginPassword = null; masterDatabaseName = null; @@ -215,6 +225,11 @@ public class ManifoldCF localConfiguration = new OverrideableManifoldCFConfiguration(); checkProperties(); + // Process ID is always local + processID = getStringProperty(processIDProperty,""); + if (processID.length() > 16) + throw new ManifoldCFException("Process ID cannot exceed 16 characters!"); + // Log file is always local File logConfigFile = getFileProperty(logConfigFileProperty); if (logConfigFile == null) @@ -277,6 +292,12 @@ public class ManifoldCF } + /** Get process ID */ + public static final String getProcessID() + { + return processID; + } + /** Get current properties. Makes no attempt to reread or interpret them. */ public static final ManifoldCFConfiguration getConfiguration() Modified: manifoldcf/trunk/framework/jetty-runner/src/main/java/org/apache/manifoldcf/jettyrunner/ManifoldCFJettyRunner.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/jetty-runner/src/main/java/org/apache/manifoldcf/jettyrunner/ManifoldCFJettyRunner.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/jetty-runner/src/main/java/org/apache/manifoldcf/jettyrunner/ManifoldCFJettyRunner.java (original) +++ manifoldcf/trunk/framework/jetty-runner/src/main/java/org/apache/manifoldcf/jettyrunner/ManifoldCFJettyRunner.java Tue Nov 26 11:45:53 2013 @@ -50,8 +50,6 @@ public class ManifoldCFJettyRunner public static final String useJettyParentClassLoaderProperty = "org.apache.manifoldcf.usejettyparentclassloader"; public static final String jettyPortProperty = "org.apache.manifoldcf.jettyport"; - public static final String agentShutdownSignal = org.apache.manifoldcf.agents.AgentRun.agentShutdownSignal; - protected Server server; public ManifoldCFJettyRunner( int port, String crawlerWarPath, String authorityServiceWarPath, String apiWarPath, boolean useParentLoader ) @@ -138,26 +136,10 @@ public class ManifoldCFJettyRunner public static void runAgents(IThreadContext tc) throws ManifoldCFException { - 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; - } - } + String processID = ManifoldCF.getProcessID(); + // Do this so we don't have to call stopAgents() ourselves. + ManifoldCF.registerAgentsShutdownHook(tc, processID); + ManifoldCF.runAgents(tc, processID); } /** @@ -216,8 +198,7 @@ public class ManifoldCFJettyRunner if (useParentClassLoader) { // Clear the agents shutdown signal. - ILockManager lockManager = LockManagerFactory.make(tc); - lockManager.clearGlobalFlag(agentShutdownSignal); + ManifoldCF.clearAgentsShutdownSignal(tc); // Do the basic initialization of the database and its schema ManifoldCF.createSystemDatabase(tc); Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java (original) +++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/IJobManager.java Tue Nov 26 11:45:53 2013 @@ -136,48 +136,68 @@ public interface IJobManager // The job queue is maintained underneath this interface, and all threads that perform // job activities need to go through this layer. - /** Reset the job queue immediately before starting up. - * If the system was shut down in the middle of a job, sufficient information should - * be around in the database to allow it to restart. However, BEFORE all the job threads - * are spun up, there needs to be a pass over the queue to bring things back to a "normal" - * state. + /** Reset the job queue for an individual process ID. + * If a node was shut down in the middle of doing something, sufficient information should + * be around in the database to allow the node's activities to be cleaned up. + *@param processID is the process ID of the node we want to clean up after. + */ + public void cleanupProcessData(String processID) + throws ManifoldCFException; + + /** Reset the job queue for all process IDs. + * If a node was shut down in the middle of doing something, sufficient information should + * be around in the database to allow the node's activities to be cleaned up. + */ + public void cleanupProcessData() + throws ManifoldCFException; + + /** Prepare to start the entire cluster. + * If there are no other nodes alive, then at the time the first node comes up, we need to + * reset the job queue for ALL processes that had been running before. This method must + * be called in addition to cleanupProcessData(). */ - public void prepareForStart() + public void prepareForClusterStart() throws ManifoldCFException; /** Reset as part of restoring document worker threads. + *@param processID is the current process ID. */ - public void resetDocumentWorkerStatus() + public void resetDocumentWorkerStatus(String processID) throws ManifoldCFException; /** Reset as part of restoring seeding threads. */ - public void resetSeedingWorkerStatus() + public void resetSeedingWorkerStatus(String processID) throws ManifoldCFException; /** Reset as part of restoring doc delete threads. + *@param processID is the current process ID. */ - public void resetDocDeleteWorkerStatus() + public void resetDocDeleteWorkerStatus(String processID) throws ManifoldCFException; /** Reset as part of restoring doc cleanup threads. + *@param processID is the current process ID. */ - public void resetDocCleanupWorkerStatus() + public void resetDocCleanupWorkerStatus(String processID) throws ManifoldCFException; /** Reset as part of restoring delete startup threads. + *@param processID is the current process ID. */ - public void resetDeleteStartupWorkerStatus() + public void resetDeleteStartupWorkerStatus(String processID) throws ManifoldCFException; /** Reset as part of restoring notification threads. + *@param processID is the current process ID. */ - public void resetNotificationWorkerStatus() + public void resetNotificationWorkerStatus(String processID) throws ManifoldCFException; /** Reset as part of restoring startup threads. + *@param processID is the current process ID. */ - public void resetStartupWorkerStatus() + public void resetStartupWorkerStatus(String processID) throws ManifoldCFException; // These methods support the "set doc priority" thread @@ -218,11 +238,12 @@ public interface IJobManager * The same marking is used as is used for documents that have been queued for worker threads. The model * is thus identical. * + *@param processID is the current process ID. *@param n is the maximum number of records desired. *@param currentTime is the current time. *@return the array of document descriptions to expire. */ - public DocumentSetAndFlags getExpiredDocuments(int n, long currentTime) + public DocumentSetAndFlags getExpiredDocuments(String processID, int n, long currentTime) throws ManifoldCFException; // This method supports the "queue stuffer" thread @@ -232,6 +253,7 @@ public interface IJobManager * pertaining to the document's handling (e.g. whether it should be refetched if the version * has not changed). * This method also marks the documents whose descriptions have be returned as "being processed". + *@param processID is the current process ID. *@param n is the number of documents desired. *@param currentTime is the current time; some fetches do not occur until a specific time. *@param interval is the number of milliseconds that this set of documents should represent (for throttling). @@ -243,7 +265,8 @@ public interface IJobManager * to being overcommitted. *@return the array of document descriptions to fetch and process. */ - public DocumentDescription[] getNextDocuments(int n, long currentTime, long interval, + public DocumentDescription[] getNextDocuments(String processID, + int n, long currentTime, long interval, BlockingDocuments blockingDocuments, PerformanceStatistics statistics, DepthStatistics scanRecord) throws ManifoldCFException; @@ -495,6 +518,7 @@ public interface IJobManager * This method is called during job startup, when the queue is being loaded. * A set of document references is passed to this method, which updates the status of the document * in the specified job's queue, according to specific state rules. + *@param processID is the current process ID. *@param jobID is the job identifier. *@param legalLinkTypes is the set of legal link types that this connector generates. *@param docIDHashes are the hashes of the local document identifiers (primary key). @@ -506,7 +530,8 @@ public interface IJobManager *@param prereqEventNames are the events that must be completed before each document can be processed. *@return true if the priority value(s) were used, false otherwise. */ - public boolean[] addDocumentsInitial(Long jobID, String[] legalLinkTypes, + public boolean[] addDocumentsInitial(String processID, + Long jobID, String[] legalLinkTypes, String[] docIDHashes, String[] docIDs, boolean overrideSchedule, int hopcountMethod, long currentTime, double[] documentPriorities, String[][] prereqEventNames) @@ -516,12 +541,14 @@ public interface IJobManager * This method is called during job startup, when the queue is being loaded, to list documents that * were NOT included by calling addDocumentsInitial(). Documents listed here are simply designed to * enable the framework to get rid of old, invalid seeds. They are not queued for processing. + *@param processID is the current process ID. *@param jobID is the job identifier. *@param legalLinkTypes is the set of legal link types that this connector generates. *@param docIDHashes are the hash values of the local document identifiers. *@param hopcountMethod is either accurate, nodelete, or neverdelete. */ - public void addRemainingDocumentsInitial(Long jobID, String[] legalLinkTypes, + public void addRemainingDocumentsInitial(String processID, + Long jobID, String[] legalLinkTypes, String[] docIDHashes, int hopcountMethod) throws ManifoldCFException; @@ -540,10 +567,11 @@ public interface IJobManager throws ManifoldCFException; /** Begin an event sequence. + *@param processID is the current process ID. *@param eventName is the name of the event. *@return true if the event could be created, or false if it's already there. */ - public boolean beginEventSequence(String eventName) + public boolean beginEventSequence(String processID, String eventName) throws ManifoldCFException; /** Complete an event sequence. @@ -578,6 +606,7 @@ public interface IJobManager * This method is called during document processing, when a document reference is discovered. * The document reference is passed to this method, which updates the status of the document * in the specified job's queue, according to specific state rules. + *@param processID is the current process ID. *@param jobID is the job identifier. *@param legalLinkTypes is the set of legal link types that this connector generates. *@param docIDHash is the local document identifier hash value. @@ -594,7 +623,8 @@ public interface IJobManager *@param prereqEventNames are the events that must be completed before the document can be processed. *@return true if the priority value was used, false otherwise. */ - public boolean addDocument(Long jobID, String[] legalLinkTypes, + public boolean addDocument(String processID, + Long jobID, String[] legalLinkTypes, String docIDHash, String docID, String parentIdentifierHash, String relationshipType, @@ -606,6 +636,7 @@ public interface IJobManager * This method is called during document processing, when a set of document references are discovered. * The document references are passed to this method, which updates the status of the document(s) * in the specified job's queue, according to specific state rules. + *@param processID is the current process ID. *@param jobID is the job identifier. *@param legalLinkTypes is the set of legal link types that this connector generates. *@param docIDHashes are the hashes of the local document identifiers. @@ -623,7 +654,8 @@ public interface IJobManager *@param prereqEventNames are the events that must be completed before each document can be processed. *@return an array of boolean values indicating whether or not the passed-in priority value was used or not for each doc id (true if used). */ - public boolean[] addDocuments(Long jobID, String[] legalLinkTypes, + public boolean[] addDocuments(String processID, + Long jobID, String[] legalLinkTypes, String[] docIDHashes, String[] docIDs, String parentIdentifierHash, String relationshipType, @@ -747,11 +779,12 @@ public interface IJobManager throws ManifoldCFException; /** Get the list of jobs that are ready for seeding. + *@param processID is the current process ID. *@param currentTime is the current time in milliseconds since epoch. *@return jobs that are active and are running in adaptive mode. These will be seeded * based on what the connector says should be added to the queue. */ - public JobSeedingRecord[] getJobsReadyForSeeding(long currentTime) + public JobSeedingRecord[] getJobsReadyForSeeding(String processID, long currentTime) throws ManifoldCFException; /** Reset a seeding job back to "active" state. @@ -761,21 +794,24 @@ public interface IJobManager throws ManifoldCFException; /** Get the list of jobs that are ready for deletion. + *@param processID is the current process ID. *@return jobs that were in the "readyfordelete" state. */ - public JobDeleteRecord[] getJobsReadyForDelete() + public JobDeleteRecord[] getJobsReadyForDelete(String processID) throws ManifoldCFException; /** Get the list of jobs that are ready for startup. + *@param processID is the current process ID. *@return jobs that were in the "readyforstartup" state. These will be marked as being in the "starting up" state. */ - public JobStartRecord[] getJobsReadyForStartup() + public JobStartRecord[] getJobsReadyForStartup(String processID) throws ManifoldCFException; /** Find the list of jobs that need to have their connectors notified of job completion. + *@param processID is the current process ID. *@return the ID's of jobs that need their output connectors notified in order to become inactive. */ - public JobNotifyRecord[] getJobsReadyForInactivity() + public JobNotifyRecord[] getJobsReadyForInactivity(String processID) throws ManifoldCFException; /** Inactivate a job, from the notification state. @@ -902,20 +938,24 @@ public interface IJobManager /** Get list of deletable document descriptions. This list will take into account * multiple jobs that may own the same document. + *@param processID is the current process ID. *@param n is the maximum number of documents to return. *@param currentTime is the current time; some fetches do not occur until a specific time. *@return the document descriptions for these documents. */ - public DocumentDescription[] getNextDeletableDocuments(int n, long currentTime) + public DocumentDescription[] getNextDeletableDocuments(String processID, + int n, long currentTime) throws ManifoldCFException; /** Get list of cleanable document descriptions. This list will take into account * multiple jobs that may own the same document. + *@param processID is the current process ID. *@param n is the maximum number of documents to return. *@param currentTime is the current time; some fetches do not occur until a specific time. *@return the document descriptions for these documents. */ - public DocumentSetAndFlags getNextCleanableDocuments(int n, long currentTime) + public DocumentSetAndFlags getNextCleanableDocuments(String processID, + int n, long currentTime) throws ManifoldCFException; /** Delete ingested document identifiers (as part of deleting the owning job). Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Carrydown.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Carrydown.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Carrydown.java (original) +++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/Carrydown.java Tue Nov 26 11:45:53 2013 @@ -39,6 +39,7 @@ import org.apache.manifoldcf.crawler.sys * <tr><td>datavaluehash</td><td>VARCHAR(40)</td><td></td></tr> * <tr><td>datavalue</td><td>LONGTEXT</td><td></td></tr> * <tr><td>isnew</td><td>CHAR(1)</td><td></td></tr> + * <tr><td>processid</td><td>VARCHAR(16)</td><td></td></tr> * </table> * <br><br> * @@ -55,6 +56,7 @@ public class Carrydown extends org.apach public static final String dataValueHashField = "datavaluehash"; public static final String dataValueField = "datavalue"; public static final String newField = "isnew"; + public static final String processIDField = "processid"; /** The standard value for the "isnew" field. Means that the link existed prior to this scan, and no new link * was found yet. */ @@ -106,6 +108,7 @@ public class Carrydown extends org.apach map.put(dataValueHashField,new ColumnDescription("VARCHAR(40)",false,true,null,null,false)); map.put(dataValueField,new ColumnDescription("LONGTEXT",false,true,null,null,false)); map.put(newField,new ColumnDescription("CHAR(1)",false,true,null,null,false)); + map.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false)); performCreate(map,null); @@ -113,13 +116,19 @@ public class Carrydown extends org.apach else { // Upgrade code goes here, if needed. + if (existing.get(processIDField) == null) + { + Map insertMap = new HashMap(); + insertMap.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false)); + performAlter(insertMap,null,null,null); + } } // Now do index management IndexDescription uniqueIndex = new IndexDescription(true,new String[]{jobIDField,parentIDHashField,childIDHashField,dataNameField,dataValueHashField}); IndexDescription jobChildDataIndex = new IndexDescription(false,new String[]{jobIDField,childIDHashField,dataNameField}); - IndexDescription newIndex = new IndexDescription(false,new String[]{newField}); + IndexDescription newIndex = new IndexDescription(false,new String[]{newField,processIDField}); Map indexes = getTableIndexes(null,null); Iterator iter = indexes.keySet().iterator(); @@ -198,8 +207,31 @@ public class Carrydown extends org.apach // /** Reset, at startup time. + *@param processID is the process ID. */ - public void reset() + public void restart(String processID) + throws ManifoldCFException + { + // Delete "new" rows + HashMap map = new HashMap(); + ArrayList list = new ArrayList(); + String query = buildConjunctionClause(list,new ClauseDescription[]{ + new UnitaryClause(newField,statusToString(ISNEW_NEW)), + new UnitaryClause(processIDField,processID)}); + performDelete("WHERE "+query,list,null); + + // Convert "existing" rows to base + map.put(newField,statusToString(ISNEW_BASE)); + list.clear(); + query = buildConjunctionClause(list,new ClauseDescription[]{ + new UnitaryClause(newField,statusToString(ISNEW_EXISTING)), + new UnitaryClause(processIDField,processID)}); + performUpdate(map,"WHERE "+query,list,null); + } + + /** Clean up after all process IDs. + */ + public void restart() throws ManifoldCFException { // Delete "new" rows @@ -216,23 +248,31 @@ public class Carrydown extends org.apach new UnitaryClause(newField,statusToString(ISNEW_EXISTING))}); performUpdate(map,"WHERE "+query,list,null); } + + /** Reset, at startup time, entire cluster + */ + public void restartCluster() + throws ManifoldCFException + { + // Does nothing + } /** Add carrydown data for a given parent/child pair. * *@return true if new carrydown data was recorded; false otherwise. */ public boolean recordCarrydownData(Long jobID, String parentDocumentIDHash, String childDocumentIDHash, - String[] documentDataNames, String[][] documentDataValueHashes, Object[][] documentDataValues) + String[] documentDataNames, String[][] documentDataValueHashes, Object[][] documentDataValues, String processID) throws ManifoldCFException { return recordCarrydownDataMultiple(jobID,parentDocumentIDHash,new String[]{childDocumentIDHash}, - new String[][]{documentDataNames},new String[][][]{documentDataValueHashes},new Object[][][]{documentDataValues})[0]; + new String[][]{documentDataNames},new String[][][]{documentDataValueHashes},new Object[][][]{documentDataValues},processID)[0]; } /** Add carrydown data to the table. */ public boolean[] recordCarrydownDataMultiple(Long jobID, String parentDocumentIDHash, String[] childDocumentIDHashes, - String[][] dataNames, String[][][] dataValueHashes, Object[][][] dataValues) + String[][] dataNames, String[][][] dataValueHashes, Object[][][] dataValues, String processID) throws ManifoldCFException { @@ -340,19 +380,30 @@ public class Carrydown extends org.apach } map.put(newField,statusToString(ISNEW_NEW)); + map.put(processIDField,processID); performInsert(map,null); noteModifications(1,0,0); insertHappened.put(childDocumentIDHash,new Boolean(true)); } else { - sb = new StringBuilder(); - sb.append("WHERE ").append(jobIDField).append("=? AND ") + sb = new StringBuilder("WHERE "); + ArrayList updateList = new ArrayList(); + sb.append(buildConjunctionClause(updateList,new ClauseDescription[]{ + new UnitaryClause(jobIDField,jobID), + new UnitaryClause(parentIDHashField,parentDocumentIDHash), + new UnitaryClause(childIDHashField,childDocumentIDHash), + new UnitaryClause(dataNameField,dataName), + (dataValueHash==null)? + new NullCheckClause(dataValueHashField,true): + new UnitaryClause(dataValueHashField,dataValueHash)})); + + /* + sb.append(jobIDField).append("=? AND ") .append(parentIDHashField).append("=? AND ") .append(childIDHashField).append("=? AND ") .append(dataNameField).append("=? AND "); - ArrayList updateList = new ArrayList(); updateList.add(jobID); updateList.add(parentDocumentIDHash); updateList.add(childDocumentIDHash); @@ -366,8 +417,10 @@ public class Carrydown extends org.apach { sb.append(dataValueHashField).append(" IS NULL"); } - + */ + map.put(newField,statusToString(ISNEW_EXISTING)); + map.put(processIDField,processID); performUpdate(map,sb.toString(),updateList,null); noteModifications(0,1,0); } @@ -483,6 +536,7 @@ public class Carrydown extends org.apach HashMap map = new HashMap(); map.put(newField,statusToString(ISNEW_BASE)); + map.put(processIDField,null); performUpdate(map,sb.toString(),newList,null); noteModifications(0,list.size(),0); Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/EventManager.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/EventManager.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/EventManager.java (original) +++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/EventManager.java Tue Nov 26 11:45:53 2013 @@ -21,6 +21,7 @@ package org.apache.manifoldcf.crawler.jo import org.apache.manifoldcf.core.interfaces.*; import org.apache.manifoldcf.crawler.interfaces.*; import org.apache.manifoldcf.crawler.interfaces.CacheKeyFactory; +import org.apache.manifoldcf.crawler.system.ManifoldCF; import java.util.*; /** This class manages the events table. @@ -33,6 +34,7 @@ import java.util.*; * <tr class="TableHeadingColor"> * <th>Field</th><th>Type</th><th>Description </th> * <tr><td>name</td><td>VARCHAR(255)</td><td>Primary Key</td></tr> +* <tr><td>processid</td><td>VARCHAR(16)</td><td></td></tr> * </table> * <br><br> * @@ -43,7 +45,8 @@ public class EventManager extends org.ap // Field names public final static String eventNameField = "name"; - + public final static String processIDField = "processid"; + /** Constructor. *@param database is the database handle. */ @@ -66,14 +69,41 @@ public class EventManager extends org.ap { HashMap map = new HashMap(); map.put(eventNameField,new ColumnDescription("VARCHAR(255)",true,false,null,null,false)); + map.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false)); performCreate(map,null); } else { // Upgrade goes here if needed + if (existing.get(processIDField) == null) + { + Map insertMap = new HashMap(); + insertMap.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false)); + performAlter(insertMap,null,null,null); + } } // Index management goes here + IndexDescription processIDIndex = new IndexDescription(false,new String[]{processIDField}); + // Get rid of unused indexes + Map indexes = getTableIndexes(null,null); + Iterator iter = indexes.keySet().iterator(); + while (iter.hasNext()) + { + String indexName = (String)iter.next(); + IndexDescription id = (IndexDescription)indexes.get(indexName); + + if (processIDIndex != null && id.equals(processIDIndex)) + processIDIndex = null; + else if (indexName.indexOf("_pkey") == -1) + // This index shouldn't be here; drop it + performRemoveIndex(indexName); + } + + // Build missing indexes + + if (processIDIndex != null) + performAddIndex(null,processIDIndex); break; } @@ -84,42 +114,45 @@ public class EventManager extends org.ap public void deinstall() throws ManifoldCFException { - beginTransaction(); - try - { - performDrop(null); - } - catch (ManifoldCFException e) - { - signalRollback(); - throw e; - } - catch (Error e) - { - signalRollback(); - throw e; - } - finally - { - endTransaction(); - } + performDrop(null); } /** Prepare for restart. + *@param processID is the processID to restart. + */ + public void restart(String processID) + throws ManifoldCFException + { + // Delete all rows in this table matching the processID + ArrayList list = new ArrayList(); + String query = buildConjunctionClause(list,new ClauseDescription[]{ + new UnitaryClause(processIDField,processID)}); + performDelete("WHERE "+query,null,null); + } + + /** Clean up after all processIDs. */ public void restart() throws ManifoldCFException { - // Delete all rows in this table. performDelete("",null,null); } - + + /** Restart cluster. + */ + public void restartCluster() + throws ManifoldCFException + { + // Does nothing + } + /** Atomically create an event - and return false if the event already exists */ - public void createEvent(String eventName) + public void createEvent(String eventName, String processID) throws ManifoldCFException { HashMap map = new HashMap(); map.put(eventNameField,eventName); + map.put(processIDField,processID); performInsert(map,null); } Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/HopCount.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/HopCount.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/HopCount.java (original) +++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/HopCount.java Tue Nov 26 11:45:53 2013 @@ -302,20 +302,37 @@ public class HopCount extends org.apache } /** Reset, at startup time. + *@param processID is the process ID. */ - public void reset() + public void restart(String processID) throws ManifoldCFException { - intrinsicLinkManager.reset(); + intrinsicLinkManager.restart(processID); } + /** Clean up after all process IDs. + */ + public void restart() + throws ManifoldCFException + { + intrinsicLinkManager.restart(); + } + + /** Restart entire cluster. + */ + public void restartCluster() + throws ManifoldCFException + { + intrinsicLinkManager.restartCluster(); + } + /** Record a references from a set of documents to the root. These will be marked as "new" or "existing", and * will have a null linktype. */ - public void recordSeedReferences(Long jobID, String[] legalLinkTypes, String[] targetDocumentIDHashes, int hopcountMethod) + public void recordSeedReferences(Long jobID, String[] legalLinkTypes, String[] targetDocumentIDHashes, int hopcountMethod, String processID) throws ManifoldCFException { - doRecord(jobID,legalLinkTypes,"",targetDocumentIDHashes,"",hopcountMethod); + doRecord(jobID,legalLinkTypes,"",targetDocumentIDHashes,"",hopcountMethod,processID); } /** Finish seed references. Seed references are special in that the only source is the root. @@ -329,19 +346,19 @@ public class HopCount extends org.apache /** Record a reference from source to target. This reference will be marked as "new" or "existing". */ public boolean recordReference(Long jobID, String[] legalLinkTypes, String sourceDocumentIDHash, String targetDocumentIDHash, String linkType, - int hopcountMethod) + int hopcountMethod, String processID) throws ManifoldCFException { - return doRecord(jobID,legalLinkTypes,sourceDocumentIDHash,new String[]{targetDocumentIDHash},linkType,hopcountMethod)[0]; + return doRecord(jobID,legalLinkTypes,sourceDocumentIDHash,new String[]{targetDocumentIDHash},linkType,hopcountMethod,processID)[0]; } /** Record a set of references from source to target. This reference will be marked as "new" or "existing". */ public boolean[] recordReferences(Long jobID, String[] legalLinkTypes, String sourceDocumentIDHash, String[] targetDocumentIDHashes, String linkType, - int hopcountMethod) + int hopcountMethod, String processID) throws ManifoldCFException { - return doRecord(jobID,legalLinkTypes,sourceDocumentIDHash,targetDocumentIDHashes,linkType,hopcountMethod); + return doRecord(jobID,legalLinkTypes,sourceDocumentIDHash,targetDocumentIDHashes,linkType,hopcountMethod,processID); } /** Complete a recalculation pass for a set of source documents. All child links that are not marked as "new" @@ -355,7 +372,7 @@ public class HopCount extends org.apache /** Do the work of recording source-target references. */ protected boolean[] doRecord(Long jobID, String[] legalLinkTypes, String sourceDocumentIDHash, String[] targetDocumentIDHashes, String linkType, - int hopcountMethod) + int hopcountMethod, String processID) throws ManifoldCFException { @@ -367,7 +384,7 @@ public class HopCount extends org.apache rval[i] = false; } - String[] newReferences = intrinsicLinkManager.recordReferences(jobID,sourceDocumentIDHash,targetDocumentIDHashes,linkType); + String[] newReferences = intrinsicLinkManager.recordReferences(jobID,sourceDocumentIDHash,targetDocumentIDHashes,linkType,processID); if (newReferences.length > 0) { // There are added links. Modified: manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/IntrinsicLink.java URL: http://svn.apache.org/viewvc/manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/IntrinsicLink.java?rev=1545624&r1=1545623&r2=1545624&view=diff ============================================================================== --- manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/IntrinsicLink.java (original) +++ manifoldcf/trunk/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/jobs/IntrinsicLink.java Tue Nov 26 11:45:53 2013 @@ -37,6 +37,7 @@ import org.apache.manifoldcf.crawler.sys * <tr><td>parentidhash</td><td>VARCHAR(40)</td><td></td></tr> * <tr><td>childidhash</td><td>VARCHAR(40)</td><td></td></tr> * <tr><td>isnew</td><td>CHAR(1)</td><td></td></tr> + * <tr><td>processid</td><td>VARCHAR(16)</td><td></td></tr> * </table> * <br><br> * @@ -61,6 +62,7 @@ public class IntrinsicLink extends org.a public static final String parentIDHashField = "parentidhash"; public static final String childIDHashField = "childidhash"; public static final String newField = "isnew"; + public static final String processIDField = "processid"; // Map from string character to link status protected static Map linkstatusMap; @@ -99,17 +101,24 @@ public class IntrinsicLink extends org.a map.put(parentIDHashField,new ColumnDescription("VARCHAR(40)",false,false,null,null,false)); map.put(childIDHashField,new ColumnDescription("VARCHAR(40)",false,true,null,null,false)); map.put(newField,new ColumnDescription("CHAR(1)",false,true,null,null,false)); + map.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false)); performCreate(map,null); } else { // Perform upgrade, if needed. + if (existing.get(processIDField) == null) + { + Map insertMap = new HashMap(); + insertMap.put(processIDField,new ColumnDescription("VARCHAR(16)",false,true,null,null,false)); + performAlter(insertMap,null,null,null); + } } // Indexes IndexDescription uniqueIndex = new IndexDescription(true,new String[]{jobIDField,parentIDHashField,linkTypeField,childIDHashField}); IndexDescription jobChildNewIndex = new IndexDescription(false,new String[]{jobIDField,childIDHashField,newField}); - IndexDescription newIndex = new IndexDescription(false,new String[]{newField}); + IndexDescription newIndex = new IndexDescription(false,new String[]{newField,processIDField}); Map indexes = getTableIndexes(null,null); Iterator iter = indexes.keySet().iterator(); @@ -179,8 +188,9 @@ public class IntrinsicLink extends org.a * of documents, and cached records of hopcount are updated only when requested, it is safest to simply * move any "new" or "new existing" links back to base state on startup. Then, the next time that page * is processed, the links will be updated properly. + *@param processID is the process to restart. */ - public void reset() + public void restart(String processID) throws ManifoldCFException { HashMap map = new HashMap(); @@ -189,14 +199,37 @@ public class IntrinsicLink extends org.a String query = buildConjunctionClause(list,new ClauseDescription[]{ new MultiClause(newField,new Object[]{ statusToString(LINKSTATUS_NEW), - statusToString(LINKSTATUS_EXISTING)})}); + statusToString(LINKSTATUS_EXISTING)}), + new UnitaryClause(processIDField,processID)}); performUpdate(map,"WHERE "+query,list,null); } + /** Clean up after all process IDs + */ + public void restart() + throws ManifoldCFException + { + HashMap map = new HashMap(); + map.put(newField,statusToString(LINKSTATUS_BASE)); + ArrayList list = new ArrayList(); + String query = buildConjunctionClause(list,new ClauseDescription[]{ + new MultiClause(newField,new Object[]{ + statusToString(LINKSTATUS_NEW), + statusToString(LINKSTATUS_EXISTING)})}); + performUpdate(map,"WHERE "+query,list,null); + } + + public void restartCluster() + throws ManifoldCFException + { + // Does nothing + } + /** Record a references from source to targets. These references will be marked as either "new" or "existing". *@return the target document ID's that are considered "new". */ - public String[] recordReferences(Long jobID, String sourceDocumentIDHash, String[] targetDocumentIDHashes, String linkType) + public String[] recordReferences(Long jobID, String sourceDocumentIDHash, + String[] targetDocumentIDHashes, String linkType, String processID) throws ManifoldCFException { HashMap duplicateRemoval = new HashMap(); @@ -253,6 +286,7 @@ public class IntrinsicLink extends org.a map.put(childIDHashField,sourceDocumentIDHash); map.put(linkTypeField,linkType); map.put(newField,statusToString(LINKSTATUS_NEW)); + map.put(processIDField,processID); performInsert(map,null); noteModifications(1,0,0); } @@ -260,6 +294,7 @@ public class IntrinsicLink extends org.a { HashMap map = new HashMap(); map.put(newField,statusToString(LINKSTATUS_EXISTING)); + map.put(processIDField,processID); ArrayList updateList = new ArrayList(); String query = buildConjunctionClause(updateList,new ClauseDescription[]{ new UnitaryClause(jobIDField,jobID), @@ -470,6 +505,7 @@ public class IntrinsicLink extends org.a { HashMap map = new HashMap(); map.put(newField,statusToString(LINKSTATUS_BASE)); + map.put(processIDField,null); StringBuilder sb = new StringBuilder("WHERE "); ArrayList newList = new ArrayList();
