/*-
* See the file LICENSE for redistribution information.
*
* Copyright (c) 2002-2005
*      Sleepycat Software.  All rights reserved.
*
* $Id: Checkpointer.java,v 1.100 2004/12/22 14:11:35 linda Exp $
*/

package com.sleepycat.je.recovery;

import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.logging.Level;

import com.sleepycat.je.CheckpointConfig;
import com.sleepycat.je.DatabaseException;
import com.sleepycat.je.DbInternal;
import com.sleepycat.je.EnvironmentStats;
import com.sleepycat.je.StatsConfig;
import com.sleepycat.je.cleaner.Cleaner;
import com.sleepycat.je.cleaner.TrackedFileSummary;
import com.sleepycat.je.cleaner.UtilizationProfile;
import com.sleepycat.je.config.EnvironmentParams;
import com.sleepycat.je.dbi.DatabaseImpl;
import com.sleepycat.je.dbi.DbConfigManager;
import com.sleepycat.je.dbi.DbTree;
import com.sleepycat.je.dbi.EnvironmentImpl;
import com.sleepycat.je.dbi.INList;
import com.sleepycat.je.dbi.MemoryBudget;
import com.sleepycat.je.log.LogManager;
import com.sleepycat.je.tree.BIN;
import com.sleepycat.je.tree.ChildReference;
import com.sleepycat.je.tree.IN;
import com.sleepycat.je.tree.Key;
import com.sleepycat.je.tree.Node;
import com.sleepycat.je.tree.SearchResult;
import com.sleepycat.je.tree.Tree;
import com.sleepycat.je.tree.WithRootLatched;
import com.sleepycat.je.utilint.DaemonThread;
import com.sleepycat.je.utilint.DbLsn;
import com.sleepycat.je.utilint.PropUtil;
import com.sleepycat.je.utilint.Tracer;

/**
 * The Checkpointer looks through the tree for internal nodes that must be
 * flushed to the log. Checkpoint flushes must be done in ascending order
 * from the bottom of the tree up.
 */
public class Checkpointer extends DaemonThread {

    private EnvironmentImpl envImpl;
    private LogManager logManager;

    /* Checkpoint sequence, initialized at recovery. */
    private long checkpointId;  

    /* 
     * How much the log should grow between checkpoints. If 0, we're
     * using time based checkpointing.
     */
    private long logSizeBytesInterval;
    private long logFileMax;
    private long timeInterval;
    private long lastCheckpointMillis;

    /* Stats */
    private int nCheckpoints;
    private long lastFirstActiveLsn;
    private long lastCheckpointStart;
    private long lastCheckpointEnd;

    private int nFullINFlush;
    private int nFullBINFlush;
    private int nDeltaINFlush;
    private int nFullINFlushThisRun;
    private int nDeltaINFlushThisRun;

    /* For future addition to stats:   private int nAlreadyEvictedThisRun;  */

    private volatile int highestFlushLevel;

    public Checkpointer(EnvironmentImpl envImpl,
                        long waitTime,
                        String name) 
        throws DatabaseException {

        super(waitTime, name, envImpl);
        this.envImpl = envImpl;
        logSizeBytesInterval = 
            envImpl.getConfigManager().getLong(
                                               EnvironmentParams.CHECKPOINTER_BYTES_INTERVAL);
        logFileMax = 
            envImpl.getConfigManager().getLong(EnvironmentParams.LOG_FILE_MAX);
        nCheckpoints = 0;
        timeInterval = waitTime;
        lastCheckpointMillis = 0;
	highestFlushLevel = IN.MIN_LEVEL;
        logManager = envImpl.getLogManager();
    }

    public int getHighestFlushLevel() {
	return highestFlushLevel;
    }

    /**
     * Figure out the wakeup period. Supplied through this static method
     * because we need to pass wakeup period to the superclass and
     * need to do the calcuation outside this constructor.
     */
    public static long getWakeupPeriod(DbConfigManager configManager) 
        throws IllegalArgumentException, DatabaseException {

        long wakeupPeriod = PropUtil.microsToMillis(configManager.getLong
                                                    (EnvironmentParams.CHECKPOINTER_WAKEUP_INTERVAL));
        long bytePeriod = configManager.getLong
            (EnvironmentParams.CHECKPOINTER_BYTES_INTERVAL);

        /* Checkpointing period must be set either by time or by log size. */
        if ((wakeupPeriod == 0) && (bytePeriod == 0)) {
            throw new IllegalArgumentException(
                                               EnvironmentParams.CHECKPOINTER_BYTES_INTERVAL.getName() +
                                               " and " +
                                               EnvironmentParams.CHECKPOINTER_WAKEUP_INTERVAL.getName() +
                                               " cannot both be 0. ");
        }

        /*
         * Checkpointing by log size takes precendence over time based period
         */
        if (bytePeriod == 0) {
            return wakeupPeriod;
        } else {
            return 0;
        }
    }

