cm.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

/*
 *  $Id: Cm.java,v 1.147 2006/05/30 21:15:03 hamada Exp $
 *
 *  Copyright (c) 2001 Sun Microsystems, Inc.  All rights reserved.
 *
 *  Redistribution and use in source and binary forms, with or without
 *  modification, are permitted provided that the following conditions
 *  are met:
 *
 *  1. Redistributions of source code must retain the above copyright
 *  notice, this list of conditions and the following disclaimer.
 *
 *  2. Redistributions in binary form must reproduce the above copyright
 *  notice, this list of conditions and the following disclaimer in
 *  the documentation and/or other materials provided with the
 *  distribution.
 *
 *  3. The end-user documentation included with the redistribution,
 *  if any, must include the following acknowledgment:
 *  "This product includes software developed by the
 *  Sun Microsystems, Inc. for Project JXTA."
 *  Alternately, this acknowledgment may appear in the software itself,
 *  if and wherever such third-party acknowledgments normally appear.
 *
 *  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA"
 *  must not be used to endorse or promote products derived from this
 *  software without prior written permission. For written
 *  permission, please contact Project JXTA at http://www.jxta.org.
 *
 *  5. Products derived from this software may not be called "JXTA",
 *  nor may "JXTA" appear in their name, without prior written
 *  permission of Sun.
 *
 *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 *  WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 *  OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 *  DISCLAIMED.  IN NO EVENT SHALL SUN MICROSYSTEMS OR
 *  ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 *  SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 *  LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
 *  USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 *  ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 *  OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 *  OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 *  SUCH DAMAGE.
 *
 *  ====================================================================
 *
 *  This software consists of voluntary contributions made by many
 *  individuals on behalf of Project JXTA.  For more
 *  information on Project JXTA, please see
 *  <http://www.jxta.org/>.
 *
 *  This license is based on the BSD license adopted by the Apache Foundation.
 *  $Id: Cm.java,v 1.147 2006/05/30 21:15:03 hamada Exp $
 */
package net.jxta.impl.cm;


import java.io.InputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.StringWriter;
import java.lang.reflect.UndeclaredThrowableException;
import java.math.BigInteger;
import java.net.URI;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Date;
import java.util.Vector;
import java.util.ResourceBundle;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import net.jxta.discovery.DiscoveryService;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.Element;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.StructuredTextDocument;
import net.jxta.protocol.SrdiMessage;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PeerGroupAdvertisement;

import net.jxta.impl.util.JxtaHash;
import net.jxta.impl.util.TimeUtils;

import net.jxta.impl.xindice.core.data.Key;
import net.jxta.impl.xindice.core.data.Record;
import net.jxta.impl.xindice.core.data.Value;
import net.jxta.impl.xindice.core.DBException;
import net.jxta.impl.xindice.core.filer.BTreeCallback;
import net.jxta.impl.xindice.core.filer.BTreeFiler;
import net.jxta.impl.xindice.core.filer.BTreeException;
import net.jxta.impl.xindice.core.indexer.IndexQuery;
import net.jxta.impl.xindice.core.indexer.NameIndexer;


/**
 * This class implements a limited document caching mechanism
 * intended to provide cache for services that have a need for cache
 * to search and exchange jxta documents.
 *
 * Only Core Services are intended to use this mechanism.
 */
public final class Cm implements Runnable {

    /**
     * Log4J Logger
     */
    private final static Logger LOG = Logger.getLogger(Cm.class.getName());

    /**
     * the name we will use for the base directory
     */
    final File ROOTDIRBASE;

    /**
     *  adv types
     */
    private static final String[] DIRNAME = { "Peers", "Groups", "Adv", "Raw" };

    // garbage collect once an hour
    public static final long DEFAULT_GC_MAX_INTERVAL = 1 * TimeUtils.ANHOUR;

    /*
     *  expiration db
     */
    private BTreeFiler cacheDB = null;
    private Indexer indexer = null;
    private final static String databaseFileName = "advertisements";

    private boolean stop = false;

    private boolean trackDeltas = false;
    private final Map deltaMap = new HashMap(3);

    /**
     * file descriptor for the root of the cm
     */
    protected File rootDir;

