RKObserverAdvocate.cc

Go to the documentation of this file.

/*
 * RKObserverAdvocate.cc
 *
 * Copyright (c) 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 RKObserverAdvocate.cc
 *
 * Implementation of the RKObserverAdvocate class.
 */

#include "config.h"

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

#include <iostream>

#include <rk.h>
#include <rk_util.h>

#include "RKObserverAdvocate.hh"

using namespace std;

#if !defined(timespecclear)
/**
 * Clear a timespec structure.
 *
 * @param tvp Pointer to the timespec object to clear.
 */
#define timespecclear(tvp)      ((tvp)->tv_sec = (tvp)->tv_nsec = 0)

/**
 * Test if a timespec has non-zero values.
 *
 * @param tvp Pointer to the timespec object to check.
 */
#define timespecisset(tvp)      ((tvp)->tv_sec || (tvp)->tv_nsec)

/**
 * Compare two timespec structures.
 *
 * @param tvp Pointer to the 'left hand side' of the comparison.
 * @param uvp Pointer to the 'right hande side' of the comparison.
 * @param cmp The comparison operator.  (e.g. '<', '>', ...)
 */
#define timespeccmp(tvp, uvp, cmp)                                      \
        (((tvp)->tv_sec == (uvp)->tv_sec) ?                             \
            ((tvp)->tv_nsec cmp (uvp)->tv_nsec) :                       \
            ((tvp)->tv_sec cmp (uvp)->tv_sec))

/**
 * Add one timespec to another.
 *
 * @param vvp Pointer to the timespec that will be added to.
 * @param uvp Pointer to the timespec to add.
 */
#define timespecadd(vvp, uvp)                                           \
        do {                                                            \
                (vvp)->tv_sec += (uvp)->tv_sec;                         \
                (vvp)->tv_nsec += (uvp)->tv_nsec;                       \
                if ((vvp)->tv_nsec >= 1000000000) {                     \
                        (vvp)->tv_sec++;                                \
                        (vvp)->tv_nsec -= 1000000000;                   \
                }                                                       \
        } while (0)

/**
 * Subtract one timespec from another.
 *
 * @param vvp Pointer to the timespec that will be subtracted from.
 * @param uvp Pointer to the timespec to subtract.
 */
#define timespecsub(vvp, uvp)                                           \
        do {                                                            \
                (vvp)->tv_sec -= (uvp)->tv_sec;                         \
                (vvp)->tv_nsec -= (uvp)->tv_nsec;                       \
                if ((vvp)->tv_nsec < 0) {                               \
                        (vvp)->tv_sec--;                                \
                        (vvp)->tv_nsec += 1000000000;                   \
                }                                                       \
        } while (0)
#endif

/**
 * Trampoline function for new threads that calls the 'run' method of
 * RKObserverAdvocate.
 *
 * @param arg The RKObserverAdvocate object.
 * @return NULL.
 */
static void *trampoline(void *arg)
{
    RKObserverAdvocate *rkoa = (RKObserverAdvocate *)arg;
    void *retval = NULL;

    require(arg != NULL);
    
    rkoa->run();

    return( retval );
}

RKObserverAdvocate::RKObserverAdvocate(void)
    throw (CORBA::SystemException)
{
    if( pthread_mutex_init(&this->rkoa_Mutex, NULL) != 0 )
    {
        ensure(0);
    }
    if( pthread_cond_init(&this->rkoa_Cond, NULL) != 0 )
    {
        ensure(0);
    }
    this->rkoa_ResourceSet = NULL_RESOURCE_SET;
    this->rkoa_Period = microsec_to_timespec(10000);
    this->rkoa_Thread = 0;
}

RKObserverAdvocate::~RKObserverAdvocate(void)
{
    this->rkoa_Thread = 0;
    if( pthread_mutex_destroy(&this->rkoa_Mutex) != 0 )
    {
        ensure(0);
    }
    if( pthread_cond_destroy(&this->rkoa_Cond) != 0 )
    {
        ensure(0);
    }
}