    /**
     * Set checkpoint id -- can only be done after recovery.
     */
    synchronized public void setCheckpointId(long lastCheckpointId) {
        checkpointId = lastCheckpointId;
    }

    public String toString() {
        StringBuffer sb = new StringBuffer();
        sb.append("<Checkpointer name=\"").append(name).append("\"/>");
        return sb.toString();
    }

    /**
     * Load stats.
     */
    public void loadStats(StatsConfig config, EnvironmentStats stat) 
        throws DatabaseException {

        stat.setNCheckpoints(nCheckpoints);
        stat.setLastCheckpointStart(lastCheckpointStart);
        stat.setLastCheckpointEnd(lastCheckpointEnd);
        stat.setLastCheckpointId(checkpointId);
        stat.setNFullINFlush(nFullINFlush);
        stat.setNFullBINFlush(nFullBINFlush);
        stat.setNDeltaINFlush(nDeltaINFlush);
        
        if (config.getClear()) {
            nCheckpoints = 0;
            nFullINFlush = 0;
            nFullBINFlush = 0;
            nDeltaINFlush = 0;
        }
    }
    
    /**
     * @return the first active LSN point of the last completed checkpoint.
     * If no checkpoint has run, return null.
     */
    public long getFirstActiveLsn() {
        return lastFirstActiveLsn;
    }

    synchronized public void clearEnv() {
        envImpl = null;
    }

    /**
     * Return the number of retries when a deadlock exception occurs.
     */
    protected int nDeadlockRetries()
        throws DatabaseException {

        return envImpl.getConfigManager().getInt
            (EnvironmentParams.CHECKPOINTER_RETRY);
    }

    /**
     * Called whenever the DaemonThread wakes up from a sleep.  
     */
    protected void onWakeup()
        throws DatabaseException {

        if (envImpl.isClosed()) {
            return;
        }

        doCheckpoint(CheckpointConfig.DEFAULT,
                     true,  // allowDeltas
                     false, // flushAll
                     "daemon");
    }

    /**
     * Determine whether a checkpoint should be run.
     * 1. If the force parameter is specified, always checkpoint. 
     * 2. If the config object specifies time or log size, use that.
     * 3. If the environment is configured to use log size based checkpointing,
     * check the log.
     * 4. Lastly, use time based checking.
     */
    private boolean isRunnable(CheckpointConfig config)
        throws DatabaseException {

        /* Figure out if we're using log size or time to determine interval.*/
        long useBytesInterval = 0;
        long useTimeInterval = 0;
        long nextLsn = DbLsn.NULL_LSN;
        try {
            if (config.getForce()) {
                return true;
            } else if (config.getKBytes() != 0) {
                useBytesInterval = config.getKBytes() << 10;
            } else if (config.getMinutes() != 0) {
                // convert to millis
                useTimeInterval = config.getMinutes() * 60 * 1000;
            } else if (logSizeBytesInterval != 0) {
                useBytesInterval = logSizeBytesInterval;
            } else {
                useTimeInterval = timeInterval;
            }

            /* 
             * If our checkpoint interval is defined by log size, check
             * on how much log has grown since the last checkpoint.
             */
            if (useBytesInterval != 0) {
                nextLsn = envImpl.getFileManager().getNextLsn();
                if (DbLsn.getNoCleaningDistance(nextLsn, lastCheckpointEnd,
						logFileMax) >=
                    useBytesInterval) {
                    return true;
                } else {
                    return false;
                }
            } else if (useTimeInterval != 0) {

                /* 
                 * Our checkpoint is determined by time.  If enough
                 * time has passed and some log data has been written,
                 * do a checkpoint.
                 */
                long lastUsedLsn = envImpl.getFileManager().getLastUsedLsn();
                if (((System.currentTimeMillis() - lastCheckpointMillis) >=
                     useTimeInterval) &&
                    (DbLsn.compareTo(lastUsedLsn, lastCheckpointEnd) != 0)) {
                    return true;
                } else {
                    return false;
                }
            } else {
                return false;
            }
        } finally {
            StringBuffer sb = new StringBuffer();
            sb.append("size interval=").append(useBytesInterval);
            if (nextLsn != DbLsn.NULL_LSN) {
                sb.append(" nextLsn=").
		    append(DbLsn.getNoFormatString(nextLsn));
            }
            if (lastCheckpointEnd != DbLsn.NULL_LSN) {
                sb.append(" lastCkpt=");
                sb.append(DbLsn.getNoFormatString(lastCheckpointEnd));
            }
            sb.append(" time interval=").append(useTimeInterval);
            sb.append(" force=").append(config.getForce());
            
            Tracer.trace(Level.FINEST,
                         envImpl, 
                         sb.toString());
        }
    }

