gt:       this( componentId, null );
context: /*******************************************************************************
   * Copyright (c) 2009  Eucalyptus Systems, Inc.
   * 
   *  This program 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, only version 3 of the License.
   * 
   * 
   *  This file 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.  See the GNU General Public License
   *  for more details.
   * 
   *  You should have received a copy of the GNU General Public License along
   *  with this program.  If not, see <http://www.gnu.org/licenses/>.
   * 
   *  Please contact Eucalyptus Systems, Inc., 130 Castilian
   *  Dr., Goleta, CA 93101 USA or visit <http://www.eucalyptus.com/licenses/>
   *  if you need additional information or have any questions.
   * 
   *  This file may incorporate work covered under the following copyright and
   *  permission notice:
   * 
   *    Software License Agreement (BSD License)
   * 
   *    Copyright (c) 2008, Regents of the University of California
   *    All rights reserved.
   * 
   *    Redistribution and use of this software in source and binary forms, with
   *    or without modification, are permitted provided that the following
   *    conditions are met:
   * 
   *      Redistributions of source code must retain the above copyright notice,
   *      this list of conditions and the following disclaimer.
   * 
   *      Redistributions in binary form must reproduce the above copyright
   *      notice, this list of conditions and the following disclaimer in the
   *      documentation and/or other materials provided with the distribution.
   * 
   *    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
   *    IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED
   *    TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
   *    PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER
   *    OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
   *    EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
   *    PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
   *    PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
   *    LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
   *    NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
   *    SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. USERS OF
   *    THIS SOFTWARE ACKNOWLEDGE THE POSSIBLE PRESENCE OF OTHER OPEN SOURCE
   *    LICENSED MATERIAL, COPYRIGHTED MATERIAL OR PATENTED MATERIAL IN THIS
   *    SOFTWARE, AND IF ANY SUCH MATERIAL IS DISCOVERED THE PARTY DISCOVERING
   *    IT MAY INFORM DR. RICH WOLSKI AT THE UNIVERSITY OF CALIFORNIA, SANTA
   *    BARBARA WHO WILL THEN ASCERTAIN THE MOST APPROPRIATE REMEDY, WHICH IN
   *    THE REGENTS DISCRETION MAY INCLUDE, WITHOUT LIMITATION, REPLACEMENT
   *    OF THE CODE SO IDENTIFIED, LICENSING OF THE CODE SO IDENTIFIED, OR
   *    WITHDRAWAL OF THE CODE CAPABILITY TO THE EXTENT NEEDED TO COMPLY WITH
   *    ANY SUCH LICENSES OR RIGHTS.
   *******************************************************************************
   * @author chris grzegorczyk <grze@eucalyptus.com>
   */
  
  package com.eucalyptus.component;
  
  import java.util.Collection;
  import java.util.Collections;
  import java.util.List;
  import java.util.Map;
  import java.util.NoSuchElementException;
  import java.util.concurrent.Callable;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.ConcurrentSkipListMap;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.Future;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import org.apache.log4j.Logger;
  import com.eucalyptus.bootstrap.Bootstrap;
  import com.eucalyptus.bootstrap.BootstrapArgs;
  import com.eucalyptus.bootstrap.Databases;
  import com.eucalyptus.bootstrap.Host;
  import com.eucalyptus.bootstrap.Hosts;
  import com.eucalyptus.component.Component.State;
  import com.eucalyptus.empyrean.DestroyServiceType;
  import com.eucalyptus.empyrean.Empyrean;
  import com.eucalyptus.empyrean.ServiceId;
  import com.eucalyptus.empyrean.ServiceTransitionType;
  import com.eucalyptus.event.ClockTick;
  import com.eucalyptus.event.EventListener;
  import com.eucalyptus.event.Listeners;
  import com.eucalyptus.records.EventRecord;
  import com.eucalyptus.records.EventType;
  import com.eucalyptus.records.Logs;
  import com.eucalyptus.system.Threads;
  import com.eucalyptus.util.Exceptions;
  import com.eucalyptus.util.Internets;
  import com.eucalyptus.util.TypeMappers;
  import com.eucalyptus.util.async.AsyncRequests;
  import com.eucalyptus.util.async.Futures;
  import com.google.common.base.Function;
  import com.google.common.base.Joiner;
  import com.google.common.base.Predicate;
  import com.google.common.base.Predicates;
  import com.google.common.base.Supplier;
  import com.google.common.base.Throwables;
  import com.google.common.collect.Collections2;
  import com.google.common.collect.Iterables;
  import com.google.common.collect.Lists;
  import com.google.common.collect.MapMaker;
  import com.google.common.primitives.Ints;
  import edu.ucsb.eucalyptus.msgs.BaseMessage;
  
  public class Topology {
    private static Logger                                         LOG          = Logger.getLogger( Topology.class );
    private static Topology                                       singleton    = null;                                                                   //TODO:GRZE:handle differently for remote case?
    private Integer                                               currentEpoch = 0;                                                                      //TODO:GRZE: get the right initial epoch value from membership bootstrap
    private final ConcurrentMap<ServiceKey, ServiceConfiguration> services     = new ConcurrentSkipListMap<Topology.ServiceKey, ServiceConfiguration>( );
    
    private enum Queue implements Function<Callable, Future> {
      INTERNAL( 1 ) {
        ServiceConfiguration internal;
        
        @Override
        ServiceConfiguration queue( ) {
          if ( this.internal == null ) {
            this.internal = ServiceConfigurations.createEphemeral( Empyrean.INSTANCE, Topology.class.getSimpleName( ), "internal",
                                                                   ServiceUris.internal( Empyrean.INSTANCE ) );
          }
          return this.internal;
        }
      },
      EXTERNAL( 32 ) {
        ServiceConfiguration external;
        
        @Override
        ServiceConfiguration queue( ) {
          if ( this.external == null ) {
            this.external = ServiceConfigurations.createEphemeral( Empyrean.INSTANCE, Topology.class.getSimpleName( ), "external",
                                                                   ServiceUris.internal( Empyrean.INSTANCE ) );
          }
          return this.external;
        }
      };
      private final int numWorkers;
      
      private Queue( final int numWorkers ) {
        this.numWorkers = numWorkers;
      }
      
      abstract ServiceConfiguration queue( );
      
      @Override
      public Future apply( final Callable call ) {
        Logs.extreme( ).debug( Topology.class.getSimpleName( ) + ": queueing " + call.toString( ) );
        Logs.extreme( ).debug( Threads.currentStackRange( 3, 9 ) );
        return Threads.enqueue( this.queue( ), this.numWorkers, call );
      }
      
      @SuppressWarnings( "unchecked" )
      public <C> Future<C> enqueue( final Callable<C> call ) {
        return this.apply( call );
      }
      
    };
    
    private enum TopologyTimer implements EventListener<ClockTick> {
      INSTANCE;
      private static final AtomicInteger counter = new AtomicInteger( 0 );
      private static final AtomicBoolean busy  = new AtomicBoolean( false );
      
      @Override
      public void fireEvent( final ClockTick event ) {
        Callable<Object> call = new Callable<Object>( ) {
          public Object call( ) {
            try {
              TimeUnit.SECONDS.sleep( 10 );
              return RunChecks.INSTANCE.call( );
            } catch ( InterruptedException ex ) {
              return Exceptions.maybeInterrupted( ex );
            } finally {
              busy.set( false );
            }
          }
        };
        if ( Hosts.isCoordinator( ) && busy.compareAndSet( false, true ) ) {
          try {
            Queue.INTERNAL.enqueue( call );
          } catch ( Exception ex ) {
            busy.set( false );
          }
        } else if ( counter.incrementAndGet( ) % 3 == 0 && busy.compareAndSet( false, true ) ) {
          try {
            Queue.INTERNAL.enqueue( call );
          } catch ( Exception ex ) {
            busy.set( false );
          }
        }
      }
      
    }
    
    private Topology( final int i ) {
      super( );
      this.currentEpoch = i;
      Listeners.register( ClockTick.class, TopologyTimer.INSTANCE );
    }
    
    private static Predicate<ServiceConfiguration> componentFilter( final Class<? extends ComponentId> c ) {
      return new Predicate<ServiceConfiguration>( ) {
        
        @Override
        public boolean apply( final ServiceConfiguration input ) {
          return input.getComponentId( ).getClass( ).equals( c );
        }
      };
    }
    
    public static void populateServices( final ServiceConfiguration config, BaseMessage msg ) {
      try {
        Predicate<ServiceConfiguration> filter = new Predicate<ServiceConfiguration>( ) {
          
          @Override
          public boolean apply( final ServiceConfiguration filterConfig ) {
            ComponentId filteredComponent = filterConfig.getComponentId( );
            ComponentId destComponent = config.getComponentId( );
            if ( filteredComponent.isDistributedService( ) ) {
              if ( destComponent.isAlwaysLocal( ) ) {
                return filterConfig.lookupState( ).ordinal( ) >= Component.State.NOTREADY.ordinal( );
              } else if ( destComponent.isPartitioned( ) && filteredComponent.isPartitioned( ) ) {
                return config.getPartition( ).equals( filterConfig.getPartition( ) );
              } else {
                return true;
              }
            } else {
              return false;
            }
          }
        };
        Function<ServiceConfiguration, ServiceId> typeMapper = TypeMappers.lookup( ServiceConfiguration.class, ServiceId.class );
        if ( Hosts.isCoordinator( ) ) {
          msg.set_epoch( Topology.epoch( ) );
          for ( ServiceConfiguration s : Topology.getInstance( ).getServices( ).values( ) ) {
            if ( !filter.apply( s ) ) {
              continue;
            } else {
              msg.get_services( ).add( typeMapper.apply( s ) );
            }
          }
          for ( Component c : Components.list( ) ) {
            for ( ServiceConfiguration s : c.services( ) ) {
              if ( !filter.apply( s ) ) {
                continue;
              } else {
                if ( !msg.get_services( ).contains( s ) && State.DISABLED.apply( s ) ) {
                  msg.get_disabledServices( ).add( typeMapper.apply( s ) );
                } else if ( !msg.get_services( ).contains( s ) && State.NOTREADY.ordinal( ) >= s.getStateMachine( ).getState( ).ordinal( ) ) {
                  msg.get_notreadyServices( ).add( typeMapper.apply( s ) );
                }
              }
            }
          }
        }
      } catch ( Exception ex ) {
        Logs.extreme( ).error( ex, ex );
      }
    }
    
    public static void touch( final ServiceTransitionType msg ) {//TODO:GRZE: @Service interceptor
      if ( !Hosts.isCoordinator( ) && ( msg.get_epoch( ) != null ) ) {
        for ( ServiceConfiguration conf : Lists.transform( msg.get_services( ), ServiceConfigurations.ServiceIdToServiceConfiguration.INSTANCE ) ) {
          if ( !conf.isVmLocal( ) ) {
            enable( conf );
          }
        }
        for ( ServiceConfiguration conf : Lists.transform( msg.get_disabledServices( ), ServiceConfigurations.ServiceIdToServiceConfiguration.INSTANCE ) ) {
          if ( !conf.isVmLocal( ) ) {
            try {
              disable( conf ).get( );
            } catch ( InterruptedException ex ) {
              Exceptions.maybeInterrupted( ex );
            } catch ( ExecutionException ex ) {
              Logs.extreme( ).error( ex, ex );
            }
          }
        }
        for ( ServiceConfiguration conf : Lists.transform( msg.get_notreadyServices( ), ServiceConfigurations.ServiceIdToServiceConfiguration.INSTANCE ) ) {
          if ( !conf.isVmLocal( ) ) {
            try {
              disable( conf ).get( );
              transition( State.NOTREADY ).apply( conf ).get( );
            } catch ( InterruptedException ex ) {
              Exceptions.maybeInterrupted( ex );
            } catch ( ExecutionException ex ) {
              Logs.extreme( ).error( ex, ex );
            }
          }
        }
        Topology.getInstance( ).currentEpoch = Ints.max( Topology.getInstance( ).currentEpoch, msg.get_epoch( ) );
      }
    }
    
    public static boolean check( final BaseMessage msg ) {
      if ( !Hosts.isCoordinator( ) && ( msg.get_epoch( ) != null ) ) {
        return Topology.getInstance( ).epoch( ) <= msg.get_epoch( );
      } else {
        return true;
      }
    }
    
    private static Topology getInstance( ) {
      if ( singleton != null ) {
        return singleton;
      } else {
        synchronized ( Topology.class ) {
          if ( singleton != null ) {
            return singleton;
          } else {
            return ( singleton = new Topology( Hosts.maxEpoch( ) ) );
          }
        }
      }
    }
    
    private Integer getEpoch( ) {
      return this.currentEpoch;
    }
    
    public static int epoch( ) {
      return Topology.getInstance( ).getEpoch( );
    }
    
    public static Function<ServiceConfiguration, Future<ServiceConfiguration>> transition( final Component.State toState ) {
      final Function<ServiceConfiguration, Future<ServiceConfiguration>> transition = new Function<ServiceConfiguration, Future<ServiceConfiguration>>( ) {
        private final List<Component.State> serializedStates = Lists.newArrayList( Component.State.ENABLED, Component.State.STOPPED );
        
        @Override
        public Future<ServiceConfiguration> apply( final ServiceConfiguration input ) {
          final Callable<ServiceConfiguration> call = Topology.callable( input, Topology.get( toState ) );
          if ( Bootstrap.isOperational( ) ) {
            final Queue workQueue = ( this.serializedStates.contains( toState )
                                                                               ? Queue.INTERNAL
                                                                               : Queue.EXTERNAL );
            return workQueue.enqueue( call );
          } else {
            try {
              return Futures.predestinedFuture( input );
            } catch ( Exception ex ) {
              return Futures.predestinedFuture( input );
            }
          }
        }
      };
      return transition;
    }
    
    private static Future<ServiceConfiguration> check( final ServiceConfiguration config ) {
      return Queue.EXTERNAL.enqueue( Topology.callable( config, Topology.check( ) ) );
    }
    
    public static Future<ServiceConfiguration> stop( final ServiceConfiguration config ) {
      return transition( State.STOPPED ).apply( config );
    }
    
    public static Future<ServiceConfiguration> destroy( final ServiceConfiguration config ) {
      try {
        ServiceConfigurations.remove( config );
      } catch ( Exception ex ) {
        LOG.error( ex );
        Logs.extreme( ).debug( ex, ex );
      }
      try {
        Topology.disable( config ).get( );
      } catch ( Exception ex ) {
        Exceptions.maybeInterrupted( ex );
        LOG.error( ex );
        Logs.extreme( ).debug( ex, ex );
      }
      try {
        Topology.stop( config ).get( );
      } catch ( Exception ex ) {
        Exceptions.maybeInterrupted( ex );
        LOG.error( ex );
        Logs.extreme( ).debug( ex, ex );
      }
      try {
        Component comp = Components.lookup( config.getComponentId( ) );
        comp.destroy( config );
      } catch ( Exception ex ) {
        Exceptions.maybeInterrupted( ex );
        LOG.error( ex );
        Logs.extreme( ).debug( ex, ex );
      }
      if ( Hosts.isCoordinator( ) ) {
        DestroyServiceType msg = new DestroyServiceType( );
        try {
          msg.getServices( ).add( TypeMappers.transform( config, ServiceId.class ) );
          for ( Host h : Hosts.list( ) ) {
            if ( !h.isLocalHost( ) && h.hasBootstrapped( ) ) { 
              try {
                AsyncRequests.sendSync( ServiceConfigurations.createEphemeral( Empyrean.INSTANCE, h.getBindAddress( ) ), msg );
              } catch ( Exception ex ) {
                Exceptions.maybeInterrupted( ex );
                LOG.error( ex );
                Logs.extreme( ).debug( ex, ex );
              }
            }
          }
        } catch ( Exception ex ) {
          LOG.error( ex );
          Logs.extreme( ).debug( ex, ex );
        }
      }
      try {
        ServiceTransitions.StateCallbacks.PROPERTIES_REMOVE.fire( config );
      } catch ( Exception ex ) {
        Exceptions.maybeInterrupted( ex );
        LOG.error( ex );
        Logs.extreme( ).debug( ex, ex );
      }
      return Futures.predestinedFuture( config );
    }
    
    public static Future<ServiceConfiguration> load( final ServiceConfiguration config ) {
      return transition( State.LOADED ).apply( config );
    }
    
    public static Future<ServiceConfiguration> start( final ServiceConfiguration config ) {
      return transition( State.DISABLED ).apply( config );
    }
    
    public static Future<ServiceConfiguration> enable( final ServiceConfiguration config ) {
      return transition( State.ENABLED ).apply( config );
    }
    
    public static Future<ServiceConfiguration> disable( final ServiceConfiguration config ) {
      return transition( State.DISABLED ).apply( config );
    }
    
    private ServiceConfiguration lookup( final ServiceKey serviceKey ) {
      return this.getServices( ).get( serviceKey );
    }
    
    private interface TransitionGuard {
      boolean tryEnable( final ServiceConfiguration config );
      
      boolean nextEpoch( );
      
      boolean tryDisable( final ServiceConfiguration config );
    }
    
    private TransitionGuard cloudControllerGuard( ) {
      return new TransitionGuard( ) {
        
        @Override
        public boolean nextEpoch( ) {
          Topology.this.currentEpoch++;
          return true;
        }
        
        @Override
        public boolean tryEnable( final ServiceConfiguration config ) {
          final ServiceKey serviceKey = ServiceKey.create( config );
          final ServiceConfiguration curr = Topology.this.getServices( ).putIfAbsent( serviceKey, config );
          if ( ( curr != null ) && !curr.equals( config ) ) {
            return false;
          } else if ( ( curr != null ) && curr.equals( config ) ) {
            return true;
          } else {
            Topology.this.currentEpoch++;
            return true;
          }
        }
        
        @Override
        public boolean tryDisable( final ServiceConfiguration config ) {
          final ServiceKey serviceKey = ServiceKey.create( config );
          return !config.equals( Topology.this.getServices( ).get( serviceKey ) )
                 || ( Topology.this.getServices( ).remove( serviceKey, config ) && this.nextEpoch( ) );
        }
        
      };
    }
    
    private TransitionGuard remoteGuard( ) {
      return new TransitionGuard( ) {
        
        @Override
        public boolean nextEpoch( ) {
          return true;
        }
        
        @Override
        public boolean tryEnable( final ServiceConfiguration config ) {
          final ServiceKey serviceKey = ServiceKey.create( config );
          final ServiceConfiguration curr = Topology.this.getServices( ).put( serviceKey, config );
          if ( ( curr != null ) && !curr.equals( config ) ) {
            transition( State.DISABLED ).apply( curr );
            return false;
          } else if ( ( curr != null ) && curr.equals( config ) ) {
            return true;
          } else {
            return true;
          }
        }
        
        @Override
        public boolean tryDisable( final ServiceConfiguration config ) {
          final ServiceKey serviceKey = ServiceKey.create( config );
          return ( Topology.this.getServices( ).remove( serviceKey, config ) || !config.equals( Topology.this.getServices( ).get( serviceKey ) ) )
                 && this.nextEpoch( );
        }
      };
    }
    
    public static class ServiceKey implements Comparable<ServiceKey> {
      private final Partition   partition;
      private final ComponentId componentId;
      
      static ServiceKey create( final ComponentId compId, final Partition partition ) {
        return new ServiceKey( compId, partition );
      }
      
      static ServiceKey create( final ServiceConfiguration config ) {
        if ( config.getComponentId( ).isPartitioned( ) && !Empyrean.class.equals( config.getComponentId( ).partitionParent( ) ) ) {
          final Partition p = Partitions.lookup( config );
          return new ServiceKey( config.getComponentId( ), p );
        } else if ( config.getComponentId( ).isAlwaysLocal( ) || ( config.getComponentId( ).isCloudLocal( ) && !config.getComponentId( ).isRegisterable( ) ) ) {
          final Partition p = Partitions.lookupInternal( config );
          return new ServiceKey( config.getComponentId( ), p );
        } else {
          return new ServiceKey( config.getComponentId( ) );
        }
      }
      
      ServiceKey( final ComponentId componentId ) {
  