storemanager.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 924 行 · 第 1/2 页

/*
 * Copyright (c) 1998-2008 Caucho Technology -- all rights reserved
 *
 * This file is part of Resin(R) Open Source
 *
 * Each copy or derived work must preserve the copyright notice and this
 * notice unmodified.
 *
 * Resin Open Source is free software; you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation; either version 2 of the License, or
 * (at your option) any later version.
 *
 * Resin Open Source is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE, or any warranty
 * of NON-INFRINGEMENT.  See the GNU General Public License for more
 * details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Resin Open Source; if not, write to the
 *
 *   Free Software Foundation, Inc.
 *   59 Temple Place, Suite 330
 *   Boston, MA 02111-1307  USA
 *
 * @author Scott Ferguson
 */

package com.caucho.server.cluster;

import com.caucho.config.types.Period;
import com.caucho.lifecycle.Lifecycle;
import com.caucho.loader.ClassLoaderListener;
import com.caucho.loader.DynamicClassLoader;
import com.caucho.loader.Environment;
import com.caucho.loader.EnvironmentClassLoader;
import com.caucho.loader.EnvironmentListener;
import com.caucho.management.server.PersistentStoreMXBean;
import com.caucho.util.Alarm;
import com.caucho.util.AlarmListener;
import com.caucho.util.L10N;
import com.caucho.util.LruCache;
import com.caucho.vfs.TempOutputStream;
import com.caucho.vfs.TempStream;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Base class for distributed stores.
 */
