RKTask.cc

Go to the documentation of this file.

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

#include "config.h"

#include <string.h>

#include <iostream>
#include <fstream>
#include <iomanip>

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

#include "RKTask.hh"

#include "rk_util.h"

using namespace std;

#define INSTR_rktask_change_cpu LRTIME_INSTR
#define INSTR_rktask_change_cpu_data "rktask_change_cpu", "The wall clock time needed to run ChangeTaskCPU"

#if defined(INSTR_rktask_change_cpu_data)
static struct iPoint INSTR_rktask_change_cpu_point = {
    INSTR_rktask_change_cpu_data
};
#endif

#if !defined(__XSTRING)
/**
 * Convert a macro argument to a string.
 *
 * @param x The macro to expand to a string.
 */
#define __XSTRING(x) __STRING(x)
#endif

/**
 * "Repair" the system by deleting any resource sets that have a CPU reserve
 * but no processes.
 */
static void rktRepairResourceSets(void)
{
#define MAX_RK_SETS 128
    static rk_resource_set_t rk_sets[MAX_RK_SETS];
    
    unsigned int lpc, actual;

    actual = rk_resource_sets_get_list(rk_sets, MAX_RK_SETS);
    for( lpc = 0; lpc < actual; lpc++ )
    {
        int proc_count = rk_resource_set_get_num_procs(rk_sets[lpc]);
        rk_reserve_t cr = rk_resource_set_get_cpu_rsv(rk_sets[lpc]);
        
        if( (proc_count == 0) && (cr != NULL_RESERVE) )
        {
            rk_resource_set_destroy(rk_sets[lpc]);
            rk_sets[lpc] = NULL_RESOURCE_SET;
        }
    }
#undef MAX_RK_SETS
}

RKTask::RKTask(const Broker::TaskParameters &tp)
    throw (CORBA::SystemException,
           Broker::DuplicateTaskParameter,
           Broker::InvalidTaskParameter,
           Broker::MissingTaskParameter)
{
    unsigned int lpc;

    gettimeofday(&this->rkt_StartTime, NULL);
    this->rkt_LastReservationLog = this->rkt_StartTime;
    this->rkt_ReservationLog = NULL;
    
    memset(&this->rkt_CPUReserveSpec, 0, sizeof(this->rkt_CPUReserveSpec));
    this->rkt_CPUReserveSpec.compute_time = microsec_to_timespec(100);
    this->rkt_CPUReserve = NULL_RESERVE;

    /* Process the arguments */
    for( lpc = 0; lpc < tp.length(); lpc++ )
    {
        if( strcasecmp(tp[lpc].name.in(), "name") == 0 )
        {
            const char *name;

            if( this->rkt_Name != NULL )
            {
                throw Broker::DuplicateTaskParameter("name");
            }
            else if( tp[lpc].value >>= name )
            {
                if( strlen(name) >= RK_NAME_LEN )
                {
                    throw Broker::InvalidTaskParameter(
                        "'name' is too long, only "
                        __XSTRING(RK_NAME_LEN)
                        " characters are allowed",
                        tp[lpc]);
                }
                else
                {
                    this->rkt_Name = name;
                }
            }
            else
            {
                throw Broker::InvalidTaskParameter("'name' not a string",
                                                   tp[lpc]);
            }
        }
        else if( (strcasecmp(tp[lpc].name.in(), "res_log") == 0) ||
                 (strcasecmp(tp[lpc].name.in(), "res-log") == 0) ||
                 (strcasecmp(tp[lpc].name.in(), "res log") == 0) )
        {
            const char *res_log;
            
            if( this->rkt_ReservationLog != NULL )
            {
                throw Broker::DuplicateTaskParameter("res log");
            }
            else if( tp[lpc].value >>= res_log )
            {
                this->rkt_ReservationLog = new ofstream(res_log);
                (*this->rkt_ReservationLog) << setfill('0');
                (*this->rkt_ReservationLog)
                    << "# start: "
                    << (long)this->rkt_StartTime.tv_sec
                    << "."
                    << setw(6) << (long)this->rkt_StartTime.tv_usec
                    << endl;
            }
            else
            {
                throw Broker::InvalidTaskParameter("'res log' not a string",
                                                   tp[lpc]);
            }
        }
    }
    
    /* Make sure we have everything we need. */
    if( this->rkt_Name == NULL )
    {
        throw Broker::MissingTaskParameter("name");
    }
}

RKTask::~RKTask()
{
    if( !CORBA::is_nil(this->rkt_Manager.in()) )
    {
        this->EndCPUScheduling();
    }
    if( this->rkt_ReservationLog != NULL )
    {
        delete this->rkt_ReservationLog;
    }
}

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

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

