diff --git a/commons/src/main/java/org/archive/bdb/BdbModule.java b/commons/src/main/java/org/archive/bdb/BdbModule.java index 25747001..48f9fe0c 100644 --- a/commons/src/main/java/org/archive/bdb/BdbModule.java +++ b/commons/src/main/java/org/archive/bdb/BdbModule.java @@ -42,7 +42,8 @@ import org.apache.commons.io.filefilter.IOFileFilter; import org.archive.checkpointing.Checkpoint; import org.archive.checkpointing.Checkpointable; import org.archive.spring.ConfigPath; -import org.archive.util.ObjectIdentityBdbCache; +import org.archive.util.IdentityCacheable; +import org.archive.util.ObjectIdentityBdbManualCache; import org.archive.util.ObjectIdentityCache; import org.archive.util.bdbje.EnhancedEnvironment; import org.springframework.beans.factory.annotation.Autowired; @@ -174,7 +175,7 @@ public class BdbModule implements Lifecycle, Checkpointable, Closeable { * according to JE FAQ * http://www.oracle.com/technology/products/berkeley-db/faq/je_faq.html#33 */ - int expectedConcurrency = 25; + int expectedConcurrency = 64; public int getExpectedConcurrency() { return expectedConcurrency; } @@ -291,20 +292,6 @@ public class BdbModule implements Lifecycle, Checkpointable, Closeable { } } - public void disposeDatabase(String name) { - DatabasePlusConfig dpc = databases.remove(name); - if (dpc == null) { - throw new IllegalStateException("No such database: " + name); - } - Database db = dpc.database; - try { - db.close(); - bdbEnvironment.removeDatabase(null, name); - } catch (DatabaseException e) { - LOGGER.log(Level.WARNING, "Error closing db " + name, e); - } - } - /** * Open a Database inside this BdbModule's environment, and * remember it for automatic close-at-module-stop. @@ -376,7 +363,7 @@ public class BdbModule implements Lifecycle, Checkpointable, Closeable { * @return * @throws DatabaseException */ - public ObjectIdentityBdbCache getOIBCCache(String dbName, boolean recycle, + public ObjectIdentityBdbManualCache getOIBCCache(String dbName, boolean recycle, Class valueClass) throws DatabaseException { if (!recycle) { @@ -386,13 +373,13 @@ public class BdbModule implements Lifecycle, Checkpointable, Closeable { // ignored } } - ObjectIdentityBdbCache oic = new ObjectIdentityBdbCache(); + ObjectIdentityBdbManualCache oic = new ObjectIdentityBdbManualCache(); oic.initialize(bdbEnvironment, dbName, valueClass, classCatalog); oiCaches.put(dbName, oic); return oic; } - public ObjectIdentityCache getObjectCache(String dbName, boolean recycle, + public ObjectIdentityCache getObjectCache(String dbName, boolean recycle, Class valueClass) throws DatabaseException { return getObjectCache(dbName, recycle, valueClass, valueClass); @@ -411,11 +398,11 @@ public class BdbModule implements Lifecycle, Checkpointable, Closeable { * @return * @throws DatabaseException */ - public ObjectIdentityCache getObjectCache(String dbName, boolean recycle, + public ObjectIdentityCache getObjectCache(String dbName, boolean recycle, Class declaredClass, Class valueClass) throws DatabaseException { @SuppressWarnings("unchecked") - ObjectIdentityCache oic = oiCaches.get(dbName); + ObjectIdentityCache oic = oiCaches.get(dbName); if(oic!=null) { return oic; } @@ -639,9 +626,12 @@ public class BdbModule implements Lifecycle, Checkpointable, Closeable { true) { @Override public void dispose() { - BdbModule.this.disposeDatabase(openName); + super.dispose(); + DatabasePlusConfig dpc = BdbModule.this.databases.remove(openName); + if (dpc == null) { + BdbModule.LOGGER.log(Level.WARNING,"No such database: " + openName); + } } - ; }; return storedMap; } diff --git a/commons/src/main/java/org/archive/util/IdentityCacheable.java b/commons/src/main/java/org/archive/util/IdentityCacheable.java new file mode 100644 index 00000000..8b746c84 --- /dev/null +++ b/commons/src/main/java/org/archive/util/IdentityCacheable.java @@ -0,0 +1,33 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * 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.util; + +import java.io.Serializable; + +/** + * Common interface for objects held in ObjectIdentityCaches. + * + * @contributor gojomo + */ +public interface IdentityCacheable extends Serializable { + public void setIdentityCache(ObjectIdentityCache cache); + public String getKey(); + public void makeDirty(); +} diff --git a/commons/src/main/java/org/archive/util/IdentityCacheableWrapper.java b/commons/src/main/java/org/archive/util/IdentityCacheableWrapper.java new file mode 100644 index 00000000..d7e8a26f --- /dev/null +++ b/commons/src/main/java/org/archive/util/IdentityCacheableWrapper.java @@ -0,0 +1,71 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * 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.util; + +import org.archive.bdb.AutoKryo; + +/** + * Wrapper allowing other objects to be held in an ObjectIdentityCache. + * + * @contributor gojomo + * @param + */ +public class IdentityCacheableWrapper implements IdentityCacheable { + private static final long serialVersionUID = 1L; + + K wrapped; + + public IdentityCacheableWrapper(String key, K wrapped) { + super(); + this.wrapped = wrapped; + this.key = key; + } + + public K get() { + return wrapped; + } + + // + // IdentityCacheable support + // + transient private ObjectIdentityCache cache; + String key; + @Override + public String getKey() { + return key; + } + + @Override + public void makeDirty() { + cache.dirtyKey(getKey()); + } + + @Override + public void setIdentityCache(ObjectIdentityCache cache) { + this.cache = cache; + } + + // + // AutoKryo suppport + // + public static void autoregisterTo(AutoKryo kryo) { + kryo.register(IdentityCacheableWrapper.class); + kryo.setRegistrationOptional(true); + } +} diff --git a/commons/src/main/java/org/archive/util/ObjectIdentityBdbCache.java b/commons/src/main/java/org/archive/util/ObjectIdentityBdbCache.java index 053929ac..95914bf9 100644 --- a/commons/src/main/java/org/archive/util/ObjectIdentityBdbCache.java +++ b/commons/src/main/java/org/archive/util/ObjectIdentityBdbCache.java @@ -73,8 +73,8 @@ import com.sleepycat.je.Environment; * @author paul baclace (conversion to ConcurrentMap) * */ -public class ObjectIdentityBdbCache -implements ObjectIdentityCache, Closeable, Serializable { +public class ObjectIdentityBdbCache +implements ObjectIdentityCache, Closeable, Serializable { private static final long serialVersionUID = 1L; private static final Logger logger = Logger.getLogger(ObjectIdentityBdbCache.class.getName()); @@ -242,6 +242,7 @@ implements ObjectIdentityCache, Closeable, Serializable { if(val != null) { // the concurrent garden path: in mem, valid cacheHit.incrementAndGet(); + val.setIdentityCache(this); return val; } } @@ -255,6 +256,7 @@ implements ObjectIdentityCache, Closeable, Serializable { V val = entry.get(); if(val != null) { cacheHit.incrementAndGet(); + val.setIdentityCache(this); return val; } } @@ -300,6 +302,7 @@ implements ObjectIdentityCache, Closeable, Serializable { // memMap.get() should be impossible logger.log(Level.SEVERE,"memMap modified outside synchronized block?", new Exception()); } + valDisk.setIdentityCache(this); return valDisk; } } @@ -412,6 +415,12 @@ implements ObjectIdentityCache, Closeable, Serializable { this.diskMap.size() + ", mem " + this.memMap.size()); } } + + @Override + public void dirtyKey(String key) { + // do nothing, because our weak/phantom trickery is supposed to + // ensure sync-to-persistence if/when dereferenced and collected + } /** An incremental, poll-based expunger. * diff --git a/commons/src/main/java/org/archive/util/ObjectIdentityBdbManualCache.java b/commons/src/main/java/org/archive/util/ObjectIdentityBdbManualCache.java new file mode 100644 index 00000000..8b728dd5 --- /dev/null +++ b/commons/src/main/java/org/archive/util/ObjectIdentityBdbManualCache.java @@ -0,0 +1,361 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * 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.util; + +import java.io.Closeable; +import java.io.File; +import java.io.Serializable; +import java.util.Iterator; +import java.util.Set; +import java.util.Map.Entry; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.logging.Level; +import java.util.logging.Logger; + +import org.archive.bdb.KryoBinding; + +import com.google.common.collect.MapEvictionListener; +import com.google.common.collect.MapMaker; +import com.sleepycat.bind.EntryBinding; +import com.sleepycat.bind.serial.StoredClassCatalog; +import com.sleepycat.bind.tuple.TupleBinding; +import com.sleepycat.collections.StoredSortedMap; +import com.sleepycat.je.Database; +import com.sleepycat.je.DatabaseConfig; +import com.sleepycat.je.DatabaseException; +import com.sleepycat.je.Environment; + +/** + * A BDB JE backed object cache. + * + * Soft references to previously-instantiated objects are held so that + * unless/until an object is garbage collected, subsequent get()s will + * return the exact same object (avoiding redundant creation or disagreement + * about canonical object state). + *

+ * The backing disk is only guaranteed to be up-to-date after a flush + * of all dirty values to disk, as can be forced by sync(). + *

+ * + *

+ * + * @author John Erik Halse + * @author stack + * @author gojomo + * @author paul baclace (conversion to ConcurrentMap) + * + */ +public class ObjectIdentityBdbManualCache +implements ObjectIdentityCache, Closeable, Serializable, MapEvictionListener { + private static final long serialVersionUID = 1L; + private static final Logger logger = + Logger.getLogger(ObjectIdentityBdbManualCache.class.getName()); + + /** The BDB JE database used for this instance. */ + protected transient Database db; + + /** in-memory map of new/recent/still-referenced-elsewhere instances */ + protected transient ConcurrentMap memMap; + + /** The Collection view of the BDB JE database used for this instance. */ + protected transient StoredSortedMap diskMap; + + protected transient ConcurrentMap dirtyItems; + + protected AtomicLong count; + + // + // USAGE STATS + // + /** Count of times we got an object from in-memory cache */ + private AtomicLong cacheHit = new AtomicLong(0); + /** Count of times the {@link ObjectIdentityBdbManualCache#get} method was called. */ + private AtomicLong countOfGets = new AtomicLong(0); + /** Count of every time disk-based map provided non-null object */ + private AtomicLong diskHit = new AtomicLong(0); + /** Count of times Supplier was used for new object */ + private AtomicLong supplierUsed = new AtomicLong(0); + /** count of {@link #sync()} use */ + transient private AtomicLong useStatsSyncUsed = new AtomicLong(0); + + /** + * Constructor. You must call + * {@link #initialize(Environment, Class, Class, StoredClassCatalog)} + * to finish construction. Construction is two-stepped to support + * reconnecting a deserialized CachedBdbMap with its backing bdbje + * database. + * + * @param dbName Name of the backing db this instance should use. + */ + public ObjectIdentityBdbManualCache() { + super(); + } + + /** + * Call this method when you have an instance when you used the + * default constructor or when you have a deserialized instance that you + * want to reconnect with an extant bdbje environment. Do not + * call this method if you used the + * {@link #CachedBdbMap(File, String, Class, Class)} constructor. + * @param env + * @param keyClass + * @param valueClass + * @param classCatalog + * @throws DatabaseException + */ + @SuppressWarnings("unchecked") + public void initialize(final Environment env, String dbName, + final Class valueClass, final StoredClassCatalog classCatalog) + throws DatabaseException { + // TODO: tune capacity for actual threads, expected size of key caches? + this.memMap = new MapMaker().concurrencyLevel(64).initialCapacity(8192).softValues().makeMap(); + this.db = openDatabase(env, dbName); + this.diskMap = createDiskMap(this.db, classCatalog, valueClass); + // TODO: add maxCapacity when upgrading to Guava r08 + this.dirtyItems = new MapMaker().concurrencyLevel(64) + .expiration(600, TimeUnit.SECONDS).evictionListener(this).makeMap(); + + this.count = new AtomicLong(diskMap.size()); + } + + @SuppressWarnings("unchecked") + protected StoredSortedMap createDiskMap(Database database, + StoredClassCatalog classCatalog, Class valueClass) { + EntryBinding keyBinding = TupleBinding.getPrimitiveBinding(String.class); + EntryBinding valueBinding = TupleBinding.getPrimitiveBinding(valueClass); + if(valueBinding == null) { + valueBinding = + new KryoBinding(valueClass); +// new SerialBinding(classCatalog, valueClass); +// new BenchmarkingBinding(new EntryBinding[] { +// new KryoBinding(valueClass), +// new RecyclingSerialBinding(classCatalog, valueClass), +// }, valueClass); + } + return new StoredSortedMap(database, keyBinding, valueBinding, true); + } + + protected Database openDatabase(final Environment environment, + final String dbName) throws DatabaseException { + DatabaseConfig dbConfig = new DatabaseConfig(); + dbConfig.setTransactional(false); + dbConfig.setAllowCreate(true); + dbConfig.setDeferredWrite(true); + return environment.openDatabase(null, dbName, dbConfig); + } + + /* (non-Javadoc) + * @see org.archive.util.ObjectIdentityCache#close() + */ + public synchronized void close() { + // Close out my bdb db. + if (this.db != null) { + try { + sync(); + this.db.sync(); + this.db.close(); + } catch (DatabaseException e) { + logger.log(Level.WARNING,"problem closing ObjectIdentityBdbCache",e); + } finally { + this.db = null; + } + } + } + + protected void finalize() throws Throwable { + close(); + super.finalize(); + } + + /* (non-Javadoc) + * @see org.archive.util.ObjectIdentityCache#get(java.lang.String) + */ + public V get(final String key) { + return getOrUse(key,null); + } + + /* (non-Javadoc) + * @see org.archive.util.ObjectIdentityCache#get(java.lang.String, org.archive.util.ObjectIdentityBdbCache) + */ + public V getOrUse(final String key, Supplier supplierOrNull) { + countOfGets.incrementAndGet(); + + if (countOfGets.get() % 10000 == 0) { + logCacheSummary(); + } + + // check mem cache + V val = memMap.get(key); + if(val != null) { + // the concurrent garden path: in memory and valid + cacheHit.incrementAndGet(); + val.setIdentityCache(this); + return val; + } + val = diskMap.get(key); + V prevVal; + if(val == null) { + // never yet created, consider creating + if(supplierOrNull==null) { + return null; + } + val = supplierOrNull.get(); + supplierUsed.incrementAndGet(); + // putting initial value directly into diskMap + // (rather than just the memMap until page-out) + // ensures diskMap.keySet() provides complete view + prevVal = diskMap.putIfAbsent(key, val); + if(prevVal!=null) { + // we lost a race; discard our local creation in favor of disk version + diskHit.incrementAndGet(); + val = prevVal; + } else { + // we uniquely added a new key + count.incrementAndGet(); + } + } else { + diskHit.incrementAndGet(); + } + + prevVal = memMap.putIfAbsent(key, val); // fill memMap or lose race gracefully + if(prevVal != null) { + val = prevVal; + } + val.setIdentityCache(this); + return val; + } + + /* (non-Javadoc) + * @see org.archive.util.ObjectIdentityCache#keySet() + */ + public Set keySet() { + return diskMap.keySet(); + } + + /** + * Summary to log, if at FINE level + */ + private void logCacheSummary() { + if (logger.isLoggable((Level.FINE))) { + logger.fine(composeCacheSummary()); + } + } + + private String composeCacheSummary() { + long totalHits = cacheHit.get() + diskHit.get(); + if (totalHits < 1) { + return ""; + } + long cacheHitPercent + = (cacheHit.get() * 100) / totalHits; + StringBuilder sb = new StringBuilder(120); + sb.append("DB name:") + .append(getDatabaseName()) + .append(", ") + .append(" hit%: ") + .append(cacheHitPercent) + .append("%, gets=") + .append(countOfGets.get()) + .append(" memHits=") + .append(cacheHit.get()) + .append(" diskHits=") + .append(diskHit.get()) + .append(" supplieds=") + .append(supplierUsed.get()) + .append(" syncs=") + .append(useStatsSyncUsed.get()); + return sb.toString(); + } + + /* (non-Javadoc) + * @see org.archive.util.ObjectIdentityCache#size() + */ + public int size() { + if(db==null) { + return 0; + } + return (int) count.get(); + } + + protected String getDatabaseName() { + String name = "DbName-Lookup-Failed"; + try { + if (this.db != null) { + name = this.db.getDatabaseName(); + } + } catch (DatabaseException e) { + // Ignore. + } + return name; + } + + /** + * Sync all in-memory map entries to backing disk store. + */ + public synchronized void sync() { + String dbName = null; + // Sync. memory and disk. + useStatsSyncUsed.incrementAndGet(); + long startTime = 0; + if (logger.isLoggable(Level.FINE)) { + dbName = getDatabaseName(); + startTime = System.currentTimeMillis(); + logger.fine(dbName + " start sizes: disk " + this.diskMap.size() + + ", mem " + this.memMap.size()); + } + + Iterator> iter = dirtyItems.entrySet().iterator(); + while(iter.hasNext()) { + Entry entry = iter.next(); + iter.remove(); + diskMap.put(entry.getKey(), entry.getValue()); + } + + try { + this.db.sync(); + } catch (DatabaseException e) { + throw new RuntimeException(e); + } + + + if (logger.isLoggable(Level.FINE)) { + logger.fine(dbName + " sync took " + + (System.currentTimeMillis() - startTime) + "ms. " + + "Finish sizes: disk " + + this.diskMap.size() + ", mem " + this.memMap.size()); + } + } + + @Override + public void dirtyKey(String key) { + V val = memMap.get(key); + if(val==null) { + logger.severe("dirty key not in memory should be impossible"); + } + dirtyItems.put(key,val); + } + + @Override + public void onEviction(String key, V val) { + diskMap.put(key, val); + } +} diff --git a/commons/src/main/java/org/archive/util/ObjectIdentityCache.java b/commons/src/main/java/org/archive/util/ObjectIdentityCache.java index 564cd995..a32dc8b9 100644 --- a/commons/src/main/java/org/archive/util/ObjectIdentityCache.java +++ b/commons/src/main/java/org/archive/util/ObjectIdentityCache.java @@ -39,18 +39,24 @@ import java.util.Set; * * @param */ -public interface ObjectIdentityCache extends Closeable { - /** get the object under the given key/name */ - public abstract V get(final K key); +public interface ObjectIdentityCache extends Closeable { + /** get the object under the given key/name -- but should not mutate + * object state*/ + public abstract V get(final String key); /** get the object under the given key/name, using (and remembering) - * the object supplied by the supplier if no prior mapping exists */ - public abstract V getOrUse(final K key, Supplier supplierOrNull); + * the object supplied by the supplier if no prior mapping exists + * -- but should not mutate object state */ + public abstract V getOrUse(final String key, Supplier supplierOrNull); /** force the persistent backend, if any, to be updated with all * live object state */ public abstract void sync(); + /** force the persistent backend, if any, to eventually be updated with + * live object state for the given key */ + public abstract void dirtyKey(final String key); + /** close/release any associated resources */ public abstract void close(); @@ -58,5 +64,5 @@ public interface ObjectIdentityCache extends Closeable { public abstract int size(); /** set of all keys */ - public abstract Set keySet(); + public abstract Set keySet(); } \ No newline at end of file diff --git a/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java b/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java index 5d5445a8..9805567d 100644 --- a/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java +++ b/commons/src/main/java/org/archive/util/ObjectIdentityMemCache.java @@ -30,8 +30,8 @@ import java.util.concurrent.ConcurrentHashMap; * @contributor gojomo * @param */ -public class ObjectIdentityMemCache -implements ObjectIdentityCache { +public class ObjectIdentityMemCache +implements ObjectIdentityCache { ConcurrentHashMap map; public ObjectIdentityMemCache() { @@ -46,8 +46,11 @@ implements ObjectIdentityCache { // do nothing } + + @Override public V get(String key) { - return map.get(key); + // all gets are mutatable + return getOrUse(key,null); } public V getOrUse(String key, Supplier supplierOrNull) { @@ -59,6 +62,7 @@ implements ObjectIdentityCache { val = prevVal; } } + val.setIdentityCache(this); return val; } @@ -74,6 +78,11 @@ implements ObjectIdentityCache { // do nothing } + @Override + public void dirtyKey(String key) { + // do nothing: memory is whole cache + } + /** * Offer raw map access for convenience of checkpoint/recovery. * @return Map diff --git a/commons/src/test/java/org/archive/settings/file/BdbModuleTest.java b/commons/src/test/java/org/archive/settings/file/BdbModuleTest.java index 3aa905fa..981e5144 100644 --- a/commons/src/test/java/org/archive/settings/file/BdbModuleTest.java +++ b/commons/src/test/java/org/archive/settings/file/BdbModuleTest.java @@ -24,7 +24,8 @@ import org.apache.commons.lang.math.RandomUtils; import org.archive.bdb.BdbModule; import org.archive.checkpointing.Checkpoint; import org.archive.spring.ConfigPath; -import org.archive.util.ObjectIdentityBdbCache; +import org.archive.util.IdentityCacheableWrapper; +import org.archive.util.ObjectIdentityBdbManualCache; import org.archive.util.Supplier; import org.archive.util.TmpDirTestCase; @@ -36,6 +37,7 @@ import org.archive.util.TmpDirTestCase; */ public class BdbModuleTest extends TmpDirTestCase { + @SuppressWarnings("unchecked") public void testDoCheckpoint() throws Exception { ConfigPath basePath = new ConfigPath("testBase",getTmpDir().getAbsolutePath()); ConfigPath bdbDir = new ConfigPath("bdb","bdb"); @@ -49,16 +51,15 @@ public class BdbModuleTest extends TmpDirTestCase { // avoid data from prior runs being mistaken for current run int randomFactor = RandomUtils.nextInt(); - ObjectIdentityBdbCache testData = - bdb.getOIBCCache("testData", false,String.class); + ObjectIdentityBdbManualCache testData = + bdb.getOIBCCache("testData", false,IdentityCacheableWrapper.class); for (int i1 = 0; i1 < 1000; i1++) { String key = String.valueOf(i1); final String value = String.valueOf(randomFactor*i1); - String cached = testData.getOrUse(key, new Supplier(){ - public String get() { - return value; - } - }); + String cached = (String)testData.getOrUse( + key, + new Supplier( + new IdentityCacheableWrapper(key, value))).get(); assertSame("unexpected prior entry",value,cached); } @@ -73,11 +74,10 @@ public class BdbModuleTest extends TmpDirTestCase { for (int i2 = 1000; i2 < 2000; i2++) { String key = String.valueOf(i2); final String value = String.valueOf(randomFactor*i2); - String cached = testData.getOrUse(key, new Supplier(){ - public String get() { - return value; - } - }); + String cached = (String)testData.getOrUse( + key, + new Supplier( + new IdentityCacheableWrapper(key, value))).get(); assertSame("unexpected prior entry",value,cached); } @@ -101,11 +101,14 @@ public class BdbModuleTest extends TmpDirTestCase { bdb2.start(); - ObjectIdentityBdbCache restoreData = - bdb2.getOIBCCache("testData",true,String.class); + ObjectIdentityBdbManualCache restoreData = + bdb2.getOIBCCache("testData",true,IdentityCacheableWrapper.class); assertEquals("unexpected size", 1000, restoreData.size()); - assertEquals("unexpected value",randomFactor*999,Integer.parseInt(restoreData.get(""+999))); + assertEquals( + "unexpected value", + randomFactor*999, + Integer.parseInt((String)restoreData.get(""+999).get())); bdb2.stop(); } diff --git a/commons/src/test/java/org/archive/util/ObjectIdentityBdbCacheTest.java b/commons/src/test/java/org/archive/util/ObjectIdentityBdbCacheTest.java index ea19e77b..de78aed6 100644 --- a/commons/src/test/java/org/archive/util/ObjectIdentityBdbCacheTest.java +++ b/commons/src/test/java/org/archive/util/ObjectIdentityBdbCacheTest.java @@ -33,7 +33,7 @@ import org.archive.util.bdbje.EnhancedEnvironment; */ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { EnhancedEnvironment env; - private ObjectIdentityBdbCache> cache; + private ObjectIdentityBdbCache>> cache; protected void setUp() throws Exception { super.setUp(); @@ -42,8 +42,8 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { FileUtils.deleteDirectory(envDir); envDir.mkdirs(); env = EnhancedEnvironment.getTestEnvironment(envDir); - this.cache = new ObjectIdentityBdbCache>(); - this.cache.initialize(env,"setUpCache",HashMap.class, env.getClassCatalog()); + this.cache = new ObjectIdentityBdbCache>>(); + this.cache.initialize(env,"setUpCache",IdentityCacheableWrapper.class, env.getClassCatalog()); } protected void tearDown() throws Exception { @@ -56,19 +56,24 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { @SuppressWarnings("unchecked") public void testReadConsistencyUnderLoad() throws Exception { - final ObjectIdentityBdbCache cbdbmap = + final ObjectIdentityBdbCache> cbdbmap = new ObjectIdentityBdbCache(); cbdbmap.initialize(env, "consistencyCache", - AtomicInteger.class, + IdentityCacheableWrapper.class, env.getClassCatalog()); try { final AtomicInteger level = new AtomicInteger(0); final int keyCount = 128 * 1024; // 128K keys final int maxLevel = 64; // initial fill - for(int i=0; i < keyCount; i++) { - cbdbmap.getOrUse(""+i, new Supplier(new AtomicInteger(level.get()))); + for(int i=0; i < keyCount; i++) { + final String key = ""+i; + cbdbmap.getOrUse( + key, + new Supplier>( + new IdentityCacheableWrapper( + key, new AtomicInteger(level.get())))); } // backward checking that all values always at level or higher new Thread() { @@ -79,7 +84,7 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { if(targetValue>=maxLevel) { break untilmax; } - assertTrue("stale value revseq key "+j,cbdbmap.get(""+j).get()>=targetValue); + assertTrue("stale value revseq key "+j,cbdbmap.get(""+j).get().get()>=targetValue); Thread.yield(); } } @@ -95,7 +100,7 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { break untilmax; } assertTrue("stale value random key "+j, - cbdbmap.get(""+j).get()>=targetValue); + cbdbmap.get(""+j).get().get()>=targetValue); Thread.yield(); } } @@ -103,7 +108,7 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { // increment all keys for(; level.get() < maxLevel; level.incrementAndGet()) { for(int k = 0; k < keyCount; k++) { - int foundValue = cbdbmap.get(""+k).getAndIncrement(); + int foundValue = cbdbmap.get(""+k).get().getAndIncrement(); assertEquals("stale value preinc key "+k, level.get(), foundValue); } if(level.get() % 10 == 0) { @@ -123,17 +128,22 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { final String key = "key"; final int upperbound = 3; // First put in empty hashmap. - for (int i = 0; i < upperbound; i++) { - this.cache.getOrUse(key + Integer.toString(i), new Supplier>(new HashMap())); + for (int i = 0; i < upperbound; i++) { + String innerKey = key + Integer.toString(i); + this.cache.getOrUse( + innerKey, + new Supplier>>( + new IdentityCacheableWrapper>( + innerKey, new HashMap()))); } // Now add value to hash map. for (int i = 0; i < upperbound; i++) { - HashMap m = this.cache.get(key + Integer.toString(i)); + HashMap m = this.cache.get(key + Integer.toString(i)).get(); m.put(key, value); } this.cache.sync(); for (int i = 0; i < upperbound; i++) { - HashMap m = this.cache.get(key + Integer.toString(i)); + HashMap m = this.cache.get(key + Integer.toString(i)).get(); String v = m.get(key); assertNotNull("value should not be null",v); assertEquals("value incorrect", value, v); @@ -157,8 +167,13 @@ public class ObjectIdentityBdbCacheTest extends TmpDirTestCase { System.gc(); // minimize effects of earlier test heap use assertEquals(0, cache.memMap.size()); assertEquals(0, cache.diskMap.size()); - for(int i=0; i < 10000; i++) { - cache.getOrUse(""+i, new Supplier>(new HashMap())); + for(int i=0; i < 10000; i++) { + String key = ""+i; + cache.getOrUse( + key, + new Supplier>>( + new IdentityCacheableWrapper>( + key, new HashMap()))); } assertEquals(cache.memMap.size(), 10000); assertEquals(cache.size(), 10000); diff --git a/commons/src/test/java/org/archive/util/ObjectIdentityBdbManualCacheTest.java b/commons/src/test/java/org/archive/util/ObjectIdentityBdbManualCacheTest.java new file mode 100644 index 00000000..12741040 --- /dev/null +++ b/commons/src/test/java/org/archive/util/ObjectIdentityBdbManualCacheTest.java @@ -0,0 +1,199 @@ +/* + * This file is part of the Heritrix web crawler (crawler.archive.org). + * + * 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.util; + +import java.io.File; +import java.util.HashMap; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.commons.io.FileUtils; +import org.apache.commons.lang.math.RandomUtils; +import org.archive.util.bdbje.EnhancedEnvironment; + +/** + * @contributor stack + * @contributor gojomo + * @version $Date: 2009-08-03 23:50:43 -0700 (Mon, 03 Aug 2009) $, $Revision: 6434 $ + */ +public class ObjectIdentityBdbManualCacheTest extends TmpDirTestCase { + EnhancedEnvironment env; + private ObjectIdentityBdbManualCache>> cache; + + protected void setUp() throws Exception { + super.setUp(); + File envDir = new File(getTmpDir(),"ObjectIdentityBdbCacheTest"); + envDir.mkdirs(); + FileUtils.deleteDirectory(envDir); + envDir.mkdirs(); + env = EnhancedEnvironment.getTestEnvironment(envDir); + this.cache = new ObjectIdentityBdbManualCache>>(); + this.cache.initialize(env,"setUpCache",IdentityCacheableWrapper.class, env.getClassCatalog()); + } + + protected void tearDown() throws Exception { + this.cache.close(); + File envDir = env.getHome(); + env.close(); + FileUtils.deleteDirectory(envDir); + super.tearDown(); + } + + @SuppressWarnings("unchecked") + public void testReadConsistencyUnderLoad() throws Exception { + final ObjectIdentityBdbManualCache> cbdbmap = + new ObjectIdentityBdbManualCache(); + cbdbmap.initialize(env, + "consistencyCache", + IdentityCacheableWrapper.class, + env.getClassCatalog()); + try { + final AtomicInteger level = new AtomicInteger(0); + final int keyCount = 128 * 1024; // 128K keys + final int maxLevel = 64; + // initial fill + for(int i=0; i < keyCount; i++) { + final String key = ""+i; + cbdbmap.getOrUse( + key, + new Supplier>( + new IdentityCacheableWrapper( + key, new AtomicInteger(level.get())))); + } + // backward checking that all values always at level or higher + new Thread() { + public void run() { + untilmax: while(true) { + for(int j=keyCount-1; j >= 0; j--) { + int targetValue = level.get(); + if(targetValue>=maxLevel) { + break untilmax; + } + assertTrue("stale value revseq key "+j,cbdbmap.get(""+j).get().get()>=targetValue); + Thread.yield(); + } + } + } + }.start(); + // random checking that all values always at level or higher + new Thread() { + public void run() { + untilmax: while(true) { + int j = RandomUtils.nextInt(keyCount); + int targetValue = level.get(); + if(targetValue>=maxLevel) { + break untilmax; + } + assertTrue("stale value random key "+j, + cbdbmap.get(""+j).get().get()>=targetValue); + Thread.yield(); + } + } + }.start(); + // increment all keys + for(; level.get() < maxLevel; level.incrementAndGet()) { + for(int k = 0; k < keyCount; k++) { + IdentityCacheableWrapper wrap = cbdbmap.get(""+k); + int foundValue = wrap.get().getAndIncrement(); + wrap.makeDirty(); + assertEquals("stale value preinc key "+k, level.get(), foundValue); + } + if(level.get() % 10 == 0) { + System.out.println("level to "+level.get()); + if(level.get()>0) { + TestUtils.forceScarceMemory(); + } + } + Thread.yield(); + } + } finally { + cbdbmap.close(); + } + // SUCCESS + } + + public void testBackingDbGetsUpdated() { + // Set up values. + final String value = "value"; + final String key = "key"; + final int upperbound = 3; + // First put in empty hashmap. + for (int i = 0; i < upperbound; i++) { + String innerKey = key + Integer.toString(i); + this.cache.getOrUse( + innerKey, + new Supplier>>( + new IdentityCacheableWrapper>( + innerKey, new HashMap()))); + } + // Now add value to hash map. + for (int i = 0; i < upperbound; i++) { + HashMap m = this.cache.get(key + Integer.toString(i)).get(); + m.put(key, value); + } + this.cache.sync(); + for (int i = 0; i < upperbound; i++) { + HashMap m = this.cache.get(key + Integer.toString(i)).get(); + String v = m.get(key); + assertNotNull("value should not be null",v); + assertEquals("value incorrect", value, v); + } + } + + /** + * Test that in scarce memory conditions, the memory map is + * expunged of otherwise unreferenced entries as expected. + * @throws InterruptedException + */ + public void xestMemMapCleared() throws InterruptedException { + TestUtils.forceScarceMemory(); + System.gc(); // minimize effects of earlier test heap use + assertEquals(0, cache.memMap.size()); + assertEquals(0, cache.diskMap.size()); + for(int i=0; i < 10000; i++) { + String key = ""+i; + cache.getOrUse( + key, + new Supplier>>( + new IdentityCacheableWrapper>( + key, new HashMap()))); + } + assertEquals(cache.memMap.size(), 10000); + assertEquals(cache.size(), 10000); + TestUtils.forceScarceMemory(); + Thread.sleep(6000); + // The 'canary' trick may make this explicit page-out, or + // a page-out riggered by a get() or put...(), unnecessary -- + // but we include anyway. + //cache.pageOutStaleEntries(); + + int countNonNull = 0; + for(String key: cache.memMap.keySet()) { + if(cache.memMap.get(key)!=null) { + countNonNull++; + } + } + System.out.println(cache.size()+","+cache.memMap.size()+","+cache.memMap.keySet().size()+","+cache.memMap.values().size()+","+countNonNull); + assertEquals("memMap not cleared", 0, cache.memMap.size()); + } + + + public static void main(String [] args) { + junit.textui.TestRunner.run(ObjectIdentityBdbManualCacheTest.class); + } +} diff --git a/engine/src/main/java/org/archive/crawler/framework/Frontier.java b/engine/src/main/java/org/archive/crawler/framework/Frontier.java index b4ff0a4a..806246dc 100644 --- a/engine/src/main/java/org/archive/crawler/framework/Frontier.java +++ b/engine/src/main/java/org/archive/crawler/framework/Frontier.java @@ -27,6 +27,7 @@ import org.archive.crawler.frontier.FrontierJournal; import org.archive.modules.CrawlURI; import org.archive.modules.deciderules.DecideRule; import org.archive.modules.fetcher.FetchStats; +import org.archive.util.IdentityCacheable; import org.archive.util.MultiReporter; import org.json.JSONException; import org.springframework.context.Lifecycle; @@ -497,7 +498,7 @@ public interface Frontier extends Lifecycle, MultiReporter { * offers the HasCrawlSubstats interface. */ public interface FrontierGroup - extends FetchStats.HasFetchStats, FetchStats.CollectsFetchStats { + extends FetchStats.HasFetchStats, FetchStats.CollectsFetchStats, IdentityCacheable { } /** diff --git a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java index 116683ce..64eb9519 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/AbstractFrontier.java @@ -607,13 +607,16 @@ public abstract class AbstractFrontier CrawlServer server = getServerCache().getServerFor(curi.getUURI()); if (server != null) { server.getSubstats().tally(curi, stage); + server.makeDirty(); } CrawlHost host = getServerCache().getHostFor(curi.getUURI()); if (host != null) { host.getSubstats().tally(curi, stage); + host.makeDirty(); } FrontierGroup group = getGroup(curi); group.tally(curi, stage); + group.makeDirty(); } protected void doJournalFinishedSuccess(CrawlURI c) { diff --git a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java index f4db3e07..d6c3a766 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/BdbFrontier.java @@ -132,18 +132,6 @@ implements Checkpointable, BeanNameAware { } - /** - * Return the work queue for the given CrawlURI's classKey. URIs - * are ordered and politeness-delayed within their 'class'. - * - * @param curi CrawlURI to base queue on - * @return the found or created BdbWorkQueue - */ - protected WorkQueue getQueueFor(CrawlURI curi) { - String classKey = curi.getClassKey(); - return getQueueFor(classKey); - } - /** * Return the work queue for the given classKey, or null * if no such queue exists. @@ -164,8 +152,12 @@ implements Checkpointable, BeanNameAware { }}); return wq; } - - + + @Override + public FrontierGroup getGroup(CrawlURI curi) { + return getQueueFor(curi.getClassKey()); + } + /** * Return list of urls. * @param marker @@ -276,7 +268,8 @@ implements Checkpointable, BeanNameAware { totalProcessedBytes.set(json.getLong("totalProcessedBytes")); } catch (JSONException e) { throw new RuntimeException(e); - } + } + // TODO: restore largestQueues topNset // restore WorkQueues to internal management queues // either retired or inactive tiers for (String key : allQueues.keySet()) { @@ -286,7 +279,7 @@ implements Checkpointable, BeanNameAware { } else if (q.getCount()>0){ deactivateQueue(q); } else { - // ensure quques that were empty-but-not-yet-recognized + // ensure queues that were empty-but-not-yet-recognized // (ready/snoozed) at time of checkpoint are reset q.noteDeactivated(); } diff --git a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java b/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java index 2a592278..cc9c8fc7 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueue.java @@ -37,7 +37,9 @@ import org.archive.modules.CrawlURI; import org.archive.modules.fetcher.FetchStats; import org.archive.modules.fetcher.FetchStats.Stage; import org.archive.util.ArchiveUtils; +import org.archive.util.IdentityCacheable; import org.archive.util.MultiReporter; +import org.archive.util.ObjectIdentityCache; /** * A single queue of related URIs to visit, grouped by a classKey @@ -47,7 +49,7 @@ import org.archive.util.MultiReporter; * @author Christian Kohlschuetter */ public abstract class WorkQueue implements Frontier.FrontierGroup, - Serializable, MultiReporter, Delayed { + Serializable, MultiReporter, Delayed, IdentityCacheable { private static final long serialVersionUID = -3199666138837266341L; private static final Logger logger = Logger.getLogger(WorkQueue.class.getName()); @@ -651,6 +653,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, public synchronized void noteDeactivated() { active = false; isManaged = true; + makeDirty(); } /** @@ -660,6 +663,7 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, public synchronized void noteExhausted() { active = false; isManaged = false; + makeDirty(); } /** @@ -679,4 +683,24 @@ public abstract class WorkQueue implements Frontier.FrontierGroup, public String toString() { return super.toString()+"("+getClassKey()+")"; } + + + // + // IdentityCacheable support + // + transient private ObjectIdentityCache cache; + @Override + public String getKey() { + return getClassKey(); + } + + @Override + public void makeDirty() { + cache.dirtyKey(getKey()); + } + + @Override + public void setIdentityCache(ObjectIdentityCache cache) { + this.cache = cache; + } } 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 c821a31d..6995e5e7 100644 --- a/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java +++ b/engine/src/main/java/org/archive/crawler/frontier/WorkQueueFrontier.java @@ -182,7 +182,7 @@ implements Closeable, /** All known queues. */ - protected ObjectIdentityCache allQueues = null; + protected ObjectIdentityCache allQueues = null; // of classKey -> ClassKeyQueue /** @@ -385,7 +385,7 @@ implements Closeable, protected void sendToQueue(CrawlURI curi) { // assert Thread.currentThread() == managerThread; - WorkQueue wq = getQueueFor(curi); + WorkQueue wq = getQueueFor(curi.getClassKey()); synchronized(wq) { int originalPrecedence = wq.getPrecedence(); wq.enqueue(this, curi); @@ -406,6 +406,7 @@ implements Closeable, } // Update recovery log. doJournalAdded(curi); + wq.makeDirty(); largestQueues.update(wq.getClassKey(), wq.getCount()); } @@ -539,6 +540,7 @@ implements Closeable, WorkQueue q = (WorkQueue)this.allQueues.get(key); if(q != null) { unretireQueue(q); + q.makeDirty(); } key = getRetiredQueues().poll(); } @@ -556,16 +558,6 @@ implements Closeable, incrementQueuedUriCount(q.getCount()); } - /** - * Return the work queue for the given CrawlURI's classKey. URIs - * are ordered and politeness-delayed within their 'class'. - * If the requested queue is not found, a new instance is created. - * - * @param curi CrawlURI to base queue on - * @return the found or created ClassKeyQueue - */ - protected abstract WorkQueue getQueueFor(CrawlURI curi); - /** * Return the work queue for the given classKey, or null * if no such queue exists. @@ -575,6 +567,7 @@ implements Closeable, */ protected abstract WorkQueue getQueueFor(String classKey); + /** * Return the next CrawlURI eligible to be processed (and presumably * visited/fetched) by a a worker thread. @@ -619,6 +612,7 @@ implements Closeable, if(readyQ.getCount()==0) { // readyQ is empty and ready: it's exhausted readyQ.noteExhausted(); + readyQ.makeDirty(); readyQ = null; continue; } @@ -635,11 +629,13 @@ implements Closeable, readyQ.setWakeTime(0); // clear obsolete wake time, if any if (readyQ.isOverSessionBudget()) { deactivateQueue(readyQ); + readyQ.makeDirty(); readyQ = null; continue; } if (readyQ.isOverTotalBudget()) { retireQueue(readyQ); + readyQ.makeDirty(); readyQ = null; continue; } @@ -698,6 +694,7 @@ implements Closeable, // receive new URI, be readied, fail not-in-process? inProcessQueues.remove(readyQ); readyQ.noteExhausted(); + readyQ.makeDirty(); readyQ = null; continue findauri; } @@ -764,6 +761,7 @@ implements Closeable, if(candidateQ.getPrecedence() > expectedPrecedence) { // queue demoted since placed; re-deactivate deactivateQueue(candidateQ); + candidateQ.makeDirty(); continue; } updateHighestWaiting(expectedPrecedence); @@ -840,6 +838,7 @@ implements Closeable, WorkQueue queue = iterSnoozed.next().getWorkQueue(WorkQueueFrontier.this); queue.setWakeTime(0); reenqueueQueue(queue); + queue.makeDirty(); iterSnoozed.remove(); } Iterator iterOverflow = snoozedOverflow.values().iterator(); @@ -847,6 +846,7 @@ implements Closeable, WorkQueue queue = iterOverflow.next().getWorkQueue(WorkQueueFrontier.this); queue.setWakeTime(0); reenqueueQueue(queue); + queue.makeDirty(); iterOverflow.remove(); snoozedOverflowCount.decrementAndGet(); } @@ -860,6 +860,7 @@ implements Closeable, while((waked = snoozedClassQueues.poll())!=null) { WorkQueue queue = waked.getWorkQueue(this); queue.setWakeTime(0); + queue.makeDirty(); reenqueueQueue(queue); } // also consider overflow (usually empty) @@ -873,6 +874,7 @@ implements Closeable, snoozedOverflowCount.decrementAndGet(); WorkQueue queue = dq.getWorkQueue(this); queue.setWakeTime(0); + queue.makeDirty(); reenqueueQueue(queue); } } @@ -924,7 +926,7 @@ implements Closeable, handleQueue(wq,curi.includesRetireDirective(),now,delay_ms); appCtx.publishEvent(new CrawlURIDispositionEvent(this,curi,DEFERRED_FOR_RETRY)); doJournalReenqueued(curi); - + wq.makeDirty(); return; // no further dequeueing, logging, rescheduling to occur } @@ -973,7 +975,8 @@ implements Closeable, long delay_ms = curi.getPolitenessDelay(); handleQueue(wq,curi.includesRetireDirective(),now,delay_ms); - + wq.makeDirty(); + if(curi.getRescheduleTime()>0) { // marked up for forced-revisit at a set time curi.processingCleanup(); @@ -1053,13 +1056,13 @@ implements Closeable, */ public long deleteURIs(String queueRegex, String uriRegex) { long count = 0; - // TODO: DANGER/ values() may not work right from CachedBdbMap Pattern queuePat = Pattern.compile(queueRegex); for (String qname: allQueues.keySet()) { if (queuePat.matcher(qname).matches()) { WorkQueue wq = getQueueFor(qname); wq.unpeek(null); count += wq.deleteMatching(this, uriRegex); + wq.makeDirty(); } } decrementQueuedCount(count); @@ -1113,6 +1116,8 @@ implements Closeable, * @param w Where to write to. */ public void shortReportLineTo(PrintWriter w) { + if (!isRunning()) return; //??? + if (this.allQueues == null) { return; } @@ -1506,7 +1511,9 @@ implements Closeable, try { KeyedProperties.loadOverridesFrom(curi); curi.setClassKey(getClassKey(curi)); - getQueueFor(curi).expend(curi.getHolderCost()); + WorkQueue wq = getQueueFor(curi.getClassKey()); + wq.expend(curi.getHolderCost()); + wq.makeDirty(); } finally { KeyedProperties.clearOverridesFrom(curi); } @@ -1523,13 +1530,7 @@ implements Closeable, * @return a constant boolean value for this class/instance */ protected abstract boolean workQueueDataOnDisk(); - - - public FrontierGroup getGroup(CrawlURI curi) { - return getQueueFor(curi); - } - - + public long averageDepth() { if(inProcessQueues==null || readyClassQueues==null || snoozedClassQueues==null) { return 0; @@ -1581,4 +1582,5 @@ implements Closeable, protected int getInProcessCount() { return inProcessQueues.size(); } + } // TODO: slim class! Suspect it should be < 800 lines, shedding budgeting/reporting \ No newline at end of file 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 467c2a62..1fa3b5d3 100644 --- a/modules/src/main/java/org/archive/modules/fetcher/DefaultServerCache.java +++ b/modules/src/main/java/org/archive/modules/fetcher/DefaultServerCache.java @@ -21,6 +21,7 @@ package org.archive.modules.fetcher; import java.io.Closeable; import java.io.Serializable; +import java.util.Set; import java.util.logging.Logger; import org.apache.commons.collections.Closure; @@ -49,13 +50,13 @@ public class DefaultServerCache extends ServerCache implements Closeable, Serial * hostname[:port] -> CrawlServer. * Set in the initialization. */ - protected ObjectIdentityCache servers = null; + protected ObjectIdentityCache servers = null; /** * hostname -> CrawlHost. * Set in the initialization. */ - protected ObjectIdentityCache hosts = null; + protected ObjectIdentityCache hosts = null; /** * Constructor. @@ -68,8 +69,8 @@ public class DefaultServerCache extends ServerCache implements Closeable, Serial - public DefaultServerCache(ObjectIdentityCache servers, - ObjectIdentityCache hosts) { + public DefaultServerCache(ObjectIdentityCache servers, + ObjectIdentityCache hosts) { this.servers = servers; this.hosts = hosts; } @@ -141,7 +142,10 @@ public class DefaultServerCache extends ServerCache implements Closeable, Serial } } - /* (non-Javadoc) + /** + * NOTE: Should not mutate the CrawlHost instance so retrieved; depending on + * the hostscache implementation, the change may not be reliably persistent. + * * @see org.archive.modules.net.ServerCache#forAllHostsDo(org.apache.commons.collections.Closure) */ public void forAllHostsDo(Closure c) { @@ -149,4 +153,8 @@ public class DefaultServerCache extends ServerCache implements Closeable, Serial c.execute(hosts.get(host)); } } + + public Set hostKeys() { + return hosts.keySet(); + } } \ No newline at end of file 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 180fed9a..fdfcee9d 100644 --- a/modules/src/main/java/org/archive/modules/net/CrawlHost.java +++ b/modules/src/main/java/org/archive/modules/net/CrawlHost.java @@ -27,7 +27,9 @@ import java.util.logging.Logger; import org.archive.bdb.AutoKryo; import org.archive.modules.fetcher.FetchStats; +import org.archive.util.IdentityCacheable; import org.archive.util.InetAddressUtil; +import org.archive.util.ObjectIdentityCache; /** * Represents a single remote "host". @@ -37,7 +39,7 @@ import org.archive.util.InetAddressUtil; * * @author gojomo */ -public class CrawlHost implements Serializable, FetchStats.HasFetchStats { +public class CrawlHost implements Serializable, FetchStats.HasFetchStats, IdentityCacheable { private static final long serialVersionUID = -5494573967890942895L; @@ -234,4 +236,23 @@ public class CrawlHost implements Serializable, FetchStats.HasFetchStats { kryo.autoregister(Inet4Address.class); kryo.setRegistrationOptional(true); } + + // + // IdentityCacheable support + // + transient private ObjectIdentityCache cache; + @Override + public String getKey() { + return getHostName(); + } + + @Override + public void makeDirty() { + cache.dirtyKey(getKey()); + } + + @Override + public void setIdentityCache(ObjectIdentityCache cache) { + this.cache = cache; + } } 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 27ba509b..b7bdfbce 100644 --- a/modules/src/main/java/org/archive/modules/net/CrawlServer.java +++ b/modules/src/main/java/org/archive/modules/net/CrawlServer.java @@ -44,6 +44,8 @@ import org.archive.modules.credential.Credential; import org.archive.modules.fetcher.FetchStats; import org.archive.net.UURI; import org.archive.net.UURIFactory; +import org.archive.util.IdentityCacheable; +import org.archive.util.ObjectIdentityCache; /** * Represents a single remote "server". @@ -53,7 +55,7 @@ import org.archive.net.UURIFactory; * * @author gojomo */ -public class CrawlServer implements Serializable, FetchStats.HasFetchStats { +public class CrawlServer implements Serializable, FetchStats.HasFetchStats, IdentityCacheable { private static final Logger logger = Logger.getLogger(CrawlServer.class.getName()); private static final long serialVersionUID = 3L; @@ -338,4 +340,23 @@ public class CrawlServer implements Serializable, FetchStats.HasFetchStats { kryo.autoregister(java.util.LinkedList.class); kryo.setRegistrationOptional(true); } + + // + // IdentityCacheable support + // + transient private ObjectIdentityCache cache; + @Override + public String getKey() { + return getName(); + } + + @Override + public void makeDirty() { + cache.dirtyKey(getKey()); + } + + @Override + public void setIdentityCache(ObjectIdentityCache cache) { + this.cache = cache; + } } 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 c6c13779..e8d389f3 100644 --- a/modules/src/main/java/org/archive/modules/net/ServerCache.java +++ b/modules/src/main/java/org/archive/modules/net/ServerCache.java @@ -19,6 +19,7 @@ package org.archive.modules.net; +import java.util.Set; import java.util.logging.Level; import java.util.logging.Logger; @@ -90,5 +91,7 @@ public abstract class ServerCache { return cs; } + abstract public Set hostKeys(); + }