abstract public class StoreManager
  implements AlarmListener, EnvironmentListener, ClassLoaderListener 
{
  private static final Logger log
    = Logger.getLogger(StoreManager.class.getName());
  private static final L10N L = new L10N(StoreManager.class);
  
  private static int DECODE[];

  private Cluster _cluster;
  private String _serverId;
  
  private HashManager _hashManager;
  
  protected int _selfIndex;
  private ClusterServer []_serverList;
  
  private Alarm _alarm;
  
  // protected long _maxIdleTime = 24 * 3600 * 1000L;
  protected long _maxIdleTime = 3 * 3600 * 1000L;
  protected long _idleCheckInterval = 5 * 60 * 1000L;

  protected boolean _isAlwaysLoad;
  protected boolean _isAlwaysSave;
  
  protected HashMap<HashKey,Store> _storeMap;
  protected LruCache<HashKey,ClusterObject> _clusterObjects;

  private final Lifecycle _lifecycle = new Lifecycle();

  //
  // statistics
  //
  
  protected volatile long _loadCount;
  protected volatile long _loadFailCount;

  protected volatile long _saveCount;
  protected volatile long _saveFailCount;

  protected StoreManager()
  {
    _hashManager = new HashManager();
    _storeMap = new HashMap<HashKey,Store>();
    
    _clusterObjects = new LruCache<HashKey,ClusterObject>(4096);
    _clusterObjects.setEnableListeners(false);
    
    _alarm = new Alarm(this);

    Environment.addClassLoaderListener(this);
  }

  /**
   * Sets the cluster.
   */
  public void setCluster(Cluster cluster)
  {
    _cluster = cluster;
  }

  /**
   * Gets the cluster.
   */
  public Cluster getCluster()
  {
    return _cluster;
  }

  /**
   * Returns the admin.
   */
  public PersistentStoreMXBean getAdmin()
  {
    return null;
  }

  /**
   * Set true if the objectStore should always try to loadImpl the object.
   */
  public void setAlwaysLoad(boolean alwaysLoad)
  {
    _isAlwaysLoad = alwaysLoad;
  }

  /**
   * Set true if the objectStore should always try to loadImpl the object.
   */
  public boolean isAlwaysLoad()
  {
    return _isAlwaysLoad;
  }

  /**
   * Set true if the objectStore should always try to objectStore the object.
   */
  public void setAlwaysSave(boolean alwaysSave)
  {
    _isAlwaysSave = alwaysSave;
  }

  /**
   * Set true if the objectStore should always try to objectStore the object.
   */
  public boolean isAlwaysSave()
  {
    return _isAlwaysSave;
  }

  /**
   * Returns the length of time an idle object can remain in the objectStore before
   * being cleaned.
   */
  public long getMaxIdleTime()
  {
    return _maxIdleTime;
  }

  /**
   * Sets the length of time an idle object can remain in the objectStore before
   * being cleaned.
   */
  public void setMaxIdleTime(Period maxIdleTime)
  {
    _maxIdleTime = maxIdleTime.getPeriod();
  }

  /**
   * Sets the idle check interval for the alarm.
   */
  public void updateIdleCheckInterval(long idleCheckInterval)
  {
    if (_idleCheckInterval > 0 && idleCheckInterval > 0
	&& idleCheckInterval < _idleCheckInterval) {
      _idleCheckInterval = idleCheckInterval;
      _alarm.queue(idleCheckInterval);
    }

    if (_idleCheckInterval >= 0 && _idleCheckInterval < 1000)
      _idleCheckInterval = 1000;
  }

  /**
   * Sets the idle check interval for the alarm.
   */
  public long getIdleCheckTime()
  {
    if (_idleCheckInterval > 0)
      return _idleCheckInterval;
    else
      return _maxIdleTime;
  }

  /**
   * Returns the length of time an idle object can remain in the objectStore before
   * being cleaned.
   */
  public long getAccessWindowTime()
  {
    long window = _maxIdleTime / 4;

    if (window < 60000L)
      return 60000L;
    else
      return window;
  }

  //
  // statistics
  //

  /**
   * Returns the objects in the objectStore
   */
  public long getObjectCount()
  {
    return -1;
  }

  /**
   * Returns the total objects loaded.
   */
  public long getLoadCount()
  {
    return _loadCount;
  }

  /**
   * Returns the objects which failed to loadImpl.
   */
  public long getLoadFailCount()
  {
    return _loadFailCount;
  }

  /**
   * Returns the total objects saved.
   */
  public long getSaveCount()
  {
    return _saveCount;
  }

  /**
   * Returns the objects which failed to save.
   */
  public long getSaveFailCount()
  {
    return _saveFailCount;
  }

  /**
   * Creates a Store.  The Store manages
   * persistent objects for a particular domain, like the sesions
   * for the /foo URL.
   *
   * @param storeId the persistent domain.
   */
  public Store createStore(String storeId, ObjectManager objectManager)
  {
    HashKey storeKey = _hashManager.generateHash(storeId);
    
    Store store = getStore(storeKey);
    store.setName(storeId);
    store.setObjectManager(objectManager);

    return store;
  }

  /**
   * Removes a Store.  The Store manages
   * persistent objects for a particular domain, like the sesions
   * for the /foo URL.
   *
   * @param storeId the persistent domain.
   */
  public Store removeStore(HashKey storeKey)
  {
    Store store = getStore(storeKey);

    store.setObjectManager(null);

    return store;
  }

  /**
   * Creates a ClusterObjectManager.  The ClusterObjectManager manages
   * persistent objects for a particular domain, like the sesions
   * for the /foo URL.
   *
   * @param storeId the persistent domain.
   */
  public Store getStore(HashKey storeKey)
  {
    synchronized (_storeMap) {
      Store store = _storeMap.get(storeKey);
      
      if (store == null) {
	store = new Store(storeKey, this);
        
	_storeMap.put(storeKey, store);
      }

      return store;
    }
  }

  /**
   * Called after any factory settings.
   */
  @PostConstruct
  public boolean init()
  {
    if (! _lifecycle.toInit())
      return false;

    _lifecycle.setName(toString());

    if (_cluster == null)
      _cluster = Cluster.getLocal();
    
    if (_cluster != null) {
      _serverId = Cluster.getServerId();
      ClusterServer selfServer = _cluster.getSelfServer();

      if (selfServer != null)
	_selfIndex = selfServer.getIndex();
      else if (_cluster.getServerList().length > 1) {
	// XXX: error?
	log.warning(L.l("cluster-store for '{0}' needs an <srun> configuration for it.",
			_serverId));
      }

      ClusterServer []serverList = _cluster.getServerList();
      
      _serverList = new ClusterServer[serverList.length];

      for (int i = 0; i < serverList.length; i++) {
	_serverList[i] = serverList[i];
      }
    }

    Environment.addEnvironmentListener(this);

    return true;
  }

  /**
   * Called to start the objectStore.
   */
  public boolean start()
    throws Exception
  {
    if (! _lifecycle.toActive())
      return false;
    
    // notify the siblings that we're awake
    if (_serverList != null) {
      ClusterServer []serverList = _serverList;

      for (int i = 0; i < serverList.length; i++) {
	ServerPool server = serverList[i].getServerPool();

	if (server == null)
	  continue;

	try {
	  ClusterStream s = server.open();
	  s.close();
	} catch (Throwable e) {
	}
      }
      
    }

    handleAlarm(_alarm);

    log.info(this + " active");

    return true;
  }

  /**
   * Called to start any invalidate processing
   */
  public boolean startUpdate()
    throws Exception
  {
    return true;
  }

  /**
   * Cleans old objects.  Living objects corresponding to the old
   * objects are not cleared, since their timeout should be less than
   * the objectStore timeout.
   */
  public void clearOldObjects()
    throws Exception
  {
  }

  /**
   * Returns true if this server is a primary for the given object id.
   */
  protected boolean isPrimary(String id)
  {
    return getPrimaryIndex(id, 0) == getSelfIndex();
  }

  /**
   * Returns the owning index.
   */
  public int getPrimaryIndex(String id, int offset)
  {
    return 0;
  }

  /**
   * Returns the backup index.
   */
  public int getSecondaryIndex(String id, int offset)
  {
    return 0;
  }

  /**
   * Returns the backup index.
   */
  public int getTertiaryIndex(String id, int offset)
  {
    return 0;
  }

  /**
   * Loads an object from the backing objectStore.
   *
   * @param obj the object to updateImpl.
   */
  abstract protected boolean load(ClusterObject clusterObject, Object obj)
    throws Exception;

  /**
   * Updates the object's objectAccess time.
   *
   * @param storeId the identifier of the storage group
   * @param obj the object to updateImpl.
   */