CORBA::ULong RKTask::Period(void)
    throw (CORBA::SystemException)
{
    CORBA::ULong retval;

    retval = timespec_to_microsec(&this->rkt_CPUReserveSpec.period);
    return( retval );
}

CORBA::ULong RKTask::Deadline(void)
    throw (CORBA::SystemException)
{
    CORBA::ULong retval;

    retval = timespec_to_microsec(&this->rkt_CPUReserveSpec.deadline);
    return( retval );
}

void RKTask::SetManager(Broker::Manager_ptr manager)
    throw (CORBA::SystemException)
{
    this->rkt_Manager = Broker::Manager::_duplicate(manager);
}

void RKTask::BeginCPUScheduling(const Broker::ScheduleParameters &cs)
    throw (CORBA::SystemException,
           Broker::DuplicateScheduleParameter,
           Broker::InvalidScheduleParameter,
           Broker::MissingScheduleParameter)
{
    CORBA::ULongLong start = 0, period = ~0, deadline = ~0;
    rk_reserve_mode_t rsv_mode = RSV_SOFT;
    unsigned int lpc;
    const char *str;
    pid_t pid = -1;
    
    if( this->rkt_CPUReserve != NULL_RESERVE )
    {
        throw CORBA::BAD_INV_ORDER();
    }
    
    /* Process the arguments. */
    for( lpc = 0; lpc < cs.length(); lpc++ )
    {
        if( strcasecmp(cs[lpc].name.in(), "reserve-mode") == 0 )
        {
            const char *str;

            if( cs[lpc].value >>= str )
            {
                if( strcasecmp(str, "soft") == 0 )
                {
                    rsv_mode = RSV_SOFT;
                }
                else if( strcasecmp(str, "hard") == 0 )
                {
                    rsv_mode = RSV_HARD;
                }
                else
                {
                    throw Broker::InvalidScheduleParameter(
                        "'reserve-mode' not 'hard' or 'soft'", cs[lpc]);
                }
            }
            else
            {
                throw Broker::InvalidScheduleParameter(
                        "'reserve-mode' not a string", cs[lpc]);
            }
        }
        else if( strcasecmp(cs[lpc].name.in(), "start") == 0 )
        {
            if( ((cs[lpc].value >>= str) && string_to_microsec(&start, str)) )
            {
            }
            else
            {
                throw Broker::InvalidScheduleParameter("'start' is not a time",
                                                       cs[lpc]);
            }
        }
        else if( strcasecmp(cs[lpc].name.in(), "pid") == 0 )
        {
            CORBA::Long c_pid;
            const char *str;

            if( pid != -1 )
            {
                throw Broker::DuplicateScheduleParameter("pid");
            }
            else if( cs[lpc].value >>= c_pid )
            {
                pid = c_pid;
            }
            else if( (cs[lpc].value >>= str) &&
                     (sscanf(str, "%d", &pid) == 1) )
            {
            }
            else
            {
                throw Broker::InvalidScheduleParameter("'pid' not a long",
                                                       cs[lpc]);
            }
            if( pid < 0 )
            {
                pid = -1;
                throw Broker::InvalidScheduleParameter("'pid' not a long",
                                                       cs[lpc]);
            }
        }
        else if( strcasecmp(cs[lpc].name.in(), "period") == 0 )
        {
            CORBA::ULong period_ul;
            
            if( period != ~0ULL )
            {
                throw Broker::DuplicateScheduleParameter("period");
            }
            else if( cs[lpc].value >>= period_ul )
            {
                period = period_ul;
            }
            else if( (cs[lpc].value >>= str) &&
                     string_to_microsec(&period, str) )
            {
            }
            else
            {
                throw Broker::InvalidScheduleParameter(
                        "'period' is not a time",
                        cs[lpc]);
            }
        }
        else if( strcasecmp(cs[lpc].name.in(), "deadline") == 0 )
        {
            CORBA::ULong deadline_ul;
            
            if( deadline != ~0ULL )
            {
                throw Broker::DuplicateScheduleParameter("deadline");
            }
            else if( cs[lpc].value >>= deadline_ul )
            {
                deadline = deadline_ul;
            }
            else if( (cs[lpc].value >>= str) &&
                     string_to_microsec(&deadline, str) )
            {
            }
            else
            {
                throw Broker::InvalidScheduleParameter(
                        "'deadline' is not a time",
                        cs[lpc]);
            }
        }
    }
    
    /* Make sure we have everything we need */
    if( period == ~0ULL )
    {
        throw Broker::MissingScheduleParameter("period");
    }
    if( pid == -1 )
    {
        throw Broker::MissingScheduleParameter("pid");
    }
    
    /* ... and derive the rest. */
    if( deadline == ~0ULL )
    {
        deadline = period;
    }

    /* Clean house first, then */
    rktRepairResourceSets();

    /* ... go ahead and find/create a resource set for the process. */
    if( (this->rkt_ResourceSet = rk_proc_get_rset(pid)) != NULL_RESOURCE_SET )
    {
        /* We'll just manage the existing resource set. */
        this->rkt_OwnsResourceSet = 0;
    }
    else if( (this->rkt_ResourceSet =
              rk_resource_set_get_by_name(this->rkt_Name.in())) !=
             NULL_RESOURCE_SET )
    {
        this->rkt_OwnsResourceSet = 0;
    }
    else if( (this->rkt_ResourceSet =
              rk_resource_set_create(this->rkt_Name.in())) !=
             NULL_RESOURCE_SET )
    {
        this->rkt_OwnsResourceSet = 1;
    }
    else
    {
        cerr << strerror(errno) << ": rk_resource_set_create" << endl;
        throw CORBA::NO_MEMORY(); // XXX
    }

    rk_reserve_t old_reserve;
    
    /* Check for an existing reserve. */
    if( (old_reserve = rk_resource_set_get_cpu_rsv(this->rkt_ResourceSet))
        != NULL_RESERVE )
    {
        if( rk_resource_set_get_num_procs(this->rkt_ResourceSet) == 0 )
        {
            /*
             * Assume something went horribly wrong.  Destroy the reserve
             * in preparation for adding our own reserve.
             */
            if( rk_cpu_reserve_delete(this->rkt_ResourceSet) == -1 )
            {
                cerr << strerror(errno)
                     << ": rk_cpu_reserve_delete"
                     << endl;
            }
        }
        else
        {
            Broker::NamedValue nv;
            
            /* Someone else is using this set... */
            this->rkt_ResourceSet = NULL_RESOURCE_SET;
            nv.name = "name";
            nv.value <<= this->rkt_Name;
            throw Broker::InvalidScheduleParameter("name", nv);
        }
    }
    if( rk_resource_set_attach_process(this->rkt_ResourceSet, pid) == 0 )
    {
        /*
         * Joy, we'll attach a CPU reservation later in BeginCPUScheduling
         */
    }
    else if( errno != EBUSY )
    {
        /* No joy, something went wrong. */
        rk_resource_set_destroy(this->rkt_ResourceSet);
        this->rkt_ResourceSet = NULL_RESOURCE_SET;
        throw CORBA::NO_MEMORY(); // XXX
    }
    else
    {
        /* This process is already attached. */
    }
    
    /* XXX Do more checking here... */
    if( (this->rkt_CPUReserveSpec.compute_time.tv_sec == 0) &&
        (this->rkt_CPUReserveSpec.compute_time.tv_nsec == 0) )
    {
        this->rkt_CPUReserveSpec.compute_time = microsec_to_timespec(100);
    }
    this->rkt_CPUReserveSpec.period = microsec_to_timespec(period);
    this->rkt_CPUReserveSpec.deadline = microsec_to_timespec(deadline);
    this->rkt_CPUReserveSpec.reserve_type.sch_mode = rsv_mode;
    this->rkt_CPUReserveSpec.reserve_type.enf_mode = rsv_mode;
    this->rkt_CPUReserveSpec.reserve_type.rep_mode = rsv_mode;
    this->rkt_CPUReserveSpec.processor = RK_ANY_CPU;

    switch( int rc = rk_cpu_reserve_create(this->rkt_ResourceSet,
                                           &this->rkt_CPUReserve,
                                           &this->rkt_CPUReserveSpec) )
    {
    case 0:
        break;
    case -1:
        cerr << strerror(errno) << ":rk_cpu_reserve_create" << endl;
        throw Broker::InvalidScheduleParameter();
    default:
        cerr << rc << endl;
        throw Broker::Internal("Unhandled return value at: "
                               __FILE__
                               ":"
                               __STRING(__LINE__));
    }
}

