src/org/objectweb/cjdbc/controller/cache/result/ResultCache.java

説明を見る。

package org.objectweb.cjdbc.controller.cache.result;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;

import org.objectweb.cjdbc.common.i18n.Translate;
import org.objectweb.cjdbc.common.log.Trace;
import org.objectweb.cjdbc.common.sql.AbstractWriteRequest;
import org.objectweb.cjdbc.common.sql.CreateRequest;
import org.objectweb.cjdbc.common.sql.ParsingGranularities;
import org.objectweb.cjdbc.common.sql.RequestType;
import org.objectweb.cjdbc.common.sql.SelectRequest;
import org.objectweb.cjdbc.common.sql.UpdateRequest;
import org.objectweb.cjdbc.common.sql.schema.DatabaseSchema;
import org.objectweb.cjdbc.common.xml.DatabasesXmlTags;
import org.objectweb.cjdbc.controller.cache.CacheException;
import org.objectweb.cjdbc.controller.cache.CacheStatistics;
import org.objectweb.cjdbc.controller.cache.result.entries.CacheEntry;
import org.objectweb.cjdbc.controller.cache.result.entries.ResultCacheEntry;
import org.objectweb.cjdbc.controller.cache.result.entries.ResultCacheEntryNoCache;
import org.objectweb.cjdbc.controller.cache.result.entries.ResultCacheEntryRelaxed;
import org.objectweb.cjdbc.controller.cache.result.schema.CacheDatabaseSchema;
import org.objectweb.cjdbc.controller.cache.result.schema.CacheDatabaseTable;
import org.objectweb.cjdbc.controller.virtualdatabase.ControllerResultSet;
import org.objectweb.cjdbc.driver.Field;

public abstract class ResultCache extends AbstractResultCache
{
  //
  // How the code is organized?
  //
  // 1. Member variables
  // 2. Constructor
  // 3. Cache management
  // 4. Transaction management
  // 5. Debug/Monitoring
  //

  // Max number of cache entries
  private int                    maxEntries;
  private long                   pendingQueryTimeout = 0;
  // queries: SQL -> ResultCacheEntry
  private HashMap                queries;
  // Pending SQL requests (String)
  private HashSet                pendingQueries;
  // The rules to apply for this cache
  private HashSet                cachingRules;
  private ResultCacheRule        defaultRule;
  private ArrayList              relaxedCache;

  // LRU (head) of cache entries for replacement
  private CacheEntry             lruHead;
  // LRU (tail) of cache entries for replacement
  private CacheEntry             lruTail;

  // Database schema
  protected CacheDatabaseSchema  cdbs;

  private CacheStatistics        stats;

  private long                   threadWakeUpTime;
  private RelaxedCacheThread     relaxedThread;
  private static final boolean[] TRUE_TRUE           = new boolean[]{true, true};

  /*
   * Constructor
   */

  public ResultCache(int maxEntries, int pendingTimeout)
  {
    this.maxEntries = maxEntries;
    this.pendingQueryTimeout = pendingTimeout;
    cdbs = null;
    stats = new CacheStatistics();
    queries = new HashMap(1000, (float) 0.75);
    pendingQueries = new HashSet();
    cachingRules = new HashSet();
    relaxedCache = new ArrayList();
    lruHead = null;
    lruTail = null;
    defaultRule = null;
    threadWakeUpTime = 0;
    relaxedThread = new RelaxedCacheThread();
    relaxedThread.setPriority(9);
    relaxedThread.start();
  }

  public int getPendingQueryTimeout()
  {
    return (int) (pendingQueryTimeout / 1000);
  }

  public void setPendingQueryTimeout(int pendingQueryTimeout)
  {
    this.pendingQueryTimeout = pendingQueryTimeout * 1000L;
  }

  public HashMap getQueries()
  {
    return this.queries;
  }

