source: gssim/trunk/src/example/gridplugin/ARGraham.java @ 5

package example.gridplugin;

import org.qcg.broker.schemas.resreqs.ExecutionTimeType;
import org.qcg.broker.schemas.resreqs.TimePeriod;
import org.qcg.broker.schemas.resreqs.TimePeriodChoice;
import gssim.schedframe.scheduling.plugin.local.GssimTimeOperations;

import java.util.List;
import java.util.Properties;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.joda.time.DateTime;

import schedframe.exceptions.ModuleException;
import schedframe.exceptions.NotAuthorizedException;
import schedframe.exceptions.ReservationException;
import schedframe.resources.ResourceProvider;
import schedframe.resources.ResourceStateDescription;
import schedframe.resources.units.Processors;
import schedframe.scheduling.JobInterface;
import schedframe.scheduling.Offer;
import schedframe.scheduling.Queue;
import schedframe.scheduling.Reservation;
import schedframe.scheduling.ResourceUsage;
import schedframe.scheduling.TaskInterface;
import schedframe.scheduling.TimeRequirements;
import schedframe.scheduling.TimeResourceAllocation;
import schedframe.scheduling.events.SchedulingEvent;
import schedframe.scheduling.plan.SchedulingPlanInterface;
import schedframe.scheduling.plan.impl.Allocation;
import schedframe.scheduling.plan.impl.Host;
import schedframe.scheduling.plan.impl.ScheduledTask;
import schedframe.scheduling.plan.impl.ScheduledTime;
import schedframe.scheduling.plan.impl.SchedulingPlan;
import schedframe.scheduling.plugin.SchedulingPluginConfiguration;
import schedframe.scheduling.plugin.configuration.DefaultConfiguration;
import schedframe.scheduling.plugin.grid.GridSchedulingPlugin;
import schedframe.scheduling.plugin.grid.JobRegistry;
import schedframe.scheduling.plugin.grid.Module;
import schedframe.scheduling.plugin.grid.ModuleList;
import schedframe.scheduling.plugin.grid.Prediction;
import schedframe.scheduling.plugin.grid.ReservationManager;
import schedframe.scheduling.plugin.local.TimeOperations;
import schedframe.scheduling.utils.ResourceParameterName;
import simulator.lists.slotList.SlotsGantt;

/**
 * 
 * @author Ariel
 * 
 * This Plugin schedules jobs using the earliest due date (EDD) policy to sort them in a queue
 * and then applies the greedy list scheduling Graham algorithm.
 * 
 * Jobs $J_i$ from the queue are assigned to one of
* clusters  $\mathcal{M}$ using a greedy list-scheduling
* algorithm based on ~\cite{graham1969bmt,mounie07}.
* To this end, the Grid scheduler queries each organization $O_k$ about
* a list of free slots $\psi_{ki} \in \Psi_k, \psi_{ki}=(t',t'',m_{ki}) , i=1..|\Psi|$.
* Parameters $t'$ and $t''$ denote start and time of a slot, respectively.
* $m_{ki}$ is a number of processors available within the slot $i$ at organization $O_k$.
* Slots are time periods within which a number of available processors is constant.
* The Grid scheduler sorts collected slots by increasing start time. The schedule is constructed
* by assigning jobs in the Grid scheduler's queue to processors in given slots in a greedy manner.
* For each slot $\psi_{kI}$ (starting from the earliest one) the scheduler chooses from its queue the
* first job $J_j$ requiring no more than $m_{ki}$ processors in all subsequent
* slots $i \geq I$ such as $t''_{ki} \geq t'_{kI} + p_j$. If such a job was find the scheduler schedules it to be started at
* $t'_{kI}$, and removes it from the queue. If there is no such a job, the scheduler apply the same procedure to the next free slot
*whose number of available processors is larger than the current one.
*/
public class ARGraham implements GridSchedulingPlugin<org.qcg.broker.schemas.schedulingplan.SchedulingPlan>{
        
        private Log log = LogFactory.getLog(ARGraham.class);
        protected TimeOperations timeOpers;
        
