org/objectweb/proactive/loadbalancing/LoadBalancer.java

/* 
 * ################################################################
 * 
 * ProActive: The Java(TM) library for Parallel, Distributed, 
 *            Concurrent computing with Security and Mobility
 * 
 * Copyright (C) 1997-2007 INRIA/University of Nice-Sophia Antipolis
 * Contact: proactive@objectweb.org
 * 
 * This library is free software; you can redistribute it and/or
 * modify it under the terms of the GNU Lesser General Public
 * License as published by the Free Software Foundation; either
 * version 2.1 of the License, or any later version.
 *  
 * This library 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
 * Lesser General Public License for more details.
 * 
 * You should have received a copy of the GNU Lesser General Public
 * License along with this library; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
 * USA
 *  
 *  Initial developer(s):               The ProActive Team
 *                        http://www.inria.fr/oasis/ProActive/contacts.html
 *  Contributor(s): 
 * 
 * ################################################################
 */ 
package org.objectweb.proactive.loadbalancing;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.Random;
import org.apache.log4j.Logger;
import org.objectweb.proactive.Body;
import org.objectweb.proactive.ProActive;
import org.objectweb.proactive.ProActiveInternalObject;
import org.objectweb.proactive.core.body.BodyMap;
import org.objectweb.proactive.core.body.LocalBodyStore;
import org.objectweb.proactive.core.body.migration.MigrationException;
import org.objectweb.proactive.core.exceptions.NonFunctionalException;
import org.objectweb.proactive.core.node.Node;
import org.objectweb.proactive.core.node.NodeException;
import org.objectweb.proactive.core.node.NodeFactory;
import org.objectweb.proactive.core.util.log.Loggers;
import org.objectweb.proactive.core.util.log.ProActiveLogger;
import org.objectweb.proactive.loadbalancing.metrics.Metric;
import org.objectweb.proactive.loadbalancing.metrics.MetricFactory;


public class LoadBalancer implements ProActiveInternalObject {
    public static Logger logger = ProActiveLogger.getLogger(Loggers.LOAD_BALANCING);
    
    protected LoadMonitor lm;
    protected Metric metric;
    protected Node myNode;
    protected ArrayList<LoadBalancer> loadBalancers;
    
    private static final int STEAL = 1;
    private static final int BALANCE = 2;
   
        protected String balancerName;
    protected Random randomizer;
        protected LoadBalancer myThis;
        protected InformationRecover informationRecover;


        public LoadBalancer(){}
        
        public LoadBalancer(MetricFactory mf){
        this.randomizer = new Random();
        this.metric = mf.getNewMetric();

       
        }
        

    public void startBalancing() {
        internalAction(BALANCE);
    }

    public void stealWork() {
        internalAction(STEAL);
    }

    
    public void sendActiveObjectsTo(Node remoteNode, double remoteRanking) {
        if (this.metric == null)
                return;
        if (this.metric.getRanking() < remoteRanking * LoadBalancingConstants.STEAL_FACTOR) { // it's better than me!
                sendActiveObjectsTo(remoteNode);
                }
        }
    
    protected void getActiveObjectsFrom (LoadBalancer remoteBalancer, double remoteRanking){
        if (this.metric == null)
                return;
                if (remoteRanking < this.metric.getRanking() * LoadBalancingConstants.BALANCE_FACTOR) { // I'm better than him!
                        remoteBalancer.sendActiveObjectsTo(myNode);                     
                }
        }
    
    public void sendActiveObjectsTo(Node destNode) {
        
        if (NodeFactory.isNodeLocal(destNode)) {
            return;
        }
        
       try {
                
                BodyMap knownBodies = LocalBodyStore.getInstance().getLocalBodies();

            if (knownBodies.size() < 1) {
                return;
            }

            java.util.Iterator bodiesIterator = knownBodies.bodiesIterator();

            int minLength = Integer.MAX_VALUE;
            Body minBody = null;
            while (bodiesIterator.hasNext()) {
                Body activeObjectBody = (Body) bodiesIterator.next();
                Object testObject = activeObjectBody.getReifiedObject();

                /********** Only some Active Objects can migrate *************/
                boolean testSerialization = !(testObject instanceof ProActiveInternalObject) && !(testObject instanceof NotLoadBalanceableObject);

                if (activeObjectBody.isAlive()) {
                    if (activeObjectBody.isActive() && testSerialization) {
                        int aoQueueLenght = activeObjectBody.getRequestQueue()
                                                            .size();
                        if (aoQueueLenght < minLength) {
                            minLength = aoQueueLenght;
                            minBody = activeObjectBody;
                        }
                    }
                }
            }

            /***********  we have the Active Object with shortest queue, so we send the migration call ********/
            if ((minBody != null) && minBody.isActive()) {
                /*logger.info("[Loadbalancer] Migrating ("+minBody.getReifiedObject().getClass().getName()+") from " +
                    myNode.getNodeInformation().getURL() + " to " +
                    destNode.getNodeInformation().getURL());
                    */
                ProActive.migrateTo(minBody, destNode, false);  
                informationRecover.register(this.getName(),this.metric.getLoad(), destNode.getNodeInformation().getURL(),minBody.getReifiedObject().getClass().getName());
            }
        } catch (IllegalArgumentException e) {
                logger.error("[LoadBalancer] "+e.getLocalizedMessage());
        } catch (SecurityException e) {
                logger.error("[LoadBalancer] Object doesn't have migrateTo method");
        } catch (MigrationException e) {
                logger.error("[LoadBalancer] Object can't migrate (?)");
                }
    }

    /*
    public String getName(){
        return balancerName;
    }
    
    public void init(ArrayList<LoadBalancer> loadBalancers, InformationRecover ir){
        try {
                        this.myNode = ProActive.getNode();
                        this.informationRecover = ir;
                } catch (NodeException e) {
                        e.printStackTrace();
                }
                this.loadBalancers = loadBalancers;
        this.myThis = (LoadBalancer) ProActive.getStubOnThis();
        this.balancerName = myNode.getNodeInformation().getHostName();
        
        // by now we use only Linux
            lm = new LoadMonitor(myThis,metric);
            new Thread(lm).start();
    } 
    
    
    private void internalAction(int action) {
        int size = loadBalancers.size();
        if (size < 1) return;
        
        int first = randomizer.nextInt(size);
        for (int i = 0; i < LoadBalancingConstants.SUBSET_SIZE  && size > 0; i++) {
                LoadBalancer remoteLb = loadBalancers.get((first+i)%size);
                try {
                        switch (action) {
                                case STEAL:
                                        remoteLb.sendActiveObjectsTo(myNode,this.metric.getRanking());
                                        break;

                                case BALANCE:
                                        remoteLb.getActiveObjectsFrom(myThis,this.metric.getRanking());
                                        break;
                                }
                        } catch (NonFunctionalException e) {
                                loadBalancers.remove((first+i)%size);
                        size--;
                        }
        }
    }
    
    public void notifyLoadBalancers(){
        LoadBalancer lb;
        LoadBalancer myThis = (LoadBalancer)ProActive.getStubOnThis();
        Iterator<LoadBalancer> it = loadBalancers.iterator();
                while (it.hasNext()) {
                        lb = it.next();
                        if(!lb.equals(this)){
                                lb.addNewBalancer(myThis);
                        }
                        lb = null;

                }
    }
    
    public void addNewBalancer(LoadBalancer lb){
        loadBalancers.add(lb);
    }

}

---