    private ThreadGroup threads = null;
    private Thread gcThread = null;
    private long gcTime = 0;
    private long gcMinInterval = 1000L * 60L;
    private long gcMaxInterval = DEFAULT_GC_MAX_INTERVAL;
    private int maxInconvenienceLevel = 1000;
    private volatile int inconvenienceLevel = 0;

   /**
    * @deprecated Legacy constructor for bench test MakeAdvertisements.java
    */
    public Cm(String areaName, boolean optimize) {
        this(new File(System.getProperty( "JXTA_HOME", ".jxta") ).toURI(), areaName);
    }

    /**
      * Constructor for cm
      *
      * @param  areaName        the name of the cm sub-dir to create
      *
      * NOTE: Default garbage interval once an hour
      */
    public Cm(URI storeRoot, String areaName) {
         // Default garbage collect once an hour
        this(Thread.currentThread().getThreadGroup(), storeRoot, areaName, DEFAULT_GC_MAX_INTERVAL, false);
     }

    /**
     * Constructor for cm
     *
     * @param  threads        the thread group
     * @param  storeRoot      persistance location
     * @param  gcinterval      garbage collect max interval
     * @param  trackDeltas     when true deltas are tracked
     */
    public Cm(ThreadGroup threads,
              URI storeRoot,
              String areaName,
              long gcinterval,
              boolean trackDeltas) {

        this.threads = threads;

        this.trackDeltas = trackDeltas;
        this.gcMaxInterval = gcinterval;
        this.gcTime = System.currentTimeMillis() + gcMaxInterval;

        ROOTDIRBASE = new File( new File( storeRoot), "cm" );

        try {
            rootDir = new File(ROOTDIRBASE, areaName);
            rootDir = new File(rootDir.getAbsolutePath());
            if (!rootDir.exists()) {
                // We need to create the directory
                if (!rootDir.mkdirs()) {
                    throw new RuntimeException("Cm cannot create directory " + rootDir);
                }
            }

            /*
             * to avoid inconsistent database state, it is highly recommended that
             * checkpoint is true by default, which causes fd.sync() on every write
             * operation.  In transitory caches such as SrdiCache it makes perfect sense
             */
            boolean chkPoint = true;
            ResourceBundle jxtaRsrcs = ResourceBundle.getBundle("net.jxta.user");
            String checkpointStr = jxtaRsrcs.getString("impl.cm.defferedcheckpoint");

            if (checkpointStr != null) {
                chkPoint = (checkpointStr.equalsIgnoreCase("true")) ? false : true;
            }

            // Storage
            cacheDB = new BTreeFiler();
            // no deffered checkpoint
            cacheDB.setSync(chkPoint);
            cacheDB.setLocation(rootDir.getAbsolutePath(), databaseFileName);

            if (!cacheDB.open()) {
                cacheDB.create();
                // now open it
                cacheDB.open();
            }

            // Index
            indexer = new Indexer(chkPoint);
            indexer.setLocation(rootDir.getAbsolutePath(), databaseFileName);

            if (!indexer.open()) {
                indexer.create();
                // now open it
                indexer.open();
            }

            if (System.getProperty("net.jxta.impl.cm.index.rebuild") != null) {
                rebuildIndex();
            }
            gcThread = new Thread(threads, this, "CM GC Thread interval : " + gcMinInterval);
            gcThread.setDaemon(true);
            gcThread.start();

            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("Instantiated Cm for: " + rootDir.getAbsolutePath());
            }

        } catch (DBException de) {
            if (LOG.isEnabledFor(Level.FATAL)) {
                LOG.fatal("Unable to Initialize databases", de);
            }
            throw new UndeclaredThrowableException(de, "Unable to Initialize databases");
        } catch (Throwable e) {
            if (LOG.isEnabledFor(Level.FATAL)) {
                LOG.fatal("Unable to create Cm", e);
            }

            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            } else if (e instanceof Error) {
                throw (Error) e;
            } else {
                throw new UndeclaredThrowableException(e, "Unable to create Cm");
            }
        }
    }

    public String toString() {
        return "CM for " + rootDir.getAbsolutePath() + "[" + super.toString() + "]";
    }

    private static String getDirName(Advertisement adv) {
        if (adv instanceof PeerAdvertisement) {
            return DIRNAME[DiscoveryService.PEER];
        } else if (adv instanceof PeerGroupAdvertisement) {
            return DIRNAME[DiscoveryService.GROUP];
        }

        return DIRNAME[DiscoveryService.ADV];
    }

    /**
     * Generates a random file name using doc hashcode
     *
     * @param  doc  to hash to generate a unique name
     * @return      String a random file name
     */
    public static String createTmpName(StructuredTextDocument doc) {
        try {
            StringWriter out = new StringWriter();

            doc.sendToWriter(out);
            out.close();
            JxtaHash digester = new JxtaHash(out.toString());
            BigInteger hash = digester.getDigestInteger();

            if (hash.compareTo(BigInteger.ZERO) < 0) {
                hash = hash.negate();
            }
            String strHash = "cm" + hash.toString(16);

            return strHash;
        } catch (IOException ex) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Exception creating tmp name: ", ex);
            }
            
            throw new IllegalStateException("Could not generate name from document");
        }
    }

    /**
     * Gets the list of all the files into the given folder
     *
     * @param  dn  contains the name of the folder
     *
     * @return Vector Strings containing the name of the
     * files
     */
    public Vector getRecords(String dn, int threshold,
            Vector values, Vector expirations) {

        return getRecords(dn, threshold, values, expirations, false);
    }

    public synchronized Vector getRecords(String dn, int threshold,
            Vector values, Vector expirations,
            boolean purge) {

        Vector res = new Vector();

        if (dn == null) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("null directory name");
            }
            return res;
        } else {
            IndexQuery iq = new IndexQuery(IndexQuery.SW, new Value(dn));

            try {
                cacheDB.query(iq, new SearchCallback(cacheDB, indexer, res, expirations, threshold, purge));
            } catch (DBException dbe) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Exception during getRecords(): ", dbe);
                }
            } catch (IOException ie) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Exception during getRecords(): ", ie);
                }
            }
            return res;
        }
    }

    public void garbageCollect() {
        // calling getRecords is good enough since it removes
        // expired entries
        Map map = indexer.getIndexers();
        Iterator it = map.keySet().iterator();
        long t0 = 0;

        while (it != null && it.hasNext()) {
            t0 = System.currentTimeMillis();
            String indexName = (String) it.next();

            getRecords(indexName, Integer.MAX_VALUE, null, null, true);
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Cm garbageCollect :" + indexName + " in :" + (System.currentTimeMillis() - t0));
            }
        }
    }

    /**
     *  Returns the relative time in milliseconds at which the file
     *  will expire.
     *
     * @param  dn  contains the name of the folder
     * @param  fn  contains the name of the file
     *
     * @return the absolute time in milliseconds at which this
     * document will expire. -1 is returned if the file is not
     * recognized or already expired.
     */
    public synchronized long getLifetime(String dn, String fn) {
        try {
            Key key = new Key(dn + "/" + fn);
            Record record = cacheDB.readRecord(key);

            if (record == null) {
                return -1;
            }
            Long life = (Long) record.getMetaData(Record.LIFETIME);

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Lifetime for :" + fn + "  " + life.toString());
            }
            if (life.longValue() < System.currentTimeMillis()) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Removing expired record :" + fn);
                }
                try {
                    remove(dn, fn);
                } catch (IOException e) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Failed to remove record", e);
                    }
                }
            }
            return TimeUtils.toRelativeTimeMillis(life.longValue());
        } catch (DBException de) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("failed to remove " + dn + "/" + fn, de);
            }
            return -1;
        }
    }

    /**
     *  Returns the maximum duration in milliseconds for which this
     *  document should cached by those other than the publisher. This
     *  value is either the cache lifetime or the remaining lifetime
     *  of the document, whichever is less.
     *
     * @param  dn  contains the name of the folder
     * @param  fn  contains the name of the file
     * @return     number of milliseconds until the file expires or -1 if the
     * file is not recognized or already expired.
     */
    public synchronized long getExpirationtime(String dn, String fn) {
        try {
            Key key = new Key(dn + "/" + fn);
            Record record = cacheDB.readRecord(key);
            long expiration = calcExpiration(record);

            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Expiration for :" + fn + "  " + expiration);
            }
            if (expiration < 0) {
                if (LOG.isEnabledFor(Level.DEBUG)) {