  public void setDatabaseSchema(DatabaseSchema dbs)
  {
    if (cdbs == null)
    {
      logger.info(Translate.get("resultcache.setting.database.schema"));
      cdbs = new CacheDatabaseSchema(dbs);
    }
    else
    { // Schema is updated, compute the diff !
      CacheDatabaseSchema newSchema = new CacheDatabaseSchema(dbs);
      ArrayList tables = cdbs.getTables();
      ArrayList newTables = newSchema.getTables();
      if (newTables == null)
      { // New schema is empty (no backend is active anymore)
        logger.info(Translate.get("resultcache.flusing.whole.cache"));
        flushCache();
        cdbs = null;
        return;
      }

      // Remove extra-tables
      for (int i = 0; i < tables.size(); i++)
      {
        CacheDatabaseTable t = (CacheDatabaseTable) tables.get(i);
        if (!newSchema.hasTable(t.getName()))
        {
          t.invalidateAll();
          cdbs.removeTable(t);
          if (logger.isInfoEnabled())
            logger.info(Translate
                .get("resultcache.removing.table", t.getName()));
        }
      }

      // Add missing tables
      int size = newTables.size();
      for (int i = 0; i < size; i++)
      {
        CacheDatabaseTable t = (CacheDatabaseTable) newTables.get(i);
        if (!cdbs.hasTable(t.getName()))
        {
          cdbs.addTable(t);
          if (logger.isInfoEnabled())
            logger.info(Translate.get("resultcache.adding.table", t.getName()));
        }
      }
    }
  }

  public void mergeDatabaseSchema(DatabaseSchema dbs)
  {
    try
    {
      logger.info(Translate.get("resultcache.merging.new.database.schema"));
      cdbs.mergeSchema(new CacheDatabaseSchema(dbs));
    }
    catch (Exception e)
    {
      logger.error(Translate.get("resultcache.error.while.merging", e));
    }
  }

  public void addCachingRule(ResultCacheRule rule)
  {
    cachingRules.add(rule);
  }

  public ResultCacheRule getDefaultRule()
  {
    return defaultRule;
  }

  public void setDefaultRule(ResultCacheRule defaultRule)
  {
    this.defaultRule = defaultRule;
  }

  private CacheBehavior getCacheBehavior(SelectRequest request)
  {
    CacheBehavior behavior = null;
    for (Iterator iter = cachingRules.iterator(); iter.hasNext();)
    {
      behavior = ((ResultCacheRule) iter.next()).matches(request);
      if (behavior != null)
      {
        break;
      }
    }
    if (behavior == null)
      behavior = defaultRule.getCacheBehavior();
    if (logger.isDebugEnabled())
      logger.debug(Translate.get("resultcache.behavior.for.request",
          new String[]{request.getSQL(), behavior.getType()}));
    return behavior;
  }

  /*
   * Cache Management
   */

  public boolean[] needInvalidate(ControllerResultSet result,
      UpdateRequest request)
  {
    HashMap updatedValues = request.getUpdatedValues();
    boolean needInvalidate = false;
    boolean needToSendQuery = false;
    String value;
    String columnName;
    try
    {
      // If we don't have exactly one row, we don't handle the optimization
      if ((result == null) || (result.getData() == null)
          || (result.getData().size() != 1))
        return TRUE_TRUE;
    }
    catch (Exception e)
    {
      return TRUE_TRUE;
    }
    Field[] fields = result.getFields();
    ArrayList data = result.getData();
    int size = fields.length;
    for (Iterator iter = updatedValues.keySet().iterator(); iter.hasNext();)
    {
      columnName = (String) iter.next();
      value = (String) updatedValues.get(columnName);
      for (int i = 0; i < size; i++)
      { // Find the corresponding column in the ResultSet by comparing column
        // names

        // We can have something like:
        //  FIRSTNAME and ADDRESS.FIRSTNAME
        if (columnName.equals(fields[i].getFieldName()))
        {
          Object o = ((Object[]) data.get(0))[i];
          if (!value.equals(o))
          {
            // The value from the cache entry is different we need to update
            // the
            // cache and the database
            return TRUE_TRUE;
          }
          else
            break;
        }
      }
      // We don't need to invalidate the cache because the columns affected are
      // different but we need to send the query to the database.
      needToSendQuery = true;
      // We don't stop here though because other columns could be updated and
      // we
      // could need invalidation
    }
    return new boolean[]{needInvalidate, needToSendQuery};
  }

