WeightedPolicyImpl.cc

Go to the documentation of this file.

/*
 * WeightedPolicyImpl.cc
 *
 * Copyright (c) 2003, 2004 The University of Utah and the Flux Group.
 * All rights reserved.
 *
 * This file is licensed under the terms of the GNU Public License.  
 * See the file "license.terms" for restrictions on redistribution 
 * of this file, and for a DISCLAIMER OF ALL WARRANTIES.
 */

/**
 * @file WeightedPolicyImpl.cc
 *
 * Implementation of the WeightedPolicyImpl class.
 */

#include "config.h"

#include <iostream>
#include <assert_pp.h>
#include <instrumentation.h>

#include "StubRealTimeTask.hh"
#include "WeightedPolicyImpl.hh"

using namespace std;

WeightedPolicyImpl::WeightedPolicyImpl(const char *name) :
    wp_Name(name)
{
    require(name != NULL);
    
    this->wp_MaxUsedCPU = DEFAULT_MAX_USED_CPU;
    this->wp_MinTaskCPU = DEFAULT_MIN_TASK_CPU;
    lnNewList((struct lnMinList *)&this->wp_List);
    this->wp_TotalWeightedComputeTime = 0;
}

WeightedPolicyImpl::~WeightedPolicyImpl()
{
}

char *WeightedPolicyImpl::Name(void)
    throw (CORBA::SystemException)
{
    CORBA::String_var retval;

    retval = this->wp_Name;
    return( retval._retn() );
}

CORBA::Float WeightedPolicyImpl::GetMaxCPUAllocation(void)
    throw (CORBA::SystemException)
{
    CORBA::Float retval;

    retval = this->wp_MaxUsedCPU;
    return( retval );
}

void WeightedPolicyImpl::SetMaxCPUAllocation(CORBA::Float value)
    throw (CORBA::SystemException)
{
    if( (value < BrokerPolicies::WeightedPolicy::CPU_ALLOCATION_MIN) ||
        (value > BrokerPolicies::WeightedPolicy::CPU_ALLOCATION_MAX) )
    {
        throw CORBA::BAD_PARAM();
    }

    this->wp_MaxUsedCPU = value;
    this->RedistributeCPU();
}

void WeightedPolicyImpl::AddTask(Broker::Task_ptr task,
                                 const Broker::ScheduleParameters &sp)
    throw (CORBA::SystemException,
           Broker::DuplicateScheduleParameter,
           Broker::InvalidScheduleParameter,
           Broker::MissingScheduleParameter)
{
    Broker::RealTimeTask_ptr rtt;
    
    /* Check inputs. */
    if( CORBA::is_nil(task) )
    {
        throw CORBA::BAD_PARAM();
    }

    // this->RepairTaskList();
    
    rtt = Broker::RealTimeTask::_narrow(task);
    if( !CORBA::is_nil(rtt) )
    {
        struct TaskData *td = NULL;
        CORBA::String_var name;
        int rc;
        
        name = rtt->Name();

        try
        {
            td = new TaskData();
            /* Initialize the node and */
            td->td_Link.ln_Pred = NULL;
            td->td_Link.ln_Succ = NULL;
            td->td_Task = Broker::RealTimeTask::_duplicate(rtt);
            td->td_Weight = 1;
            
            /* ... cache some values. */
            td->td_Schedule.setParameters(sp);
            td->td_ComputeTime = td->td_RequestedTime = 0;
        }
        catch(const Broker::DuplicateScheduleParameter &e)
        {
            delete td;
            throw e;
        }
        catch(const Broker::InvalidScheduleParameter &e)
        {
            delete td;
            throw e;
        }
        catch(const Broker::MissingScheduleParameter &e)
        {
            delete td;
            throw e;
        }
        
        if( (rc = rts_grab_schedulable(&td->td_Schedulable,
                                       td->td_Schedule.rts_PID,
                                       name.in())) == -1 )
        {
            cerr << "rts_grab_schedulable: " << strerror(rc) << endl;
        }
        else
        {
            /* Finally, add it to the tail of the list. */
            lnAddTail((struct lnMinList *)&this->wp_List, &td->td_Link);
            td = NULL;
        }

        if( td != NULL )
        {
            delete td;
            throw Broker::InvalidScheduleParameter();
        }
    }
}

