Minimize transient garbage volume by optimized serialization based on the 'Kryo' library

* AutoKryo
    extension of Kryo to allow classes to control their own registration, trigger registration of associated classes, and deserialize classes without no-arg constructors
* KryoBinding
    binding for use with BDB that uses AutoKryo serialization for a 2X-4X reduction in byte[] size
* UURI
    improved serialization via Externalizable and Kryo's CustomSerialization methods
* BdbModule
    discard deprecated CachedBDBMap option
    (getObjectCache) extend with both declaredClass and valueClass (for when map values are specializations of the declared type, as with frontier.allQueues)
    adjust type declarations
* ObjectIdentityBdbCache
    use KryoBinding rather than SerialBinding
* CachedBdbMapTest
    discarded
* BdbFrontier, BdbServerCache, StatisticsTracker
    adjust type declarations, objectCache creation
* BdbMultipleWorkQueues
    use KryoBinding rather than (Recycling)SerialBinding
* BdbWorkQueue, CrawlServer, CrawlHost, CrawlURI
    add autoregister support
* LinkContext
    public for kryo registration
This commit is contained in:
gojomo committed 2010-11-05 23:23:58 +00:00
1 parent f4a01f1155
commit 0627a67dce
16 files changed
+360 -275

No files matched your search

@@ -153,9 +153,9 @@ implements Checkpointable, BeanNameAware {
WorkQueue wq = allQueues.getOrUse(
classKey,
new Supplier<WorkQueue>() {
public WorkQueue get() {
public BdbWorkQueue get() {
String qKey = new String(classKey); // ensure private minimal key
WorkQueue q = new BdbWorkQueue(qKey, BdbFrontier.this);
BdbWorkQueue q = new BdbWorkQueue(qKey, BdbFrontier.this);
q.setTotalBudget(getQueueTotalBudget());
getQueuePrecedencePolicy().queueCreated(q);
return q;
@@ -257,7 +257,7 @@ implements Checkpointable, BeanNameAware {
@Override
protected void initAllQueues() throws DatabaseException {
boolean isRecovery = (recoveryCheckpoint != null);
this.allQueues = bdb.getObjectCache("allqueues", isRecovery, WorkQueue.class);
this.allQueues = bdb.getObjectCache("allqueues", isRecovery, WorkQueue.class, BdbWorkQueue.class);
if(isRecovery) {
JSONObject json = recoveryCheckpoint.loadJson(beanName);
try {
@@ -31,9 +31,11 @@ import javax.management.openmbean.CompositeDataSupport;
import javax.management.openmbean.OpenDataException;
import org.apache.commons.collections.Closure;
import org.archive.bdb.KryoBinding;
import org.archive.modules.CrawlURI;
import org.archive.util.ArchiveUtils;
import com.sleepycat.bind.EntryBinding;
import com.sleepycat.bind.serial.StoredClassCatalog;
import com.sleepycat.je.Cursor;
import com.sleepycat.je.Database;
@@ -65,7 +67,7 @@ public class BdbMultipleWorkQueues {
private Database pendingUrisDB = null;
/** Supporting bdb serialization of CrawlURIs */
private RecyclingSerialBinding<CrawlURI> crawlUriBinding;
private EntryBinding<CrawlURI> crawlUriBinding;
/**
* Create the multi queue in the given environment.
@@ -80,7 +82,14 @@ public class BdbMultipleWorkQueues {
throws DatabaseException {
this.pendingUrisDB = db;
crawlUriBinding =
new RecyclingSerialBinding<CrawlURI>(classCatalog, CrawlURI.class);
new KryoBinding<CrawlURI>(CrawlURI.class);
// new RecyclingSerialBinding<CrawlURI>(classCatalog, CrawlURI.class);
// new BenchmarkingBinding<CrawlURI>(new EntryBinding[] {
// new KryoBinding<CrawlURI>(CrawlURI.class,true),
// new KryoBinding<CrawlURI>(CrawlURI.class,false),
// new RecyclingSerialBinding<CrawlURI>(classCatalog, CrawlURI.class),
// });
}
/**
@@ -21,10 +21,14 @@ package org.archive.crawler.frontier;
import java.io.IOException;
import java.io.Serializable;
import java.io.UnsupportedEncodingException;
import java.util.HashSet;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.archive.bdb.AutoKryo;
import org.archive.crawler.frontier.precedence.SimplePrecedenceProvider;
import org.archive.modules.CrawlURI;
import org.archive.modules.fetcher.FetchStats;
import org.archive.util.ArchiveUtils;
import com.sleepycat.je.DatabaseEntry;
@@ -166,4 +170,14 @@ implements Serializable {
return e.getMessage();
}
}
// Kryo support
public static void autoregisterTo(AutoKryo kryo) {
kryo.register(BdbWorkQueue.class);
kryo.autoregister(FetchStats.class);
kryo.autoregister(HashSet.class);
kryo.autoregister(SimplePrecedenceProvider.class);
kryo.autoregister(byte[].class);
kryo.setRegistrationOptional(true);
}
}
@@ -331,8 +331,9 @@ public class StatisticsTracker
new ObjectIdentityMemCache<AtomicLong>(); // temp dummy
/** Keep track of URL counts per host per seed */
protected ObjectIdentityCache<String,ConcurrentMap<String,AtomicLong>> sourceHostDistribution =
new ObjectIdentityMemCache<ConcurrentMap<String,AtomicLong>>(); // temp dummy;
@SuppressWarnings("unchecked")
protected ObjectIdentityCache<String,ConcurrentMap> sourceHostDistribution =
new ObjectIdentityMemCache<ConcurrentMap>(); // temp dummy;
/* Keep track of 'top' hosts for live reports */
protected TopNSet hostsDistributionTop;
@@ -503,6 +504,10 @@ public class StatisticsTracker
return snapshot;
}
public LinkedList<CrawlStatSnapshot> getSnapshots() {
return snapshots;
}
public CrawlStatSnapshot getLastSnapshot() {
CrawlStatSnapshot snap = snapshots.peek();
return snap == null ? getSnapshot() : snap;
@@ -840,12 +845,13 @@ public class StatisticsTracker
}
}
@SuppressWarnings("unchecked")
protected void saveSourceStats(String source, String hostname) {
synchronized(sourceHostDistribution) {
ConcurrentMap<String,AtomicLong> hostUriCount =
sourceHostDistribution.getOrUse(
source,
new Supplier<ConcurrentMap<String,AtomicLong>>() {
new Supplier<ConcurrentMap>() {
public ConcurrentMap<String, AtomicLong> get() {
return new ConcurrentHashMap<String,AtomicLong>();
}});
@@ -1121,6 +1127,7 @@ public class StatisticsTracker
// CrawlController's only interest is in knowing that a Checkpoint is
// being recovered
public void startCheckpoint(Checkpoint checkpointInProgress) {}
@SuppressWarnings("unchecked")
public void doCheckpoint(Checkpoint checkpointInProgress) throws IOException {
JSONObject json = new JSONObject();
try {