serverpool.java

来自「RESIN 3.2 最新源码」· Java 代码 · 共 1,285 行 · 第 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.ConfigException;
import com.caucho.management.server.ServerConnectorMXBean;
import com.caucho.util.L10N;
import com.caucho.util.Alarm;
import com.caucho.util.QDate;
import com.caucho.vfs.*;
import com.caucho.server.resin.*;

import javax.management.ObjectName;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * A pool of connections to a Resin server.
 */
public class ServerPool
{
  private static final Logger log
    = Logger.getLogger(ServerPool.class.getName());
  private static final L10N L = new L10N(ServerPool.class);

  private static final int ST_NEW = 0;
  private static final int ST_STANDBY = 1;
  private static final int ST_SESSION_ONLY = 2;
  // the following 5 are the active states
  private static final int ST_STARTING = 3;
  private static final int ST_WARMUP = 4;
  private static final int ST_BUSY = 5;
  private static final int ST_FAIL = 6;
  private static final int ST_ACTIVE = 7;
  private static final int ST_CLOSED = 8;

  // number of chunks in the throttling
  private static final int WARMUP_MAX = 16;
  private static final int WARMUP_MIN = -16;
  private static final int []WARMUP_CONNECTION_MAX
    = new int[] { 1, 1, 1, 1, 2, 2, 2, 2, 4, 4, 8, 8, 16, 32, 64, 128 };

  private final String _serverId;
  private final String _targetId;

  private final String _address;
  private final int _port;
  private final boolean _isSecure;

  private String _debugId;
  
  private Path _tcpPath;

  private int _maxConnections = Integer.MAX_VALUE / 2;

  private long _loadBalanceConnectTimeout = 5000;
  private long _loadBalanceSocketTimeout = 30000;
  private long _loadBalanceIdleTime = 10000;
  private long _loadBalanceRecoverTime = 15000;
  private long _loadBalanceWarmupTime = 60000;
  private int _loadBalanceWeight = 100;
  
  private ClusterStream []_idle = new ClusterStream[64];
  private volatile int _idleHead;
  private volatile int _idleTail;
  private int _idleSize = 16;

  private int _streamCount;

  private long _warmupChunkTime;
  
  private long _failRecoverTime;
  private long _failChunkTime;

  private volatile int _state = ST_NEW;

  // current connection count
  private volatile int _activeCount;
  private volatile int _startingCount;

  private volatile int _loadBalanceAllocateCount;

  // numeric value representing the throttle state
  private volatile int _warmupState;
  
  // load management data
  private volatile long _lastFailConnectTime;
  private volatile long _dynamicFailRecoverTime = 1000L;
  
  private volatile long _lastFailTime;
  private volatile long _lastBusyTime;

  private volatile long _failTime;
  private volatile long _firstSuccessTime;
  private volatile long _lastSuccessTime;
  private volatile long _prevSuccessTime;
  private volatile double _latencyFactor;

  // statistics
  private volatile long _keepaliveCountTotal;
  private volatile long _connectCountTotal;
  private volatile long _failCountTotal;
  private volatile long _busyCountTotal;

  private volatile double _cpuLoadAvg;
  private volatile long _cpuSetTime;

  public ServerPool(String serverId,
		    String targetId,
		    String address,
		    int port,
		    boolean isSecure)
  {
    _serverId = serverId;
    _targetId = targetId;
    _debugId = _serverId + "->" + _targetId;
    _address = address;
    _port = port;
    _isSecure = isSecure;
  }

  public ServerPool(String serverId,
		    ClusterServer server)
  {
    this(serverId,
	 server.getId(),
	 server.getClusterPort().getAddress(),
	 server.getClusterPort().getPort(),
	 server.getClusterPort().isSSL());

    _loadBalanceConnectTimeout = server.getLoadBalanceConnectTimeout();
    _loadBalanceSocketTimeout = server.getLoadBalanceSocketTimeout();
    _loadBalanceIdleTime = server.getLoadBalanceIdleTime();
    _loadBalanceRecoverTime = server.getLoadBalanceRecoverTime();
    _loadBalanceWarmupTime = server.getLoadBalanceWarmupTime();
    _loadBalanceWeight = server.getLoadBalanceWeight();
  }

  /**
   * Returns the user-readable id of the target server.
   */
  public String getId()
  {
    return _targetId;
  }

  /**
   * Returns the debug id.
   */
  public String getDebugId()
  {
    return _debugId;
  }

  /**
   * Returns the hostname of the target server.
   */
  public String getAddress()
  {
    return _address;
  }

