basicresourcepool.java

c3p0数据库连接池实现源码

	    exampleResource = resc;
    }

    // should NOT be called from synchronized method
    private void synchronousRemoveArbitraryResource()
    { 
	Object removeMe = null;

	synchronized ( this )
	    {
		if (unused.size() > 0)
		    {
			removeMe = unused.get(0);
			managed.remove(removeMe);
			unused.remove(removeMe);
		    }
		else
		    {
			Set checkedOut = cloneOfManaged().keySet();
			if ( checkedOut.isEmpty() )
			    {
				unexpectedBreak();
				logger.severe("A pool from which a resource is requested to be removed appears to have no managed resources?!");
			    }
			else
			    excludeResource( checkedOut.iterator().next() );
		    }
	    }

	if (removeMe != null)
	    destroyResource( removeMe, true );
    }

    private void removeResource(Object resc)
    { removeResource( resc, false ); }

    private void removeResource(Object resc, boolean synchronous)
    {
	managed.remove(resc);
	unused.remove(resc);
	destroyResource(resc, synchronous);
	asyncFireResourceRemoved( resc, false, managed.size(), unused.size(), excluded.size() );
	if (Debug.DEBUG && Debug.TRACE == Debug.TRACE_MAX) trace();
	//System.err.println("RESOURCE REMOVED!");
    }

    //when we want to conceptually remove a checked
    //out resource from the pool
    private void excludeResource(Object resc)
    {
	managed.remove(resc);
	excluded.add(resc);
	if (Debug.DEBUG && unused.contains(resc) )
	    throw new InternalError( "We should only \"exclude\" checked-out resources!" );
	asyncFireResourceRemoved( resc, true, managed.size(), unused.size(), excluded.size() );
    }

    private void removeTowards( int new_sz )
    {
	int num_to_remove = managed.size() - new_sz;
	int count = 0;
	for (Iterator ii = cloneOfUnused().iterator(); 
	     ii.hasNext() && count < num_to_remove; 
	     ++count)
	    {
		Object resc = ii.next();
		removeResource( resc );
	    }
    }

    private void cullExpiredAndUnused()
    {
	for ( Iterator ii = cloneOfUnused().iterator(); ii.hasNext(); )
	    {
		Object resc = ii.next();
		if ( isExpired( resc ) )
		    {
			if (Debug.DEBUG && logger.isLoggable( MLevel.FINER ))
			    logger.log( MLevel.FINER, "Removing expired resource: " + resc + " [" + this + "]");

			target_pool_size = Math.max( min, target_pool_size - 1 ); //idling out a resource resources the target size to match

			//System.err.println("c3p0-JENNIFER: removing expired resource: " + resc + " [" + this + "]");
			removeResource( resc );
		    }
	    }
	ensureMinResources();
    }

    private void checkIdleResources()
    {
	List u = cloneOfUnused();
	for ( Iterator ii = u.iterator(); ii.hasNext(); )
	    {
		Object resc = ii.next();
		if ( idleCheckResources.add( resc ) )
		    taskRunner.postRunnable( new AsyncTestIdleResourceTask( resc ) );
	    }

	if (Debug.DEBUG && Debug.TRACE == Debug.TRACE_MAX) trace();
    }

    private boolean isExpired( Object resc )
    {
	if (max_resource_age > 0)
	    {
		Date d = (Date) managed.get( resc );
		long now = System.currentTimeMillis();
		long age = now - d.getTime();
		boolean expired = ( age > max_resource_age );

		if ( Debug.DEBUG && Debug.TRACE == Debug.TRACE_MAX && logger.isLoggable( MLevel.FINEST ) )
		    {
			if (expired)
			    logger.log(MLevel.FINEST, 
				       "EXPIRED resource: " + resc + " ---> age: " + age + 
				       "   max: " + max_resource_age + " [" + this + "]");
			else
			    logger.log(MLevel.FINEST, 
				       "resource age is okay: " + resc + " ---> age: " + age + 
				       "   max: " + max_resource_age + " [" + this + "]");
		    }
		return expired;
	    }
	else
	    return false; 
    }

//     private boolean resourcesInIdleCheck()
//     { return idleCheckresources.size() > 0; }