void RKTask::EndCPUScheduling(void)
    throw (CORBA::SystemException)
{
    if( this->rkt_CPUReserve == NULL_RESERVE )
    {
        throw CORBA::BAD_INV_ORDER();
    }

    if( this->rkt_ReservationLog != NULL )
    {
        struct timeval tv, st, diff;
        
        timersub(&this->rkt_LastReservationLog,
                 &this->rkt_StartTime,
                 &diff);
        st.tv_sec = 0;
        st.tv_usec = 0;
        gettimeofday(&tv, NULL);
        timersub(&tv, &this->rkt_StartTime, &diff);
        (*this->rkt_ReservationLog)
            << (long)diff.tv_sec
            << "."
            << setw(6) << (long)diff.tv_usec
            << " "
            << (long)st.tv_sec
            << "."
            << setw(6) << (long)st.tv_usec
            << endl;
        this->rkt_LastReservationLog = tv;
    }
    
    rk_cpu_reserve_delete(this->rkt_ResourceSet);
    this->rkt_CPUReserve = NULL_RESERVE;
    if( this->rkt_OwnsResourceSet )
    {
        rk_resource_set_destroy(this->rkt_ResourceSet);
    }
    this->rkt_ResourceSet = NULL_RESOURCE_SET;
}

CORBA::ULong RKTask::GetComputeTime(void)
    throw (CORBA::SystemException)
{
    CORBA::ULong retval;

    retval = timespec_to_microsec(&this->rkt_CPUReserveSpec.compute_time);
    return( retval );
}