  public void addToCache(SelectRequest request, ControllerResultSet result)
      throws CacheException
  {
    String sqlQuery = request.getSQL();

    // Sanity checks
    if (request.getCacheAbility() == RequestType.UNCACHEABLE)
      throw new CacheException(Translate.get("resultcache.uncacheable.request",
          sqlQuery));

    if (result == null)
      throw new CacheException(Translate.get("resultcache.null.result",
          sqlQuery));

    synchronized (pendingQueries)
    {
      // Remove the pending query from the list and wake up
      // all waiting queries
      if (pendingQueries.remove(sqlQuery))
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("resultcache.removing.pending.query",
              sqlQuery));
        pendingQueries.notifyAll();
      }
      else
        logger.warn(Translate.get("resultcache.removing.pending.query.failed",
            sqlQuery));

      // Check against streamable ResultSets
      if (result.hasMoreData())
      {
        logger.info(Translate.get("resultcache.streamed.resultset", request
            .getSQLShortForm(20)));
        return;
      }

      if (logger.isDebugEnabled())
        logger.debug(Translate.get("resultcache.adding.query", sqlQuery));

      CacheEntry ce;
      synchronized (queries)
      {
        // Check first that the query is not already in the cache
        ce = (ResultCacheEntry) queries.get(sqlQuery);
        if (ce == null)
        {
          // Not in cache, add this entry
          // check the rule
          CacheBehavior behavior = getCacheBehavior(request);
          ce = behavior.getCacheEntry(request, result, this);
          if (ce instanceof ResultCacheEntryNoCache)
            return;

          // Test size of cache
          if (maxEntries > 0)
          {
            int size = queries.size();
            if (size >= maxEntries)
              // LRU replacement policy: Remove the oldest cache entry
              removeOldest();
          }
          // Add to the cache
          queries.put(sqlQuery, ce);
          if (ce instanceof ResultCacheEntryRelaxed)
          {
            ResultCacheEntryRelaxed qcer = (ResultCacheEntryRelaxed) ce;
            synchronized (relaxedThread)
            {
              relaxedCache.add(qcer);
              if (qcer.getDeadline() < threadWakeUpTime
                  || threadWakeUpTime == 0)
              {
                relaxedThread.notify();
              }
            }
          }
        }
        else
        { // Oh, oh, already in cache ...
          if (ce.isValid())
            logger.warn(Translate.get(
                "resultcache.modifying.result.valid.entry", sqlQuery));
          ce.setResult(result);
        }

        // Update LRU
        if (lruHead != null)
        {
          lruHead.setPrev(ce);
          ce.setNext(lruHead);
          ce.setPrev(null);
        }
        if (lruTail == null)
          lruTail = ce;
        lruHead = ce; // This is also fine if LRUHead == null
      }
      processAddToCache(ce);
    }
  }

  protected abstract void processAddToCache(CacheEntry qe);

  public CacheEntry getFromCache(SelectRequest request,
      boolean addToPendingQueries)
  {
    stats.addSelect();

    if (request.getCacheAbility() == RequestType.UNCACHEABLE)
    {
      stats.addUncacheable();
      return null;
    }

    String sqlQuery = request.getSQL();

    // Check if we have the same query pending
    synchronized (pendingQueries)
    {
      if (addToPendingQueries)
      {
        long timeout = pendingQueryTimeout;
        // Yes, wait for the result
        // As we use a single lock for all pending queries, we use a
        // while to re-check that this wake-up was for us!
        while (pendingQueries.contains(sqlQuery))
        {
          try
          {
            if (logger.isDebugEnabled())
              logger.debug(Translate.get("resultcache.waiting.pending.query",
                  sqlQuery));

            if (timeout > 0)
            {
              long start = System.currentTimeMillis();
              pendingQueries.wait(pendingQueryTimeout);
              long end = System.currentTimeMillis();
              timeout = timeout - (end - start);
              if (timeout <= 0)
              {
                logger.warn(Translate.get("resultcache.pending.query.timeout"));
                break;
              }
            }
            else
              pendingQueries.wait();
          }
          catch (InterruptedException e)
          {
            logger.warn(Translate.get("resultcache.pending.query.timeout"));
            break;
          }
        }
      }

      // Check the cache
      ResultCacheEntry ce;
      synchronized (queries)
      {
        ce = (ResultCacheEntry) queries.get(sqlQuery);
        if (ce == null)
        //if ((ce == null) || !ce.isValid())
        { // Cache miss or dirty entry
          if (addToPendingQueries)
          {
            pendingQueries.add(sqlQuery);
            // Add this query to the pending queries
            if (logger.isDebugEnabled())
            {
              logger.debug(Translate.get("resultcache.cache.miss"));
              logger.debug(Translate.get(
                  "resultcache.adding.to.pending.queries", sqlQuery));
            }
          }
          return null;
        }
        else
        { // Cache hit (must update LRU)
          // Move cache entry to head of LRU
          CacheEntry before = ce.getPrev();
          if (before != null)
          {
            CacheEntry after = ce.getNext();
            before.setNext(after);
            if (after != null)
              after.setPrev(before);
            else
              // We were the tail, update the tail
              lruTail = before;
            ce.setNext(lruHead);
            ce.setPrev(null);
            if (lruHead != ce)
              lruHead.setPrev(ce);
            lruHead = ce;
          }
          // else it was already the LRU head
        }
      }

      if (ce.getResult() == null)
      {
        if (addToPendingQueries)
        {
          pendingQueries.add(sqlQuery);
          // Add this query to the pending queries
          if (logger.isDebugEnabled())
          {
            logger.debug(Translate.get("resultcache.cache.miss"));
            logger.debug(Translate.get("resultcache.adding.to.pending.queries",
                sqlQuery));
          }
        }
        if (ce.isValid() && logger.isInfoEnabled())
          logger.info(Translate.get("resultcache.valid.entry.without.result",
              ce.getRequest().getSQL()));
      }
      else
      {
        if (logger.isDebugEnabled())
          logger.debug(Translate.get("resultcache.cache.hit", sqlQuery));
        stats.addHits();
      }

      return ce;
    }
  }

  public void removeFromCache(SelectRequest request)
  {
    String sqlQuery = request.getSQL();

    if (logger.isDebugEnabled())
      logger.debug("Removing from cache: " + sqlQuery);

    synchronized (queries)
    {
      // Remove from the cache
      ResultCacheEntry ce = (ResultCacheEntry) queries.remove(sqlQuery);
      if (ce == null)
        return; // Was not in the cache!
      else
      {
        // Update result set
        ce.setResult(null);
        // Update LRU
        CacheEntry before = ce.getPrev();
        CacheEntry after = ce.getNext();
        if (before != null)
        {
          before.setNext(after);
          if (after != null)
            after.setPrev(before);
          else
            // We were the tail, update the tail
            lruTail = before;
        }
        else
        { // We are the LRUHead
          lruHead = ce.getNext();
          if (after != null)
            after.setPrev(null);
          else
            // We were the tail, update the tail
            lruTail = before;
        }
      }
    }
  }

  public void removeFromPendingQueries(SelectRequest request)
  {
    String sqlQuery = request.getSQL();

    synchronized (pendingQueries)
    {
      // Remove the pending query from the list and wake up
      // all waiting queries
      if (pendingQueries.remove(sqlQuery))
        pendingQueries.notifyAll();
    }
  }

  public abstract boolean isUpdateNecessary(UpdateRequest request)
      throws CacheException;

  public void writeNotify(AbstractWriteRequest request) throws CacheException
  {
    // Update the stats
    if (request.isInsert())
      stats.addInsert();
    else if (request.isUpdate())
      stats.addUpdate();
    else if (request.isDelete())
      stats.addDelete();
    else if (request.isCreate())
    {
      stats.addCreate();
      // Create: we only need to update the schema
      if (parsingGranularity != ParsingGranularities.NO_PARSING)
        cdbs.addTable(new CacheDatabaseTable(((CreateRequest) request)
            .getDatabaseTable()));
      return;
    }
    else if (request.isDrop())
    {
      stats.addDrop();
      // Drop: we need to update the schema
      if (parsingGranularity != ParsingGranularities.NO_PARSING)
      {
        // Invalidate the cache entries associated with this table
        CacheDatabaseTable cdt = cdbs.getTable(request.getTableName());
        if (cdt != null)
        {
          cdt.invalidateAll();
          cdbs.removeTable(cdt);
          return;
        }
        // else: the table was not previously cached
        // (no previous 'select' requests on the table).
      }
    }
    else
    {
      stats.addUnknown();
    }
    if (logger.isDebugEnabled())
      logger.debug("Notifying write " + request.getSQL());

    processWriteNotify(request);
  }

  protected abstract void processWriteNotify(AbstractWriteRequest request);

  public void flushCache()
  {
    synchronized (queries)
    { // Invalidate the whole cache until it is empty
      while (!queries.isEmpty())
      {
        Iterator iter = queries.values().iterator();
        ((ResultCacheEntry) iter.next()).invalidate();
      }
    }

    synchronized (pendingQueries)
    { // Clean pending queries to unblock everyone if some queries/backends
      // remained in an unstable state.
      pendingQueries.clear();
      pendingQueries.notifyAll();
    }

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("resultcache.cache.flushed"));
  }

  public long getCacheSize()
  {
    // No need to synchronize, the implementation returns an int
    return queries.size();
  }

  private void removeOldest()
  {
    if (lruTail == null)
      return;
    // Update the LRU
    ResultCacheEntry oldce = (ResultCacheEntry) lruTail;
    lruTail = lruTail.getPrev();
    if (lruTail != null)
      lruTail.setNext(null);

    if (logger.isDebugEnabled())
      logger.debug(Translate.get("resultcache.removing.oldest.cache.entry",
          oldce.getRequest().getSQL()));

    /*
     * We remove the query from the hashtable so that the garbage collector can
     * do its job. We need to remove the query from the queries HashTable first
     * in case we invalidate an eager cache entry that will call removeFromCache
     * (and will try to update the LRU is the entry is still in the queries
     * HashTable). So, to be compatible with all type of cache entries: 1.
     * queries.remove(ce) 2. ce.invalidate
     */
    queries.remove(oldce.getRequest().getSQL());

    if (oldce.isValid())
    {
      oldce.setResult(null);
      oldce.invalidate();
    }

    stats.addRemove();
  }

  public int getParsingGranularity()
  {
    return this.parsingGranularity;
  }

  public abstract String getName();

  //
  // Transaction management
  //

  public void commit(long transactionId) throws CacheException
  {
    // Ok, the transaction has commited, nothing to do
  }

  public void rollback(long transactionId) throws CacheException
  {
    logger.info(Translate.get("resultcache.flushing.cache.cause.rollback",
        transactionId));
    flushCache();
  }

  /*
   * Debug/Monitoring
   */

  public String[][] getCacheData() throws CacheException
  {
    try
    {
      synchronized (queries)
      {
        String[][] data = new String[queries.size()][];
        int count = 0;
        for (Iterator iter = queries.values().iterator(); iter.hasNext(); count++)
        {
          ResultCacheEntry qe = (ResultCacheEntry) iter.next();
          if (qe != null)
          {
            data[count] = qe.toStringTable();
          }
        }
        return data;
      }
    }
    catch (Exception e)
    {
      logger.error(Translate.get("resultcache.error.retrieving.cache.data", e));
      throw new CacheException(e.getMessage());
    }
  }

  public String[][] getCacheStatsData() throws CacheException
  {
    String[][] data = new String[1][];
    String[] stat = stats.getCacheStatsData();
    data[0] = new String[stat.length + 1];
    for (int i = 0; i < stat.length; i++)
      data[0][i] = stat[i];
    data[0][data[0].length - 1] = "" + queries.size();
    return data;
  }

  public CacheStatistics getCacheStatistics()
  {
    return stats;
  }

  protected String getXmlImpl()
  {
    StringBuffer info = new StringBuffer();
    info.append("<" + DatabasesXmlTags.ELT_ResultCache + " "
        + DatabasesXmlTags.ATT_pendingTimeout + "=\"" + pendingQueryTimeout
        + "\" " + DatabasesXmlTags.ATT_maxNbOfEntries + "=\"" + maxEntries
        + "\" " + DatabasesXmlTags.ATT_granularity + "=\"" + getName() + "\">");
    info.append("<" + DatabasesXmlTags.ELT_DefaultResultCacheRule + " "
        + DatabasesXmlTags.ATT_timestampResolution + "=\""
        + defaultRule.getTimestampResolution() / 1000 + "\">");
    info.append(defaultRule.getCacheBehavior().getXml());
    info.append("</" + DatabasesXmlTags.ELT_DefaultResultCacheRule + ">");
    for (Iterator iter = cachingRules.iterator(); iter.hasNext();)
      info.append(((ResultCacheRule) iter.next()).getXml());
    info.append("</" + DatabasesXmlTags.ELT_ResultCache + ">");
    return info.toString();
  }

  private final class RelaxedCacheThread extends Thread
  {
    int             refreshCacheRate = 60;
    int             refreshCacheTime = 60 / refreshCacheRate;
    private Trace   logger           = Trace
                                         .getLogger("org.objectweb.cjdbc.controller.cache.cache.RelaxedThread");

    private boolean isKilled         = false;

    public RelaxedCacheThread()
    {
      super("RelaxedCacheThread");
    }

    public RelaxedCacheThread(int refreshCacheRate)
    {
      this();
      this.refreshCacheRate = refreshCacheRate;
    }

    public void run()
    {
      ResultCacheEntryRelaxed entry;
      long now;
      long sleep;
      // Keep trace of relaxed cache entries to delete
      ArrayList toRemoveFromRelaxedCache = new ArrayList();
      while (!isKilled)
      {
        synchronized (this)
        {
          try
          {
            threadWakeUpTime = 0;
            if (relaxedCache.isEmpty())
            { // Nothing in the cache, just sleep!
              if (logger.isDebugEnabled())
                logger.debug(Translate
                    .get("relaxedcachethread.cache.empty.sleeping"));
              wait();
            }
            else
            { // Look for first deadline
              now = System.currentTimeMillis();
              clean : for (Iterator iter = relaxedCache.iterator(); iter
                  .hasNext();)
              {
                entry = (ResultCacheEntryRelaxed) iter.next();
                if (entry.getDeadline() < now)
                {
                  if (entry.isDirty() || !entry.getKeepIfNotDirty())
                  {
                    {
                      toRemoveFromRelaxedCache.add(entry);
                    }
                    continue clean;
                  }
                  else
                    entry.setDeadline(now + entry.getTimeout());
                }
                if (threadWakeUpTime == 0
                    || (entry.getDeadline() < threadWakeUpTime))
                  threadWakeUpTime = entry.getDeadline();
              }

              // Clean up all the entries from the relaxed cache
              for (int i = 0; i < toRemoveFromRelaxedCache.size(); i++)
              {
                entry = (ResultCacheEntryRelaxed) toRemoveFromRelaxedCache
                    .get(i);
                if (logger.isDebugEnabled())
                  logger.debug(Translate.get(
                      "relaxedcachethread.remove.entry.from.cache", entry
                          .getRequest().getSQL()));
                removeFromCache(entry.getRequest());
                relaxedCache.remove(toRemoveFromRelaxedCache.get(i));
                toRemoveFromRelaxedCache.remove(i);
              }
              if (threadWakeUpTime == 0)
              { // All entries were dirty and not kept in the cache, therefore
                // there is no next deadline. (and no cache entry to wait for)
                continue;
              }
              else
              { // Sleep until the next deadline
                sleep = (threadWakeUpTime - now) / 1000 + refreshCacheTime;
                if (logger.isDebugEnabled())
                {
                  logger.debug(Translate.get("relaxedcachethread.sleeping",
                      sleep));
                }
                sleep = (sleep) * 1000;
                wait(sleep);
              }
            }
          }
          catch (Exception e)
          {
            logger.warn(e.getMessage(), e);
          }
        }
      }
    }
  }

}

---