    /**
     * The real work to do a checkpoint. This may be called by the checkpoint
     * thread when waking up, or it may be invoked programatically through the
     * api.
     * 
     * @param allowDeltas if true, this checkpoint may opt to log BIN deltas
     *       instead of the full node.
     * @param flushAll if true, this checkpoint must flush all the way to
     *       the top of the dbtree, instead of stopping at the highest level
     *       last modified.
     * @param invokingSource a debug aid, to indicate who invoked this 
     *       checkpoint. (i.e. recovery, the checkpointer daemon, the cleaner,
     *       programatically)
     */
    public synchronized void doCheckpoint(CheckpointConfig config,
					  boolean allowDeltas,
					  boolean flushAll,
					  String invokingSource) 
        throws DatabaseException {

	if (!isRunnable(config)) {
	    return;
	}

        /*
         * If there are cleaned files to be deleted, flush an extra level to
         * write out the parents of cleaned nodes.  This ensures that the
         * no node will contain the LSN of a cleaned files.
         */
        boolean flushExtraLevel = false;
        Set cleanedFiles = null;
        Cleaner cleaner = envImpl.getCleaner();
        if (cleaner != null) {
            cleanedFiles = cleaner.getCleanedFiles();
            if (cleanedFiles != null) {
                flushExtraLevel = true;
            }
        }

        lastCheckpointMillis = System.currentTimeMillis();
        resetPerRunCounters();

        /* Get the next checkpoint id. */
        checkpointId++;
        nCheckpoints++;
        boolean success = false;
        boolean traced = false;
        try {

	    /* 
	     * Eviction can run during checkpoint as long as it follows the
	     * same rules for using provisional logging and for propagating
	     * logging of the checkpoint dirty set up the tree. We have to lock
	     * out the evictor after the logging of checkpoint start until
	     * we've selected the dirty set and decided on the highest level to
	     * be flushed. See SR 11163, 11349.
	     */
	    long checkpointStart = DbLsn.NULL_LSN;
	    long firstActiveLsn = DbLsn.NULL_LSN;
	    SortedMap dirtyMap = null;
	    synchronized (envImpl.getEvictor()) {
		/* Log the checkpoint start. */
		CheckpointStart startEntry =
		    new CheckpointStart(checkpointId, invokingSource);
		checkpointStart = logManager.log(startEntry);

		/* 
		 * Remember the first active LSN -- before this position in the
		 * log, there are no active transactions at this point in time.
		 */
		firstActiveLsn = envImpl.getTxnManager().getFirstActiveLsn();

		if (firstActiveLsn == DbLsn.NULL_LSN) {
		    firstActiveLsn = checkpointStart;
		} else {
		    if (DbLsn.compareTo(checkpointStart, firstActiveLsn) < 0) {
			firstActiveLsn = checkpointStart;
		    }
		}
		/* Find the dirty set. */
		dirtyMap = selectDirtyINs(flushAll, flushExtraLevel);
	    }

            /* Flush IN nodes. */
            flushDirtyNodes(dirtyMap, flushAll, allowDeltas,
                            flushExtraLevel, checkpointStart);

            /*
             * Flush utilization info AFTER flushing IN nodes to reduce the
             * inaccuracies caused by the sequence FileSummaryLN-LN-BIN.
             */
            flushUtilizationInfo();

            CheckpointEnd endEntry =
                new CheckpointEnd(invokingSource,
                                  checkpointStart,
                                  envImpl.getRootLsn(),
                                  firstActiveLsn,
                                  Node.getLastId(),
                                  envImpl.getDbMapTree().getLastDbId(),
                                  envImpl.getTxnManager().getLastTxnId(),
                                  checkpointId);

            /* 
             * Log checkpoint end and update state kept about the last
             * checkpoint location. Send a trace message *before* the
             * checkpoint end log entry. This is done  so that the  normal
             * trace  message doesn't affect the time-based isRunnable()
             * calculation, which only issues a checkpoint if a log record
             * has been written since the last checkpoint.
             */
            trace(envImpl, invokingSource, true);
            traced = true;

            /*
             * Always flush to ensure that cleaned files are not referenced,
             * and to ensure that this checkpoint is not wasted if we crash.
             */
            lastCheckpointEnd = 
            	logManager.logForceFlush(endEntry,
            	                 	 true); // fsync required
            lastFirstActiveLsn = firstActiveLsn;
            lastCheckpointStart = checkpointStart;

	    /* 
	     * Reset the highestFlushLevel so evictor activity knows there's no
	     * further requirement for provisional logging. SR 11163.
	     */
	    highestFlushLevel = IN.MIN_LEVEL;

            success = true;

            if (cleaner != null && cleanedFiles != null) {
                cleaner.deleteCleanedFiles(cleanedFiles);
            }
        } catch (DatabaseException e) {
            Tracer.trace(envImpl, "Checkpointer", "doCheckpoint",
                         "checkpointId=" + checkpointId, e);
            throw e;
        } finally {
            if (!traced) {
                trace(envImpl, invokingSource, success);
            }
        }
    }

