Author: kwright
Date: Wed Nov 27 11:16:27 2013
New Revision: 1545996
URL: http://svn.apache.org/r1545996
Log:
Move document prioritization to docbins table, and out of QueueTracker.
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/bins/BinManager.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/QueueTracker.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/PriorityCalculator.java
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/bins/BinManager.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/bins/BinManager.java?rev=1545996&r1=1545995&r2=1545996&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/bins/BinManager.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/bins/BinManager.java
Wed Nov 27 11:16:27 2013
@@ -122,6 +122,8 @@ public class BinManager extends org.apac
IResultRow row = result.getRow(0);
Double value = (Double)row.getValue(binCounterField);
double rval = value.doubleValue();
+ if (rval < newBinValue)
+ rval = newBinValue;
HashMap map = new HashMap();
map.put(binCounterField,new Double(rval+1.0));
performUpdate(map," WHERE "+query,params,null);
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/QueueTracker.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/QueueTracker.java?rev=1545996&r1=1545995&r2=1545996&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/QueueTracker.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/interfaces/QueueTracker.java
Wed Nov 27 11:16:27 2013
@@ -60,10 +60,6 @@ public class QueueTracker
/** These are the accumulated performance averages for all connections etc.
*/
protected final PerformanceStatistics performanceStatistics = new
PerformanceStatistics();
- /** These are the bin counts for a prioritization pass.
- * This hash table is keyed by bin, and contains DoubleBinCount objects as
values */
- protected final Map<String,DoubleBinCount> binCounts = new
HashMap<String,DoubleBinCount>();
-
/** These are the bin counts for tracking the documents that are on
* the active queue, but are not being processed yet */
protected final Map<String,BinCount> queuedBinCounts = new
HashMap<String,BinCount>();
@@ -71,6 +67,9 @@ public class QueueTracker
/** These are the bin counts for active threads */
protected final Map<String,BinCount> activeBinCounts = new
HashMap<String,BinCount>();
+ /** The locker for the minimum depth calculation */
+ protected final Integer minimumDepthLock = new Integer(0);
+
/** The "minimum depth" - which is the smallest bin count of the last
document queued. This helps guarantee that documents that are
* newly discovered don't wind up with high priority, but instead wind up
about the same as the currently active document priority. */
protected double currentMinimumDepth = 0.0;
@@ -90,9 +89,8 @@ public class QueueTracker
*/
public void beginReset()
{
- synchronized (binCounts)
+ synchronized (minimumDepthLock)
{
- binCounts.clear();
currentMinimumDepth = 0.0;
resetInProgress = true;
}
@@ -102,7 +100,7 @@ public class QueueTracker
/** Finish the reset operation */
public void endReset()
{
- synchronized (binCounts)
+ synchronized (minimumDepthLock)
{
resetInProgress = false;
}
@@ -220,7 +218,7 @@ public class QueueTracker
*/
public void assessMinimumDepth(Double[] binNamesSet)
{
- synchronized (binCounts)
+ synchronized (minimumDepthLock)
{
// Ignore all numbers until reset is complete
if (!resetInProgress)
@@ -343,223 +341,16 @@ public class QueueTracker
return rval;
}
- /** This is a made-up constant, originally based on 100 documents/second,
but adjusted downward as a result of experimentation and testing, which is
described as "T" below.
- */
- private final static double minMsPerFetch = 50.0;
-
- /** Calculate a document priority value. Priorities are reversed, and in
log space, so that
- * zero (0.0) is considered the highest possible priority, and larger
priority values are considered lower in actual priority.
- *@param binNames are the global bins to which the document belongs.
- *@param connection is the connection, from which the throttles may be
obtained. More highly throttled connections are given
- * less favorable priority.
- *@return the priority value, based on recent history. Also updates
statistics atomically.
- */
- public double calculatePriority(String[] binNames, IRepositoryConnection
connection)
- {
-
- // NOTE: We must be sure to adjust the return value by the factor
calculated due to performance; a slower throttle rate
- // should yield a lower priority. In theory it should be possible to
calculate an adjusted priority pretty exactly,
- // on the basis that the fetch rates of two distinct bins should grant
priorities such that:
- //
- // (n documents) / (the rate of fetch (docs/millisecond) of the first
bin) = milliseconds for the first bin
- //
- // should equal:
- //
- // (m documents) / (the rate of fetch of the second bin) = milliseconds
for the second bin
- //
- // ... and then assigning priorities so that after a given number of
document priorities are assigned from the first bin, the
- // corresponding (*m/n) number of document priorities would get assigned
for the second bin.
- //
- // Suppose the maximum fetch rate for the document is F fetches per
millisecond. If the document priority assigned for the Bth
- // bin member is -log(1/(1+B)) for a document fetched with no throttling
whatsoever,
- // then we want the priority to be -log(1/(1+k)) for a throttled bin,
where k is chosen so that:
- // k = B * ((T + 1/F)/T) = B * (1 + 1/TF)
- // ... where T is the time taken to fetch a single document that has no
throttling at all.
- // For the purposes of this exercise, a value of 100 doc/sec, or T=10ms.
- //
- // Basically, for F = 0, k should be infinity, and for F = infinity, k
should be B.
-
- // First, calculate the document's max fetch rate, in fetches per
millisecond. This will be used to adjust the priority, and
- // also when resetting the bin counts.
- double[] maxFetchRates = calculateMaxFetchRates(binNames,connection);
-
- // For each bin, we will be calculating the bin count scale factor, which
is what we multiply the bincount by to adjust for the
- // throttling on that bin.
- double[] binCountScaleFactors = new double[binNames.length];
- double[] weightedMinimumDepths = new double[binNames.length];
-
- // Before calculating priority, calculate some factors that will allow us
to determine the proper starting value for a bin.
-
- // First thing to do is to reset the bin values based on the current
minimum.
- for (int i = 0; i < binNames.length; i++)
- {
- String binName = binNames[i];
- // Remember, maxFetchRate is in fetches per ms.
- double maxFetchRate = maxFetchRates[i];
-
- // Calculate (and save for later) the scale factor for this bin.
- double binCountScaleFactor;
- if (maxFetchRate == 0.0)
- binCountScaleFactor = Double.POSITIVE_INFINITY;
- else
- binCountScaleFactor = 1.0 + 1.0 / (minMsPerFetch * maxFetchRate);
- binCountScaleFactors[i] = binCountScaleFactor;
- weightedMinimumDepths[i] = currentMinimumDepth / binCountScaleFactor;
- }
-
- double highestAdjustedCount = 0.0;
- synchronized (binCounts)
- {
- // Find the bin with the largest effective count, and use that for the
document's priority.
- // (This of course assumes that the slowest throttle is the one that
wins.)
- for (int i = 0; i < binNames.length; i++)
- {
- String binName = binNames[i];
- double binCountScaleFactor = binCountScaleFactors[i];
- double weightedMinimumDepth = weightedMinimumDepths[i];
-
- double thisCount = 0.0;
- DoubleBinCount bc = binCounts.get(binName);
- if (bc != null)
- thisCount = bc.getValue();
- else
- {
- bc = new DoubleBinCount();
- binCounts.put(binName,bc);
- }
-
- // Adjust the count, if needed, so that we are not assigning
priorities greater than the current level we are
- // grabbing documents at
- if (thisCount < weightedMinimumDepth)
- {
-
- if (Logging.scheduling.isDebugEnabled())
- Logging.scheduling.debug("Resetting value of bin '"+binName+"' to
"+new Double(weightedMinimumDepth).toString()+"(scale factor is "+new
Double(binCountScaleFactor)+")");
-
- // Set a new bin value
- bc.setValue(weightedMinimumDepth);
- }
- bc.increment();
-
- double adjustedCount;
- // Use the scale factor already calculated above to yield a priority
that is adjusted for the fetch rate.
- if (binCountScaleFactor == Double.POSITIVE_INFINITY)
- adjustedCount = Double.POSITIVE_INFINITY;
- else
- adjustedCount = thisCount * binCountScaleFactor;
- if (adjustedCount > highestAdjustedCount)
- highestAdjustedCount = adjustedCount;
- }
- }
-
- // Calculate the proper log value
- double returnValue;
-
- if (highestAdjustedCount == Double.POSITIVE_INFINITY)
- returnValue = Double.POSITIVE_INFINITY;
- else
- returnValue = Math.log(1.0 + highestAdjustedCount);
-
- if (Logging.scheduling.isDebugEnabled())
- {
- StringBuilder sb = new StringBuilder();
- int k = 0;
- while (k < binNames.length)
- {
- sb.append(binNames[k++]).append(" ");
- }
- Logging.scheduling.debug("Document with bins ["+sb.toString()+"] given
priority value "+new Double(returnValue).toString());
- }
-
-
- return returnValue;
- }
-
- /** Calculate the maximum fetch rate for a given set of bins for a given
connection.
- * This is used to adjust the final priority of a document.
+ /** Get the minimum depth.
*/
- protected static double[] calculateMaxFetchRates(String[] binNames,
IRepositoryConnection connection)
- {
- ThrottleLimits tl = new ThrottleLimits(connection);
- return tl.getMaximumRates(binNames);
- }
-
- /** This class represents the throttle limits out of the connection
specification */
- protected static class ThrottleLimits
- {
- protected List<ThrottleLimitSpec> specs = new
ArrayList<ThrottleLimitSpec>();
-
- public ThrottleLimits(IRepositoryConnection connection)
- {
- String[] throttles = connection.getThrottles();
- int i = 0;
- while (i < throttles.length)
- {
- try
- {
- specs.add(new
ThrottleLimitSpec(throttles[i],(double)connection.getThrottleValue(throttles[i])));
- }
- catch (PatternSyntaxException e)
- {
- }
- i++;
- }
- }
-
- public double[] getMaximumRates(String[] binNames)
- {
- double[] rval = new double[binNames.length];
- for (int j = 0 ; j < binNames.length ; j++)
- {
- String binName = binNames[j];
- double maxRate = Double.POSITIVE_INFINITY;
- for (ThrottleLimitSpec spec : specs)
- {
- Pattern p = spec.getRegexp();
- Matcher m = p.matcher(binName);
- if (m.find())
- {
- double rate = spec.getMaxRate();
- // The direction of this inequality reflects the fact that the
throttling is conservative when more rules are present.
- if (rate < maxRate)
- maxRate = rate;
- }
- }
- rval[j] = maxRate;
- }
- return rval;
- }
-
- }
-
- /** This is a class which describes an individual throttle limit, in fetches
per millisecond. */
- protected static class ThrottleLimitSpec
+ public double getMinimumDepth()
{
- /** Regexp */
- protected final Pattern regexp;
- /** The fetch limit for all bins matching that regexp, in fetches per
millisecond */
- protected final double maxRate;
-
- /** Constructor */
- public ThrottleLimitSpec(String regexp, double maxRate)
- throws PatternSyntaxException
+ synchronized (minimumDepthLock)
{
- this.regexp = Pattern.compile(regexp);
- this.maxRate = maxRate;
- }
-
- /** Get the regexp. */
- public Pattern getRegexp()
- {
- return regexp;
- }
-
- /** Get the max count */
- public double getMaxRate()
- {
- return maxRate;
+ return currentMinimumDepth;
}
}
+
/** This is the class which allows a mutable integer count value to be saved
in the bincount table.
*/
@@ -606,42 +397,5 @@ public class QueueTracker
}
}
- /** This is the class which allows a mutable integer count value to be saved
in the bincount table.
- */
- protected static class DoubleBinCount
- {
- /** The count */
- protected double count = 0.0;
-
- /** Create */
- public DoubleBinCount()
- {
- }
-
- public DoubleBinCount duplicate()
- {
- DoubleBinCount rval = new DoubleBinCount();
- rval.count = this.count;
- return rval;
- }
-
- /** Increment the counter */
- public void increment()
- {
- count += 1.0;
- }
-
- /** Set the value */
- public void setValue(double count)
- {
- this.count = count;
- }
-
- /** Get the value */
- public double getValue()
- {
- return count;
- }
- }
}
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java?rev=1545996&r1=1545995&r2=1545996&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ManifoldCF.java
Wed Nov 27 11:16:27 2013
@@ -932,6 +932,9 @@ public class ManifoldCF extends org.apac
// will therefore block that from occurring, until the reset is complete.
try
{
+ // Does this need to be in a transaction??? Can it fail and require
retries?
+ binManager.reset();
+
// Reprioritize all documents in the jobqueue, 1000 at a time
Map<String,IRepositoryConnection> connectionMap = new
HashMap<String,IRepositoryConnection>();
Modified:
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/PriorityCalculator.java
URL:
http://svn.apache.org/viewvc/manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/PriorityCalculator.java?rev=1545996&r1=1545995&r2=1545996&view=diff
==============================================================================
---
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/PriorityCalculator.java
(original)
+++
manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/PriorityCalculator.java
Wed Nov 27 11:16:27 2013
@@ -21,6 +21,8 @@ package org.apache.manifoldcf.crawler.sy
import org.apache.manifoldcf.core.interfaces.*;
import org.apache.manifoldcf.agents.interfaces.*;
import org.apache.manifoldcf.crawler.interfaces.*;
+import java.util.*;
+import java.util.regex.*;
/** This class calculates a document priority given all the required inputs.
* It is not thread safe, but calls classes that are (e.g. QueueTracker).
@@ -29,9 +31,13 @@ public class PriorityCalculator implemen
{
public static final String _rcsid = "@(#)$Id$";
+ /** This is a made-up constant, originally based on 100 documents/second,
but adjusted downward as a result of experimentation and testing, which is
described as "T" below.
+ */
+ private final static double minMsPerFetch = 50.0;
+
protected final QueueTracker queueTracker;
protected final IRepositoryConnection connection;
- protected final String[] documentBins;
+ protected final String[] binNames;
protected final IBinManager binManager;
/** Constructor. */
@@ -39,19 +45,201 @@ public class PriorityCalculator implemen
{
this.queueTracker = queueTracker;
this.connection = connection;
- this.documentBins = documentBins;
+ this.binNames = documentBins;
this.binManager = binManager;
}
- /** Compute the document priority given an actual bincounter value.
- *@return the document priority.
+ /** Calculate a document priority value. Priorities are reversed, and in
log space, so that
+ * zero (0.0) is considered the highest possible priority, and larger
priority values are considered lower in actual priority.
+ *@param binNames are the global bins to which the document belongs.
+ *@param connection is the connection, from which the throttles may be
obtained. More highly throttled connections are given
+ * less favorable priority.
+ *@return the priority value, based on recent history. Also updates
statistics atomically.
*/
@Override
public double getDocumentPriority()
throws ManifoldCFException
{
- // MHL to use database bin tables
- return queueTracker.calculatePriority(documentBins,connection);
+
+ // NOTE: We must be sure to adjust the return value by the factor
calculated due to performance; a slower throttle rate
+ // should yield a lower priority. In theory it should be possible to
calculate an adjusted priority pretty exactly,
+ // on the basis that the fetch rates of two distinct bins should grant
priorities such that:
+ //
+ // (n documents) / (the rate of fetch (docs/millisecond) of the first
bin) = milliseconds for the first bin
+ //
+ // should equal:
+ //
+ // (m documents) / (the rate of fetch of the second bin) = milliseconds
for the second bin
+ //
+ // ... and then assigning priorities so that after a given number of
document priorities are assigned from the first bin, the
+ // corresponding (*m/n) number of document priorities would get assigned
for the second bin.
+ //
+ // Suppose the maximum fetch rate for the document is F fetches per
millisecond. If the document priority assigned for the Bth
+ // bin member is -log(1/(1+B)) for a document fetched with no throttling
whatsoever,
+ // then we want the priority to be -log(1/(1+k)) for a throttled bin,
where k is chosen so that:
+ // k = B * ((T + 1/F)/T) = B * (1 + 1/TF)
+ // ... where T is the time taken to fetch a single document that has no
throttling at all.
+ // For the purposes of this exercise, a value of 100 doc/sec, or T=10ms.
+ //
+ // Basically, for F = 0, k should be infinity, and for F = infinity, k
should be B.
+
+ // First, calculate the document's max fetch rate, in fetches per
millisecond. This will be used to adjust the priority, and
+ // also when resetting the bin counts.
+ double[] maxFetchRates = calculateMaxFetchRates(binNames,connection);
+
+ // For each bin, we will be calculating the bin count scale factor, which
is what we multiply the bincount by to adjust for the
+ // throttling on that bin.
+ double[] binCountScaleFactors = new double[binNames.length];
+ double[] weightedMinimumDepths = new double[binNames.length];
+
+ // Before calculating priority, calculate some factors that will allow us
to determine the proper starting value for a bin.
+ double currentMinimumDepth = queueTracker.getMinimumDepth();
+
+ // First thing to do is to reset the bin values based on the current
minimum.
+ for (int i = 0; i < binNames.length; i++)
+ {
+ String binName = binNames[i];
+ // Remember, maxFetchRate is in fetches per ms.
+ double maxFetchRate = maxFetchRates[i];
+
+ // Calculate (and save for later) the scale factor for this bin.
+ double binCountScaleFactor;
+ if (maxFetchRate == 0.0)
+ binCountScaleFactor = Double.POSITIVE_INFINITY;
+ else
+ binCountScaleFactor = 1.0 + 1.0 / (minMsPerFetch * maxFetchRate);
+ binCountScaleFactors[i] = binCountScaleFactor;
+ weightedMinimumDepths[i] = currentMinimumDepth / binCountScaleFactor;
+ }
+
+ double highestAdjustedCount = 0.0;
+ // Find the bin with the largest effective count, and use that for the
document's priority.
+ // (This of course assumes that the slowest throttle is the one that wins.)
+ for (int i = 0; i < binNames.length; i++)
+ {
+ String binName = binNames[i];
+ double binCountScaleFactor = binCountScaleFactors[i];
+ double weightedMinimumDepth = weightedMinimumDepths[i];
+
+ double thisCount =
binManager.getIncrementBinValue(binName,weightedMinimumDepth);
+ double adjustedCount;
+ // Use the scale factor already calculated above to yield a priority
that is adjusted for the fetch rate.
+ if (binCountScaleFactor == Double.POSITIVE_INFINITY)
+ adjustedCount = Double.POSITIVE_INFINITY;
+ else
+ adjustedCount = thisCount * binCountScaleFactor;
+ if (adjustedCount > highestAdjustedCount)
+ highestAdjustedCount = adjustedCount;
+ }
+
+ // Calculate the proper log value
+ double returnValue;
+
+ if (highestAdjustedCount == Double.POSITIVE_INFINITY)
+ returnValue = Double.POSITIVE_INFINITY;
+ else
+ returnValue = Math.log(1.0 + highestAdjustedCount);
+
+ if (Logging.scheduling.isDebugEnabled())
+ {
+ StringBuilder sb = new StringBuilder();
+ int k = 0;
+ while (k < binNames.length)
+ {
+ sb.append(binNames[k++]).append(" ");
+ }
+ Logging.scheduling.debug("Document with bins ["+sb.toString()+"] given
priority value "+new Double(returnValue).toString());
+ }
+
+
+ return returnValue;
+ }
+
+
+ /** Calculate the maximum fetch rate for a given set of bins for a given
connection.
+ * This is used to adjust the final priority of a document.
+ */
+ protected static double[] calculateMaxFetchRates(String[] binNames,
IRepositoryConnection connection)
+ {
+ ThrottleLimits tl = new ThrottleLimits(connection);
+ return tl.getMaximumRates(binNames);
+ }
+
+ /** This class represents the throttle limits out of the connection
specification */
+ protected static class ThrottleLimits
+ {
+ protected List<ThrottleLimitSpec> specs = new
ArrayList<ThrottleLimitSpec>();
+
+ public ThrottleLimits(IRepositoryConnection connection)
+ {
+ String[] throttles = connection.getThrottles();
+ int i = 0;
+ while (i < throttles.length)
+ {
+ try
+ {
+ specs.add(new
ThrottleLimitSpec(throttles[i],(double)connection.getThrottleValue(throttles[i])));
+ }
+ catch (PatternSyntaxException e)
+ {
+ }
+ i++;
+ }
+ }
+
+ public double[] getMaximumRates(String[] binNames)
+ {
+ double[] rval = new double[binNames.length];
+ for (int j = 0 ; j < binNames.length ; j++)
+ {
+ String binName = binNames[j];
+ double maxRate = Double.POSITIVE_INFINITY;
+ for (ThrottleLimitSpec spec : specs)
+ {
+ Pattern p = spec.getRegexp();
+ Matcher m = p.matcher(binName);
+ if (m.find())
+ {
+ double rate = spec.getMaxRate();
+ // The direction of this inequality reflects the fact that the
throttling is conservative when more rules are present.
+ if (rate < maxRate)
+ maxRate = rate;
+ }
+ }
+ rval[j] = maxRate;
+ }
+ return rval;
+ }
+
+ }
+
+ /** This is a class which describes an individual throttle limit, in fetches
per millisecond. */
+ protected static class ThrottleLimitSpec
+ {
+ /** Regexp */
+ protected final Pattern regexp;
+ /** The fetch limit for all bins matching that regexp, in fetches per
millisecond */
+ protected final double maxRate;
+
+ /** Constructor */
+ public ThrottleLimitSpec(String regexp, double maxRate)
+ throws PatternSyntaxException
+ {
+ this.regexp = Pattern.compile(regexp);
+ this.maxRate = maxRate;
+ }
+
+ /** Get the regexp. */
+ public Pattern getRegexp()
+ {
+ return regexp;
+ }
+
+ /** Get the max count */
+ public double getMaxRate()
+ {
+ return maxRate;
+ }
}
}
