// DBPortPool.java

/**

* Copyright (C) 2008 10gen Inc.

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*/

package com.mongodb;

import com.mongodb.util.ConnectionPoolStatisticsBean;

import com.mongodb.util.SimplePool;

import com.mongodb.util.management.JMException;

import com.mongodb.util.management.MBeanServerFactory;

import java.io.InterruptedIOException;

import java.util.ArrayList;

import java.util.Collections;

import java.util.HashMap;

import java.util.List;

import java.util.Map;

import java.util.concurrent.Semaphore;

import java.util.concurrent.atomic.AtomicInteger;

import java.util.logging.Level;

/**

* This class is NOT part of the public API. Be prepared for non-binary compatible changes in minor releases.

*/

public class DBPortPool extends SimplePool<DBPort> {

public String getHost() {

return _addr.getHost();

}

public int getPort() {

return _addr.getPort();

}

public synchronized ConnectionPoolStatisticsBean getStatistics() {

return new ConnectionPoolStatisticsBean(getTotal(), getInUse(), getInUseConnections());

}

private InUseConnectionBean[] getInUseConnections() {

List<InUseConnectionBean> inUseConnectionInfoList = new ArrayList<InUseConnectionBean>();

long currentNanoTime = System.nanoTime();

for (DBPort port : _out) {

inUseConnectionInfoList.add(new InUseConnectionBean(port, currentNanoTime));

}

return inUseConnectionInfoList.toArray(new InUseConnectionBean[inUseConnectionInfoList.size()]);

}

static class Holder {

Holder( MongoOptions options ){

_options = options;

}

DBPortPool get( ServerAddress addr ){

DBPortPool p = _pools.get( addr );

if (p != null)

return p;

synchronized (_pools) {

p = _pools.get( addr );

if (p != null) {

return p;

}

p = createPool(addr);

_pools.put( addr , p);

try {

String on = createObjectName(addr);

if (MBeanServerFactory.getMBeanServer().isRegistered(on)) {

MBeanServerFactory.getMBeanServer().unregisterMBean(on);

Bytes.LOGGER.log(Level.INFO, "multiple Mongo instances for same host, jmx numbers might be off");

}

MBeanServerFactory.getMBeanServer().registerMBean(p, on);

} catch (JMException e) {

Bytes.LOGGER.log(Level.WARNING, "JMX registration error: " + e +

"\nConsider setting com.mongodb.MongoOptions.alwaysUseMBeans property to true." +

"\nContinuing...");

} catch (java.security.AccessControlException e) {

Bytes.LOGGER.log(Level.WARNING, "JMX registration error: " + e +

"\nContinuing...");

}

}

return p;

}

private DBPortPool createPool(final ServerAddress addr) {

if (isJava5 || _options.isAlwaysUseMBeans()) {

return new Java5MongoConnectionPool(addr, _options);

} else {

return new MongoConnectionPool(addr, _options);

}

}

void close(){

synchronized ( _pools ){

for ( DBPortPool p : _pools.values() ){

p.close();

try {

String on = createObjectName( p._addr );

if ( MBeanServerFactory.getMBeanServer().isRegistered(on) ){

MBeanServerFactory.getMBeanServer().unregisterMBean(on);

}

} catch ( JMException e ){

Bytes.LOGGER.log( Level.WARNING , "jmx de-registration error, continuing" , e );

}

}

}

}

private String createObjectName( ServerAddress addr ) {

String name = "com.mongodb:type=ConnectionPool,host=" + addr.toString().replace( ":" , ",port=" ) + ",instance=" + _serial;

if ( _options.description != null )

name += ",description=" + _options.description;

return name;

}

static {

isJava5 = System.getProperty("java.version").startsWith("1.5");

}

final MongoOptions _options;

final Map<ServerAddress,DBPortPool> _pools = Collections.synchronizedMap( new HashMap<ServerAddress,DBPortPool>() );

final int _serial = nextSerial.incrementAndGet();

// we use this to give each Holder a different mbean name

static AtomicInteger nextSerial = new AtomicInteger(0);

static final boolean isJava5;

}

// ----

public static class NoMoreConnection extends MongoInternalException {

private static final long serialVersionUID = -4415279469780082174L;

NoMoreConnection( String msg ){

super( msg );

}

}

public static class SemaphoresOut extends NoMoreConnection {

private static final long serialVersionUID = -4415279469780082174L;

private static final String message = "Concurrent requests for database connection have exceeded limit";

SemaphoresOut(){

super( message );

}

SemaphoresOut(int numPermits){

super( message + " of " + numPermits);

}

}

public static class ConnectionWaitTimeOut extends NoMoreConnection {

private static final long serialVersionUID = -4415279469780082174L;

ConnectionWaitTimeOut(int timeout) {

super("Connection wait timeout after " + timeout + " ms");

}

}

// ----

DBPortPool( ServerAddress addr , MongoOptions options ){

super( "DBPortPool-" + addr.toString() + ", options = " + options.toString() , options.connectionsPerHost );

_options = options;

_addr = addr;

_waitingSem = new Semaphore( _options.connectionsPerHost * _options.threadsAllowedToBlockForConnectionMultiplier );

}

protected long memSize( DBPort p ){

return 0;

}

@Override

protected int pick( int recommended, boolean couldCreate ){

int id = System.identityHashCode(Thread.currentThread());

for (int i = _avail.size() - 1; i >= 0; i--){

if ( _avail.get(i)._lastThread == id )

return i;

}

return couldCreate ? -1 : recommended;

}

/**

* @return

* @throws MongoException

*/

@Override

public DBPort get() {

DBPort port = null;

if ( ! _waitingSem.tryAcquire() )

throw new SemaphoresOut(_options.connectionsPerHost * _options.threadsAllowedToBlockForConnectionMultiplier);

try {

port = get( _options.maxWaitTime );

} catch (InterruptedException e) {

throw new MongoInterruptedException(e);

} finally {

_waitingSem.release();

}

if ( port == null )

throw new ConnectionWaitTimeOut( _options.maxWaitTime );

port._lastThread = System.identityHashCode(Thread.currentThread());

return port;

}

// return true if the exception is recoverable

boolean gotError( Exception e ){

if (e instanceof java.nio.channels.ClosedByInterruptException){

// this is probably a request that is taking too long

// so usually doesn't mean there is a real db problem

return true;

}

if ( e instanceof InterruptedIOException){

// we don't want to clear the port pool for a connection timing out or interrupted

return true;

}

Bytes.LOGGER.log( Level.WARNING , "emptying DBPortPool to " + getServerAddress() + " b/c of error" , e );

// force close all sockets

List<DBPort> all = new ArrayList<DBPort>();

while ( true ){

try {

DBPort temp = get(0);

if ( temp == null )

break;

all.add( temp );

} catch (InterruptedException interruptedException) {

throw new MongoInterruptedException(interruptedException);

}

}

for ( DBPort p : all ){

p.close();

done(p);

}

return false;

}

@Override

public void cleanup( DBPort p ){

p.close();

}

@Override

protected DBPort createNew(){

return new DBPort( _addr , this , _options );

}

public ServerAddress getServerAddress() {

return _addr;

}

final MongoOptions _options;

final private Semaphore _waitingSem;

final ServerAddress _addr;

boolean _everWorked = false;

}