void WeightedPolicyImpl::RemoveTask(Broker::Task_ptr task)
    throw (CORBA::SystemException)
{
    Broker::RealTimeTask_var rtt;
    struct TaskData *td;
    int rc;
    
    /* Find the task, */
    rtt = Broker::RealTimeTask::_narrow(task);
    if( (td = this->FindTaskData(rtt.in())) == NULL )
    {
        throw CORBA::BAD_PARAM();
    }
    
    /* ... remove the node, */
    lnRemove((struct lnMinNode *)&td->td_Link);
    if( (rc = rts_clear_schedule(td->td_Schedulable)) == -1 )
    {
        cerr << "rts_clear_schedule: " << strerror(rc) << endl;
    }
    if( (rc = rts_release_schedulable(td->td_Schedulable)) == -1 )
    {
        cerr << "rts_release_schedulable: " << strerror(rc) << endl;
    }
    this->wp_TotalWeightedComputeTime -=
        (td->td_Weight * td->td_RequestedTime);
    /* ... give its compute time to the others, and */
    this->RedistributeCPU();
    /* ... delete it. */
    delete td;
}

void WeightedPolicyImpl::Activate(const Broker::TaskList &tl)
    throw (CORBA::SystemException)
{
    /** @todo implement */
}

void WeightedPolicyImpl::Deactivate(void)
    throw (CORBA::SystemException)
{
    /** @todo implement */
}

Broker::TaskList *WeightedPolicyImpl::GetTaskList(void)
    throw (CORBA::SystemException)
{
    Broker::TaskList_var retval;
    struct TaskData *td;
    unsigned int lpc = 0;
    
    retval = new Broker::TaskList();
    retval->length(lnCountNodes((struct lnMinList *)&this->wp_List));
    td = (struct TaskData *)this->wp_List.lh_Head;
    while( td->td_Link.ln_Succ != NULL )
    {
        (*retval)[lpc] = Broker::Task::_duplicate(td->td_Task.in());
        lpc += 1;
        td = (struct TaskData *)td->td_Link.ln_Succ;
    }
    return( retval._retn() );
}

void WeightedPolicyImpl::SetTaskWeight(Broker::Task_ptr task,
                                       CORBA::UShort weight)
    throw (CORBA::SystemException)
{
    Broker::RealTimeTask_var rtt;
    struct TaskData *td;

    rtt = Broker::RealTimeTask::_narrow(task);
    if( CORBA::is_nil(rtt.in()) )
    {
        throw CORBA::BAD_PARAM();
    }
    if( weight == 0 )
    {
        throw CORBA::BAD_PARAM();
    }

    /* Find the task, */
    if( (td = this->FindTaskData(rtt.in())) == NULL )
    {
        throw CORBA::BAD_PARAM();
    }

    this->wp_TotalWeightedComputeTime -=
        (td->td_Weight * td->td_RequestedTime);
    
    /* ... set the new weight, and */
    td->td_Weight = weight;
    
    this->wp_TotalWeightedComputeTime +=
        (td->td_Weight * td->td_RequestedTime);
    /* ... redistribute the reservations. */
    this->RedistributeCPU();
}

CORBA::UShort WeightedPolicyImpl::GetTaskWeight(Broker::Task_ptr task)
    throw (CORBA::SystemException)
{
    Broker::RealTimeTask_var rtt;
    CORBA::UShort retval;
    struct TaskData *td;

    rtt = Broker::RealTimeTask::_narrow(task);
    if( CORBA::is_nil(rtt.in()) )
    {
        throw CORBA::BAD_PARAM();
    }
    if( (td = this->FindTaskData(rtt.in())) == NULL )
    {
        throw CORBA::BAD_PARAM();
    }

    retval = td->td_Weight;
    
    return( retval );
}