    private void trace(EnvironmentImpl envImpl,
                       String invokingSource, boolean success ) {
        StringBuffer sb = new StringBuffer();
        sb.append("Checkpoint ").append(checkpointId);
        sb.append(": source=" ).append(invokingSource);
        sb.append(" success=").append(success);
        sb.append(" nFullINFlushThisRun=").append(nFullINFlushThisRun);
        sb.append(" nDeltaINFlushThisRun=").append(nDeltaINFlushThisRun);
        Tracer.trace(Level.INFO, envImpl, sb.toString());
    }

    /**
     * Flush a FileSummaryLN node for each TrackedFileSummary that is
     * currently active.  Tell the UtilizationProfile about the updated file
     * summary.
     */
    private void flushUtilizationInfo()
        throws DatabaseException {

        if (!DbInternal.getCheckpointUP
	    (envImpl.getConfigManager().getEnvironmentConfig())) {
            return; // only disabled for unittests
        }
        
        UtilizationProfile profile = envImpl.getUtilizationProfile();

        TrackedFileSummary[] activeFiles =
            envImpl.getUtilizationTracker().getTrackedFiles();

        for (int i = 0; i < activeFiles.length; i += 1) {
            TrackedFileSummary trackedFile = activeFiles[i];
            profile.putFileSummary(trackedFile);
        }
    }

    /**
     * Flush the nodes in order, from the lowest level to highest level.  As a
     * flush dirties its parent, add it to the dirty map, thereby cascading the
     * writes up the tree. If flushAll wasn't specified, we need only cascade
     * up to the highest level set at the start of checkpointing.
     *
     * Note that all but the top level INs and the BINDeltas are logged
     * provisionally. That's because we don't need to process lower INs because
     * the higher INs will end up pointing at them.
     */
    private void flushDirtyNodes(SortedMap dirtyMap,
				 boolean flushAll,
				 boolean allowDeltas,
                                 boolean flushExtraLevel,
                                 long checkpointStart)
        throws DatabaseException {

        while (dirtyMap.size() > 0) {

            /* Work on one level's worth of nodes in ascending level order. */
            Integer currentLevel = (Integer) dirtyMap.firstKey();
            boolean logProvisionally =
                (currentLevel.intValue() != highestFlushLevel);

            Set nodeSet = (Set) dirtyMap.get(currentLevel);
            Iterator iter = nodeSet.iterator();

            /* 
             * Add this level's references to the budget. To be more precise,
             * we should update the memory budget when we create the dirty
             * map, but it's easier to do it by level here.
             */
            MemoryBudget mb = envImpl.getMemoryBudget();
            long levelSize = nodeSet.size() *
                MemoryBudget.CHECKPOINT_REFERENCE_SIZE;            
            mb.updateCacheMemoryUsage(levelSize);
                
            /* Flush all those nodes */
            while (iter.hasNext()) {
                CheckpointReference targetRef =
                    (CheckpointReference) iter.next();

                /* 
                 * Check if the db is still valid since INs of deleted
                 * databases are left on the in-memory tree until the post
                 * commit processing of a Environment.removeDatabase occurs.
                 */
                if (!(targetRef.db.getIsDeleted())) {
                    flushIN(targetRef, dirtyMap, currentLevel.intValue(),
                            logProvisionally, allowDeltas, checkpointStart);
                }

                iter.remove();
            }

            /* We're done with this level. */
            dirtyMap.remove(currentLevel);
            mb.updateCacheMemoryUsage(0 - levelSize);

            /* We can stop at this point. */
            if (currentLevel.intValue() == highestFlushLevel) {
                break;
            }
        }
    }

