ManagerImpl.cc

Go to the documentation of this file.

/*
 * ManagerImpl.cc
 *
 * Copyright (c) 2003 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 <assert_pp.h>

#include "ManagerImpl.hh"

ManagerImpl::ManagerImpl(const char *name)
{
}

ManagerImpl::~ManagerImpl()
{
}

void ManagerImpl::RepairTaskList(void)
{
    unsigned int lpc;

    /* Walk the list, */
    for( lpc = 0; lpc < this->mi_Tasks.length(); )
    {
        Broker::Task_ptr task = this->mi_Tasks[lpc];
        
        try
        {
            CORBA::String_var name;

            /* ... do an RPC, and */
            name = task->Name();
            /* ... move to the next index. */
            lpc += 1;
        }
        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::SystemException &e)
                {
                    cerr << "Error while repairing: "
                         << e
                         << " file: "
                         << __FILE__
                         << " line: "
                         << __LINE__
                         << endl;
                }
            }
            /* ... then remove it. */
            this->RemoveTaskAt(lpc);
            task = Broker::Task::_nil(); /* 'task' is dead after removal. */
        }
    }
}

int ManagerImpl::TaskIndex(Broker::Task_ptr task)
{
    unsigned int lpc;
    int retval = -1;

    require(!CORBA::is_nil(task));
    
    for( lpc = 0; (lpc < this->mi_Tasks.length()) && (retval == -1); lpc++ )
    {
        if( this->mi_Tasks[lpc]->_is_equivalent(task) )
        {
            retval = lpc;
        }
    }
    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()) )
    {
        this->mi_CurrentPolicy->Deactivate();
    }
    this->mi_CurrentPolicy = Broker::Policy::_duplicate(policy);
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        this->mi_CurrentPolicy->Activate(this->mi_Tasks);
    }
}

void ManagerImpl::AddTask(Broker::Task_ptr task,
                          const Broker::ScheduleParameters &sp)
    throw (CORBA::SystemException,
           Broker::DuplicateScheduleParameter,
           Broker::InvalidScheduleParameter,
           Broker::MissingScheduleParameter)
{
    int slot;

    if( CORBA::is_nil(task) )
    {
        throw CORBA::BAD_PARAM();
    }

    this->RepairTaskList();

    if( this->TaskIndex(task) != -1 )
    {
        throw CORBA::BAD_PARAM();
    }
    
    slot = this->mi_Tasks.length();
    this->mi_Tasks.length(slot + 1);
    this->mi_Tasks[slot] = Broker::Task::_duplicate(task);

    try
    {
        task->BeginCPUScheduling(this->_this(), sp);
        if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
        {
            this->mi_CurrentPolicy->AddTask(task, sp);
        }
    }
    catch(const CORBA::Exception &e)
    {
        this->mi_Tasks.length(slot);
        task->EndCPUScheduling();
        throw e;
    }
}

void ManagerImpl::RemoveTaskAt(unsigned int index)
{
    unsigned int lpc;
    
    require(index < this->mi_Tasks.length());

    /* Fill the gap created by the removal and */
    for( lpc = (unsigned int)index;
         lpc < (this->mi_Tasks.length() - 1);
         lpc++ )
    {
        this->mi_Tasks[lpc] = this->mi_Tasks[lpc + 1];
    }
    /* ... decrement the length. */
    this->mi_Tasks.length(this->mi_Tasks.length() - 1);
}

void ManagerImpl::RemoveTask(Broker::Task_ptr task)
    throw (CORBA::SystemException)
{
    int slot;
    
    if( CORBA::is_nil(task) )
    {
        throw CORBA::BAD_PARAM();
    }

    /* Find the task, */
    if( (slot = this->TaskIndex(task)) == -1 )
    {
        throw CORBA::BAD_PARAM();
    }

    /* ... remove it from our list, */
    this->RemoveTaskAt(slot);

    try
    {
        /* ... tell the task, and */
        task->EndCPUScheduling();
    }
    catch(const CORBA::SystemException &e)
    {
        cerr << e << endl; /* Atleast log the failure. */
    }
    
    /* ... tell the policy. */
    if( !CORBA::is_nil(this->mi_CurrentPolicy.in()) )
    {
        this->mi_CurrentPolicy->RemoveTask(task);
    }
}

Broker::TaskList *ManagerImpl::GetTaskList(void)
    throw (CORBA::SystemException)
{
    Broker::TaskList_var retval;

    retval = new Broker::TaskList(this->mi_Tasks);
    return( retval._retn() );
}

void ManagerImpl::ChangeTaskCPU(Broker::RealTimeTask_ptr task,
                                CORBA::ULong ct,
                                CORBA::ULong status,
                                CORBA::ULong 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. */
        this->mi_CurrentPolicy->ChangeTaskCPU(task, ct, status, advice);
    }
    else
    {
        /* No policy, just set the time and return. */
        try
        {
            task->SetComputeTime(advice);
        }
        catch(...)
        {
            /* XXX */
        }
    }
}

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. */
        obj = orb->string_to_object(name);
        retval = Broker::Task::_narrow(obj.in());
        if( CORBA::is_nil(retval.in()) )
        {
            throw CORBA::BAD_PARAM();
        }
    }
    return( retval._retn() );
}

---