ManagerImpl.cc

Go to the documentation of this file.

/*
 * ManagerImpl.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 ManagerImpl.cc
 *
 * Implementation of the ManagerImpl class.
 */

#include "config.h"

#include <iostream>

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

#include "ManagerImpl.hh"

using namespace std;

#define INSTR_manager_change_cpu LRTIME_INSTR
#define INSTR_manager_change_cpu_data "manager_change_cpu", "The wall clock time needed to run ChangeTaskCPU"

#if defined(INSTR_manager_change_cpu_data)
static struct iPoint INSTR_manager_change_cpu_point = {
    INSTR_manager_change_cpu_data
};
#endif

ManagerImpl::ManagerImpl(const char *name)
{
    static int initialized = 0;

    if( !initialized )
    {
        atexit(iPrintPointsAtExit);
        initialized = 1;
    }
    lnNewList(&this->mi_Tasks);
}

ManagerImpl::~ManagerImpl()
{
}

void ManagerImpl::RepairTaskList(void)
{
    struct TaskNode *curr, *next;

    curr = (struct TaskNode *)this->mi_Tasks.lh_Head;
    while( curr->tn_Link.ln_Succ != NULL )
    {
        Broker::Task_ptr task = curr->tn_Task.in();
        
        next = (struct TaskNode *)curr->tn_Link.ln_Succ;
        
        try
        {
            CORBA::String_var name;

            /* ... do an RPC, and */
            name = task->Name();
            /* ... move to the next node. */
        }
        catch(const CORBA::TRANSIENT &e)
        {
            /* Oops, task is toast, try to tell the policy and */
            if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
            {
                try
                {
                    this->mi_CurrentPolicy->RemoveTask(task);
                }
                catch(const CORBA::BAD_PARAM &e)
                {
                    /* Ignore... */
                }
                catch(const CORBA::SystemException &e)
                {
                    cerr << "Error while repairing: "
                         << e
                         << " file: "
                         << __FILE__
                         << " line: "
                         << __LINE__
                         << endl;
                }
            }
            /* ... then remove it. */
            lnRemove(&curr->tn_Link);
            task = Broker::Task::_nil(); /* 'task' is dead after removal. */

            delete curr;
            curr = NULL;
        }
        
        curr = next;
    }
}

struct ManagerImpl::TaskNode *ManagerImpl::FindTask(Broker::Task_ptr task)
{
    struct TaskNode *curr, *retval = NULL;

    curr = (struct TaskNode *)this->mi_Tasks.lh_Head;
    while( (curr->tn_Link.ln_Succ != NULL) && (retval == NULL) )
    {
        if( curr->tn_Task->_is_equivalent(task) )
        {
            retval = curr;
        }
        curr = (struct TaskNode *)curr->tn_Link.ln_Succ;
    }
    return( retval );
}

Broker::Policy_ptr ManagerImpl::CurrentPolicy(void)
    throw (CORBA::SystemException)
{
    Broker::Policy_var retval;

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

void ManagerImpl::CurrentPolicy(Broker::Policy_ptr policy)
    throw (CORBA::SystemException)
{
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        struct TaskNode *tn;

        tn = (struct TaskNode *)this->mi_Tasks.lh_Head;
        while( tn->tn_Link.ln_Succ != NULL )
        {
            this->mi_CurrentPolicy->RemoveTask(tn->tn_Task.in());
            tn = (struct TaskNode *)tn->tn_Link.ln_Succ;
        }
    }
    this->mi_CurrentPolicy = Broker::Policy::_duplicate(policy);
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        struct TaskNode *tn;

        tn = (struct TaskNode *)this->mi_Tasks.lh_Head;
        while( tn->tn_Link.ln_Succ != NULL )
        {
            this->mi_CurrentPolicy->AddTask(tn->tn_Task.in(),
                                            tn->tn_ScheduleParameters);
            tn = (struct TaskNode *)tn->tn_Link.ln_Succ;
        }
    }
}