void RKTask::SetComputeTime(CORBA::ULong usecs)
    throw (CORBA::SystemException)
{
    struct timespec old_ct;

    if( usecs == 0 )
    {
        throw CORBA::BAD_PARAM();
    }
    
    old_ct = this->rkt_CPUReserveSpec.compute_time;
    this->rkt_CPUReserveSpec.compute_time = microsec_to_timespec(usecs);

    if( this->rkt_CPUReserve != NULL_RESERVE )
    {
        switch( rk_cpu_reserve_ctl(this->rkt_ResourceSet,
                                   &this->rkt_CPUReserveSpec) )
        {
        case 0:
            if( this->rkt_ReservationLog != NULL )
            {
                struct timeval tv, st, diff;
                
                timersub(&this->rkt_LastReservationLog,
                         &this->rkt_StartTime,
                         &diff);
                st.tv_sec = usecs / 1000000;
                st.tv_usec = usecs % 1000000;
                gettimeofday(&tv, NULL);
                timersub(&tv, &this->rkt_StartTime, &diff);
                (*this->rkt_ReservationLog)
                    << (long)diff.tv_sec
                    << "."
                    << setw(6) << (long)diff.tv_usec
                    << " "
                    << (long)st.tv_sec
                    << "."
                    << setw(6) << (long)st.tv_usec
                    << endl;
                this->rkt_LastReservationLog = tv;
            }
            break;
        case -1:
            switch( errno )
            {
            case EINVAL:
                ensure(0);
                break;
            case ENODEV:
                ensure(0);
                break;
            case ENOSPC:
                cerr << "** SetComputeTime failed! "
                     << usecs
                     << endl;
                break;
            case EFAULT:
                ensure(0);
                break;
            case EACCES:
                cerr << "** No permission to change compute time? "
                     << usecs
                     << endl;
                break;
            case ENOSYS:
                ensure(0);
                break;
            }
            this->rkt_CPUReserveSpec.compute_time = old_ct;
            throw CORBA::BAD_PARAM();
        }
    }
    else
    {
        /* We're not scheduling at the moment, just remember the value. */
        if( this->rkt_ReservationLog != NULL )
        {
            struct timeval tv, st, diff;
            
            timersub(&this->rkt_LastReservationLog,
                     &this->rkt_StartTime,
                     &diff);
            st.tv_sec = usecs / 1000000;
            st.tv_usec = usecs % 1000000;
            gettimeofday(&tv, NULL);
            timersub(&tv, &this->rkt_StartTime, &diff);
            (*this->rkt_ReservationLog)
                << (long)diff.tv_sec
                << "."
                << setw(6) << (long)diff.tv_usec
                << " "
                << (long)st.tv_sec
                << "."
                << setw(6) << (long)st.tv_usec
                << endl;
            this->rkt_LastReservationLog = tv;
        }
    }
}

void RKTask::ReportCPU(Broker::RealTimeTask_ptr rtt,
                       CORBA::ULong status,
                       CORBA::ULong advice)
    throw (CORBA::SystemException)
{
    INSTR_rktask_change_cpu(&INSTR_rktask_change_cpu_point, {
        CORBA::ULong ct;
        
        if( CORBA::is_nil(this->rkt_Manager.in()) )
        {
            throw CORBA::BAD_INV_ORDER();
        }
        if( CORBA::is_nil(rtt) )
        {
            throw CORBA::BAD_PARAM();
        }
        if( advice == 0 )
        {
            throw CORBA::BAD_PARAM();
        }
        
        ct = timespec_to_microsec(&this->rkt_CPUReserveSpec.compute_time);
        this->rkt_Manager->ChangeTaskCPU(rtt, ct, status, advice);
    });
}

---