diff --git a/commons/src/main/java/org/archive/bdb/BdbModule.java b/commons/src/main/java/org/archive/bdb/BdbModule.java index e7cdeaf2..d9d6ec68 100644 --- a/commons/src/main/java/org/archive/bdb/BdbModule.java +++ b/commons/src/main/java/org/archive/bdb/BdbModule.java @@ -21,35 +21,32 @@ package org.archive.bdb; import java.io.Closeable; import java.io.File; -import java.io.FileOutputStream; -import java.io.FilenameFilter; +import java.io.FileFilter; import java.io.IOException; -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; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; import java.util.logging.Level; import java.util.logging.Logger; import org.apache.commons.io.FileUtils; -import org.archive.checkpointing.CheckpointRecovery; +import org.apache.commons.io.filefilter.FileFilterUtils; +import org.apache.commons.io.filefilter.IOFileFilter; import org.archive.checkpointing.Checkpointable; -import org.archive.checkpointing.RecoverAction; +import org.archive.crawler.framework.Checkpoint; import org.archive.spring.ConfigPath; import org.archive.util.CachedBdbMap; import org.archive.util.ObjectIdentityBdbCache; import org.archive.util.ObjectIdentityCache; import org.archive.util.bdbje.EnhancedEnvironment; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.Lifecycle; import com.sleepycat.bind.EntryBinding; @@ -61,10 +58,8 @@ import com.sleepycat.je.Database; import com.sleepycat.je.DatabaseConfig; import com.sleepycat.je.DatabaseException; import com.sleepycat.je.DatabaseNotFoundException; -import com.sleepycat.je.DbInternal; import com.sleepycat.je.EnvironmentConfig; -import com.sleepycat.je.dbi.EnvironmentImpl; -import com.sleepycat.je.utilint.DbLsn; +import com.sleepycat.je.util.DbBackup; /** * Utility module for managing a shared BerkeleyDB-JE environment @@ -219,11 +214,21 @@ Serializable, Closeable { if (isRunning()) { return; } + try { - setUp(getDir().getFile(), getCachePercent(), true, getUseSharedCache()); + boolean isRecovery = false; + if(recoveryCheckpoint!=null) { + isRecovery = true; + doRecover(); + } + + setup(getDir().getFile(), getCachePercent(), !isRecovery, getUseSharedCache()); } catch (DatabaseException e) { throw new IllegalStateException(e); + } catch (IOException e) { + throw new IllegalStateException(e); } + shutdownHook = new BdbShutdownHook(this); Runtime.getRuntime().addShutdownHook(shutdownHook); } @@ -239,7 +244,7 @@ Serializable, Closeable { close(); } - private void setUp(File f, int cachePercent, boolean create, boolean sharedCache) + protected void setup(File f, int cachePercent, boolean create, boolean sharedCache) throws DatabaseException { EnvironmentConfig config = new EnvironmentConfig(); config.setAllowCreate(create); @@ -260,6 +265,10 @@ Serializable, Closeable { // triple this value to 6K because stats show many faults config.setConfigParam("je.log.faultReadSize", "6144"); f.mkdirs(); + + // to support checkpoints, prevent BDB's cleaner from deleting log files + config.setConfigParam("je.cleaner.expunge", "false"); + this.bdbEnvironment = new EnhancedEnvironment(f, config); this.classCatalog = this.bdbEnvironment.getClassCatalog(); @@ -449,71 +458,21 @@ Serializable, Closeable { out.defaultWriteObject(); } - // TODO:FIXME: restore functionality - @SuppressWarnings("unchecked") - private void readObject(ObjectInputStream in) - throws IOException, ClassNotFoundException { - in.defaultReadObject(); - if (in instanceof CheckpointRecovery) { -// CheckpointRecovery cr = (CheckpointRecovery)in; -// path = cr.translatePath(path); -// cr.setState(this, DIR, path); - } - try { - setUp(getDir().getFile(), getCachePercent(), false, getUseSharedCache()); -// for (CachedBdbMap map: bigMaps.values()) { -// map.initialize( -// this.bdbEnvironment, -// null, -// null, -// map.getKeyClass(), -// map.getValueClass(), -// this.classCatalog); -// } - for (DatabasePlusConfig dpc: databases.values()) { - dpc.database = bdbEnvironment.openDatabase(null, - dpc.name, dpc.config.toDatabaseConfig()); - } - } catch (DatabaseException e) { - IOException io = new IOException(); - io.initCause(e); - throw io; - } - this.shutdownHook = new BdbShutdownHook(this); - Runtime.getRuntime().addShutdownHook(shutdownHook); - } - + public void startCheckpoint(Checkpoint checkpointInProgress) {} @SuppressWarnings("unchecked") - public void checkpoint(File dir, List actions) - throws IOException { - if (checkpointCopyLogs) { - actions.add(new BdbRecover(getDir().getFile().getAbsolutePath())); - } + public void doCheckpoint(Checkpoint checkpointInProgress) throws IOException { // First sync objectCaches for (ObjectIdentityCache oic : oiCaches.values()) { oic.sync(); } - EnvironmentConfig envConfig; try { // sync all databases for (DatabasePlusConfig dbc: databases.values()) { dbc.database.sync(); } - envConfig = bdbEnvironment.getConfig(); - } catch (DatabaseException e) { - IOException io = new IOException(); - io.initCause(e); - throw io; - } - final List bkgrdThreads = Arrays.asList(new String [] - {"je.env.runCheckpointer", "je.env.runCleaner", - "je.env.runINCompressor"}); - try { - // Disable background threads - setBdbjeBkgrdThreads(envConfig, bkgrdThreads, "false"); // Do a force checkpoint. Thats what a sync does (i.e. doSync). CheckpointConfig chkptConfig = new CheckpointConfig(); chkptConfig.setForce(true); @@ -531,109 +490,69 @@ Serializable, Closeable { // in the log, and then applying the delta to it. This can be // pretty slow, since it is potentially a large amount of // random I/O." - chkptConfig.setMinimizeRecoveryTime(true); + // chkptConfig.setMinimizeRecoveryTime(true); bdbEnvironment.checkpoint(chkptConfig); LOGGER.fine("Finished bdb checkpoint."); - - // From the sleepycat folks: A trick for flipping db logs. - EnvironmentImpl envImpl = - DbInternal.envGetEnvironmentImpl(bdbEnvironment); - long firstFileInNextSet = - DbLsn.getFileNumber(envImpl.forceLogFileFlip()); - // So the last file in the checkpoint is firstFileInNextSet - 1. - // Write manifest of all log files into the bdb directory. - final String lastBdbCheckpointLog = - getBdbLogFileName(firstFileInNextSet - 1); - processBdbLogs(dir, lastBdbCheckpointLog); - LOGGER.fine("Finished processing bdb log files."); + + DbBackup dbBackup = new DbBackup(bdbEnvironment); + try { + dbBackup.startBackup(); + + File envCpDir = new File(dir.getFile(),checkpointInProgress.getName()); + envCpDir.mkdirs(); + File logfilesList = new File(envCpDir,"jdbfiles.manifest"); + FileUtils.writeLines(logfilesList, + Arrays.asList(dbBackup.getLogFilesInBackupSet())); + LOGGER.fine("Finished processing bdb log files."); + } finally { + dbBackup.endBackup(); + } } catch (DatabaseException e) { - IOException io = new IOException(); - io.initCause(e); - throw io; - } finally { - // Restore background threads. - setBdbjeBkgrdThreads(envConfig, bkgrdThreads, "true"); + throw new IOException(e); } } - - - private void processBdbLogs(final File checkpointDir, - final String lastBdbCheckpointLog) throws IOException { - File bdbDir = getBdbSubDirectory(checkpointDir); - if (!bdbDir.exists()) { - bdbDir.mkdir(); - } - PrintWriter pw = new PrintWriter(new FileOutputStream(new File( - checkpointDir, "bdbje-logs-manifest.txt"))); - try { - // Don't copy any beyond the last bdb log file (bdbje can keep - // writing logs after checkpoint). - boolean pastLastLogFile = false; - Set srcFilenames = null; - do { - FilenameFilter filter = new FilenameFilter() { - public boolean accept(File dir, String name) { - return name != null - && name.toLowerCase().endsWith(".jdb"); - } - }; - - srcFilenames = - new HashSet(Arrays.asList(getDir().getFile().list(filter))); - List tgtFilenames = Arrays.asList(bdbDir.list(filter)); - if (tgtFilenames != null && tgtFilenames.size() > 0) { - srcFilenames.removeAll(tgtFilenames); - } - if (srcFilenames.size() > 0) { - // Sort files. - srcFilenames = new TreeSet(srcFilenames); - int count = 0; - for (final Iterator i = srcFilenames.iterator(); - i.hasNext() && !pastLastLogFile;) { - String name = (String) i.next(); - if (this.checkpointCopyLogs) { - FileUtils.copyDirectory(new File(getDir().getFile(), name), - new File(bdbDir, name)); - } - pw.println(name); - if (name.equals(lastBdbCheckpointLog)) { - // We're done. - pastLastLogFile = true; - } - count++; - } - if (LOGGER.isLoggable(Level.FINE)) { - LOGGER.fine("Copied " + count); - } - } - } while (!pastLastLogFile && srcFilenames != null && - srcFilenames.size() > 0); - } finally { - pw.close(); - } - } - - - private void setBdbjeBkgrdThreads(final EnvironmentConfig config, - final List threads, final String setting) { - for (final Iterator i = threads.iterator(); i.hasNext();) { - config.setConfigParam((String)i.next(), setting); + @SuppressWarnings("unchecked") + protected void doRecover() throws IOException { + File logfilesList = new File( + new File(dir.getFile(),recoveryCheckpoint.getName()), + "jdbfiles.manifest"); + Set retainLogfiles = new HashSet(FileUtils.readLines(logfilesList)); + IOFileFilter filter = FileFilterUtils.orFileFilter( + FileFilterUtils.suffixFileFilter(".jdb"), + FileFilterUtils.suffixFileFilter(".del")); + filter = FileFilterUtils.makeFileOnly(filter); + for(File f : dir.getFile().listFiles((FileFilter)filter)) { + if(retainLogfiles.contains(f.getName())) { + retainLogfiles.remove(f.getName()); + continue; + } + String undelName = f.getName().replace(".del", ".jdb"); + if(retainLogfiles.contains(undelName)) { + if(!f.renameTo(new File(f.getParentFile(),undelName))) { + throw new IOException("Unable to rename " + f + " to " + + undelName); + } + retainLogfiles.remove(f.getName()); + } + // not needed; move aside + org.archive.util.FileUtils.moveAsideIfExists(f); + // TODO: log/warn of ruined later checkpoints? } + if(retainLogfiles.size()>0) { + // some needed files weren't present + LOGGER.severe("needed log files missing: "+retainLogfiles); + } + } - - private String getBdbLogFileName(final long index) { - String lastBdbLogFileHex = Long.toHexString(index); - StringBuffer buffer = new StringBuffer(); - for (int i = 0; i < (8 - lastBdbLogFileHex.length()); i++) { - buffer.append('0'); - } - buffer.append(lastBdbLogFileHex); - buffer.append(".jdb"); - return buffer.toString(); + public void finishCheckpoint(Checkpoint checkpointInProgress) {} + + Checkpoint recoveryCheckpoint; + @Autowired(required=false) + public void setRecoveryCheckpoint(Checkpoint checkpoint) { + this.recoveryCheckpoint = checkpoint; } - public void close() { close2(); @@ -671,12 +590,6 @@ Serializable, Closeable { LOGGER.log(Level.SEVERE, "Error closing environment.", e); } } - - - private static File getBdbSubDirectory(File checkpointDir) { - return new File(checkpointDir, "bdbje-logs"); - } - public Database getDatabase(String name) { DatabasePlusConfig dpc = databases.get(name); @@ -686,26 +599,6 @@ Serializable, Closeable { return dpc.database; } - - private static class BdbRecover implements RecoverAction { - - private static final long serialVersionUID = 1L; - - private String path; - - public BdbRecover(String path) { - this.path = path; - } - - public void recoverFrom(File checkpointDir, - CheckpointRecovery recovery) throws Exception { - File bdbDir = getBdbSubDirectory(checkpointDir); - path = recovery.translatePath(path); - FileUtils.copyDirectory(bdbDir, new File(path)); - } - } - - private static class BdbShutdownHook extends Thread { final private BdbModule bdb; 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 afd32868..1f3659c5 100644 --- a/commons/src/test/java/org/archive/settings/file/BdbModuleTest.java +++ b/commons/src/test/java/org/archive/settings/file/BdbModuleTest.java @@ -19,82 +19,94 @@ package org.archive.settings.file; -import java.io.File; -import java.util.HashMap; -import java.util.Map; -import java.util.concurrent.ConcurrentMap; - import org.apache.commons.io.FileUtils; +import org.apache.commons.lang.math.RandomUtils; import org.archive.bdb.BdbModule; -import org.archive.checkpointing.DefaultCheckpointRecovery; +import org.archive.crawler.framework.Checkpoint; import org.archive.spring.ConfigPath; +import org.archive.util.ObjectIdentityBdbCache; +import org.archive.util.Supplier; import org.archive.util.TmpDirTestCase; /** - * @author pjack + * Test BdbModule. + * + * @contributor pjack + * @contributor gojomo */ public class BdbModuleTest extends TmpDirTestCase { - public void testCheckpoint() throws Exception { - doCheckpoint(); - - File first = new File(getTmpDir(), "first"); -// File checkpointDir = new File(getTmpDir(), "checkpoint"); - - File second = new File(getTmpDir(), "second"); - FileUtils.deleteDirectory(second); - - DefaultCheckpointRecovery cr = new DefaultCheckpointRecovery("job"); - cr.getFileTranslations().put(first.getAbsolutePath(), - second.getAbsolutePath()); -// SheetManager mgr2 = Checkpointer.recover(checkpointDir, cr); -// BdbModule bdb2 = (BdbModule)mgr2.getRoot().get("module"); -// Map testData2 = bdb2.getBigMap("testData", false, -// String.class, String.class); - Map map1 = new HashMap(); - for (int i = 0; i < 1000; i++) { - map1.put(String.valueOf(i), String.valueOf(i * 2)); - } + public void testDoCheckpoint() throws Exception { + ConfigPath basePath = new ConfigPath("testBase",getTmpDir().getAbsolutePath()); + ConfigPath bdbDir = new ConfigPath("bdb","bdb"); + bdbDir.setBase(basePath); + FileUtils.deleteDirectory(bdbDir.getFile()); -// Map map2 = dump(testData2); -// assertEquals(map1, map2); - } - - - private void doCheckpoint() throws Exception { - File first = new File(getTmpDir(), "first"); - FileUtils.deleteDirectory(first); - - File firstState = new File(first, "state"); -// MemorySheetManager mgr = new MemorySheetManager(); - BdbModule bdb = new BdbModule(); -// mgr.getRoot().put("module", bdb); - bdb.setDir(new ConfigPath("test",firstState.getAbsolutePath())); -// mgr.getGlobalSheet().set(bdb, BdbModule.DIR, firstState.getAbsolutePath()); + bdb.setDir(bdbDir); bdb.start(); + + // avoid data from prior runs being mistaken for current run + int randomFactor = RandomUtils.nextInt(); - BdbModule.BdbConfig config = new BdbModule.BdbConfig(); - config.setAllowCreate(true); - bdb.openManagedDatabase("testOpen", config, false); - - ConcurrentMap testData = bdb.getCBMMap("testData", false, - String.class, String.class); - for (int i = 0; i < 1000; i++) { - assertNull("unexpected prior entry", testData.putIfAbsent(String.valueOf(i), String.valueOf(i * 2))); + ObjectIdentityBdbCache testData = + bdb.getOIBCCache("testData", false,String.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; + } + }); + assertSame("unexpected prior entry",value,cached); } - File checkpointDir = new File(getTmpDir(), "checkpoint"); - checkpointDir.mkdirs(); -// Checkpointer.checkpoint(mgr, checkpointDir); + Checkpoint checkpointInProgress = new Checkpoint(); + ConfigPath checkpointsPath = new ConfigPath("checkpoints","checkpoints"); + checkpointsPath.setBase(basePath); + checkpointInProgress.generateFrom(checkpointsPath,998); + + bdb.doCheckpoint(checkpointInProgress); + String firstCheckpointName = checkpointInProgress.getName(); + + 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; + } + }); + assertSame("unexpected prior entry",value,cached); + } + + checkpointInProgress = new Checkpoint(); + checkpointInProgress.generateFrom(checkpointsPath,999); + + bdb.doCheckpoint(checkpointInProgress); + bdb.stop(); + + BdbModule bdb2 = new BdbModule(); + bdb2.setDir(bdbDir); + + Checkpoint recoveryCheckpoint = new Checkpoint(); + ConfigPath recoverPath = new ConfigPath("recover",firstCheckpointName); + recoverPath.setBase(basePath); + recoveryCheckpoint.setCheckpointDir(recoverPath); + recoveryCheckpoint.afterPropertiesSet(); + + bdb2.setRecoveryCheckpoint(recoveryCheckpoint); + + bdb2.start(); + + ObjectIdentityBdbCache restoreData = + bdb2.getOIBCCache("testData",true,String.class); + + assertEquals("unexpected size", 1000, restoreData.size()); + assertEquals("unexpected value",randomFactor*999,Integer.parseInt(restoreData.get(""+999))); + + bdb2.stop(); } - -// private Map dump(Map src) { -// HashMap dest = new HashMap(); -// for (String k: src.keySet()) { -// dest.put(k, src.get(k)); -// } -// return dest; -// } }