void
RKObserverAdvocate::BeginCPUScheduling(const Broker::ScheduleParameters &sp)
    throw (CORBA::SystemException,
           Broker::DuplicateScheduleParameter,
           Broker::InvalidScheduleParameter,
           Broker::MissingScheduleParameter)
{
    unsigned long long period = ~0;
    unsigned int lpc;
    const char *str;
    pid_t pid = -1;
    
    if( CORBA::is_nil(this->dm_RemoteObject.in()) )
    {
        throw CORBA::BAD_INV_ORDER();
    }

    if( this->rkoa_ResourceSet != NULL_RESOURCE_SET )
    {
        throw CORBA::BAD_INV_ORDER();
    }

    /* Process the arguments */
    for( lpc = 0; lpc < sp.length(); lpc++ )
    {
        if( strcasecmp(sp[lpc].name.in(), "pid") == 0 )
        {
            CORBA::Long c_pid;
            const char *str;

            if( pid != -1 )
            {
                throw Broker::DuplicateScheduleParameter("pid");
            }
            else if( sp[lpc].value >>= c_pid )
            {
                pid = c_pid;
            }
            else if( (sp[lpc].value >>= str) &&
                     (sscanf(str, "%d", &pid) == 1) )
            {
            }
            else
            {
                throw Broker::InvalidScheduleParameter("'pid' not a long",
                                                       sp[lpc]);
            }
            if( pid < 0 )
            {
                pid = -1;
                throw Broker::InvalidScheduleParameter("'pid' not a long",
                                                       sp[lpc]);
            }
        }
        else if( strcasecmp(sp[lpc].name.in(), "period") == 0 )
        {
            CORBA::ULong period_ul;
            
            if( period != ~0ULL )
            {
                throw Broker::DuplicateScheduleParameter("period");
            }
            else if( sp[lpc].value >>= period_ul )
            {
                period = period_ul;
            }
            else if( (sp[lpc].value >>= str) &&
                     string_to_microsec(&period, str) )
            {
            }
            else
            {
                throw Broker::InvalidScheduleParameter(
                        "'period' is not a time",
                        sp[lpc]);
            }
        }
    }

    if( pid < -1 )
    {
        throw CORBA::BAD_PARAM();
    }
    
    this->dm_RemoteObject->BeginCPUScheduling(sp);

    this->rkoa_Period = microsec_to_timespec(period);

    this->rkoa_Advocate = Broker::RealTimeTask::_duplicate(this->_this());
    
    {
        this->rkoa_ResourceSet = rk_proc_get_rset(pid);

        ensure(this->rkoa_ResourceSet != NULL_RESOURCE_SET); // XXX
    }
    
    if( pthread_create(&this->rkoa_Thread, NULL, trampoline, this) != 0 )
    {
        this->dm_RemoteObject->EndCPUScheduling();
        throw CORBA::NO_MEMORY();
    }
}

void RKObserverAdvocate::EndCPUScheduling(void)
    throw (CORBA::SystemException)
{
    pthread_t pt;
    void *rc;
    
    if( pthread_mutex_lock(&this->rkoa_Mutex) != 0 )
    {
        ensure(0);
    }
    {
        this->rkoa_ResourceSet = NULL_RESOURCE_SET;
        pt = this->rkoa_Thread;
        this->rkoa_Thread = 0;
        if( pthread_cond_signal(&this->rkoa_Cond) != 0 )
        {
            ensure(0);
        }
    }
    if( pthread_mutex_unlock(&this->rkoa_Mutex) != 0 )
    {
        ensure(0);
    }
    if( pthread_join(pt, &rc) != 0 )
    {
        ensure(0);
    }

    this->dm_RemoteObject->EndCPUScheduling();
}

void RKObserverAdvocate::run(void)
{
    struct rk_resource_set_proc proc_elements[128];
    struct rk_resource_set_proc_cache pc;
    Broker::KeyedReportParameters krp;
    unsigned long long last_usage = 0;
    struct rk_resource_set_usage rsu;
    struct timespec abstime;
    pid_t my_pid;
    int rc;

    pc.data = proc_elements;
    pc.used = 0;
    pc.length = 128;
    memset(&rsu, 0, sizeof(rsu));
    my_pid = getpid();
    if( rk_resource_set_detach_process(rk_proc_get_rset(my_pid),
                                       my_pid) == -1 )
    {
        perror("rk_resource_set_detach_process");
    }
    if( rk_resource_set_attach_process(this->rkoa_ResourceSet,
                                       my_pid) == -1 )
    {
        perror("rk_resource_set_attach_process");
    }
    clock_gettime(CLOCK_REALTIME, &abstime);
    if( (rc = pthread_mutex_lock(&this->rkoa_Mutex)) != 0 )
    {
        cerr << "pthread_mutex_lock: " << rc << endl;
        ensure(0);
    }
    do {
        if( this->rkoa_ResourceSet == NULL_RESOURCE_SET )
        {
            /* ... */
        }
        else
        {
            int rc;

            if( (rc = rk_resource_set_get_usage(this->rkoa_ResourceSet,
                                                &rsu,
                                                &pc)) != 0 )
            {
                cerr << "rk_resource_set_get_usage: "
                     << strerror(rc)
                     << endl;
            }
            else
            {
                try
                {
                    unsigned long long total_usage;
                    Broker::CPUReserve status;

                    total_usage = rsu.inactive_cpu_usage +
                        rsu.active_cpu_usage;
                    status.Period = timespec_to_microsec(&this->rkoa_Period);
                    status.Compute = (CORBA::ULong)
                        (total_usage - last_usage) + 50;
                    /*
                     * NOTE: We must use PassCPU with rkoa_Advocate here
                     * because using _this() will return an object reference in
                     * the default POA.
                     */
                    this->PassCPU(this->rkoa_Advocate, status, status, krp);
                    last_usage = total_usage;
                }
                catch(const CORBA::SystemException &e)
                {
                    cerr << "System exception: " << e << endl;
                }
                catch(...)
                {
                    cerr << "Caught unknown exception..." << endl;
                }
            }
            timespecadd(&abstime, &this->rkoa_Period);
            rc = pthread_cond_timedwait(&this->rkoa_Cond,
                                        &this->rkoa_Mutex,
                                        &abstime);
            switch( rc )
            {
            case 0:
                /* Time to exit. */
                break;
            case ETIMEDOUT:
                /* Expected. */
                break;
            default:
                cerr << "pthread_cond_timedwait: " << rc << endl;
                ensure(0);
                break;
            }
        }
    } while( this->rkoa_Thread != 0 );
    if( pthread_mutex_unlock(&this->rkoa_Mutex) != 0 )
    {
        ensure(0);
    }
    rk_resource_set_proc_cache_release(&pc);
}

---