        public SchedulingPlanInterface<org.qcg.broker.schemas.schedulingplan.SchedulingPlan> schedule(
                        SchedulingEvent event, Queue<? extends JobInterface<?>> jobQueue,
                        Queue<? extends TaskInterface<?>> taskQueue,
                        JobRegistry jobRegistry, ModuleList moduleList, Prediction prediction)
                        throws Exception {
                
                ReservationManager reservManager = null;
                for(int i = 0; i < moduleList.size() && reservManager == null; i++){
                        Module m = moduleList.get(i);
                        switch(m.getType()){
                                case RESERVATION_MANAGER: reservManager = (ReservationManager) m;
                                        break;
                        }
                }

                // empty scheduling plan
                SchedulingPlan sd = new SchedulingPlan();

                // getting offers
                DateTime startPeriod = timeOpers.getTime();
                DateTime endPeriod = new DateTime(Long.MAX_VALUE);
                
                ExecutionTimeType ett = new ExecutionTimeType();
                        TimePeriod tp = new TimePeriod();
                        tp.setPeriodStart(startPeriod.toDate());
                                TimePeriodChoice tpc = new TimePeriodChoice();
                                tpc.setPeriodEnd(endPeriod.toDate());
                        tp.setTimePeriodChoice(tpc);
                ett.setTimePeriod(tp);

                TimeRequirements timeRequirements = new TimeRequirements(ett);
                
                List<Offer> offers = reservManager.getOffer(timeRequirements, null, null);
                
                System.out.print("Number of offers " + offers.size());

                if (offers.size() < 1)
                        return sd;

                // in this example, there is only one computing resource, 
                // therefore only one offer is expected
                Offer offer = offers.get(0);
                ResourceProvider provider = offer.getProvider(); 
                int maxCpuCnt = offer.get(0).getAllocatedResource().
                                                                        getResourceUnit(ResourceParameterName.CPUCOUNT).
                                                                        getAmount();
                
                SlotsGantt slotGantt = new SlotsGantt(offer);

                int i = 0;
                while (taskQueue.size() > 0 && i < slotGantt.size()) {
                        DateTime startTime = slotGantt.get(i).getStart();
                        int j = 0;
                        while (j < taskQueue.size()) {
                                TaskInterface<?> task = taskQueue.get(j);
                                
                                // resource requirements
                                int taskSize = Double.valueOf(task.getCpuCntRequest()).intValue();
                                Processors ruReq = new Processors("compRes1", maxCpuCnt, taskSize);
                                ResourceStateDescription resourceState = new ResourceStateDescription(provider);
                                resourceState.addResourceUnit(ruReq);
                                
                                // end of task
                                DateTime endTask = new DateTime(startTime.plus(task.getExpectedDuration()));

                                // check slot
                                TimeResourceAllocation taskSlot = new TimeResourceAllocation(resourceState, startTime, endTask);
                                if (slotGantt.checkSlot(taskSlot)) {
                                        slotGantt.addSlot(taskSlot);
                                        taskQueue.remove(j); // pull out of the queue
                                        
                                        Reservation reservation = reserveResources(reservManager, provider,
                                                        taskSlot);

                                        // submit task to created reservation
                                        // add to scheduling plan
                                        ScheduledTask taskSD = new ScheduledTask();
                                        taskSD.setJobId(task.getJobId());
                                        taskSD.setTaskId(task.getId());

                                        Allocation taskAllocation = new Allocation();
                                        taskAllocation.setProcessesCount(taskSize);

                                        Host host = new Host();
                                        host.setHostname(provider.getProviderId());
                                        taskAllocation.setHost(host);
                                        taskAllocation.setReservation(reservation);

                                        taskSD.addAllocation(taskAllocation);

                                        ScheduledTime scheduledTime = new ScheduledTime();
                                        scheduledTime.setStart(startTime.toDate());
                                        scheduledTime.setEnd(endTask.toDate());
                                        taskSD.setScheduledTime(scheduledTime);

                                        sd.addTask(taskSD);

                                } else
                                        j++;
                        }
                        i++;
                }
        
                return sd;
        }
        
        protected Reservation reserveResources(ReservationManager reservManager, 
                                                                                        ResourceProvider provider, 
                                                                                        TimeResourceAllocation taskSlot) {

                ResourceUsage resourceUsage = new ResourceUsage();
                resourceUsage.setProvider(provider);
                resourceUsage.add(taskSlot);
                
                // set initial reservation
                log.debug("GLOBAL: creating reservation for offer: " + taskSlot + " provider: " + provider);
                List<Reservation> reservations = null;
                try {
                        reservations = reservManager.createReservation(resourceUsage, null);
                } catch (ModuleException e) {
                        log.error("GLOBAL: crerating reservation fails");
                        return null;
                }

                // only one reservation is expected
                Reservation reservation = reservations.get(0);
                
                if(reservation == null)
                        return null;
                
                String reservationId = reservation.getId();
                log.debug("GLOBAL: reservationID: "+reservationId);
                
                
                // commit initial reservation
                try {
                        reservation = reservManager.commitReservation(reservation, null);
                } catch (ModuleException e) {
                        log.error("GLOBAL: reservation " + reservationId + " was not commited successfully");
                        return null;
                }
                
                log.debug("GLOBAL: reservation "+ reservation.getId() + " commited");
                
                return reservation;
        }

        public SchedulingPluginConfiguration getConfiguration() {
                return DefaultConfiguration.forGridPlugin();
        }

        public String getPluginName() {
                return getClass().getName();
        }

        public void initPlugin(Properties properties) {
                timeOpers = new GssimTimeOperations();
        }

}