  /**
   * Gets the port of the target server.
   */
  public int getPort()
  {
    return _port;
  }

  /**
   * The socket timeout when connecting to the target server.
   */
  public long getLoadBalanceConnectTimeout()
  {
    return _loadBalanceConnectTimeout;
  }

  /**
   * The socket timeout when connecting to the target server.
   */
  public void setLoadBalanceConnectTimeout(long timeout)
  {
    _loadBalanceConnectTimeout = timeout;
  }

  /**
   * The socket timeout when reading from the target server.
   */
  public long getLoadBalanceSocketTimeout()
  {
    return _loadBalanceSocketTimeout;
  }

  /**
   * The socket timeout when reading from the target server.
   */
  public void setLoadBalanceSocketTimeout(long timeout)
  {
    _loadBalanceSocketTimeout = timeout;
  }

  /**
   * How long the connection can be cached in the free pool.
   */
  public long getLoadBalanceIdleTime()
  {
    return _loadBalanceIdleTime;
  }

  /**
   * How long the connection can be cached in the free pool.
   */
  public void setLoadBalanceIdleTime(long timeout)
  {
    _loadBalanceIdleTime = timeout;
  }

  /**
   * Returns how long the connection will be treated as dead.
   */
  public void setLoadBalanceRecoverTime(long timeout)
  {
    _loadBalanceRecoverTime = timeout;
  }

  /**
   * Returns the time in milliseconds for the slow start throttling.
   */
  public void setLoadBalanceWarmupTime(long timeout)
  {
    _loadBalanceWarmupTime = timeout;
  }

  /**
   * The load balance weight.
   */
  public int getLoadBalanceWeight()
  {
    return _loadBalanceWeight;
  }

  /**
   * The load balance weight.
   */
  public void setLoadBalanceWeight(int weight)
  {
    _loadBalanceWeight = weight;
  }

  /**
   * Initialize
   */
  public void init()
    throws Exception
  {
    _warmupChunkTime = _loadBalanceWarmupTime / WARMUP_MAX;
    if (_warmupChunkTime <= 0)
      _warmupChunkTime = 1;
    
    _failChunkTime = _loadBalanceRecoverTime / WARMUP_MAX;
    if (_failChunkTime <= 0)
      _failChunkTime = 1;

    _state = ST_STARTING;
    
    String address = getAddress();

    if (address == null)
      address = "localhost";

    HashMap<String,Object> attr = new HashMap<String,Object>();
    attr.put("connect-timeout", new Long(_loadBalanceConnectTimeout));

    if (_isSecure)
      _tcpPath = Vfs.lookup("tcps://" + address + ":" + _port, attr);
    else
      _tcpPath = Vfs.lookup("tcp://" + address + ":" + _port, attr);
  }

  /**
   * Returns the number of active connections.
   */
  public int getActiveCount()
  {
    return _activeCount;
  }

  /**
   * Returns the number of idle connections.
   */
  public int getIdleCount()
  {
    return (_idleHead - _idleTail + _idle.length) % _idle.length;
  }

  /**
   * Returns the number of load balance allocations
   */
  public int getLoadBalanceAllocateCount()
  {
    return _loadBalanceAllocateCount;
  }

  /**
   * Allocate a connection for load balancing.
   */
  public void allocateLoadBalance()
  {
    synchronized (this) {
      _loadBalanceAllocateCount++;
    }
  }

  /**
   * Free a connection for load balancing.
   */
  public void freeLoadBalance()
  {
    synchronized (this) {
      _loadBalanceAllocateCount--;
    }
  }

  /**
   * Returns the total number of successful socket connections
   */
  public long getConnectCountTotal()
  {
    return _connectCountTotal;
  }

  /**
   * Returns the number of times a keepalive connection has been used.
   */
  public long getKeepaliveCountTotal()
  {
    return _keepaliveCountTotal;
  }

  /**
   * Returns the total number of failed connect attempts.
   */
  public long getFailCountTotal()
  {
    return _failCountTotal;
  }

  /**
   * Returns the time of the last failure.
   */
  public Date getLastFailTime()
  {
    return new Date(_lastFailTime);
  }

  /**
   * Returns the time of the last failure.
   */
  public Date getLastFailConnectTime()
  {
    return new Date(_lastFailConnectTime);
  }

  /**
   * Returns the time of the last failure.
   */
  public long getLastSuccessTime()
  {
    return _lastSuccessTime;
  }

  /**
   * Returns the latency factory
   */
  public double getLatencyFactor()
  {
    return _latencyFactor;
  }

  /**
   * Returns the count of busy connections.
   */
  public long getBusyCountTotal()
  {
    return _busyCountTotal;
  }