//     private int countAvailable()
//     { return unused.size() - idleCheckResources.size(); }

    private void ensureStartResources()
    { recheckResizePool(); }

    private void ensureMinResources()
    { recheckResizePool(); }

    private boolean attemptRefurbishResourceOnCheckout( Object resc )
    {
	try
	    { 
		mgr.refurbishResourceOnCheckout(resc); 
		return true;
	    }
	catch (Exception e)
	    {
		//uh oh... bad resource...
		if (Debug.DEBUG) 
		    {
			//e.printStackTrace();
			if (logger.isLoggable( MLevel.FINE ))
			    logger.log( MLevel.FINE, "A resource could not be refurbished on checkout.", e );
		    }
		return false;
	    }
    }

    private boolean attemptRefurbishResourceOnCheckin( Object resc )
    {
	try
	    { 
		mgr.refurbishResourceOnCheckin(resc); 
		return true;
	    }
	catch (Exception e)
	    {
		//uh oh... bad resource...
		if (Debug.DEBUG) 
		    {
			//e.printStackTrace();
			if (logger.isLoggable( MLevel.FINE ))
			    logger.log( MLevel.FINE, "A resource could not be refurbished on checkin.", e );
		    }
		return false;
	    }
    }

    private void ensureNotBroken() throws ResourcePoolException
    {
	if (broken) 
	    throw new ResourcePoolException("Attempted to use a closed or broken resource pool");
    }

    private void trace()
    {
	if ( logger.isLoggable( MLevel.FINEST ) )
 	    {
   		String exampleResStr = ( exampleResource == null ?
   					 "" :
   					 " (e.g. " + exampleResource +")");
   		logger.finest("trace " + this + " [managed: " + managed.size() + ", " +
   			      "unused: " + unused.size() + ", excluded: " +
   			      excluded.size() + ']' + exampleResStr );
	    }
    }

    private final HashMap cloneOfManaged()
    { return (HashMap) managed.clone(); }

    private final LinkedList cloneOfUnused()
    { return (LinkedList) unused.clone(); }

    private final HashSet cloneOfExcluded()
    { return (HashSet) excluded.clone(); }

    /*
     *  task we post to separate thread to acquire
     *  pooled resources
     */
    class AcquireTask implements Runnable
    {
	boolean success = false;
	int     num;

	public AcquireTask() 
	{ incrementPendingAcquires(); }

	public void run()
	{
	    try
		{
		    Exception lastException = null;
		    for (int i = 0; shouldTry( i ); ++i)
			{
			    try
				{
				    if (i > 0)
					Thread.sleep(acq_attempt_delay); 
				    
				    //we don't want this call to be sync'd
				    //on the pool, so that resource acquisition
				    //does not interfere with other pool clients.
				    BasicResourcePool.this.doAcquire();

				    success = true;
				}
			    catch (Exception e)
				{
				    if (Debug.DEBUG) 
					{
					    //e.printStackTrace();
					    if (logger.isLoggable( MLevel.FINE ))
						logger.log( MLevel.FINE, "An exception occurred while acquiring a resource.", e );
					}
				    lastException = e;
				}
			}
		    if (!success) 
			{
			    if ( logger.isLoggable( MLevel.WARNING ) )
				{
				    logger.log( MLevel.WARNING,
						this + " -- Acquisition Attempt Failed!!! Clearing pending acquires. " +
						"While trying to acquire a needed new resource, we failed " +
						"to succeed more than the maximum number of allowed " +
						"acquisition attempts (" + num_acq_attempts + "). " + 
						(lastException == null ? "" : "Last acquisition attempt exception: "),
						lastException);
				}
			    if (break_on_acquisition_failure)
				{
				    //System.err.println("\tTHE RESOURCE POOL IS PERMANENTLY BROKEN!");
				    if ( logger.isLoggable( MLevel.SEVERE ) )
					logger.severe("A RESOURCE POOL IS PERMANENTLY BROKEN! [" + this + "]");
				    unexpectedBreak();
				}
			    else
				forceKillAcquires();
			}
		    else
			recheckResizePool();
		}
	    catch ( ResourceClosedException e ) // one of our async threads died
		{
		    //e.printStackTrace();
		    if ( Debug.DEBUG )
			{
			    if ( logger.isLoggable( MLevel.FINE ) )
				logger.log( MLevel.FINE, "a resource pool async thread died.", e );
			}
		    unexpectedBreak();
		}
	    catch (InterruptedException e) //from force kill acquires
		{
		    if ( logger.isLoggable( MLevel.WARNING ) )
			{
			    logger.log( MLevel.WARNING,
					BasicResourcePool.this + " -- Thread unexpectedly interrupted while waiting for stale acquisition attempts to die.",
					e );
			}

// 		    System.err.println(BasicResourcePool.this + " -- Thread unexpectedly interrupted while waiting for stale acquisition attempts to die.");
// 		    e.printStackTrace();

		    recheckResizePool();
		}
	    finally
		{ decrementPendingAcquires(); }
	}

	private boolean shouldTry(int attempt_num)
	{
	    //try if we haven't already succeeded
	    //and someone hasn't signalled that our resource source is down
	    //and not max attempts is set,
	    //or we are less than the set limit
	    return 
		!success && 
		!isForceKillAcquiresPending() &&
		(num_acq_attempts <= 0 || attempt_num < num_acq_attempts);
	}
    }

    /*
     *  task we post to separate thread to remove
     *  unspecified pooled resources
     *
     *  TODO: do removal and destruction synchronously
     *        but carefully not synchronized during the
     *        destruction of the resource.
     */
    class RemoveTask implements Runnable
    {
	public RemoveTask() 
	{ incrementPendingRemoves(); }

	public void run()
	{
	    try
		{
		    synchronousRemoveArbitraryResource();
		    recheckResizePool();
		}
	    finally
		{ decrementPendingRemoves(); }
	}
    }

    class CullTask extends TimerTask
    {
	public void run()
	{
	    try
		{
		    if (Debug.DEBUG && Debug.TRACE >= Debug.TRACE_MED && logger.isLoggable( MLevel.FINER ))
			logger.log( MLevel.FINER, "Checking for expired resources - " + new Date() + " [" + BasicResourcePool.this + "]");
		    synchronized ( BasicResourcePool.this )
			{ cullExpiredAndUnused(); }
		}
	    catch ( ResourceClosedException e ) // one of our async threads died
		{
		    if ( Debug.DEBUG )
			{
			    if ( logger.isLoggable( MLevel.FINE ) )
				logger.log( MLevel.FINE, "a resource pool async thread died.", e );
			}
		    unexpectedBreak();
		}
	}
    }

    // this is run by a single-threaded timer, so we don't have
    // to worry about multiple threads executing the task at the same 
    // time 
    class CheckIdleResourcesTask extends TimerTask
    {
	public void run()
	{
	    try
		{
		    //System.err.println("c3p0-JENNIFER: refurbishing idle resources - " + new Date() + " [" + BasicResourcePool.this + "]");
		    if (Debug.DEBUG && Debug.TRACE >= Debug.TRACE_MED && logger.isLoggable(MLevel.FINER))
			logger.log(MLevel.FINER, "Refurbishing idle resources - " + new Date() + " [" + BasicResourcePool.this + "]");
		    synchronized ( BasicResourcePool.this )
			{ checkIdleResources(); }
		}
	    catch ( ResourceClosedException e ) // one of our async threads died
		{
		    //e.printStackTrace();
		    if ( Debug.DEBUG )
			{
			    if ( logger.isLoggable( MLevel.FINE ) )
				logger.log( MLevel.FINE, "a resource pool async thread died.", e );
			}
		    unexpectedBreak();
		}
	}
    }

    class AsyncTestIdleResourceTask implements Runnable
    {
	// unchanging after ctor
	Object resc;

	// protected by this' lock
	boolean pending = true;
	boolean failed;

	AsyncTestIdleResourceTask( Object resc )
	{ this.resc = resc; }

// 	synchronized boolean pending()
// 	{ return pending; }

// 	synchronized boolean failed()
// 	{
// 	    if (pending)
// 		throw new RuntimeException(this + " You bastard! You can't check if the test failed wile it's pending!");
// 	    return 
// 		failed;
// 	}

// 	synchronized void unpend()
// 	{ pending = false; }

// 	synchronized void setFailed( boolean f )
// 	{ this.failed = f; }

	public void run()
	{
	    try
		{
		    boolean failed;
		    try
			{ 
			    mgr.refurbishIdleResource( resc ); 
			    failed = false;

			    //trace();
			    //Thread.sleep(1000); //DEBUG: make sure collision detection works
			}
		    catch ( Exception e )
			{
			    //System.err.println("c3p0: An idle resource is broken and will be purged.");
			    //System.err.print("c3p0 [broken resource]: ");
			    //e.printStackTrace();

			    if ( logger.isLoggable( MLevel.WARNING ) )
				logger.log( MLevel.WARNING, "BasicResourcePool: An idle resource is broken and will be purged.", e);

			    failed = true;
			}
		    
		    synchronized (BasicResourcePool.this)
			{
			    if ( failed )
				{
				    if ( managed.keySet().contains( resc ) ) //resc might have been culled as expired while we tested
					{
					    removeResource( resc ); 
					    ensureMinResources();
					}
				}
			}
		}
	    finally
		{
		    synchronized (BasicResourcePool.this)
			{
			    idleCheckResources.remove( resc );
			    BasicResourcePool.this.notifyAll();
			}
		}
	}
    }

//     static class CheckInProgressResourceHolder
//     {
// 	Object checkResource;

// 	public synchronized void setCheckResource( Object resc )
// 	{ 
// 	    this.checkResource = resc; 
// 	    this.notifyAll();
// 	}

// 	public void unsetCheckResource()
// 	{ setCheckResource( null ); }

// 	/**
// 	 * @return true if we actually had to wait
// 	 */
// 	public synchronized boolean awaitNotInCheck( Object resc )
// 	{
// 	    boolean had_to_wait = false;
// 	    boolean set_interrupt = false;
// 	    while ( checkResource == resc )
// 		{
// 		    try
// 			{
// 			    had_to_wait = true;
// 			    this.wait(); 
// 			}
// 		    catch ( InterruptedException e )
// 			{ 
// 			    e.printStackTrace();
// 			    set_interrupt = true;
// 			}
// 		}
// 	    if ( set_interrupt )
// 		Thread.currentThread().interrupt();
// 	    return had_to_wait;
// 	}
//     }
}