diff --git a/commons/src/main/java/org/archive/bdb/BdbModule.java b/commons/src/main/java/org/archive/bdb/BdbModule.java index 2655b6a0..8e959553 100644 --- a/commons/src/main/java/org/archive/bdb/BdbModule.java +++ b/commons/src/main/java/org/archive/bdb/BdbModule.java @@ -28,6 +28,7 @@ import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.io.PrintWriter; import java.io.Serializable; +import java.math.BigInteger; import java.util.ArrayList; import java.util.Arrays; import java.util.HashSet; @@ -37,6 +38,7 @@ import java.util.Map; import java.util.Set; import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.logging.Level; import java.util.logging.Logger; @@ -207,6 +209,19 @@ Serializable, Closeable { this.checkpointCopyLogs = checkpointCopyLogs; } + /** + * Expected number of concurrent threads; used to tune nLockTables + * according to JE FAQ + * http://www.oracle.com/technology/products/berkeley-db/faq/je_faq.html#33 + */ + int expectedConcurrency = 25; + public int getExpectedConcurrency() { + return expectedConcurrency; + } + public void setExpectedConcurrency(int expectedConcurrency) { + this.expectedConcurrency = expectedConcurrency; + } + private transient EnhancedEnvironment bdbEnvironment; private transient StoredClassCatalog classCatalog; @@ -256,6 +271,14 @@ Serializable, Closeable { config.setCachePercent(cachePercent); config.setSharedCache(sharedCache); + // we take the advice literally from... + // http://www.oracle.com/technology/products/berkeley-db/faq/je_faq.html#33 + long nLockTables = expectedConcurrency-1; + while(!BigInteger.valueOf(nLockTables).isProbablePrime(Integer.MAX_VALUE)) { + nLockTables--; + } + config.setConfigParam("je.lock.nLockTables", Long.toString(nLockTables)); + f.mkdirs(); this.bdbEnvironment = new EnhancedEnvironment(f, config); @@ -329,7 +352,7 @@ Serializable, Closeable { } - public Map getBigMap(String dbName, boolean recycle, + public ConcurrentMap getBigMap(String dbName, boolean recycle, Class key, Class value) throws DatabaseException { @SuppressWarnings("unchecked") @@ -359,7 +382,7 @@ Serializable, Closeable { } - + // TODO:FIXME: restore functionality @SuppressWarnings("unchecked") private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { @@ -374,8 +397,10 @@ Serializable, Closeable { for (CachedBdbMap map: bigMaps.values()) { map.initialize( this.bdbEnvironment, - map.getKeyClass(), - map.getValueClass(), + null, + null, +// map.getKeyClass(), +// map.getValueClass(), this.classCatalog); } for (DatabasePlusConfig dpc: databases.values()) { diff --git a/commons/src/main/java/org/archive/util/CachedBdbMap.java b/commons/src/main/java/org/archive/util/CachedBdbMap.java index d0278b60..32833e65 100644 --- a/commons/src/main/java/org/archive/util/CachedBdbMap.java +++ b/commons/src/main/java/org/archive/util/CachedBdbMap.java @@ -16,11 +16,9 @@ * See the License for the specific language governing permissions and * limitations under the License. */ - package org.archive.util; import java.io.File; -import java.io.IOException; import java.io.Serializable; import java.lang.ref.PhantomReference; import java.lang.ref.Reference; @@ -33,9 +31,17 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Level; import java.util.logging.Logger; +import org.archive.crawler.framework.CrawlController; + import com.sleepycat.bind.EntryBinding; import com.sleepycat.bind.serial.SerialBinding; import com.sleepycat.bind.serial.StoredClassCatalog; @@ -44,6 +50,7 @@ import com.sleepycat.collections.StoredSortedMap; import com.sleepycat.je.Database; import com.sleepycat.je.DatabaseConfig; import com.sleepycat.je.DatabaseException; +import com.sleepycat.je.DeadlockException; import com.sleepycat.je.Environment; import com.sleepycat.je.EnvironmentConfig; @@ -52,16 +59,41 @@ import com.sleepycat.je.EnvironmentConfig; * holding a cache of soft referenced objects. That is objects are not written * to disk until they are not referenced by any other object and therefore can be * Garbage Collected. + *

+ * BDB Java Edition is actually a btree. Flush to disk can be forced by sync(). + *

+ * To ensure that changes/mutations to values in this map are coherent and + * consistent at the application level, it is assumed that the application + * level only mutates values that are in this map and does not retain references + * to values longer than necessary. This allows mappings to be persisted + * during GC without explicit transactions or write operations. + *

+ * There are two styles of CachedBdbMap usage: + *

+ * 1. single threaded (or externally synchronized) activity that uses any + * Map and ConcurrentMap methods available and specifically requires remove(). + *

+ * 2. concurrent, high volume, accretive-only activity that uses + * {@link #putIfAbsent}, but not the {@link #put}, the {@link #replace} + * or 1 arg {@link #remove} methods. + * The concurrent {@link #replace} methods can be used if application level logic + * can rule out surprise unmapping of values between threads. + * This usage style does not require locking memMap and diskMap together + * to guarantee cache coherence and consistency. + *

+ * Both styles rely on an internal expunge operation (or the explicit + * {@link #sync()}) to save changes to values. + * The single threaded case can also use {@link #put} on top of an + * existing entry as long as no thread retains the previous value instance. * * @author John Erik Halse * @author stack * @author gojomo + * @author paul baclace (conversion to ConcurrentMap) * */ -@SuppressWarnings("unchecked") public class CachedBdbMap extends AbstractMap -implements Map, Serializable { - +implements ConcurrentMap, Serializable { private static final long serialVersionUID = -8655539411367047332L; private static final Logger logger = @@ -70,6 +102,21 @@ implements Map, Serializable { /** The database name of the class definition catalog.*/ private static final String CLASS_CATALOG = "java_class_catalog"; + /** Number of attempts to tolerate for put(). + * In BDB, simple write lock contention (non-deadlock) arrives as + * DeadlockException even though we are not using explicit transactions. + * For details see discussion at + * http://forums.oracle.com/forums/thread.jspa?threadID=521898&tstart=-1 + * Value here should be a function of + * setLockTimeout() in {@link CrawlController#setupBdb()} + * and CrawlController.getToeCount(); since CrawlController is not + * accessible from this util class, a value suitable for average + * disk performance and 100 TOE threads is use. + * (In practice, {@link #put} contention of statistics is where + * timeouts were found.) + */ + private static final int BDB_LOCK_ATTEMPT_TOLERANCE = 24; + /** * A map of BDB JE Environments so that we reuse the Environment for * databases in the same directory. @@ -87,43 +134,162 @@ implements Map, Serializable { /** The Collection view of the BDB JE database used for this instance. */ protected transient StoredSortedMap diskMap; - /** The softreferenced cache */ - private transient Map> memMap; + /** The softreferenced cache of diskMap. + * Policy is: a memMap value is always correct *if present*. + * The diskMap value is only correct if there is no memMap value + * (and any phantom memMap values have been persisted.) + * That is, if memMap has a value, then that value is the most + * up to date. If memMap does not hold a value, then diskMap has the + * most up to date value. + *

+ * For diskMap size monitoring and concurrency support, the following + * invariant about keys is maintained: If a key is present in memMap, + * it must also present in diskMap (although the value in diskMap may + * not be the most recent, as above policy describes). + *

+ * The key presence invariant and the value coherence policy are maintained + * by using the "happens after" sense in the Java memory model by + * transitivity between diskMap and memMap. + *

+ * The clients of this class "only see one value instance" at a time so + * that multiple, independent get(k) calls return the same object instance. + *

+ * Strategy Notes about using ConcurrentMap semantics to + * implement the desired policy (see method comments for details): + *

+ * 

+     * 1. Swap in/First insert: diskMap, 
+     *     then memMap // putIfAbsent() assures atomicity.
+     * 2. Value mutation: only one value instance per key in memMap 
+     *     is maintained so identity compare of value works and all clients 
+     *     operate on the same object.
+     *     This implies that methods which change the instance 
+     *     (replace() methods) are not compatible with this design and
+     *     should only be used if instance of CachedBdbMap is used 
+     *     by a single thread.
+     *     Because content of referent is mutated, mapping is not changed;
+     *    2.1. BDB JE operation assumption: calling get(k) twice returns
+     *     values that are equals(), but not necessarily == by identity
+     *     ( diskMap.get(k).equals(diskMap.get(k)) is true, but 
+     *      diskMap.get(k) != diskMap.get(k) (might be, but not guaranteed)).
+     * 3. Swap out/flush: diskMap update, then memMap remove: 
+     *     diskMap.put(k,v2); // exclusive only if performed when 
+     *     processing ref queue during expunge;
+     *     Avoid expunge races with a countdown latch in SoftEntry;
+     *     memMap.remove(k,v2); if memMap race lost, then no harm done
+     *     (notice that (k,v2) could match in either diskMap or memMap,
+     *     or both, or neither).
+     *    3.1. Only expunge does an update of an existing diskMap entry for 
+     *      concurrent methods.
+     *    3.2. CachedBdbMap.get() must be able to return referent or otherwise
+     *      effect a swap-in if get() is invoked during a swap out of a
+     *      particular key-value entry.
+     * 4. sync() to disk: synchronized sync() // fully locked, no races.
+     *
+ *

+ * See the "find or create" style cache + * {@link ServerCache#getServerFor(CandidateURI)} for the major + * high-concurrency client of this class in terms of number of objects + * stored and performance impact. + */ + private transient ConcurrentHashMap> memMap; protected transient ReferenceQueue refQueue; /** The number of objects in the diskMap StoredMap. * (Package access for unit testing.) */ - protected int diskMapSize = 0; + protected AtomicInteger diskMapSize = new AtomicInteger(0); + // The following atomic counters are used for reporting + // code branches that can occur. + // Some of these are frequent while others are rare or depend on + // application level and GC behavior. + // If problems occur, these provide diagnostics such as whether usage went + // outside the bounds of the above described expected "two styles". + + /** count of expunge already done */ + final transient private AtomicInteger expungeStatsNullPhantom + = new AtomicInteger(0); + + /** count of {@link #putIfAbsent} / {@link #remove} races, + * if they occur */ + final transient private AtomicInteger expungeStatsTransientCond + = new AtomicInteger(0); + + /** count of {@link SoftEntry#awaitExpunge()) */ + final transient private AtomicInteger expungeStatsAwaitExpunge + = new AtomicInteger(0); + + /** count of expunge already done to see if they occur */ + final transient private AtomicInteger expungeStatsNullValue + = new AtomicInteger(0); + + /** count of expunge of entries not in memMap to see if they occur */ + final transient private AtomicInteger expungeStatsNotInMap + = new AtomicInteger(0); + + /** static count {@link SoftEntry#awaitExpunge()) timeouts to see if + * they occur */ + final static private AtomicInteger expungeStatsAwaitTimeout + = new AtomicInteger(0); + + /** static count of swap-in timeouts */ + final static private AtomicInteger expungeStatsSwapIn + = new AtomicInteger(0); + + /** static count of swap-in re-tries timeouts to see if they occur */ + final static private AtomicInteger expungeStatsSwapInRetry + = new AtomicInteger(0); + + /** static count of expunge put() to BDB (implies disk) */ + final static private AtomicInteger expungeStatsDiskPut + = new AtomicInteger(0); + + /** count of one arg {@link #remove } use */ + final transient private AtomicInteger useStatsRemove1Used + = new AtomicInteger(0); + + /** count of two arg {@link #remove} use */ + final transient private AtomicInteger useStatsRemove2Used + = new AtomicInteger(0); + + /** count of {@link #replace} (2 or 3 arg) use */ + final transient private AtomicInteger useStatsReplaceUsed + = new AtomicInteger(0); + + /** count of {@link #put} use */ + final transient private AtomicInteger useStatsPutUsed + = new AtomicInteger(0); + + /** count of {@link #putIfAbsent} use */ + final transient private AtomicInteger useStatsPutIfUsed + = new AtomicInteger(0); + + /** count of {@link #sync()} use */ + final transient private AtomicInteger useStatsSyncUsed + = new AtomicInteger(0); + /** * Count of times we got an object from in-memory cache. */ - private long cacheHit = 0; + private AtomicLong cacheHit = new AtomicLong(0); /** - * Count of times the {@link CachedBdbMap#get(Object)} method was called. + * Count of times the {@link CachedBdbMap#get} method was called. */ - private long countOfGets = 0; + private AtomicLong countOfGets = new AtomicLong(0); /** * Count of every time we went to the disk-based map AND we found an * object (Doesn't include accesses that came back null). */ - private long diskHit = 0; + private AtomicLong diskHit = new AtomicLong(0); /** * Name of bdbje db. */ private String dbName = null; - - private Class keyClass; - - - private Class valueClass; - - /** * Reference to the Reference#referent Field. */ @@ -131,6 +297,7 @@ implements Map, Serializable { static { // We need access to the referent field in the PhantomReference. // For more on this trick, see + // // http://www.javaspecialists.co.za/archive/Issue098.html and for // discussion: // http://www.theserverside.com/tss?service=direct/0/NewsThread/threadViewer.markNoisy.link&sp=l29865&sp=l146901 @@ -144,6 +311,14 @@ implements Map, Serializable { } } + /** internal behavior characterization flag. + * log a warning when operations are performed that could violate the + * "only one reference" design rule. The problem to avoid is + * mutating an unmapped value instance that would not be persisted. + * (warning is emitted at most once per instance) + */ + final private boolean LOG_ERROR_ON_DESIGN_VIOLATING_METHODS=false; + /** * Simple structure to keep needed information about a DB Environment. */ @@ -210,7 +385,8 @@ implements Map, Serializable { EnvironmentConfig cfg = this.dbEnvironment.environment.getConfig(); logger.info("BdbConfiguration: Cache percentage " + cfg.getCachePercent() + ", cache size " + cfg.getCacheSize() + - ", Map size: " + size()); + ", Map size: " + size() + " cfg=" + cfg); + System.out.println(" cfg=" + cfg); //DEBUG } } @@ -226,37 +402,69 @@ implements Map, Serializable { * @param classCatalog * @throws DatabaseException */ - @SuppressWarnings("unchecked") public synchronized void initialize(final Environment env, final Class keyClass, final Class valueClass, final StoredClassCatalog classCatalog) throws DatabaseException { initializeInstance(); this.db = openDatabase(env, this.dbName); - this.keyClass = keyClass; - this.valueClass = valueClass; this.diskMap = createDiskMap(this.db, classCatalog, keyClass, valueClass); } - - public Class getKeyClass() { - return keyClass; - } - - - public Class getValueClass() { - return valueClass; - } - /** * Do any instance setup. * This method is used by constructors and when deserializing an instance. */ protected void initializeInstance() { - this.memMap = new HashMap>(); + // ToDo: initial capacity should be related to number of seeds, max depth, max docs + this.memMap = new ConcurrentHashMap>( + 8192, // initial capacity + 0.9f, // acceptable load factor + 64 // est. number of concurrent threads + ); this.refQueue = new ReferenceQueue(); + startExpunger(); } + private void startExpunger() { + /* + SettingsHandler ambientSettingsHandler = null; + try { + ambientSettingsHandler + = SettingsHandler.getThreadContextSettingsHandler(); + } catch (RuntimeException absorbed) { + } + if (ambientSettingsHandler != null) { + this.expunger = new Expunger("Expunger_" + dbName, refQueue, + logger, ambientSettingsHandler); + this.expunger.setDaemon(true); + this.expunger.setPriority(Thread.MAX_PRIORITY - 1); + this.expunger.start(); + } + */ + } + + /** + * + * @return true if expunder thread was stopped, false if no expunger is + * running. + */ + private boolean stopExpunger() { + /* + if (expunger != null) { + expunger.interrupt(); + try { + expunger.join(); + } catch (InterruptedException ignored) { + } + expunger = null; + return true; + } + */ + return false; + } + + @SuppressWarnings("unchecked") protected StoredSortedMap createDiskMap(Database database, StoredClassCatalog classCatalog, Class keyClass, Class valueClass) { EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(keyClass); @@ -267,9 +475,73 @@ implements Map, Serializable { if(valueBinding == null) { valueBinding = new SerialBinding(classCatalog, valueClass); } + // BDB JE is btree only, so Sorted Map is no additional work return new StoredSortedMap(database, keyBinding, valueBinding, true); } + /** + * Conditionally await for an entry to be expunged. + * If the given entry is in the process of being expunged, this + * will block until it is complete and then return true. + * If the entry should be expunged, this will try to do the expunge. + * If the entry is not being expunged, + * then this will return immediately with a value of false. + * If this thread is interrupted, the condition is cleared and + * the method returns immediately. + *

+ * For transitive "happens after" correctness invariant to be true, + * these operations must occur in the following order in one thread: + * 

+     *  1. if (entry.get() == null) // cleared by GC
+     *  2. if (entry.startExpunge()) // thread has exclusive expunge
+     *  3. diskMap.put(k, entry.getPhantom().doctoredGet())
+     *  4. memMap.remove(k, entry)
+     *  5. entry.clearPhantom()
+     * 
+     *  Concurrent threads interested in the same entry will execute:
+     *  
+     *  1. if (entry.get() == null) // cleared by GC
+     *  2. if (! entry.startExpunge()) // thread cannot do expunge
+     *  3. if (entry.expungeStarted()) 
+     *  4. entry.awaitExpunge();
+     *
+ * @param entry + * @return true if entry was expunged, cleared, and no longer in memMap. + */ + private boolean isExpunged(final SoftEntry entry) { + boolean ret = false; + + V val = entry.get(); + if (val == null) { + if (entry.startExpunge()) { // try do the expunge now + // got exclusive expunge access + expungeStaleEntry(entry, true); + return true; + } else if (entry.expungeStarted()) { // wait for expunge + expungeStatsAwaitExpunge.incrementAndGet(); + expungeStaleEntries(); + while (true) { + // the proper way to handle interrupted + try { + ret = entry.awaitExpunge(); + return ret; + } catch (InterruptedException ex) { + if (Thread.interrupted()) { + break; // be lively + } + } + } + return ret; + } else { + logger.warning("could not start expunge, and expunge was never started and val is null"); + return false; + } + } else { // entry not collected, not expunged + return false; + } + } + + /** * Get the database environment for a physical directory where data will be * stored. @@ -290,9 +562,9 @@ implements Map, Serializable { envConfig.setAllowCreate(true); envConfig.setTransactional(false); - // We're doing the caching ourselves so setting these at the lowest - // possible level. - envConfig.setCachePercent(1); +// // We're doing the caching ourselves so setting these at the lowest +// // possible level. +// envConfig.setCachePercent(1); DbEnvironmentEntry env = new DbEnvironmentEntry(); try { env.environment = new Environment(dbDir, envConfig); @@ -325,6 +597,7 @@ implements Map, Serializable { } public synchronized void close() throws DatabaseException { + stopExpunger(); // Close out my bdb db. if (this.db != null) { try { @@ -368,66 +641,472 @@ implements Map, Serializable { throw new UnsupportedOperationException(); } - public synchronized V get(final Object object) { + public V get(final Object object) { K key = toKey(object); - countOfGets++; + countOfGets.incrementAndGet(); expungeStaleEntries(); - if (countOfGets % 10000 == 0) { + if (countOfGets.get() % 10000 == 0) { + dumpExtraStats(); logCacheSummary(); } + + // check mem cache + V val = _getMem(key); + if (val != null) { // cache hit + return val; + } + + // check backing diskMap + V valDisk = _getDisk(key); // disk io can occur here + return valDisk; + } + + @SuppressWarnings("unchecked") + private V _getDisk(K key) { + V v = diskMapGet(key); // disk io can occur here + if (v != null) { + diskHit.incrementAndGet(); + // + // cache in memMap (swap in) + // + expungeStatsSwapIn.incrementAndGet(); + SoftEntry newEntry = new SoftEntry(key, v, refQueue); + do { + SoftEntry existingEntry = + (SoftEntry) memMap.putIfAbsent(key, newEntry); // unchecked cast + if (existingEntry != null) { + V prevValue = (V) existingEntry.get(); + if (prevValue != null) { + // another thread beat this thread to memMap insert for swap-in + newEntry.clearPhantom(); // instance not needed + return prevValue; + } else { // SoftEntry referent has been collected + // another thread beat this thread in memMap insert for swap-in + // AND existingEntry was cleared before we could see it. + // This condition requires 2 competing threads working on + // the same entry, and a full GC to be interleaved. + // This is like a simultaneous collision between + // 2 cars and an airplane. + // Wait for expunge to finish + + if (isExpunged(existingEntry)) { + // existingEntry is no longer in memMap + expungeStatsSwapInRetry.incrementAndGet(); + continue; // re-try insert into memMap + } else { + try { + // stuck memMap entry + Thread.sleep(0L, 100000); + } catch (InterruptedException ignored) { + } + logger.warning("Swap-In Retry"); + continue; // re-try insert into memMap + } + } + } + return v; + } while (true); + } + return null; + } + + private V _getMem(K key) { + do { SoftEntry entry = memMap.get(key); if (entry != null) { V val = entry.get(); // get & hold, so not cleared pre-return if (val != null) { - cacheHit++; + cacheHit.incrementAndGet(); return val; } - // Explicitly clear this entry from referencequeue since its - // value is null. - expungeStaleEntry(entry); + // null SoftEntry.get() means entry is being GC-ed + // need to wait for expunge to finish and then retry + // which will lead to a diskMap.get() + if (isExpunged(entry)) { + // entry has been removed, lookup again + continue; // tail recursion optimization + } else { + logger.warning("entry not expunged AND value is null"); + return null; } - - // check backing diskMap - V v = diskMapGet(key); - if (v != null) { - diskHit++; - memMap.put(key, new SoftEntry(key, v, refQueue)); } - return v; + return null; + } while (true); } /** - * Info to log, if at FINE level, on every get() + * Info to log, if at FINE level */ private void logCacheSummary() { - if (!logger.isLoggable((Level.FINE))) { - return; - } - if (cacheHit + diskHit == 0) { - // Preent division by zero below - return; - } - try { - long cacheHitPercent = (cacheHit * 100) / (cacheHit + diskHit); - logger.fine("DB name: " + this.db.getDatabaseName() - + ", Cache Hit: " + cacheHitPercent - + "%, Not in map: " + (countOfGets - (cacheHit + diskHit)) - + ", Total number of gets: " + countOfGets); - } catch (DatabaseException e) { - // This is just for logging so ignore DB Exceptions + if (logger.isLoggable((Level.FINE))) { + logger.fine(composeCacheSummary()); } } - public synchronized V put(K key, V value) { - V prevVal = get(key); - memMap.put(key, new SoftEntry(key, value, refQueue)); - diskMap.put(key,value); // dummy - if(prevVal==null) { - diskMapSize++; + private String composeCacheSummary() { + long totalHits = cacheHit.get() + diskHit.get(); + if (totalHits < 1) { + return ""; } - return prevVal; + long cacheHitPercent + = (cacheHit.get() * 100) / totalHits; + StringBuilder sb = new StringBuilder(120); + try { + String sname = this.db.getDatabaseName(); + sb.append("DB name:").append(sname).append(", "); + } catch (DatabaseException ignore) { + } + sb.append(" Cache Hit: ") + .append(cacheHitPercent) + .append("%, Not in map: ") + .append((countOfGets.get() - totalHits)) + .append(", Total number of gets: ") + .append(countOfGets.get()); + return sb.toString(); + } + + /** Map.put() implementation. + *

+ * Warning: This method violates the "only expose one value instance" + * design rule for this class. Multiple instances means that it is + * undetermined as to which instance will be saved to disk last. + * This method can be safely used when only one thread has access to + * this instance of CachedBdbMap. + *

If possible, use {@link #putIfAbsent} instead. + * + * 

+     * Preconditions: (diskMap.containsKey(), memMap.containsKey()) are:
+     * (*,*) put() cannot assure "only expose one value instance", so
+     *     all cases are the same: put to diskMap then memMap; clear any 
+     *     pre-existing SoftEntry to prevent expunge of old value.
+     * 
+     * PostConditions:
+     * (T,T) both memMap and diskMap will have entries for given key; if 
+     *     null is returned, then this is the first time the key has an
+     *     entry, otherwise the previous value in diskMap will not be 
+     *     updated until expunge. 
+     *
+ * + * @param key + * @param value + * @return previous value or null. + */ + @SuppressWarnings("unchecked") + @Override + public V put(K key, V value) { + V prevMemValue = null; + V prevDiskValue = null; + int pu = useStatsPutUsed.incrementAndGet(); + if (pu == 1 && LOG_ERROR_ON_DESIGN_VIOLATING_METHODS) { + logger.warning("design violating put() used on dbName=" + dbName); + } + int attemptTolerance = BDB_LOCK_ATTEMPT_TOLERANCE; + while (--attemptTolerance > 0) { + try { + prevDiskValue = (V)diskMap.put(key,value);// unchecked cast, disk io + break; + } catch (Exception e) { + if (e instanceof com.sleepycat.util.RuntimeExceptionWrapper + && e.getCause() instanceof DeadlockException + && attemptTolerance > 0) { + if (attemptTolerance == BDB_LOCK_ATTEMPT_TOLERANCE - 1) { + // emit once on first retry + logger.warning("BDB implicit transaction timeout while " + + "waiting to acquire lock, retrying; " + + " dbName=" + dbName + " key=" + key); + } + continue; + } else { + logger.warning("unexpected exception: " + e); + throw new RuntimeException(e); + } + } + } + if (prevDiskValue == null) { + diskMapSize.incrementAndGet(); + } + SoftEntry prevEntry = + memMap.put(key, new SoftEntry(key, value, refQueue)); + if (prevEntry != null) { + prevMemValue = prevEntry.get(); + // prevent previous SoftEntry from being expunged + prevEntry.clearPhantom(); + } + + if (prevMemValue != null) return prevMemValue; + return prevDiskValue; } + /** Replace entry for key only if currently mapped to given value. + * To maintain the illusion of an invisible cache, + * if the memMap has no mapping, the diskMap entry, if any, must + * be swapped-in to memMap in order to see if the given oldValue + * matches. + * + *

Possible disk io to swap in from diskMap. + *

+ * Warning: This method violates the "only expose one value instance" + * design rule for this class. Multiple instances means that it is + * undetermined as to which instance will be saved to disk last. + * This method can be safely used when only one thread has access to + * this instance of CachedBdbMap. + * 

+     * Preconditions: (diskMap.containsKey(), memMap.containsKey()) are:
+     * (F,F) nothing to replace. 
+     *    PostCondition: (F,F) no replace.
+     * (T,F) value must be swapped-in, do replace(), 
+     *    PostCondition: (T,T) if
+     *    replace occurred or not (value might have been replaced), diskMap
+     *    need not be updated because expunge will do that.
+     * (*,T) normal swapped-in condition (we can assume that the method which
+     *    put a mapping to memMap took care of creating one for diskMap), 
+     *    memMap value is most recent and expunge can take care of 
+     *    updating diskMap, so we only do replace() on memMap; 
+     *    PostCondition: (*,T).
+     *
+ * @return true if the replace was peformed (only if the previous mapping + * of key was to given oldValue). + */ + @SuppressWarnings("unchecked") + //@Override + public boolean replace(K key, V oldValue, V newValue) { + int ru = useStatsReplaceUsed.incrementAndGet(); + if (ru == 1 && LOG_ERROR_ON_DESIGN_VIOLATING_METHODS) { + // warn on non-accretive use + logger.warning("design violating replace(,,) used on dbName=" + dbName); + } + + // make the ref wrappers + SoftEntry newEntry = new SoftEntry(key, newValue, refQueue); + SoftEntry oldEntry = new SoftEntry(key, oldValue, refQueue); + + try { + boolean memReplaced = memMap.replace(key, oldEntry, newEntry); + if (memReplaced) { // case (*,T) + newEntry = null; // skip neutralize in finally + return true; + } else { // case (?,F), does diskMap have a mapping? + boolean diskReplaced = diskMap.replace(key, oldValue, newValue); + if (diskReplaced) { // case (T,F) + // swap-in + memMap.put(key, newEntry); + newEntry = null; // skip neutralize in finally + return true; + } else { // case (F,F) + return false; // nothing to replace + } + } + } finally { + // neutralize any unused ref wrappers + if (oldEntry != null) { + oldEntry.clearPhantom(); + } + if (newEntry != null) { + newEntry.clearPhantom(); + } + } + } + + /** Replace entry for key only if currently mapped to some value. + * To maintain the illusion of an invisible cache, + * the diskMap value must be read to see if oldValue + * matches. The replace() operation is performed on memCache + * and then on diskCache, without synchronizing over the pair of + * operations. + *

Possible disk io to swap in from diskMap. + * + *

+ * Warning: This method violates the "only expose one value instance" + * design rule for this class. Multiple instances means that it is + * undetermined as to which instance will be saved to disk last. + * This method can be safely used when only one thread has access to + * this instance of CachedBdbMap. + * 

+     * Preconditions: (diskMap.containsKey(), memMap.containsKey()) are:
+     * (F,F) nothing to replace. PostCondition: (F,F) no change.
+     * (T,F) value must be swapped-in, do replace(), PostCondition: (T,T) if
+     *    replace occurred or not (value might have been replaced), diskMap
+     *    need not be updated because expunge will do that.
+     * (*,T) normal swapped-in condition (we can assume that the method which
+     *    put a mapping to memMap took care of creating one for diskMap), 
+     *    memMap value is most recent and expunge can take care of updating 
+     *    diskMap, so we only do replace() on memMap.
+     *
+ * @return previous value if the replace was peformed on either memMap + * or diskMap, the previous value in memMap is returned if non-null, + * otherwise the previous from diskMap is returned, if no match in + * either map, then null is returned. + */ + @SuppressWarnings("unchecked") + //@Override + public V replace(K key, V value) { + int ru = useStatsReplaceUsed.incrementAndGet(); + if (ru == 1 && LOG_ERROR_ON_DESIGN_VIOLATING_METHODS) { + // warn on non-accretive use + logger.warning("design violating replace(,) used on dbName=" + dbName); + } + SoftEntry newEntry = new SoftEntry(key, value, refQueue); + V prevMemValue = null; + SoftEntry prevEntry = (SoftEntry)memMap.replace(key, newEntry); // unchecked cast + if (prevEntry != null) { // case (*,T) AND matched + prevMemValue = (V)prevEntry.get(); + // at this point an old softEntry and phantom is ready to GC but + // it has an old value that expunge() will write to disk; + // hence the warning in comments above. + return prevMemValue; + } else { // case (?,F), no match in memMap, but does diskMap match? + V prevDiskValue = (V)diskMap.replace(key, value); // unchecked cast + if (prevDiskValue == null) { // case (F,F) + // key not mapped previously + newEntry.clearPhantom(); // instance not used + return null; // nothing to replace + } else { // case (T,F) + // swap in + memMap.put(key, newEntry); + return prevDiskValue; + } + } + } + + /** A composite putIfAbsent() over memMap and diskMap. + * If the specified key is not already associated with any value, + * associate it with the given value. + * By modifying memMap and diskMap. + *

+ * diskMap is not updated every time a value is mutated because this + * cached disk store attempts to be invisible to client code. However, + * This method will cause diskMap to be updated if there was no previous + * because this was the case for v1.14.2 and earlier. (It also means + * diskMap and memMap, both ConcurrentMaps, can be coherently maintained + * from the perspective of a one-instance-only style cache with + * diskMap updated at expunge by always modifying diskMap first here. + * The disk map size estimate is also reasonably correct if diskMap + * always has any mapping that is in memMap.) + * + *

+ * Swap in/First insert: diskMap.putIfAbsent(), + * then memMap.putIfAbsent(), putIfAbsent() assures atomicity; + * 

+     * Preconditions: (diskMap.containsKey(), memMap.containsKey(SoftEntry)) are:
+     * (F,F) initial starting conditions: is absent, put to diskMap then memMap.
+     * (F,T) transient remove(): await other thread to finish with memMap, 
+     *     then proceed as (F,F).
+     * (T,F) reloadable from disk or in process of inserting first time:
+     *     not absent.
+     * (T,T) normal swapped in condition: not absent.
+     * 
+     * PostConditions:
+     * (T,T) both memMap and diskMap will have entries for given key (if 
+     *     null is returned, then the value given is the one mapped).
+     *
+ * @param key + * @param value the value to which the given key should map, but only + * if there is existing mapping for key. + * @return + */ + @SuppressWarnings("unchecked") + //@Override + public V putIfAbsent(K key, V value) { + expungeStaleEntries(); // possible disk io + V existingDiskValue = null; + int attemptTolerance = BDB_LOCK_ATTEMPT_TOLERANCE; + while (--attemptTolerance > 0) { + try { + existingDiskValue = (V)diskMap.putIfAbsent(key,value);// unchecked cast, disk io + break; + } catch (Exception e) { + if (e instanceof com.sleepycat.util.RuntimeExceptionWrapper + && e.getCause() instanceof DeadlockException + && attemptTolerance > 0) { + if (attemptTolerance == BDB_LOCK_ATTEMPT_TOLERANCE - 1) { + // emit once on first retry + logger.warning("BDB implicit transaction timeout while " + + "waiting to acquire lock, retrying; " + + " dbName=" + dbName + " key=" + key); + } + continue; + } else { + logger.warning("unexpected exception: " + e); + throw new RuntimeException(e); + } + } + } + if (existingDiskValue == null) { // insert to diskMap succeeded + // case (F,*) + diskMapSize.incrementAndGet(); + useStatsPutIfUsed.incrementAndGet(); + // next, put mapping into cache, if not already there + SoftEntry newEntry = new SoftEntry(key, value, refQueue); + SoftEntry prevEntry = + (SoftEntry) memMap.putIfAbsent(key, newEntry); + if (prevEntry != null) { + // case (F,T) transient remove(): await other thread to finish, + // then proceed as (F,F). + // diskMap insert succeeded but insert to memMap was skipped. + // This branch occurs when a simultaneous + // remove() is occurring. + V prevMemValue = prevEntry.get(); + expungeStatsTransientCond.getAndIncrement(); + if (prevMemValue == null && isExpunged(prevEntry)) { + prevEntry = (SoftEntry) memMap.putIfAbsent(key, newEntry); + if (prevEntry == null) { + // this put "happens after" the concurrent expunge or remove() + return null; // put succeeded + } else { + prevMemValue = prevEntry.get(); + } + } + newEntry.clearPhantom(); // unused instance + // other thread won, return value it set + return prevMemValue; + } else { + // case (F,F) + // both inserts to diskMap and then memMap succeeded + return null; // put succeeded + } + } else { // case (T,*) already exists in diskMap, so not absent. + // No new entry inserted, but current value must be returned. + V retValue = get(key); // swap in, if necessary + if (retValue == null) { + // this can only happen if there is a race with this.remove() + // that just removed the key we inserted into diskMap. + // It is always possible that another thread could remove + // an entry before the calling client sees it, but this is + // unlikely by design. + logger.warning("unusual put/remove activity for key=" + key); + } + return retValue; + } + } + + /** Helper for remove() methods. + * Neutralize a map entry so that diskMap is not updated and remove + * the entry from memMap. This prevents expunge ("swap out") of + * the given entry. + * @param prevEntry SoftEntry to remove from memMap. + * @param key the key to match. + */ + private void neutralizeMemMapEntry(SoftEntry prevEntry, final Object key) { + do { + if (prevEntry.expungeStarted()) { + // GC or expunge already in progress. + isExpunged(prevEntry); + memMap.remove(key, prevEntry); // possibly redundant + break; + } else { + // reserve expunge, then clear it + if (prevEntry.startExpunge()) { // get exclusive access + memMap.remove(key, prevEntry); + prevEntry.clearPhantom(); // prevent expunge on old entry + break; + } else { + // another thread started expunge + continue; + } + } + } while (true); + } + /** * Note that a call to this method CLOSEs the underlying bdbje. * This instance is no longer of any use. It must be re-initialized. @@ -435,26 +1114,138 @@ implements Map, Serializable { * Map, this is only opportunity for cleanup. */ public synchronized void clear() { + dumpExtraStats(); this.memMap.clear(); - if (db != null) { - this.diskMap.clear(); - this.diskMapSize = 0; - try { - close(); - } catch (DatabaseException e) { - e.printStackTrace(); - } + this.diskMap.clear(); + this.diskMapSize.set(0); + try { + close(); + } catch (DatabaseException e) { + e.printStackTrace(); } } - public synchronized V remove(final Object key) { - V prevValue = get(key); - memMap.remove(key); - expungeStaleEntries(); - diskMap.remove(key); - diskMapSize--; + /** Remove mapping for the given key. + * A mapping does not need to be swapped in to be removed. + * Matching entry is removed from memMap and diskMap. + * 
+     * Preconditions: (diskMap.containsKey(), memMap.containsKey(SoftEntry)) are:
+     * (F,*) nothing to remove if diskMap has no entry.
+     * (T,F) remove from diskMap, await expunge in memMap, if in progress.
+     * (T,T) remove from diskMap, clear phantom to prevent expunge of memMap.
+     * 
+     * PostConditions:
+     * (F,F) both memMap and diskMap will NOT have entries for given key (if 
+     *     null is returned, then nothing was removed).
+     *
+ * @param key + * @return old value or null if nothing removed. + */ + @Override + @SuppressWarnings("unchecked") + public V remove(final Object key) { + V prevDiskValue = null; + // lookup memMap so we can use remove(K,V) later + SoftEntry prevEntry = (SoftEntry) memMap.get(key); // unchecked + + // map mutating operations always change diskMap first + if ((prevDiskValue = (V) diskMap.remove(key)) != null) { // unchecked + diskMapSize.decrementAndGet(); + int ru = useStatsRemove1Used.incrementAndGet(); + if (ru == 1) { + // warn on non-accretive use + logger.warning("remove() used for dbName=" + dbName); + } + } + // regardless of whether prevDiskValue is null, memMap remove + // must occur to ensure consistency. + + if (prevEntry == null) { + // nothing to remove from memMap + return prevDiskValue; + } + V prevValue = prevEntry.get(); + neutralizeMemMapEntry(prevEntry, key); + if (prevValue == null) { + return prevDiskValue; + } else { return prevValue; } + } + + /** remove item matching both the key and value. + * Matching entry is removed from memMap and diskMap. + * A mapping does not need to be swapped in to be removed. + * If the entry matching key is swapped in, then value is used + * to find and remove the entry in memMap (the value is of type V), + * and diskMap.remove(K) is used to remove the persistent entry. + * + * Otherwise, a diskMap.remove(K,V) is performed. + * 
+     * Preconditions: (diskMap.containsKey() AND value.equals(V), 
+     *        memMap.containsKey(SoftEntry) AND value.equals(V) AND
+     *        swapped-in) are:
+     * (F,F) nothing to do.
+     * (F,T) diskMap.remove(K,V) fails, but memMap.remove(K,V) still possible
+     * (T,F) remove from diskMap, await expunge in memMap, if in progress.
+     * (T,T) remove from diskMap, clear phantom to prevent expunge of memMap.
+     * 
+     * PostConditions:
+     * (F,F) both memMap and diskMap will NOT have entries for given key (if 
+     *     false is returned, then nothing was removed).
+     *
+ * @param key key used for matching mapping to remove. + * @param value value used for matching mapping to remove. + * @return true if entry removed. + */ + //@Override + public boolean remove(Object key, Object value) { // ConcurrentMap version + SoftEntry entry = null; + V memValue = null; + + // first get cache entry since we want to inhibit expunge + // if the remove succeeds on memMap + entry = (SoftEntry) memMap.get(key); // unchecked cast + if (entry != null) { + // entry found for key + memValue = entry.get(); + if (memValue != null && memValue.equals(value)) { + // case (*,T) AND swapped in + // swapped-in value matches + int r2u = useStatsRemove2Used.incrementAndGet(); + if (r2u == 1 && LOG_ERROR_ON_DESIGN_VIOLATING_METHODS) { + logger.warning("design violating remove(,) used for dbName=" + + dbName); + } + // map mutating operations always change diskMap first + Object obj = diskMap.remove(key); + if (obj != null) { + diskMapSize.decrementAndGet(); + } + neutralizeMemMapEntry(entry, key); // also removes entry from memMap + return true; + } else { // case (*,F) AND swapped in + // if swapped in and value does not match, then + // diskMap is *not* checked since it's value is + // not necessarily up to date (when swapped in) + return false; + } + } + // already covered cases: swapped in + // cases: (*,F) AND swapped out + boolean diskMapFound = diskMap.remove(key, value); + if (diskMapFound) { // case (T,F) AND swapped out + int r2u = useStatsRemove2Used.incrementAndGet(); + if (r2u == 1 && LOG_ERROR_ON_DESIGN_VIOLATING_METHODS) { + logger.warning("design violating remove(,) used for dbName=" + + dbName); + } + diskMapSize.decrementAndGet(); + return true; + } else { + return false; // case (F,F) AND swapped out + } + } public synchronized boolean containsKey(Object key) { if (quickContainsKey(key)) { @@ -463,11 +1254,17 @@ implements Map, Serializable { return diskMap.containsKey(key); } - public synchronized boolean quickContainsKey(Object key) { - expungeStaleEntries(); + private boolean quickContainsKey(Object key) { + expungeStaleEntries(); // note: expunge is not quick return memMap.containsKey(key); } + /** + * This method is not supported. + * @deprecated + * @param value + * @return + */ public synchronized boolean containsValue(Object value) { if (quickContainsValue(value)) { return true; @@ -475,14 +1272,14 @@ implements Map, Serializable { return diskMap.containsValue(value); } - public synchronized boolean quickContainsValue(Object value) { - expungeStaleEntries(); + private boolean quickContainsValue(Object value) { + expungeStaleEntries(); // note: expunge is not quick // FIXME this isn't really right, as memMap is of SoftEntries return memMap.containsValue(value); } public int size() { - return diskMapSize; + return diskMapSize.get(); } protected String getDatabaseName() { @@ -502,34 +1299,38 @@ implements Map, Serializable { * When done, the memory map will be cleared and all entries stored * on disk. */ + @SuppressWarnings("unchecked") public synchronized void sync() { String dbName = null; // Sync. memory and disk. + useStatsSyncUsed.incrementAndGet(); + boolean expungerWasRunning = stopExpunger(); long startTime = 0; if (logger.isLoggable(Level.INFO)) { dbName = getDatabaseName(); startTime = System.currentTimeMillis(); - logger.info(dbName + " start sizes: disk " + this.diskMapSize + + logger.info(dbName + " start sizes: disk " + this.diskMapSize.get() + ", mem " + this.memMap.size()); } expungeStaleEntries(); - LinkedList stale = new LinkedList(); - for (Iterator i = this.memMap.keySet().iterator(); i.hasNext();) { - Object key = i.next(); - SoftEntry entry = (SoftEntry) memMap.get(key); + LinkedList> stale = new LinkedList>(); + for (Iterator i = this.memMap.keySet().iterator(); i.hasNext();) { + K key = i.next(); + SoftEntry entry = memMap.get(key); if (entry != null) { // Get & hold so not cleared pre-return. - Object value = entry.get(); + V value = entry.get(); if (value != null) { - this.diskMap.put(key, value); + expungeStatsDiskPut.incrementAndGet(); + this.diskMap.put(key, value); // unchecked cast } else { stale.add(entry); } } } // for any entries above that had been cleared, ensure expunged - for (SoftEntry entry : stale) { - expungeStaleEntry(entry); + for (SoftEntry entry : stale) { + expungeStaleEntry(entry, false); } // force sync of deferred-writes @@ -544,28 +1345,86 @@ implements Map, Serializable { logger.info(dbName + " sync took " + (System.currentTimeMillis() - startTime) + "ms. " + "Finish sizes: disk " + - this.diskMapSize + ", mem " + this.memMap.size()); + this.diskMapSize.get() + ", mem " + this.memMap.size()); + dumpExtraStats(); + } + if (expungerWasRunning) { + startExpunger(); + } + } + + /** log at INFO level, interesting stats if non-zero. */ + private void dumpExtraStats() { + if (logger.isLoggable(Level.INFO)) { + // emit if FINE or has interesting values + if (logger.isLoggable(Level.FINE) || + (expungeStatsNullPhantom.get() + + expungeStatsTransientCond.get() + + expungeStatsAwaitExpunge.get() + + expungeStatsNullValue.get() + + expungeStatsNotInMap.get() + + expungeStatsAwaitTimeout.get() + + expungeStatsSwapIn.get() + + expungeStatsSwapInRetry.get() + + useStatsRemove1Used.get() + + useStatsRemove2Used.get() + + useStatsReplaceUsed.get()) > 0) { + logger.info(composeCacheSummary() + " " + + expungeStatsNullPhantom + "=NullPhantom " + + expungeStatsTransientCond + "=TransientCond " + + expungeStatsAwaitExpunge + "=AwaitExpunge " + + expungeStatsNullValue + "=NullValue " + + expungeStatsNotInMap+ "=NotInMap " + + expungeStatsAwaitTimeout+ "=AwaitTimeout " + + expungeStatsSwapIn+ "=SwapIn " + + diskHit+ "=diskHit " + + cacheHit+ "=cacheHit " + + countOfGets+ "=countOfGets " + + expungeStatsSwapInRetry+ "=SwapInRetry " + + expungeStatsDiskPut + "=DiskPut " + + useStatsRemove1Used + "=Remove1Used " + + useStatsRemove2Used + "=Remove2Used " + + useStatsReplaceUsed + "=ReplaceUsed " + + useStatsPutUsed + "=PutUsed " + + useStatsPutIfUsed+ "=PutIfUsed " + + useStatsSyncUsed+ "=SyncUsed " + + diskMapSize.get()+ "=diskMapSize " + + memMap.size()+ "=memMap.size " + ); + } } } + /** An incremental, poll-based expunger. + * See #Expunger for dedicated thread expunger. + */ + @SuppressWarnings("unchecked") private void expungeStaleEntries() { int c = 0; - for(SoftEntry entry; (entry = refQueuePoll()) != null;) { - expungeStaleEntry(entry); + long startTime = System.currentTimeMillis(); + for(SoftEntry entry; (entry = (SoftEntry)refQueuePoll()) != null;) { + expungeStaleEntry(entry, false); c++; } if (c > 0 && logger.isLoggable(Level.FINER)) { + long endTime = System.currentTimeMillis(); try { logger.finer("DB: " + db.getDatabaseName() + ", Expunged: " - + c + ", Diskmap size: " + diskMapSize - + ", Cache size: " + memMap.size()); + + c + ", Diskmap size: " + diskMapSize.get() + + ", Cache size: " + memMap.size() + + ", in "+(endTime-startTime)+"ms"); } catch (DatabaseException e) { - // Just for logging so ignore Exceptions + logger.log(Level.FINER,"exception while logging",e); } } } - private void expungeStaleEntry(SoftEntry entry) { + /** Expunge an entry from cache (memMap) and put the key,value pair to diskMap. + * This is the only way entries are written to BDB, except for {@link #sync()} + * Possible disk io. + */ + @SuppressWarnings({ "unchecked", "unused" }) + private synchronized void expungeStaleEntry_synchronized(SoftEntry entry) { // If phantom already null, its already expunged -- probably // because it was purged directly first from inside in // {@link #get(String)} and then it went on the poll queue and @@ -576,15 +1435,106 @@ implements Map, Serializable { } // If the object that is in memMap is not the one passed here, then // memMap has been changed -- probably by a put on top of this entry. - if (memMap.get(entry.getPhantom().getKey()) == entry) { + if (memMap.get(entry.getPhantom().getKey()) == entry) { // NOTE: identity compare memMap.remove(entry.getPhantom().getKey()); + diskMap.put(entry.getPhantom().getKey(), entry.getPhantom().doctoredGet()); } entry.clearPhantom(); } - private class PhantomEntry extends PhantomReference { + /** Expunge an entry from memMap while updating diskMap. + * Concurrent implementation. + * + * This thread is the only thread performing an expunge of the given + * entry (if it or the calling thread sees a true SoftEntry.startExpunge()), + * so we can assume exclusive access to the given entry. + * + * 3. Swap out/flush: diskMap update, then memMap remove: + * diskMap.put(k,v2); // exclusive only if performed when processing ref queue + * memMap.remove(k,v2); if memMap race lost, then no harm done; + * 3.1. Only expunge does an update of an existing diskMap entry (except + * for sync()). + * Possible disk io as data is written to BDB map. + * 
+     * Preconditions: (diskMap.containsKey(K), memMap.containsKey(SoftEntry)) are:
+     * (F,F) remove() or replace was used.
+     * (F,T) transient while remove() or replace in progress.
+     * (T,F) normal condition, memMap value is most recent
+     * (T,T) normal condition, memMap value is cleared SoftReference and 
+     *     the attached PhantomEntry doctoredGet() has most recent value.
+     * 
+     * PostConditions:
+     * (T,F)  for K key, memMap will not have the given mapping and diskMap 
+     *     will have a mapping to the value held by the phantom value 
+     *     held by given entry, if entry was not already expunged. 
+     *
+ * The PostCondition only applies if exclusive expunge access is obtained. + * If another thread is already doing the expunge of given entry, this + * will not wait for that to finish. + * + * @param entry a SoftEntry obtained from refQueuePoll(). + * @param alreadyExclusive if true, then assume this thread already has + * SoftEntry.startExpunge()==true. + */ + @SuppressWarnings("unchecked") + private void expungeStaleEntry(SoftEntry entry, final boolean alreadyExclusive) { // concurrent version + // If phantom already null, its already expunged -- probably + // because it was purged directly first from inside in + // {@link #get(String)} and then it went on the poll queue and + // when it came off inside in expungeStaleEntries, this method + // was called again. + V phantomValue = null; + // keep this ref on stack so it cannot be nulled in mid-operation + PhantomEntry phantom = (PhantomEntry)entry.getPhantom(); // unchecked cast + if (phantom == null) { + expungeStatsNullPhantom.incrementAndGet(); + return; // nothing to do + } else { + // recover hidden value + phantomValue = (V)phantom.doctoredGet(); // unchecked cast + } + if (alreadyExclusive || entry.startExpunge()) { // exclusive expunge + try { + // Still in memMap? + // If the object that is in memMap is not the one passed here, then + // memMap has been changed -- probably by a put on top of this entry.] + if (memMap.get(phantom.getKey()) + == entry) { // NOTE: intentional identity compare + // given instance entry still in memMap; + // as long as CachedBDBMap.remove(), CachedBDBMap.put(), + // and CachedBDBMap.replace() are not used + // by application, then only this thread is doing the expunge. + + // we have the key and if we have a phantom Value, then + // the diskMap can be updated. + // diskMap MUST be updated *before* entry.clearPhantom(). + // Because this thread is the only one expunging the given entry.getKey() + // we can use diskMap.put(). + if (phantomValue != null) { + diskMap.put(phantom.getKey(), phantomValue); // unchecked cast + expungeStatsDiskPut.incrementAndGet(); + } else { + expungeStatsNullValue.incrementAndGet(); + } + // remove entry that matches key and value; + // since only this thread has the exclusive expunge job + // for entry, we will either remove the exact object + // or remove no object (non-interference). + memMap.remove(phantom.getKey(), entry); + // at this point: memMap no longer contains the given entry + } else { // not in map (a soft assert failure) + this.expungeStatsNotInMap.incrementAndGet(); + } + } finally { // an exclusive expunge must clear entry + // but only after diskMap and memMap are updated + entry.clearPhantom(); + } + } + } + + private static class PhantomEntry extends PhantomReference { private final Object key; public PhantomEntry(Object key, T referent) { @@ -598,7 +1548,7 @@ implements Map, Serializable { * PhantomReference to return the actual referent value. See notes * at {@link #referentField}; */ - public Object doctoredGet() { + final public Object doctoredGet() { try { // Here we use the referentField saved off on static // initialization of this class to get at this References' @@ -612,41 +1562,242 @@ implements Map, Serializable { /** * @return Returns the key. */ - public Object getKey() { + final public Object getKey() { return this.key; } + + /** + * @param obj + * @return true if the referents are equals(). + */ + @Override + @SuppressWarnings("unchecked") + final public boolean equals(Object obj) { + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + final PhantomEntry other = (PhantomEntry) obj; // unchecked cast + if (this.doctoredGet() != other.doctoredGet() + && (this.doctoredGet() == null + || !this.doctoredGet().equals(other.doctoredGet()))) { + return false; + } + return true; + } + + @Override + final public int hashCode() { + return this.key != null ? this.key.hashCode() : 0; + } + } - private class SoftEntry extends SoftReference { - private PhantomEntry phantom; + /** + * SoftReference cache entry. + * "Expunge" is the process of flushing entries in the cache (memMap) + * to diskMap after GC determines they are no longer references. + * A PhantomReference is used to hold the key and value as a last + * chance before GC hook that can effect the update of diskMap. + *

+ * The coordinated process to do expunge is: + * 

+     *     SoftEntry se = map.get();
+     *     if (se.startExpunge()) {
+     *         ;// update diskMap here
+     *         map.remove(); // remove SoftEntry after diskMap update
+     *         se.clearPhantom(); // clear after remove mapping
+     *     } else { // another thread does expunge
+     *     }
+     *
+ * The coordinated process to wait for another thread to do expunge: + * 
+     *     SoftEntry se = map.get();
+     *     if (se.awaitExpunge()) {
+     *         ;// wait occurred (if you care)
+     *     } 
+     *     ;// postcondition: se was removed from map
+     *
+ * Entries are not recycled. + */ + private static class SoftEntry extends SoftReference { + private PhantomEntry phantom; + /** count down to expunged; if null, then expunge has not + * been started. See {@link #startExpunge()} and + * {@link #expungeStarted()} + */ + private CountDownLatch expunged = null; + + @SuppressWarnings("unchecked") public SoftEntry(Object key, T referent, ReferenceQueue q) { super(referent, q); - this.phantom = new PhantomEntry(key, referent); + this.phantom = new PhantomEntry(key, referent); // unchecked cast + } + + public T get() { + // ensure visibility + synchronized (this) { + return super.get(); + } } /** + * Synchronized so this can be atomic w.r.t. + * {@link #clearPhantom()}, {@link #expungeStarted()}, and + * {@link #startExpunge()}. * @return Returns the phantom reference. */ - public PhantomEntry getPhantom() { + final public synchronized PhantomEntry getPhantom() { return this.phantom; } - public void clearPhantom() { + /** + * @return true if this entry is at end of lifecycle. + */ + final public boolean isCleared() { + return getPhantom() == null; + } + + /** clear referent and end-of-life-cycle this instance. + * Do not clearPhantom() until after diskMap updated and after + * this entry is removed from memMap (because threads waiting + * for expunge to finish will likely do a memMap.get() + * or putIfAbsent() and + * there should be no residue so a fresh entry can be created + * when the value is swapped in from diskMap. + *

Causes any threads that called isExpunged() on this + * instance to continue. + * + * + *

Idempotent. + * + * After this method returns, + * {@link #clearPhantom()} will return true. + * Synchronized so this can be atomic w.r.t. + * {@link #getPhantom()}, {@link #expungeStarted()}, and + * {@link #startExpunge()}. */ + final public synchronized void clearPhantom() { + if (this.phantom != null) { this.phantom.clear(); this.phantom = null; + } super.clear(); + if (expungeStarted()) { + expunged.countDown(); } } - private void readObject(java.io.ObjectInputStream stream) - throws IOException, ClassNotFoundException { - stream.defaultReadObject(); - initializeInstance(); - if (logger.isLoggable(Level.FINE)) { - logger.fine(getDatabaseName() + " diskMapSize: " + diskMapSize); + /** + * Synchronized so this can be atomic w.r.t. + * {@link #clearPhantom()}, {@link #getPhantom()}, and + * {@link #startExpunge()}. + * No side-effects. + * @return true if expunge was ever started on this entry, + * expunge may or may not be finished and #isCleared() + * may or may not be true, + * see {@link #awaitExpunge()} + * @see {@link CachedBdbMap#isExpunged(SoftEntry)} + */ + final public synchronized boolean expungeStarted() { + if (expunged != null) return true; + else return false; + } + + /** Can the calling thread perform the expunge on this? + * Side Effect: if true is returned, sets up latch for any + * threads that want to await for expunge. + * + * Synchronized so this can be atomic w.r.t. + * {@link #clearPhantom()}, {@link #expungeStarted()}, and + * {@link #getPhantom()}. + * @return true if calling thread has exclusive access to + * expunge this instance; do not call this unless + * going to do the expunge; true is only returned + * once. + */ + final public synchronized boolean startExpunge() { + if (isCleared() || expungeStarted()) { + return false; + } else { + expunged = new CountDownLatch(1); + return true; + } + } + + /** If this entry is in the process of being expunged + * (when {@link #expungeStarted()} is true), then this will wait + * until the expunge is finished. If expunge was not started + * or was finished, this method returns immediately. + * Just in case, 1sec timeout is used. + * @throws java.lang.InterruptedException + * @return true if expunge is completed (or timeout) or this + * entry was never used, false if expunge is not happening. + */ + final public boolean awaitExpunge() throws InterruptedException { + if (isCleared()) { + return true; // expunge completed, or entry was unused + } + if (expungeStarted()) { + long longerThanFullGCsec = 60L; + if (expunged.await(longerThanFullGCsec, TimeUnit.SECONDS)) { + return true; // finished + } else { // timeout + // just in case + expungeStatsAwaitTimeout.incrementAndGet(); + Logger.getLogger(CachedBdbMap.class.getName()) + .warning("timeout at " + + longerThanFullGCsec + " sec"); + throw new InterruptedException("timeout"); + } + } else { + return false; // not started + } + } + + /** + * + * @param obj + * @return true if the references are equal, and if references differ, + * if phantoms are == or equals(), return true. + */ + @Override + @SuppressWarnings("unchecked") + public boolean equals(Object obj) { + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + if (this == obj) { // same reference + return true; + } + final SoftEntry other = (SoftEntry) obj; // unchecked cast + if (this.phantom != other.phantom + && (this.phantom == null + || !this.phantom.equals(other.phantom))) { + return false; + } + return true; + } + + @Override + public int hashCode() { + return this.phantom != null ? this.phantom.hashCode() : 0; + } + + public String toString() { + if (phantom != null) { + return "SoftEntry(key=" + phantom.getKey().toString() + ")"; + } else { + return "SoftEntry()"; + } } } + @@ -661,7 +1812,7 @@ implements Map, Serializable { } @SuppressWarnings("unchecked") - private SoftEntry refQueuePoll() { - return (SoftEntry)refQueue.poll(); + private SoftEntry refQueuePoll() { + return (SoftEntry)refQueue.poll(); } } diff --git a/commons/src/main/java/org/archive/util/LongWrapper.java b/commons/src/main/java/org/archive/util/LongWrapper.java deleted file mode 100644 index b91b69b9..00000000 --- a/commons/src/main/java/org/archive/util/LongWrapper.java +++ /dev/null @@ -1,45 +0,0 @@ -/* LongWrapper - * - * $Id$ - * - * Copyright (C) 2003 Internet Archive. - * - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Heritrix is free software; you can redistribute it and/or modify - * it under the terms of the GNU Lesser Public License as published by - * the Free Software Foundation; either version 2.1 of the License, or - * any later version. - * - * Heritrix is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser Public License for more details. - * - * You should have received a copy of the GNU Lesser Public License - * along with Heritrix; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - */ -package org.archive.util; - -import java.io.Serializable; - -/** - * Wraps a long. Used in place of Long so that when we extract it from a - * Collection we can modify the long value without creating a new object. - * This way we don't have to rewrite the Collection to update one of the - * stored longs. - * @author Kristinn Sigurdsson - */ -public class LongWrapper implements Serializable { - - private static final long serialVersionUID = -6537350490019555280L; - - public long longValue; - public LongWrapper(long initial){ - this.longValue = initial; - } - public long getLongValue() { - return longValue; - } -} diff --git a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java index 3ebc6013..bbe1d910 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java @@ -35,13 +35,13 @@ import java.lang.ref.SoftReference; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; import java.util.Iterator; import java.util.Map; import java.util.Queue; import java.util.SortedMap; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.DelayQueue; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; @@ -216,7 +216,7 @@ ApplicationContextAware { /** All known queues. */ - protected Map allQueues = null; + protected ConcurrentMap allQueues = null; // of classKey -> ClassKeyQueue /** @@ -287,8 +287,8 @@ ApplicationContextAware { && getQueueAssignmentPolicy().maximumNumberOfKeys() >= 0 && getQueueAssignmentPolicy().maximumNumberOfKeys() <= MAX_QUEUES_TO_HOLD_ALLQUEUES_IN_MEMORY) { - this.allQueues = Collections.synchronizedMap( - new HashMap()); + this.allQueues = + new ConcurrentHashMap(701, .9f, 100); } else { this.initAllQueues(); } diff --git a/engine/src/main/java/org/archive/crawler/reporting/HostsReport.java b/engine/src/main/java/org/archive/crawler/reporting/HostsReport.java index 9cd1f70b..2e55b4f9 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/HostsReport.java +++ b/engine/src/main/java/org/archive/crawler/reporting/HostsReport.java @@ -22,10 +22,10 @@ package org.archive.crawler.reporting; import java.io.PrintWriter; import java.util.Iterator; import java.util.SortedMap; +import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.collections.Closure; import org.archive.modules.net.CrawlHost; -import org.archive.util.LongWrapper; /** * The "Hosts Report", tallies by host. @@ -38,16 +38,16 @@ public class HostsReport extends Report { public void write(final PrintWriter writer) { // TODO: use CrawlHosts for all stats; only perform sorting on // manageable number of hosts - SortedMap hd = stats.getReverseSortedHostsDistribution(); + SortedMap hd = stats.getReverseSortedHostsDistribution(); // header writer.print("[#urls] [#bytes] [host] [#robots] [#remaining]\n"); for (Iterator i = hd.keySet().iterator(); i.hasNext();) { // Key is 'host'. String key = (String) i.next(); CrawlHost host = stats.serverCache.getHostFor(key); - LongWrapper val = (LongWrapper)hd.get(key); + AtomicLong val = (AtomicLong)hd.get(key); writeReportLine(writer, - ((val==null)?"-":val.longValue), + ((val==null)?"-":val.get()), stats.getBytesPerHost(key), key, host.getSubstats().getRobotsDenials(), diff --git a/engine/src/main/java/org/archive/crawler/reporting/MimetypesReport.java b/engine/src/main/java/org/archive/crawler/reporting/MimetypesReport.java index 80068a00..45d6ffc3 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/MimetypesReport.java +++ b/engine/src/main/java/org/archive/crawler/reporting/MimetypesReport.java @@ -22,8 +22,7 @@ package org.archive.crawler.reporting; import java.io.PrintWriter; import java.util.Iterator; import java.util.TreeMap; - -import org.archive.util.LongWrapper; +import java.util.concurrent.atomic.AtomicLong; /** * The "Mimetypes Report", tallies by MIME type. @@ -36,11 +35,11 @@ public class MimetypesReport extends Report { public void write(PrintWriter writer) { // header writer.print("[#urls] [#bytes] [mime-types]\n"); - TreeMap fd = stats.getReverseSortedCopy(stats.getFileDistribution()); + TreeMap fd = stats.getReverseSortedCopy(stats.getFileDistribution()); for (Iterator i = fd.keySet().iterator(); i.hasNext();) { Object key = i.next(); // Key is mime type. - writer.print(Long.toString(((LongWrapper)fd.get(key)).longValue)); + writer.print(Long.toString(((AtomicLong)fd.get(key)).get())); writer.print(" "); writer.print(Long.toString(stats.getBytesPerFileType((String)key))); writer.print(" "); diff --git a/engine/src/main/java/org/archive/crawler/reporting/ResponseCodeReport.java b/engine/src/main/java/org/archive/crawler/reporting/ResponseCodeReport.java index fbd15322..78c4907e 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/ResponseCodeReport.java +++ b/engine/src/main/java/org/archive/crawler/reporting/ResponseCodeReport.java @@ -21,8 +21,7 @@ package org.archive.crawler.reporting; import java.io.PrintWriter; import java.util.Iterator; import java.util.TreeMap; - -import org.archive.util.LongWrapper; +import java.util.concurrent.atomic.AtomicLong; /** * The "Response Codes Report", tallies by response/disposition code. @@ -36,13 +35,13 @@ public class ResponseCodeReport extends Report { // header writer.print("[rescode] [#urls]\n"); - TreeMap scd = + TreeMap scd = stats.getReverseSortedCopy(stats.getStatusCodeDistribution()); for (Iterator i = scd.keySet().iterator(); i.hasNext();) { Object key = i.next(); writer.print((String)key); writer.print(" "); - writer.print(Long.toString(((LongWrapper)scd.get(key)).longValue)); + writer.print(Long.toString(((AtomicLong)scd.get(key)).get())); writer.print("\n"); } } diff --git a/engine/src/main/java/org/archive/crawler/reporting/SourceTagsReport.java b/engine/src/main/java/org/archive/crawler/reporting/SourceTagsReport.java index 01d7c4b0..ee3f81bf 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/SourceTagsReport.java +++ b/engine/src/main/java/org/archive/crawler/reporting/SourceTagsReport.java @@ -22,8 +22,7 @@ import java.io.PrintWriter; import java.util.Iterator; import java.util.Map; import java.util.SortedMap; - -import org.archive.util.LongWrapper; +import java.util.concurrent.atomic.AtomicLong; /** * The "Source Report", tallies of source tags (usually seeds) by host. @@ -39,20 +38,20 @@ public class SourceTagsReport extends Report { // for each source for (Iterator i = stats.sourceHostDistribution.keySet().iterator(); i.hasNext();) { String sourceKey = i.next(); - Map hostCounts = - (Map)stats.sourceHostDistribution.get(sourceKey); + Map hostCounts = + (Map)stats.sourceHostDistribution.get(sourceKey); // sort hosts by #urls - SortedMap sortedHostCounts = + SortedMap sortedHostCounts = stats.getReverseSortedHostCounts(hostCounts); // for each host for (Iterator j = sortedHostCounts.keySet().iterator(); j.hasNext();) { Object hostKey = j.next(); - LongWrapper hostCount = (LongWrapper) hostCounts.get(hostKey); + AtomicLong hostCount = (AtomicLong) hostCounts.get(hostKey); writer.print(sourceKey.toString()); writer.print(" "); writer.print(hostKey.toString()); writer.print(" "); - writer.print(hostCount.longValue); + writer.print(hostCount.get()); writer.print("\n"); } } diff --git a/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java b/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java index 6d3515ad..2d73a84a 100644 --- a/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java +++ b/engine/src/main/java/org/archive/crawler/reporting/StatisticsTracker.java @@ -26,11 +26,8 @@ import java.io.FileWriter; import java.io.IOException; import java.io.PrintWriter; import java.io.Serializable; -import java.util.Collections; import java.util.Comparator; import java.util.Date; -import java.util.HashMap; -import java.util.Hashtable; import java.util.Iterator; import java.util.LinkedList; import java.util.List; @@ -39,9 +36,12 @@ import java.util.SortedMap; import java.util.TreeMap; import java.util.TreeSet; import java.util.Vector; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.logging.Level; import java.util.logging.Logger; @@ -60,7 +60,6 @@ import org.archive.modules.seeds.SeedModule; import org.archive.net.UURI; import org.archive.spring.ConfigPath; import org.archive.util.ArchiveUtils; -import org.archive.util.LongWrapper; import org.archive.util.MimetypeUtils; import org.archive.util.PaddingStringBuffer; import org.springframework.beans.BeansException; @@ -286,28 +285,29 @@ public class StatisticsTracker // TODO: fortify these against key explosion with bigmaps like other tallies /** Keep track of the file types we see (mime type -> count) */ - protected Hashtable mimeTypeDistribution - = new Hashtable(); - protected Hashtable mimeTypeBytes - = new Hashtable(); + protected ConcurrentMap mimeTypeDistribution + = new ConcurrentHashMap(); + protected ConcurrentMap mimeTypeBytes + = new ConcurrentHashMap(); /** Keep track of fetch status codes */ - protected Hashtable statusCodeDistribution - = new Hashtable(); + protected ConcurrentMap statusCodeDistribution + = new ConcurrentHashMap(); /** Keep track of hosts. * - * Each of these Maps are individually unsynchronized, and cannot - * be trivially synchronized with the Collections wrapper. Thus - * their synchronized access is enforced by this class. */ - protected Map hostsDistribution = Collections.emptyMap(); // temp dummy - protected Map hostsBytes = Collections.emptyMap(); // temp dummy - protected Map hostsLastFinished = Collections.emptyMap(); // temp dummy + protected ConcurrentMap hostsDistribution = + new ConcurrentHashMap(); // temp dummy + protected ConcurrentMap hostsBytes = + new ConcurrentHashMap(); // temp dummy + protected ConcurrentMap hostsLastFinished = + new ConcurrentHashMap(); // temp dummy /** Keep track of URL counts per host per seed */ protected - Map> sourceHostDistribution = Collections.emptyMap(); // temp dummy; + ConcurrentMap> sourceHostDistribution = + new ConcurrentHashMap>(); // temp dummy; /* Keep track of 'top' hosts for live reports */ protected TopNSet hostsDistributionTop; @@ -317,7 +317,8 @@ public class StatisticsTracker /** * Record of seeds' latest actions. */ - protected Map processedSeedsRecords = Collections.emptyMap(); + protected ConcurrentMap processedSeedsRecords = + new ConcurrentHashMap(); long seedsTotal = -1; long seedsCrawled = -1; @@ -339,11 +340,11 @@ public class StatisticsTracker isRunning = true; try { this.sourceHostDistribution = bdb.getBigMap("sourceHostDistribution", - false, String.class, HashMap.class); + false, String.class, ConcurrentMap.class); this.hostsDistribution = bdb.getBigMap("hostsDistribution", - false, String.class, LongWrapper.class); + false, String.class, AtomicLong.class); this.hostsBytes = bdb.getBigMap("hostsBytes", false, String.class, - LongWrapper.class); + AtomicLong.class); this.hostsLastFinished = bdb.getBigMap("hostsLastFinished", false, String.class, Long.class); this.processedSeedsRecords = bdb.getBigMap("processedSeedsRecords", @@ -517,10 +518,10 @@ public class StatisticsTracker * encountered mime types. Key/value pairs represent * mime type -> count. *

- * Note: All the values are wrapped with a {@link LongWrapper LongWrapper} + * Note: All the values are wrapped with a {@link AtomicLong AtomicLong} * @return mimeTypeDistribution */ - public Hashtable getFileDistribution() { + public ConcurrentMap getFileDistribution() { return mimeTypeDistribution; } @@ -529,16 +530,12 @@ public class StatisticsTracker * Increment a counter for a key in a given HashMap. Used for various * aggregate data. * - * As this is used to change Maps which depend on StatisticsTracker - * for their synchronization, this method should only be invoked - * from a a block synchronized on 'this'. - * - * @param map The HashMap + * @param map The Map or ConcurrentMap * @param key The key for the counter to be incremented, if it does not * exist it will be added (set to 1). If null it will * increment the counter "unknown". */ - protected static void incrementMapCount(Map map, + protected static void incrementMapCount(ConcurrentMap map, String key) { incrementMapCount(map,key,1); } @@ -547,9 +544,6 @@ public class StatisticsTracker * Increment a counter for a key in a given HashMap by an arbitrary amount. * Used for various aggregate data. The increment amount can be negative. * - * As this is used to change Maps which depend on StatisticsTracker - * for their synchronization, this method should only be invoked - * from a a block synchronized on 'this'. * * @param map * The HashMap @@ -560,29 +554,25 @@ public class StatisticsTracker * @param increment * The amount to increment counter related to the key. */ - protected static void incrementMapCount(Map map, + protected static void incrementMapCount(ConcurrentMap map, String key, long increment) { if (key == null) { key = "unknown"; } - Object o = map.get(key); - if (o == null) { - // Considered normal - map.put(key, new LongWrapper(increment)); - } else if (o instanceof LongWrapper) { - LongWrapper lw = (LongWrapper)o; - lw.longValue += increment; - } else { - // Abnormal - logger.severe("Resetting " + key + ": Expected LongWrapper but got " - + o.getClass().getName()); - map.put(key, new LongWrapper(increment)); - } + AtomicLong lw = (AtomicLong)map.get(key); + if(lw == null) { + lw = new AtomicLong(); + AtomicLong prevVal = map.putIfAbsent(key, lw); + if(prevVal != null) { + lw = prevVal; + } + } + lw.addAndGet(increment); } /** * Sort the entries of the given HashMap in descending order by their - * values, which must be longs wrapped with LongWrapper. + * values, which must be longs wrapped with AtomicLong. *

* Elements are sorted by value from largest to smallest. Equal values are * sorted in an arbitrary, but consistent manner by their keys. Only items @@ -591,19 +581,17 @@ public class StatisticsTracker * If the passed-in map requires access to be synchronized, the caller * should ensure this synchronization. * - * @param mapOfLongWrapperValues - * Assumes values are wrapped with LongWrapper. + * @param mapOfAtomicLongValues + * Assumes values are wrapped with AtomicLong. * @return a sorted set containing the same elements as the map. */ - public TreeMap getReverseSortedCopy( - final Map mapOfLongWrapperValues) { - TreeMap sortedMap = - new TreeMap(new Comparator() { + public TreeMap getReverseSortedCopy( + final Map mapOfAtomicLongValues) { + TreeMap sortedMap = + new TreeMap(new Comparator() { public int compare(String e1, String e2) { - long firstVal = mapOfLongWrapperValues.get(e1). - longValue; - long secondVal = mapOfLongWrapperValues.get(e2). - longValue; + long firstVal = mapOfAtomicLongValues.get(e1).get(); + long secondVal = mapOfAtomicLongValues.get(e2).get(); if (firstVal < secondVal) { return 1; } @@ -615,13 +603,13 @@ public class StatisticsTracker } }); try { - sortedMap.putAll(mapOfLongWrapperValues); + sortedMap.putAll(mapOfAtomicLongValues); } catch (UnsupportedOperationException e) { - Iterator i = mapOfLongWrapperValues.keySet().iterator(); + Iterator i = mapOfAtomicLongValues.keySet().iterator(); for (;i.hasNext();) { // Ok. Try doing it the slow way then. String key = i.next(); - sortedMap.put(key, mapOfLongWrapperValues.get(key)); + sortedMap.put(key, mapOfAtomicLongValues.get(key)); } } return sortedMap; @@ -633,11 +621,11 @@ public class StatisticsTracker * val represents (string)code -> (integer)count. * * Note: All the values are wrapped with a - * {@link LongWrapper LongWrapper} + * {@link AtomicLong AtomicLong} * * @return statusCodeDistribution */ - public Hashtable getStatusCodeDistribution() { + public ConcurrentMap getStatusCodeDistribution() { return statusCodeDistribution; } @@ -651,10 +639,7 @@ public class StatisticsTracker * -1 will be returned. */ public long getHostLastFinished(String host){ - Long l = null; - synchronized(hostsLastFinished){ - l = (Long)hostsLastFinished.get(host); - } + Long l = (Long)hostsLastFinished.get(host); return (l != null)? l.longValue(): -1; } @@ -664,9 +649,7 @@ public class StatisticsTracker * @return the accumulated number of bytes downloaded from a given host */ public long getBytesPerHost(String host){ - synchronized(hostsBytes){ - return getReportValue(hostsBytes, host); - } + return getReportValue(hostsBytes, host); } /** @@ -733,32 +716,32 @@ public class StatisticsTracker protected void saveSourceStats(String source, String hostname) { synchronized(sourceHostDistribution) { - HashMap hostUriCount = + ConcurrentMap hostUriCount = sourceHostDistribution.get(source); if (hostUriCount == null) { - hostUriCount = new HashMap(); - sourceHostDistribution.put(source, hostUriCount); + hostUriCount = new ConcurrentHashMap(); + ConcurrentMap prevVal = + sourceHostDistribution.putIfAbsent(source, hostUriCount); + if(prevVal != null) { + hostUriCount = prevVal; + } } incrementMapCount(hostUriCount, hostname); } } protected void saveHostStats(String hostname, long size) { - synchronized(hostsDistribution){ - incrementMapCount(hostsDistribution, hostname); - hostsDistributionTop.update( - hostname, getReportValue(hostsDistribution, hostname)); - } - synchronized(hostsBytes){ - incrementMapCount(hostsBytes, hostname, size); - hostsBytesTop.update(hostname, - getReportValue(hostsBytes, hostname)); - } - synchronized(hostsLastFinished){ - long time = new Long(System.currentTimeMillis()); - hostsLastFinished.put(hostname, time); - hostsLastFinishedTop.update(hostname, time); - } + incrementMapCount(hostsDistribution, hostname); + hostsDistributionTop.update( + hostname, getReportValue(hostsDistribution, hostname)); + + incrementMapCount(hostsBytes, hostname, size); + hostsBytesTop.update(hostname, + getReportValue(hostsBytes, hostname)); + + long time = new Long(System.currentTimeMillis()); + hostsLastFinished.put(hostname, time); + hostsLastFinishedTop.update(hostname, time); } public void crawledURINeedRetry(CrawlURI curi) { @@ -821,7 +804,7 @@ public class StatisticsTracker SeedRecord sr = (SeedRecord) processedSeedsRecords.get(seed); if(sr==null) { sr = new SeedRecord(seed,"Seed has not been processed"); -// processedSeedsRecords.put(seed,sr); + // no need to retain synthesized record } sortedSet.add(sr); } @@ -834,8 +817,8 @@ public class StatisticsTracker * * @return SortedMap of hosts distribution */ - public SortedMap getReverseSortedHostCounts( - Map hostCounts) { + public SortedMap getReverseSortedHostCounts( + Map hostCounts) { synchronized(hostCounts){ return getReverseSortedCopy(hostCounts); } @@ -846,7 +829,7 @@ public class StatisticsTracker * (largest first) order. * @return SortedMap of hosts distribution */ - public SortedMap getReverseSortedHostsDistribution() { + public SortedMap getReverseSortedHostsDistribution() { synchronized(hostsDistribution){ return getReverseSortedCopy(hostsDistribution); } @@ -916,7 +899,7 @@ public class StatisticsTracker logNote("CRAWL CHECKPOINTING TO " + cpDir.toString()); } - private long getReportValue(Map map, String key) { + private long getReportValue(Map map, String key) { if (key == null) { return -1; } @@ -924,11 +907,11 @@ public class StatisticsTracker if (o == null) { return -2; } - if (!(o instanceof LongWrapper)) { - throw new IllegalStateException("Expected LongWrapper but got " + if (!(o instanceof AtomicLong)) { + throw new IllegalStateException("Expected AtomicLong but got " + o.getClass() + " for " + key); } - return ((LongWrapper)o).longValue; + return ((AtomicLong)o).get(); } public void onApplicationEvent(ApplicationEvent event) { diff --git a/engine/src/main/java/org/archive/crawler/util/TopNSet.java b/engine/src/main/java/org/archive/crawler/util/TopNSet.java index 91c595de..01040298 100644 --- a/engine/src/main/java/org/archive/crawler/util/TopNSet.java +++ b/engine/src/main/java/org/archive/crawler/util/TopNSet.java @@ -43,7 +43,7 @@ public class TopNSet implements Serializable { set = new HashMap(size); } - public void update(String key, long value){ + public synchronized void update(String key, long value){ if(set.containsKey(key)) { // Update the value of an existing key set.put(key,value); diff --git a/modules/src/main/java/org/archive/modules/fetcher/DefaultServerCache.java b/modules/src/main/java/org/archive/modules/fetcher/DefaultServerCache.java index 831ae265..57036671 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/DefaultServerCache.java +++ b/modules/src/main/java/org/archive/modules/fetcher/DefaultServerCache.java @@ -1,32 +1,28 @@ -/* ServerCache - * - * Created on Nov 19, 2004 +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). * - * Copyright (C) 2004 Internet Archive. - * - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Heritrix is free software; you can redistribute it and/or modify - * it under the terms of the GNU Lesser Public License as published by - * the Free Software Foundation; either version 2.1 of the License, or - * any later version. - * - * Heritrix is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser Public License for more details. - * - * You should have received a copy of the GNU Lesser Public License - * along with Heritrix; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. + * + * The IA 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.archive.modules.fetcher; import java.io.Closeable; import java.io.Serializable; -import java.util.Map; -import java.util.Hashtable; -import java.util.logging.Level; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import java.util.logging.Logger; import org.apache.commons.collections.Closure; @@ -43,11 +39,8 @@ import org.archive.net.UURI; * @version $Date$, $Revision$ */ public class DefaultServerCache implements ServerCache, Closeable, Serializable { - - private static final long serialVersionUID = 1L; - private static Logger logger = Logger.getLogger(DefaultServerCache.class.getName()); @@ -56,25 +49,27 @@ public class DefaultServerCache implements ServerCache, Closeable, Serializable * hostname[:port] -> CrawlServer. * Set in the initialization. */ - protected Map servers = null; + protected ConcurrentMap servers = null; /** * hostname -> CrawlHost. * Set in the initialization. */ - protected Map hosts = null; + protected ConcurrentMap hosts = null; /** * Constructor. */ public DefaultServerCache() { - this(new Hashtable(), new Hashtable()); + this( + new ConcurrentHashMap(), + new ConcurrentHashMap()); } - public DefaultServerCache(Map servers, - Map hosts) { + public DefaultServerCache(ConcurrentMap servers, + ConcurrentMap hosts) { this.servers = servers; this.hosts = hosts; } @@ -85,21 +80,14 @@ public class DefaultServerCache implements ServerCache, Closeable, Serializable * @return CrawlServer instance that matches the passed server name. */ public synchronized CrawlServer getServerFor(String serverKey) { - CrawlServer cserver = (CrawlServer)this.servers.get(serverKey); - return (cserver != null)? cserver: createServerFor(serverKey); - } - - protected CrawlServer createServerFor(String s) { - CrawlServer cserver = (CrawlServer)this.servers.get(s); - if (cserver != null) { - return cserver; - } - // Ensure key is private object - String skey = new String(s); - cserver = new CrawlServer(skey); - servers.put(skey,cserver); - if (logger.isLoggable(Level.FINER)) { - logger.finer("Created server " + s); + CrawlServer cserver = servers.get(serverKey); + if(cserver==null) { + String skey = new String(serverKey); // ensure private minimal key + cserver = new CrawlServer(skey); + CrawlServer prevVal = servers.putIfAbsent(skey, cserver); + if(prevVal!=null) { + cserver = prevVal; + } } return cserver; } @@ -137,23 +125,14 @@ public class DefaultServerCache implements ServerCache, Closeable, Serializable if (hostname == null || hostname.length() == 0) { return null; } - CrawlHost host = (CrawlHost)this.hosts.get(hostname); - return (host != null)? host: createHostFor(hostname); - } - - protected CrawlHost createHostFor(String hostname) { - if (hostname == null || hostname.length() == 0) { - return null; - } - CrawlHost host = (CrawlHost)this.hosts.get(hostname); - if (host != null) { - return host; - } - String hkey = new String(hostname); - host = new CrawlHost(hkey); - this.hosts.put(hkey, host); - if (logger.isLoggable(Level.FINE)) { - logger.fine("Created host " + hostname); + CrawlHost host = hosts.get(hostname); + if(host == null) { + String hkey = new String(hostname); // ensure private minimal key + host = new CrawlHost(hkey); + CrawlHost prevVal = hosts.putIfAbsent(hkey, host); + if(prevVal!=null) { + host = prevVal; + } } return host; } diff --git a/modules/src/main/java/org/archive/modules/net/CrawlHost.java b/modules/src/main/java/org/archive/modules/net/CrawlHost.java index 4789929f..4c96299e 100644 --- a/modules/src/main/java/org/archive/modules/net/CrawlHost.java +++ b/modules/src/main/java/org/archive/modules/net/CrawlHost.java @@ -1,26 +1,22 @@ -/* Copyright (C) 2003 Internet Archive. +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). * - * This file is part of the Heritrix web crawler (crawler.archive.org). + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. * - * Heritrix is free software; you can redistribute it and/or modify - * it under the terms of the GNU Lesser Public License as published by - * the Free Software Foundation; either version 2.1 of the License, or - * any later version. + * The IA 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 * - * Heritrix is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser Public License for more details. + * http://www.apache.org/licenses/LICENSE-2.0 * - * You should have received a copy of the GNU Lesser Public License - * along with Heritrix; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA - * - * CrawlHost.java - * Created on Aug 5, 2003 - * - * $Header$ + * 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.archive.modules.net; import java.io.Serializable; @@ -153,6 +149,28 @@ public class CrawlHost implements Serializable, FetchStats.HasFetchStats { return "CrawlHost<" + hostname + "(ip:" + ip + ")>"; } + @Override + public int hashCode() { + return this.hostname != null ? this.hostname.hashCode() : 0; + } + + @Override + public boolean equals(Object obj) { + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + final CrawlHost other = (CrawlHost) obj; + if (this.hostname != other.hostname // identity compare + && (this.hostname == null + || !this.hostname.equals(other.hostname))) { + return false; + } + return true; + } + /** * Get the host name. * @return Returns the host name. diff --git a/modules/src/main/java/org/archive/modules/net/CrawlServer.java b/modules/src/main/java/org/archive/modules/net/CrawlServer.java index 2890d110..b706833f 100644 --- a/modules/src/main/java/org/archive/modules/net/CrawlServer.java +++ b/modules/src/main/java/org/archive/modules/net/CrawlServer.java @@ -117,7 +117,28 @@ public class CrawlServer implements Serializable, FetchStats.HasFetchStats { public String toString() { return "CrawlServer("+server+")"; } + @Override + public int hashCode() { + return this.server != null ? this.server.hashCode() : 0; + } + @Override + public boolean equals(Object obj) { + if (obj == null) { + return false; + } + if (getClass() != obj.getClass()) { + return false; + } + final CrawlServer other = (CrawlServer) obj; + if (this.server != other.server // identity compare + && (this.server == null + || !this.server.equals(other.server))) { + return false; + } + return true; + } + /** Update the robots exclusion policy. * * @param curi the crawl URI containing the fetched robots.txt diff --git a/modules/src/main/java/org/archive/modules/net/ServerCache.java b/modules/src/main/java/org/archive/modules/net/ServerCache.java index 1a7cd0a5..4efb50d5 100644 --- a/modules/src/main/java/org/archive/modules/net/ServerCache.java +++ b/modules/src/main/java/org/archive/modules/net/ServerCache.java @@ -1,27 +1,22 @@ -/* CrawlHostCache.java +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). * - * $Id$ + * Licensed to the Internet Archive (IA) by one or more individual + * contributors. * - * Created on Nov 21, 2006 + * The IA 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 * - * Copyright (C) 2006 Internet Archive. + * http://www.apache.org/licenses/LICENSE-2.0 * - * This file is part of the Heritrix web crawler (crawler.archive.org). - * - * Heritrix is free software; you can redistribute it and/or modify - * it under the terms of the GNU Lesser Public License as published by - * the Free Software Foundation; either version 2.1 of the License, or - * any later version. - * - * Heritrix is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU Lesser Public License for more details. - * - * You should have received a copy of the GNU Lesser Public License - * along with Heritrix; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA + * 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.archive.modules.net; import org.apache.commons.collections.Closure;