  /**
   * Returns the time of the last busy.
   */
  public Date getLastBusyTime()
  {
    return new Date(_lastBusyTime);
  }

  /**
   * Sets the CPU load avg (from backend).
   */
  public void setCpuLoadAvg(double load)
  {
    _cpuSetTime = Alarm.getCurrentTime();
    _cpuLoadAvg = load;
  }

  /**
   * Gets the CPU load avg
   */
  public double getCpuLoadAvg()
  {
    double avg = _cpuLoadAvg;
    long time = _cpuSetTime;

    long now = Alarm.getCurrentTime();

    if (now - time < 10000L)
      return avg;
    else
      return avg * 10000L / (now - time);
  }

  /**
   * Returns true if the server is active.
   */
  public final boolean isActive()
  {
    switch (_state) {
    case ST_ACTIVE:
      return true;

    case ST_STANDBY:
    case ST_CLOSED:
      return false;

    case ST_FAIL:
      return (_failTime + _failRecoverTime <= Alarm.getCurrentTime());
      
    default:
      return false;
    }
  }
  
  /**
   * Returns true if the server is dead.
   */
  public boolean isDead()
  {
    return ! isActive();
  }

  /**
   * Enable the client
   */
  public void enable()
  {
    start();
  }

  /**
   * Disable the client
   */
  public void disable()
  {
    stop();
  }

  /**
   * Returns the lifecycle state.
   */
  public String getState()
  {
    updateWarmup();
    
    switch (_state) {
    case ST_NEW:
      return "init";
    case ST_STANDBY:
      return "standby";
    case ST_SESSION_ONLY:
      return "session-only";
    case ST_STARTING:
      return "starting";
    case ST_WARMUP:
      return "warmup";
    case ST_BUSY:
      return "busy";
    case ST_FAIL:
      return "fail";
    case ST_ACTIVE:
      return "active";
    case ST_CLOSED:
      return "closed";
    default:
      return "unknown(" + _state + ")";
    }
  }

  /**
   * Returns true if the server can open a connection.
   */
  public boolean canOpenSoftOrRecycle()
  {
    return getIdleCount() > 0 || canOpenSoft();
  }

  /**
   * Returns true if the server can open a connection.
   */
  public boolean canOpenSoft()
  {
    int state = _state;

    if (state == ST_ACTIVE)
      return true;
    else if (ST_STARTING <= state && state < ST_ACTIVE) {
      long now = Alarm.getCurrentTime();

      if (now < _lastFailConnectTime + _dynamicFailRecoverTime) {
	return false;
      }

      int warmupState = _warmupState;

      if (warmupState < 0) {
	return (_failTime - warmupState * _failChunkTime < now);
      }
      else if (WARMUP_MAX <= warmupState)
	return true;

      int connectionMax = WARMUP_CONNECTION_MAX[warmupState];

      int idleCount = getIdleCount();
      int activeCount = _activeCount + _startingCount;
      int totalCount = activeCount + idleCount;
      
      return totalCount < connectionMax;
    }
    else {
      return false;
    }
  }

  /**
   * Return true if active.
   */
  public boolean isEnabled()
  {
    int state = _state;
    
    return ST_STARTING <= state && state <= ST_ACTIVE;
  }
  
  private void toActive()
  {
    synchronized (this) {
      if (_state < ST_CLOSED)
	_state = ST_ACTIVE;
    }
  }
  
  public void toBusy()
  {
    _lastBusyTime = Alarm.getCurrentTime();
    _firstSuccessTime = 0;
    
    synchronized (this) {
      _busyCountTotal++;
      
      if (_state < ST_CLOSED)
	_state = ST_BUSY;
    }
  }
  
  public void toFail()
  {
    _failTime = Alarm.getCurrentTime();
    _lastFailTime = _failTime;
    _firstSuccessTime = 0;
    
    synchronized (this) {
      _failCountTotal++;
      
      if (_state < ST_CLOSED)
	_state = ST_FAIL;
    }

    clearRecycle();
  }

  /**
   * Called when the socket read/write fails.
   */
  public void failSocket()
  {
    synchronized (this) {
      _failCountTotal++;
      
      long now = Alarm.getCurrentTime();
      _firstSuccessTime = 0;

      // only degrade one per 100ms
      if (now - _failTime >= 100) {
	_warmupState--;
	_failTime = now;
	_lastFailTime = _failTime;
      }

      if (_warmupState < WARMUP_MIN)
	_warmupState = WARMUP_MIN;
      
      if (_state < ST_CLOSED)
	_state = ST_FAIL;
    }