    /**
     * Scan the INList for all dirty INs. Arrange them in level sorted 
     * map for level ordered flushing.
     */
    private SortedMap selectDirtyINs(boolean flushAll,
                                     boolean flushExtraLevel) 
        throws DatabaseException {

        SortedMap newDirtyMap = new TreeMap();

        INList inMemINs = envImpl.getInMemoryINs();
        inMemINs.latchMajor();

        /* 
	 * Opportunistically recalculate the environment wide memory
	 * count.  Incurs no extra cost because we're walking the IN
	 * list anyway.  Not the best in terms of encapsulation as
	 * prefereably all memory calculations are done in
	 * MemoryBudget, but done this way to avoid any extra
	 * latching.
	 */
        long totalSize = 0;
        MemoryBudget mb = envImpl.getMemoryBudget();

        try {
            Iterator iter = inMemINs.iterator();
            while (iter.hasNext()) {
                IN in = (IN) iter.next();
                in.latch();
                try {
                    totalSize = mb.accumulateNewUsage(in, totalSize);

                    if (in.getDirty()) {
                        Integer level = new Integer(in.getLevel());
                        Set dirtySet;
                        if (newDirtyMap.containsKey(level)) {
                            dirtySet = (Set) newDirtyMap.get(level);
                        } else {
                            dirtySet = new HashSet();
                            newDirtyMap.put(level, dirtySet);
                        }
                        dirtySet.add(
                             new CheckpointReference(in.getDatabase(),
                                                     in.getNodeId(),
                                                     in.containsDuplicates(),
                                                     in.isDbRoot(),
                                                     in.getMainTreeKey(),
                                                     in.getDupTreeKey()));
                    }
                } finally {
                    in.releaseLatch();
                }
            }

            // Later release: refresh env count.
            // mb.refreshCacheMemoryUsage(totalSize);

            /* 
             * If we're flushing all for cleaning, we must flush to the point
             * that there are no nodes with LSNs in the cleaned files. 
             */
            if (newDirtyMap.size() > 0) {
                if (flushAll) {
                    highestFlushLevel =
			envImpl.getDbMapTree().getHighestLevel();
                } else {
                    highestFlushLevel =
                        ((Integer) newDirtyMap.lastKey()).intValue();
                    if (flushExtraLevel) {
                        highestFlushLevel += 1;
                    }
                }
            } else {
		highestFlushLevel = IN.MAX_LEVEL;
	    }
        } finally {
            inMemINs.releaseMajorLatchIfHeld();
        }

        return newDirtyMap;
    }
    