void ManagerImpl::AddTask(Broker::Task_ptr task,
                          const Broker::ScheduleParameters &sp)
    throw (CORBA::SystemException,
           Broker::DuplicateScheduleParameter,
           Broker::InvalidScheduleParameter,
           Broker::MissingScheduleParameter)
{
    if( CORBA::is_nil(task) )
    {
        throw CORBA::BAD_PARAM();
    }

    /* Make sure our list is sane and */
    this->RepairTaskList();

    /* ... the task is not already in the list. */
    if( this->FindTask(task) != NULL )
    {
        throw CORBA::BAD_PARAM();
    }

    /* Make a new node, */
    auto_ptr<struct TaskNode> tn(new TaskNode());

    /* ... fill out the object, */
    tn->tn_Task = Broker::Task::_duplicate(task);
    tn->tn_ScheduleParameters = new Broker::ScheduleParameters(sp);

    /* ... set things in motion, and */
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        this->mi_CurrentPolicy->AddTask(task, sp);
    }
    tn->tn_Task->BeginCPUScheduling(sp);
    tn->tn_Task->SetManager(this->_this());
    
    /* ... transfer ownership of 'tn' from this frame to the list. */
    lnAddTail(&this->mi_Tasks, &(tn.release())->tn_Link);
}

void ManagerImpl::RemoveTask(Broker::Task_ptr task)
    throw (CORBA::SystemException)
{
    struct TaskNode *tn;
    
    if( CORBA::is_nil(task) )
    {
        throw CORBA::BAD_PARAM();
    }

    /* Find the task, */
    if( (tn = this->FindTask(task)) == NULL )
    {
        throw CORBA::BAD_PARAM();
    }

    /* ... remove it from our list, and */
    lnRemove(&tn->tn_Link);

    /* ... tell the policy. */
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        this->mi_CurrentPolicy->RemoveTask(task);
    }
    
    delete tn;
}

Broker::TaskList *ManagerImpl::GetTaskList(void)
    throw (CORBA::SystemException)
{
    Broker::TaskList_var retval;
    unsigned int lpc = 0;
    struct TaskNode *tn;

    retval = new Broker::TaskList();
    retval->length(lnCountNodes(&this->mi_Tasks));
    tn = (struct TaskNode *)this->mi_Tasks.lh_Head;
    while( tn->tn_Link.ln_Succ != NULL )
    {
        (*retval)[lpc] = Broker::Task::_duplicate(tn->tn_Task.in());
        lpc += 1;
        tn = (struct TaskNode *)tn->tn_Link.ln_Succ;
    }
    return( retval._retn() );
}

Broker::CPUReserve ManagerImpl::ChangeTaskCPU(Broker::RealTimeTask_ptr task,
                                              const Broker::CPUReserve &advice)
    throw (CORBA::SystemException, Broker::InvalidState)
{
    if( CORBA::is_nil(task) )
    {
        throw CORBA::BAD_PARAM();
    }
    
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        /* Have a policy, just pass it on. */
        return this->mi_CurrentPolicy->ChangeTaskCPU(task, advice);
    }
    else
    {
        Broker::CPUReserve retval;

        retval = advice;
        retval.Compute = 0;
        return retval;
    }
}

Broker::Task_ptr ManagerImpl::ResolveTask(CORBA::ORB_ptr orb,
                                          Broker::TaskList &tl,
                                          const char *name)
    throw (CORBA::SystemException)
{
    Broker::Task_var retval;
    unsigned int lpc;

    require(!CORBA::is_nil(orb));
    require(name != NULL);

    /* First try finding the task, by name, in the list. */
    for( lpc = 0; (lpc < tl.length()) && CORBA::is_nil(retval.in()); lpc++ )
    {
        try
        {
            CORBA::String_var task_name;
            
            task_name = tl[lpc]->Name();
            if( strcmp(task_name.in(), name) == 0 )
            {
                retval = Broker::Task::_duplicate(tl[lpc]);
            }
        }
        catch(const CORBA::SystemException &e)
        {
            /* ... */
        }
    }
    if( CORBA::is_nil(retval.in()) )
    {
        CORBA::Object_var obj;

        /* No luck finding it in the list, check if it is an IOR. */
        try
        {
            obj = orb->string_to_object(name);
        }
        catch(const CORBA::SystemException &e)
        {
            throw CORBA::BAD_PARAM();
        }
        retval = Broker::Task::_narrow(obj.in());
        if( CORBA::is_nil(retval.in()) )
        {
            throw CORBA::BAD_PARAM();
        }
    }
    return( retval._retn() );
}

---