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/27 21:09:29 UTC
svn commit: r1546178 -
/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
Author: kwright
Date: Wed Nov 27 20:09:29 2013
New Revision: 1546178
URL: http://svn.apache.org/r1546178
Log:
Prioritization tracking class
Added:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java (with props)
Added: manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
URL: http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java?rev=1546178&view=auto
==============================================================================
--- manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java (added)
+++ manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java Wed Nov 27 20:09:29 2013
@@ -0,0 +1,238 @@
+/* $Id$ */
+
+/**
+* Licensed to the Apache Software Foundation (ASF) under one or more
+* contributor license agreements. See the NOTICE file distributed with
+* this work for additional information regarding copyright ownership.
+* The ASF licenses this file to You under the Apache License, Version 2.0
+* (the "License"); you may not use this file except in compliance with
+* the License. You may obtain a copy of the License at
+*
+* http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+package org.apache.manifoldcf.crawler.system;
+
+import org.apache.manifoldcf.core.interfaces.*;
+import org.apache.manifoldcf.agents.interfaces.*;
+import org.apache.manifoldcf.crawler.interfaces.*;
+import java.util.*;
+import java.io.*;
+
+/** This class tracks cluster-wide reprioritization operations.
+* These operations are driven forward by whatever thread needs them,
+* and are completed if those processes die by the threads that clean up
+* after the original process.
+*/
+public class ReprioritizationTracker
+{
+ public static final String _rcsid = "@(#)$Id$";
+
+ protected final static String trackerWriteLock = "_REPR_TRACKER_LOCK_";
+ protected final static String trackerProcessIDResource = "_REPR_TRACKER_PID_";
+ protected final static String trackerTimestampResource = "_REPR_TIMESTAMP_";
+
+ /** Lock manager */
+ protected final ILockManager lockManager;
+
+ /** Constructor.
+ */
+ public ReprioritizationTracker(IThreadContext threadContext)
+ throws ManifoldCFException
+ {
+ lockManager = LockManagerFactory.make(threadContext);
+ }
+
+ /** Start a reprioritization activity.
+ *@param prioritizationTime is the timestamp of the prioritization.
+ *@param processID is the process ID of the process performing/waiting for the prioritization
+ * to complete.
+ */
+ public void startReprioritization(long prioritizationTime, String processID)
+ throws ManifoldCFException
+ {
+ lockManager.enterWriteLock(trackerWriteLock);
+ try
+ {
+ Long currentTime = readTime();
+ String currentProcessID = readProcessID();
+ if (currentTime != null && currentProcessID != null)
+ {
+ // Already a reprioritization in progress.
+ if (prioritizationTime <= currentTime.longValue())
+ return;
+ }
+ writeTime(new Long(prioritizationTime));
+ writeProcessID(processID);
+ }
+ finally
+ {
+ lockManager.leaveWriteLock(trackerWriteLock);
+ }
+ }
+
+
+ /** Retrieve the current reprioritization time stamp. This should be obtained before
+ * performing any prioritization steps.
+ *@return the current prioritization timestamp, or null if no prioritization is in effect.
+ */
+ public Long checkReprioritizationInProgress()
+ throws ManifoldCFException
+ {
+ lockManager.enterWriteLock(trackerWriteLock);
+ try
+ {
+ Long currentTime = readTime();
+ String currentProcessID = readProcessID();
+ if (currentTime == null || currentProcessID == null)
+ return null;
+ return currentTime;
+ }
+ finally
+ {
+ lockManager.leaveWriteLock(trackerWriteLock);
+ }
+ }
+
+ /** Complete a reprioritization activity. Prioritization will be marked as complete
+ * only if the processID matches the one that started the current reprioritization.
+ *@param processID is the process ID of the process completing the prioritization.
+ */
+ public void doneReprioritization(String processID)
+ throws ManifoldCFException
+ {
+ lockManager.enterWriteLock(trackerWriteLock);
+ try
+ {
+ Long currentTime = readTime();
+ String currentProcessID = readProcessID();
+ if (currentTime != null && currentProcessID != null && currentProcessID.equals(processID))
+ {
+ // Null out the fields
+ writeTime(null);
+ writeProcessID(null);
+ }
+ }
+ finally
+ {
+ lockManager.leaveWriteLock(trackerWriteLock);
+ }
+ }
+
+ /** Check if the specified processID is the one performing reprioritization.
+ *@param processID is the process ID to check.
+ *@return true if the specified processID is declared as being responsible for the
+ * current reprioritization, false if no prioritization in progress or a different process is involved.
+ */
+ public boolean isSpecifiedProcessReprioritizing(String processID)
+ throws ManifoldCFException
+ {
+ lockManager.enterWriteLock(trackerWriteLock);
+ try
+ {
+ Long currentTime = readTime();
+ String currentProcessID = readProcessID();
+ return (currentTime != null && currentProcessID != null && currentProcessID.equals(processID));
+ }
+ finally
+ {
+ lockManager.leaveWriteLock(trackerWriteLock);
+ }
+ }
+
+ // Protected methods
+
+ /** Read time.
+ *@return the time, or null if none.
+ */
+ protected Long readTime()
+ throws ManifoldCFException
+ {
+ byte[] timeData = lockManager.readData(trackerTimestampResource);
+ if (timeData == null || timeData.length != 8)
+ return null;
+
+ long rval = ((long)timeData[0]) & 0xffL +
+ (((long)timeData[1]) << 8) & 0xff00L +
+ (((long)timeData[2]) << 16) & 0xff0000L +
+ (((long)timeData[3]) << 24) & 0xff000000L +
+ (((long)timeData[4]) << 32) & 0xff00000000L +
+ (((long)timeData[5]) << 40) & 0xff0000000000L +
+ (((long)timeData[6]) << 48) & 0xff000000000000L +
+ (((long)timeData[7]) << 56) & 0xff00000000000000L;
+
+ return new Long(rval);
+ }
+
+ /** Write time.
+ *@param time is the time to write.
+ */
+ protected void writeTime(Long timeValue)
+ throws ManifoldCFException
+ {
+ if (timeValue == null)
+ lockManager.writeData(trackerTimestampResource, null);
+ else
+ {
+ long time = timeValue.longValue();
+ byte[] timeData = new byte[8];
+ timeData[0] = (byte)(time & 0xffL);
+ timeData[1] = (byte)((time >> 8) & 0xffL);
+ timeData[2] = (byte)((time >> 16) & 0xffL);
+ timeData[3] = (byte)((time >> 24) & 0xffL);
+ timeData[4] = (byte)((time >> 32) & 0xffL);
+ timeData[5] = (byte)((time >> 40) & 0xffL);
+ timeData[6] = (byte)((time >> 48) & 0xffL);
+ timeData[7] = (byte)((time >> 56) & 0xffL);
+ lockManager.writeData(trackerTimestampResource, timeData);
+ }
+ }
+
+ /** Read process ID.
+ *@return processID, or null if none.
+ */
+ protected String readProcessID()
+ throws ManifoldCFException
+ {
+ byte[] processIDData = lockManager.readData(trackerProcessIDResource);
+ if (processIDData == null)
+ return null;
+ try
+ {
+ return new String(processIDData, "utf-8");
+ }
+ catch (UnsupportedEncodingException e)
+ {
+ throw new RuntimeException(e.getMessage(),e);
+ }
+ }
+
+ /** Write process ID.
+ *@param processID is the process ID to write.
+ */
+ protected void writeProcessID(String processID)
+ throws ManifoldCFException
+ {
+ if (processID == null)
+ lockManager.writeData(trackerProcessIDResource, null);
+ else
+ {
+ try
+ {
+ byte[] processIDData = processID.getBytes("utf-8");
+ lockManager.writeData(trackerProcessIDResource, processIDData);
+ }
+ catch (UnsupportedEncodingException e)
+ {
+ throw new RuntimeException(e.getMessage(),e);
+ }
+ }
+ }
+
+}
+
Propchange: manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange: manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
------------------------------------------------------------------------------
svn:keywords = Id