    /** 
     * Flush the target IN.
     */
    private void flushIN(CheckpointReference targetRef,
                         Map dirtyMap,
                         int currentLevel,
                         boolean logProvisionally,
                         boolean allowDeltas,
                         long checkpointStart)
        throws DatabaseException {

        Tree tree = targetRef.db.getTree();
        boolean targetWasRoot = false;
        if (targetRef.isDbRoot) {
            /* We're trying to flush the root. */
            RootFlusher flusher = new RootFlusher(targetRef.db,
                                                  logManager,
                                                  targetRef.nodeId);
            tree.withRootLatched(flusher);
            boolean flushed = flusher.getFlushed();

            /* 
             * If this target isn't the root anymore, we'll have to
             * handled it like a regular node.
             */
            targetWasRoot = flusher.stillRoot();
            
            /* 
             * Update the tree's owner, whether it's the env root or the
             * dbmapping tree.
             */
            if (flushed) {
                DbTree dbTree = targetRef.db.getDbEnvironment().getDbMapTree();
                dbTree.modifyDbRoot(targetRef.db);
                nFullINFlushThisRun++;
                nFullINFlush++;
            }
        }

        /* 
         * The following attempt to flush applies to two cases:
         * (1) the target was not ever the root
         * (2) the target was the root, when the checkpoint dirty set was 
         * assembled but is not the root now.
         */
        if (!targetWasRoot) {
            /*
             * The "isRoot" param is used to stop a search in
             * BIN.descendOnParentSearch and is passed as false (never stop).
             */ 
            SearchResult result =
                tree.getParentINForChildIN(targetRef.nodeId,
                                           targetRef.containsDuplicates,
                                           false,  // isRoot
                                           targetRef.mainTreeKey,
                                           targetRef.dupTreeKey,
                                           false,  // exact match
                                           null,   // tracking list
                                           false); // doFetch

            /* 
             * We must make sure that every IN that was selected for the 
             * checkpointer's dirty IN set at the beginning of checkpoint is
             * written into the log and can be properly accessed from 
             * ancestors. However, we have to take care for cases where the
             * evictor has written out a member of this dirty set before the
             * checkpointer got to it. See SR 10249.
             *
             * If no possible parent is found, the compressor may have deleted
             * this item before we got to processing it.
             */
            if (result.parent != null) {
                boolean mustLogParent = false;
                try {
                    if (result.exactParentFound) {
                        /* 
                         * If the child has already been evicted, don't
                         * refetch it.
                         */
                        IN renewedTarget =
                            (IN) result.parent.getTarget(result.index);

                        if (renewedTarget == null) {
                            /* nAlreadyEvictedThisRun++;  -- for future */
                            mustLogParent = true;
                        } else {
                            mustLogParent =
                                logTargetAndUpdateParent(renewedTarget,
                                                         result.parent,
                                                         result.index,
                                                         allowDeltas,
                                                         checkpointStart,
                                                         logProvisionally);
                        }
                    } else {
                        /* result.exactParentFound was false. */
                        if (result.childNotResident) {

                            /* 
                             * But it was because the child wasn't resident.
                             * To be on the safe side, we'll put the parent 
                             * into the dirty set to be logged when that level
                             * is processed.
                             *
                             * Only do this if the parent we found is at a
                             * higher level that the child.  This ensures that
                             * the non-exact search does not find a sibling
                             * rather than a parent. [#11555]
                             */
                            if (result.parent.getLevel() > currentLevel) {
                                mustLogParent = true;
                            }
                            /* nAlreadyEvictedThisRun++; -- for future. */
                        }
                    }

                    if (mustLogParent) {
                        addToDirtyMap(dirtyMap, result.parent);
                    }
                } finally {
                    result.parent.releaseLatch();
                }
            }
        }
    }

    private boolean logTargetAndUpdateParent(IN target,
                                             IN parent,
                                             int index,
                                             boolean allowDeltas,
                                             long checkpointStart,
                                             boolean logProvisionally) 
        throws DatabaseException {

        target.latch();
        long newLsn = DbLsn.NULL_LSN;
        boolean mustLogParent = true;
        try {
            if (target.getDirty()) {
                if (allowDeltas) {

                    /* 
                     * Note that target decides whether to log a delta. Only
                     * BINs that fall into the required percentages and have
                     * not been cleaned will be logged with a delta.
                     */
                    newLsn = target.logAllowDeltas(logManager,
                                                   logProvisionally);
                    if (newLsn == DbLsn.NULL_LSN) {
                        nDeltaINFlushThisRun++;
                        nDeltaINFlush++;

                        /* 
                         * If this BIN was already logged after checkpoint
                         * start and before this point (i.e. by an eviction),
                         * we must make sure that the last full version is
                         * accessible from ancestors. We can skip logging
                         * parents only if this is the first logging of this
                         * node in the checkpoint interval.
                         */
                        long lastFullLsn =  target.getLastFullVersion();
                        if (DbLsn.compareTo(lastFullLsn,
                                            checkpointStart) < 0) {
                            mustLogParent = false;
                        }
                    }
                } else {
                    newLsn = target.log(logManager, logProvisionally);
                }
            }
        } finally {
            target.releaseLatch();
        }

        /* Update parent if logging occurred */
        if (newLsn != DbLsn.NULL_LSN) {
            nFullINFlushThisRun++;
            nFullINFlush++;
            if (target instanceof BIN) {
                nFullBINFlush++;
            }
            parent.updateEntry(index, newLsn);
        }
        
        return mustLogParent;
    }