void WeightedPolicyImpl::RedistributeCPU(void)
{
    struct lnMinList change_list;
    struct ChangeNode *cn;
    struct TaskData *curr;

    lnNewList(&change_list);
    /* Produce a change list that */
    curr = (struct TaskData *)this->wp_List.lh_Head;
    while( curr->td_Link.ln_Succ != NULL )
    {
        unsigned long long actual_cpu_time, max_cpu_time;

        /* Compute the weighted CPU time, */
        max_cpu_time = (unsigned long long)
            ((this->wp_MaxUsedCPU * (float)curr->td_Schedule.rts_Deadline));
        
        actual_cpu_time = (unsigned long long)
            (((curr->td_Weight * curr->td_RequestedTime) * max_cpu_time) /
             this->wp_TotalWeightedComputeTime);
        
        if( actual_cpu_time > curr->td_RequestedTime )
        {
            /* no overload, no need to compress... */
            actual_cpu_time = curr->td_RequestedTime;
        }

        /* ... update the change node, and */
        cn = &curr->td_ChangeLink;
        cn->cn_Parent = curr;
        cn->cn_NewTime = actual_cpu_time;

        /* add it to the change list depending on how it changes. */
        if( actual_cpu_time < curr->td_ComputeTime )
        {
            /*
             * Put decreases at the front of the change list so there will
             * CPU to give away later on.
             */
            lnAddHead(&change_list, &cn->cn_Link);
        }
        else if( actual_cpu_time == curr->td_ComputeTime )
        {
            /* no change... */
        }
        else if( actual_cpu_time > curr->td_ComputeTime )
        {
            /*
             * Put increases at the back of the change list so we do not try to
             * take CPU that is not yet available.
             */
            lnAddTail(&change_list, &cn->cn_Link);
        }
        else 
        {
            assert(0);
        }
        curr = (struct TaskData *)curr->td_Link.ln_Succ;
    }
    curr = NULL;

    /* ... takes away CPU before giving to the other tasks. */
    cn = (struct ChangeNode *)change_list.lh_Head;
    while( cn->cn_Link.ln_Succ != NULL )
    {
        struct TaskData *td;

        td = cn->cn_Parent;
        if( cn->cn_NewTime > 0 )
        {
            struct rts_schedule rs;
            int rc;
            
            rs.rs_Period = microsec_to_timespec(td->td_Schedule.rts_Period);
            rs.rs_ComputeTime = microsec_to_timespec(cn->cn_NewTime);
            rs.rs_ReserveMode = td->td_Schedule.rts_ReserveMode;
            if( (rc = rts_set_schedule(td->td_Schedulable, &rs)) == -1 )
            {
                cerr << "rts_set_schedule: " << strerror(rc) << endl;
            }
            else
            {
                cn->cn_Parent->td_ComputeTime = cn->cn_NewTime;
            }
        }
        else
        {
            int rc;
            
            if( (rc = rts_clear_schedule(td->td_Schedulable)) == -1 )
            {
                cerr << "rts_set_schedule: " << strerror(rc) << endl;
            }
        }
        cn = (struct ChangeNode *)cn->cn_Link.ln_Succ;
    }
}

void WeightedPolicyImpl::RepairTaskList(void)
{
    struct TaskData *curr, *succ;
    int do_redist = 0;

    /* Walk the task list to make sure it is sane and then */
    curr = (struct TaskData *)this->wp_List.lh_Head;
    while( curr->td_Link.ln_Succ != NULL )
    {
        succ = (struct TaskData *)curr->td_Link.ln_Succ;
#if 0
        try
        {
            /* Force an RPC, to make sure the object is still alive. */
            curr->td_Task->GetComputeTime();
        }
        catch(const CORBA::TRANSIENT &e)
        {
            /* The object is dead, remove it, and */
            lnRemove((struct lnMinNode *)&curr->td_Link);
            /* ... delete it. */
            delete curr;
            do_redist = 1;
        }
#endif
        curr = succ;
    }
    
    /* ... redistribute the CPU if there was a change. */
    if( do_redist )
    {
        this->RedistributeCPU();
    }
}

struct WeightedPolicyImpl::TaskData *
WeightedPolicyImpl::FindTaskData(Broker::RealTimeTask_ptr task)
{
    struct TaskData *curr, *retval = NULL;

    require(task != NULL);
    
    curr = (struct TaskData *)this->wp_List.lh_Head;
    while( (retval == NULL) && (curr->td_Link.ln_Succ != NULL) )
    {
        if( task->_is_equivalent(curr->td_Task.in()) )
        {
            retval = curr;
        }
        curr = (struct TaskData *)curr->td_Link.ln_Succ;
    }

    if( retval == NULL )
    {
        cerr << "Unknown task" << endl;
        throw Broker::Internal("FindTaskData failed to find task");
    }
    
    return( retval );
}

Broker::CPUReserve
WeightedPolicyImpl::ChangeTaskCPU(Broker::RealTimeTask_ptr task,
                                  const Broker::CPUReserve &advice)
    throw (CORBA::SystemException, Broker::InvalidState)
{
    Broker::CPUReserve retval = advice;
    struct TaskData *td;

    require(task != NULL);
    
    if( (td = this->FindTaskData(task)) != NULL )
    {
        td->td_Schedule.rts_Period = advice.Period;
        this->wp_TotalWeightedComputeTime -=
            (td->td_Weight * td->td_RequestedTime);
        td->td_RequestedTime = advice.Compute;
        this->wp_TotalWeightedComputeTime +=
            (td->td_Weight * td->td_RequestedTime);
        this->RedistributeCPU();
        retval.Compute = td->td_ComputeTime;
    }
    else
    {
        throw Broker::InvalidState("AddTask must be called first");
    }
    return retval;
}

---