manifoldcf-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kwri...@apache.org
Subject svn commit: r1546328 - /manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
Date Thu, 28 Nov 2013 09:57:01 GMT
Author: kwright
Date: Thu Nov 28 09:57:00 2013
New Revision: 1546328

URL: http://svn.apache.org/r1546328
Log:
Add code for tracking minimum depth from QueueTracker.

Modified:
    manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java

Modified: 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=1546328&r1=1546327&r2=1546328&view=diff
==============================================================================
--- manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
(original)
+++ manifoldcf/branches/CONNECTORS-781/framework/pull-agent/src/main/java/org/apache/manifoldcf/crawler/system/ReprioritizationTracker.java
Thu Nov 28 09:57:00 2013
@@ -36,9 +36,11 @@ public class ReprioritizationTracker
   protected final static String trackerWriteLock = "_REPR_TRACKER_LOCK_";
   protected final static String trackerProcessIDResource = "_REPR_TRACKER_PID_";
   protected final static String trackerTimestampResource = "_REPR_TIMESTAMP_";
+  protected final static String trackerMinimumDepthResource = "_REPR_MINDEPTH_";
   
   /** Lock manager */
   protected final ILockManager lockManager;
+  protected final IBinManager binManager;
   
   /** Constructor.
   */
@@ -46,6 +48,7 @@ public class ReprioritizationTracker
     throws ManifoldCFException
   {
     lockManager = LockManagerFactory.make(threadContext);
+    binManager = BinManagerFactory.make(threadContext);
   }
   
   /** Start a reprioritization activity.
@@ -69,6 +72,24 @@ public class ReprioritizationTracker
       }
       writeTime(new Long(prioritizationTime));
       writeProcessID(processID);
+      try
+      {
+        binManager.reset();
+      }
+      catch (Throwable e)
+      {
+        writeTime(null);
+        writeProcessID(null);
+        if (e instanceof Error)
+          throw (Error)e;
+        else if (e instanceof RuntimeException)
+          throw (RuntimeException)e;
+        else if (e instanceof ManifoldCFException)
+          throw (ManifoldCFException)e;
+        else
+          throw new RuntimeException("Unknown exception: "+e.getClass().getName()+": "+e.getMessage(),e);
+      }
+      writeMinimumDepth(0.0);
     }
     finally
     {
@@ -145,6 +166,78 @@ public class ReprioritizationTracker
     }
   }
   
+  /** Assess the current minimum depth.
+  * This method is called to provide information about the priorities of the documents being
currently
+  * queued.  It is the case that it is unoptimal to assign document priorities that are fundamentally
higher than this value,
+  * because then the new documents will be preferentially queued, and the goal of distributing
documents across bins will not be
+  * adequately met.
+  *@param binNamesSet is the current set of priorities we see on the queuing operation.
+  */
+  public void assessMinimumDepth(Double[] binNamesSet)
+    throws ManifoldCFException
+  {
+    double newMinPriority = Double.MAX_VALUE;
+    for (Double binValue : binNamesSet)
+    {
+      if (binValue.doubleValue() < newMinPriority)
+        newMinPriority = binValue.doubleValue();
+    }
+
+    if (newMinPriority != Double.MAX_VALUE)
+    {
+
+      lockManager.enterWriteLock(trackerWriteLock);
+      try
+      {
+        Long reproTime = readTime();
+        String processID = readProcessID();
+        if (reproTime == null || processID == null)
+        {
+          double currentMinimumDepth = readMinimumDepth();
+
+          // Convert minPriority to minDepth.
+          // Note that this calculation does not take into account anything having to do
with connection rates, throttling,
+          // or other adjustment factors.  It allows us only to obtain the "raw" minimum
depth: the depth without any
+          // adjustments.
+          double newMinDepth = Math.exp(newMinPriority)-1.0;
+
+          if (newMinDepth > currentMinimumDepth)
+          {
+            writeMinimumDepth(newMinDepth);
+            if (Logging.scheduling.isDebugEnabled())
+              Logging.scheduling.debug("Setting new minimum depth value to "+new Double(currentMinimumDepth).toString());
+          }
+          else
+          {
+            if (newMinDepth < currentMinimumDepth && Logging.scheduling.isDebugEnabled())
+              Logging.scheduling.debug("Minimum depth value seems to have been set too high
too early! currentMin = "+new Double(currentMinimumDepth).toString()+"; queue value = "+new
Double(newMinDepth).toString());
+          }
+        }
+      }
+      finally
+      {
+        lockManager.leaveWriteLock(trackerWriteLock);
+      }
+    }
+  }
+
+  /** Retrieve current minimum depth.
+  *@return the current minimum depth to use.
+  */
+  public double getMinimumDepth()
+    throws ManifoldCFException
+  {
+    lockManager.enterReadLock(trackerWriteLock);
+    try
+    {
+      return readMinimumDepth();
+    }
+    finally
+    {
+      lockManager.leaveReadLock(trackerWriteLock);
+    }
+  }
+  
   // Protected methods
   
   /** Read time.
@@ -234,5 +327,45 @@ public class ReprioritizationTracker
     }
   }
 
+  /** Read minimum depth.
+  *@return the minimum depth.
+  */
+  protected double readMinimumDepth()
+    throws ManifoldCFException
+  {
+    byte[] data = lockManager.readData(trackerMinimumDepthResource);
+    if (data == null || data.length != 8)
+      return 0.0;
+    long dataLong = ((long)data[0]) & 0xffL +
+      (((long)data[1]) << 8) & 0xff00L +
+      (((long)data[2]) << 16) & 0xff0000L +
+      (((long)data[3]) << 24) & 0xff000000L +
+      (((long)data[4]) << 32) & 0xff00000000L +
+      (((long)data[5]) << 40) & 0xff0000000000L +
+      (((long)data[6]) << 48) & 0xff000000000000L +
+      (((long)data[7]) << 56) & 0xff00000000000000L;
+
+    return Double.longBitsToDouble(dataLong);
+  }
+  
+  /** Write minimum depth.
+  *@param the minimum depth.
+  */
+  protected void writeMinimumDepth(double depth)
+    throws ManifoldCFException
+  {
+    long dataLong = Double.doubleToLongBits(depth);
+    byte[] data = new byte[8];
+    data[0] = (byte)(dataLong & 0xffL);
+    data[1] = (byte)((dataLong >> 8) & 0xffL);
+    data[2] = (byte)((dataLong >> 16) & 0xffL);
+    data[3] = (byte)((dataLong >> 24) & 0xffL);
+    data[4] = (byte)((dataLong >> 32) & 0xffL);
+    data[5] = (byte)((dataLong >> 40) & 0xffL);
+    data[6] = (byte)((dataLong >> 48) & 0xffL);
+    data[7] = (byte)((dataLong >> 56) & 0xffL);
+    lockManager.writeData(trackerMinimumDepthResource,data);
+  }
+  
 }
 



Mime
View raw message