    /*
     * RootFlusher lets us write out the root IN within the root latch.
     */
    private static class RootFlusher implements WithRootLatched {
        private DatabaseImpl db;
        private boolean flushed;
        private boolean stillRoot;
        private LogManager logManager;
        private long targetNodeId;

        RootFlusher(DatabaseImpl db,
                    LogManager logManager,
                    long targetNodeId) {
            this.db = db;
            flushed = false;
            this.logManager = logManager;
            this.targetNodeId = targetNodeId;
            stillRoot = false;
        }

        /**
         * Flush the rootIN if dirty.
         */
        public IN doWork(ChildReference root) 
            throws DatabaseException {

	    if (root == null) {
		return null;
	    }
            IN rootIN = (IN) root.fetchTarget(db, null);
            rootIN.latch();
            try {
                if (rootIN.getNodeId() == targetNodeId) {

                    /* 
		     * stillRoot handles the situation where the root was
		     * split after it was place in the checkpointer's dirty
		     * set.
                     */
                    stillRoot = true;
                    if (rootIN.getDirty()) {
                        long newLsn = rootIN.log(logManager);
                        root.setLsn(newLsn);
                        flushed = true;
                    }
                }
            } finally {
                rootIN.releaseLatch();
            }                    
            return null;
        }

        boolean getFlushed() {
            return flushed;
        }

        boolean stillRoot() {
            return stillRoot;
        }
    }

    /**
     * Add a node to the dirty map. The dirty map is keyed by level (Integers)
     * and holds sets of IN references.
     */
    private void addToDirtyMap(Map dirtyMap, IN in) {
        Integer inLevel = new Integer(in.getLevel());
        Set inSet = (Set) dirtyMap.get(inLevel);
        
        /* If this level doesn't exist in the map yet, make a new entry. */
        if (inSet == null) {
            inSet = new HashSet();
            dirtyMap.put(inLevel, inSet);
        }    
        
        /* Add to the set. */
        inSet.add(new CheckpointReference(in.getDatabase(),
                                          in.getNodeId(),
                                          in.containsDuplicates(),
                                          in.isDbRoot(),
                                          in.getMainTreeKey(),
                                          in.getDupTreeKey()));
    }

    /**
     * Reset per-run counters.
     */
    private void resetPerRunCounters() {
        nFullINFlushThisRun = 0;
        nDeltaINFlushThisRun = 0;
        /* nAlreadyEvictedThisRun = 0; -- for future */
    }

    /* 
     * CheckpointReferences are used to identify nodes that must be flushed as
     * part of the checkpoint. We don't keep an actual reference to the node
     * because that prevents nodes from being GC'ed during checkpoint.
     *
     * Using a checkpointReference introduces a window between the point when
     * the checkpoint dirty set is created and when the node is flushed. Some
     * of the fields saved in the reference are immutable: db, nodeId,
     * containsDuplicates. The others are not and we have to handle potential 
     * change:
     *
     * isDbRoot: it's possible for isDbRoot to go from true->false, but not 
     *         false->true. True->false is handled by the flushIN method
     *         by finding the root and checking if it is the target.
     * mainTreeKey, dupTreeKey: These can change only in the event of a 
     *         split. If they do, there is the chance that the checkpointer
     *         will find the wrong node to flush, but that's okay because
     *         the split guarantees flushing to the root, so the target will
     *         be properly logged within the checkpoint period.
     */
    static class CheckpointReference {
        DatabaseImpl db;
        long nodeId;
        boolean containsDuplicates;
        boolean isDbRoot;
        Key mainTreeKey;
        Key dupTreeKey;

        CheckpointReference(DatabaseImpl db,
                            long nodeId,
                            boolean containsDuplicates,
                            boolean isDbRoot,
                            Key mainTreeKey,
                            Key dupTreeKey) {
            this.db = db;
            this.nodeId = nodeId;
            this.containsDuplicates = containsDuplicates;
            this.isDbRoot = isDbRoot;
            this.mainTreeKey = mainTreeKey;
            this.dupTreeKey = dupTreeKey;
        }

        public boolean equals(Object o) {
            if (!(o instanceof CheckpointReference)) {
                return false;
            }

            CheckpointReference other = (CheckpointReference) o;
            return nodeId == other.nodeId;
        }

        public int hashCode() {
            return (int) nodeId;
        }
    }
}
