proc_advocate.cc

Go to the documentation of this file.

/*
 * proc_advocate.c
 *
 * 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 proc_advocate.cc
 *
 * Wrapper used to attach unmodified programs to the Broker.
 */

#include "config.h"

#include <stdlib.h>
#include <string.h>

#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>

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

#include <iostream>

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

#include "BrokerC.h"
#include "RealTimeTaskImpl.hh"
#include "MaxDecayTaskAdvocate.hh"

using namespace std;

#define INSTR_proc_len LRTIME_INSTR
#define INSTR_proc_len_data "proc_len", "the wall clock time"

#if defined(INSTR_proc_len_data)
static struct iPoint INSTR_proc_len_point = {
    INSTR_proc_len_data,
};
#endif

#if defined(INSTR_proc_advocate_periodic_data)
static double periodic_history[1000];
static struct iPoint INSTR_proc_advocate_periodic_point = {
    INSTR_proc_advocate_periodic_data,
    0,
    IPOINT_INIT_HISTORY(periodic_history),
};
#endif

#if defined(INSTR_proc_advocate_rpc_data)
#undef INSTR_proc_advocate_rpc
#define INSTR_proc_advocate_rpc LRTIME_INSTR
static struct iPoint INSTR_proc_advocate_rpc_point = {
    INSTR_proc_advocate_rpc_data,
};
#endif

#if defined(INSTR_proc_advocate_periods_data)
#undef INSTR_proc_advocate_periods
#define INSTR_proc_advocate_periods LRTIME_INSTR
static double periods_history[1000];
static struct iPoint INSTR_proc_advocate_periods_point = {
    INSTR_proc_advocate_periods_data,
    0,
    IPOINT_INIT_HISTORY(periods_history),
};
#endif

enum {
    PAB_DONE,
    PAB_CREATED_RESOURCE_SET,
    PAB_NO_OBSERVE,
};

/*
 * Flags for the pa_data.pa_Flags field.
 *
 * PAF_DONE - Stop monitoring the resource set/child process.
 */
enum {
    PAF_DONE = (1L << PAB_DONE),
    PAF_CREATED_RESOURCE_SET = (1L << PAB_CREATED_RESOURCE_SET),
    PAF_NO_OBSERVE = (1L << PAB_NO_OBSERVE),
};

/*
 * Global data for the tool.
 *
 * pa_Flags - Holds the PAF_ flags.
 * pa_ManagerIOR - The IOR of the Broker::Manager object to use.
 * pa_TaskIOR - The IOR of the Broker::RealTimeTask object to send reports to.
 * pa_TaskName - The name of the task to create.
 * pa_ChildPID - The PID of the wrapped process.
 * pa_ChildPeriod - The period to use when making reports.
 * pa_Advocate - The Broker::RealTimeTask object to send reports to.
 * pa_ResourceSet - The resource set for this process and the child.
 * pa_LastUsage - The last recorded CPU usage, in microseconds.
 */
static struct {
    PortableServer::POA_var pa_POA;
    unsigned long pa_Flags;
    const char *pa_ManagerIOR;
    const char *pa_TaskIOR;
    const char *pa_TaskName;
    pid_t pa_ChildPID;
    CORBA::ULong pa_ChildPeriod;
    Broker::RealTimeTask_ptr pa_Advocate;
    rk_resource_set_t pa_ResourceSet;
    unsigned long long pa_LastUsage;
    Broker::KeyedReportParameters pa_KeyedReportParameters;
} pa_data;

/**
 * Handle a SIGCHLD signal.  This function will be called when the child exits.
 *
 * @param sig The actual signal number received.
 */
static void sigchld(int sig)
{
    require(sig == SIGCHLD);
    
    pa_data.pa_Flags |= PAF_DONE;
}

/**
 * A signal handler that passes the received signal on to the child process.
 *
 * @param sig The actual signal number received.
 */
static void sigpass(int sig)
{
    require(pa_data.pa_ChildPID > 0);
    require((sig == SIGINT) || (sig == SIGTERM));
    
    kill(pa_data.pa_ChildPID, sig);
}

/**
 * The SIGALRM signal handler.  This function will be called at the end of
 * every period to report CPU usage to the advocate.
 *
 * @param sig The actual signal number received.
 */
static void sigalrm(int sig)
{
    struct rk_resource_set_usage rsu;
    int rc;

    require(pa_data.pa_ResourceSet != NULL_RESOURCE_SET);
    require(!CORBA::is_nil(pa_data.pa_Advocate));
    require(sig == SIGALRM);

    INSTR_proc_advocate_periods(&INSTR_proc_advocate_periods_point, {
        memset(&rsu, 0, sizeof(rsu));
        if( (rc = rk_resource_set_get_usage(pa_data.pa_ResourceSet,
                                            &rsu,
                                            NULL)) == 0 )
        {
            try
            {
                Broker::CPUReserve status;

                status.Period = pa_data.pa_ChildPeriod;
                status.Compute = (CORBA::ULong)
                    (rsu.active_cpu_usage - pa_data.pa_LastUsage);
                pa_data.pa_LastUsage = rsu.active_cpu_usage;
                pa_data.pa_Advocate->
                    ReportCPU(status,
                              status,
                              pa_data.pa_KeyedReportParameters);
            }
            catch(const CORBA::SystemException &e)
            {
                cerr << "System exception: " << e << endl;
                pa_data.pa_Flags |= PAF_DONE;
            }
            catch(...)
            {
                cerr << "Caught unknown exception..." << endl;
                pa_data.pa_Flags |= PAF_DONE;
            }
        }
        else
        {
            cerr << "rk_resource_set_get_usage: "
                 << strerror(rc)
                 << endl;
            pa_data.pa_Flags |= PAF_DONE;
        }
    });
}

/**
 * Print out the usage statement to standard error.
 *
 * @param prog_name The program name.
 */
static void paUsage(const char *prog_name)
{
    require(pa_data.pa_ManagerIOR != NULL);
    require(prog_name != NULL);

    cerr << "CPU Broker wrapper for managing unmodified programs.\n"
         << "Usage: "
         << prog_name
         << " [options] -- <command> [argument ...]\n"
         << endl
         << "Options:\n"
         << "\t-h\t\tThis help message.\n"
         << "\t-V\t\tShow the version number.\n"

         << "\t-o\t\tDo not observe the child process or make reports." << endl

         << "\t-m <ior>\tThe Broker::Manager IOR."
         << " (Default: " << pa_data.pa_ManagerIOR << ")\n"

         << "\t-t <ior>\tThe Broker::RealTimeTask IOR.\n"

         << "\t-n <name>\tThe name for the task to create.\n"

         << "\t-P <period>\tThe period as a time value."
         << " (Default: " << pa_data.pa_ChildPeriod << "us)\n"

         << endl
        
         << "Package: " << PACKAGE_STRING << endl
         << "Contact: " << PACKAGE_BUGREPORT << endl

        ;
}

/**
 * Process the command line options.
 *
 * @param argc_inout Reference to main's argc variable.  On return, the
 * variable will contain the number of arguments remaining after option
 * processing.
 * @param argv_inout Reference to main's argv variable.  On return, the
 * variable will contain the remaining argument values.
 * @return True if the options were processed correctly, false otherwise.
 */
static int paProcessOptions(int &argc_inout, char **&argv_inout)
{
    int ch, retval = 0;
    char *prog_name;
    char **argv;
    int argc;

    argc = argc_inout;
    argv = argv_inout;
    prog_name = argv[0];
    while( ((ch = getopt(argc, argv, "hVn:m:t:P:o")) != -1) && (retval == 0) )
    {
        switch( ch )
        {
        case 'm':
            if( strlen(optarg) == 0 )
            {
                cerr << prog_name << ": Manager IOR is empty" << endl;
                retval = 1;
            }
            else
            {
                pa_data.pa_ManagerIOR = optarg;
            }
            break;
        case 't':
            if( strlen(optarg) == 0 )
            {
                cerr << prog_name << ": Task IOR is empty" << endl;
                retval = 1;
            }
            else if( pa_data.pa_TaskName != NULL )
            {
                cerr << prog_name
                     << ": The -t and -n options are mutually exclusive"
                     << endl;
                retval = 1;
            }
            else
            {
                pa_data.pa_TaskIOR = optarg;
            }
            break;
        case 'n':
            if( strlen(optarg) == 0 )
            {
                cerr << prog_name << ": Task name is empty" << endl;
                retval = 1;
            }
            else if( pa_data.pa_TaskIOR != NULL )
            {
                cerr << prog_name
                     << ": The -t and -n options are mutually exclusive"
                     << endl;
                retval = 1;
            }
            else
            {
                pa_data.pa_TaskName = optarg;
            }
            break;
        case 'o':
            pa_data.pa_Flags |= PAF_NO_OBSERVE;
            break;
        case 'P':
            {
                unsigned long long usecs;
                
                /* Period for the CPU reservation. */
                if( string_to_microsec(&usecs, optarg) )
                {
                    pa_data.pa_ChildPeriod = usecs;
                }
                else
                {
                    cerr << prog_name
                         << ": -P options requires a time value\n";
                    retval = 1;
                }
            }
            break;
        case 'V':
            cerr << PACKAGE_VERSION << endl;
            retval = -1;
            break;
        case 'h':
        case '?':
        default:
            retval = 1;
            break;
        }
    }

    if( (pa_data.pa_TaskIOR == NULL) && (pa_data.pa_TaskName == NULL) )
    {
        cerr << prog_name
             << ": no task name or IOR specified\n";
        retval = 1;
    }
    
    argc_inout -= optind;
    argv_inout += optind;
    /* Optionally skip the '--' that is used to terminate the option list. */
    if( (argc_inout > 0) && (strcmp(argv_inout[0], "--") == 0) )
    {
        argc_inout -= 1;
        argv_inout += 1;
    }
    return( retval );
}

/**
 * Create the advocate and begin CPU scheduling based on the command line
 * parameters.
 *
 * @param orb Pointer to the ORB instance.
 * @param manager The manager for this machine.
 * @param task_ior The IOR of the Broker::RealTimeTask to use or NULL if a
 * task should be created.
 * @param task_name The name of the Broker::RealTimeTask to create or NULL if
 * an existing task should be used.
 * @return A pointer to the allocated advocate.
 */
static Broker::RealTimeTask_ptr paGetAdvocate(CORBA::ORB_ptr orb,
                                              Broker::Manager_ptr manager,
                                              const char *task_ior,
                                              const char *task_name)
{
    Broker::RealTimeTask_var retval;
    CORBA::Object_var obj;

    require(!CORBA::is_nil(orb));
    require(!CORBA::is_nil(manager));
    require((task_ior != NULL) || (task_name != NULL));

    if( task_ior != NULL )
    {
        obj = orb->string_to_object(task_ior);
        retval = Broker::RealTimeTask::_narrow(obj.in());
        if( CORBA::is_nil(retval.in()) )
        {
            cerr << "Invalid task factory IOR: "
                 << task_ior
                 << endl;
            throw CORBA::BAD_PARAM();
        }
        pa_data.pa_TaskName = retval->Name();
    }
    else if( task_name != NULL )
    {
        try
        {
            PortableServer::ObjectId_var oid;
            Broker::RealTimeTask_var rtt;
            MaxDecayTaskAdvocate *mdta;
            Broker::TaskParameters tp;
            RealTimeTaskImpl *rtti;
            CORBA::Any value;
            
            tp.length(1);
            tp[0].name = "name";
            tp[0].value <<= task_name;
            rtti = new RealTimeTaskImpl(tp);
            oid = pa_data.pa_POA->activate_object(rtti);
            rtt = Broker::RealTimeTask::
                _narrow(pa_data.pa_POA->id_to_reference(oid.in()));
            mdta = new MaxDecayTaskAdvocate();
            value <<= rtt.in();
            mdta->SetDelegateAttribute("remote-object", value);
            oid = pa_data.pa_POA->activate_object(mdta);
            retval = Broker::RealTimeTask::
                _narrow(pa_data.pa_POA->id_to_reference(oid.in()));
        }
        catch(const Broker::DuplicateTaskParameter &e)
        {
            cerr << e << endl;
            ensure(0);
        }
        catch(const Broker::InvalidTaskParameter &e)
        {
            cerr << e << endl;
            ensure(0);
        }
        catch(const Broker::MissingTaskParameter &e)
        {
            cerr << e << endl;
            ensure(0);
        }
    }
    else
    {
        ensure(0);
    }

    if( (pa_data.pa_ResourceSet = rk_proc_get_rset(getpid())) !=
        NULL_RESOURCE_SET )
    {
    }
    else if( (pa_data.pa_ResourceSet =
              rk_resource_set_get_by_name(pa_data.pa_TaskName)) !=
             NULL_RESOURCE_SET )
    {
    }
    else if( (pa_data.pa_ResourceSet =
              rk_resource_set_create((char *)pa_data.pa_TaskName)) ==
             NULL_RESOURCE_SET )
    {
        cerr << "rk_resource_set_create: " << strerror(errno) << endl;
    }
    else
    {
        pa_data.pa_Flags |= PAF_CREATED_RESOURCE_SET;
    }
    
    try
    {
        struct rk_resource_set_usage rsu;
        Broker::ScheduleParameters sp;

        memset(&rsu, 0, sizeof(rsu));
        sp.length(2);
        sp[0].name = "period";
        sp[0].value <<= pa_data.pa_ChildPeriod;
        sp[1].name = "pid";
        sp[1].value <<= (CORBA::Long)getpid();
        manager->AddTask(retval.in(), sp);
        if( rk_resource_set_get_usage(pa_data.pa_ResourceSet,
                                      &rsu,
                                      NULL) == 0 )
        {
            pa_data.pa_LastUsage = rsu.active_cpu_usage;
        }
    }
    catch(const Broker::DuplicateScheduleParameter &e)
    {
        cerr << e << endl;
        cerr << e.name << endl;
        throw CORBA::UNKNOWN();
    }
    catch(const Broker::InvalidScheduleParameter &e)
    {
        cerr << e << endl;
        cerr << e.message << endl;
        throw CORBA::UNKNOWN();
    }
    catch(const Broker::MissingScheduleParameter &e)
    {
        cerr << e << endl;
        cerr << e.name << endl;
        throw CORBA::UNKNOWN();
    }
    
    return( retval._retn() );
}

/**
 * The parent portion of the fork(2) between pa and the monitored utility.
 * This function will attach itself to a resource set to ensure that it has
 * some CPU time to work and then wait for gkrellm connections or the child's
 * death.
 *
 * @param rtt The Broker::RealTimeTask that is managing this process.
 * @return The return code for main().
 *
 * @sa paChildPart
 */
static int paParentPart(Broker::RealTimeTask_ptr rtt)
{
    int retval = EXIT_FAILURE;
    struct itimerval itv;
    struct sigaction sa;
    sigset_t sigmask;

    require(pa_data.pa_ChildPID > 0);
    require(!CORBA::is_nil(rtt));
    
    /*
     * We use signals as a primitive event system, so any that we use and are
     * not passed on to the child need to be blocked before setting up the
     * handlers.  Once everything has been setup, we will use sigsuspend(2) to
     * atomically unblock and wait for the signals to arrive.
     */
    sigemptyset(&sigmask);
    sigaddset(&sigmask, SIGALRM);
    sigaddset(&sigmask, SIGCHLD);
    sigaddset(&sigmask, SIGINT);
    sigaddset(&sigmask, SIGTERM);
    if( sigprocmask(SIG_BLOCK, &sigmask, NULL) < 0 )
    {
        perror("sigprocmask");

        ensure(0);
    }

    /* Setup the signal handlers. */
    sa.sa_mask = sigmask;
    sa.sa_flags = 0;
#if defined(SA_RESTART)
    sa.sa_flags |= SA_RESTART;
#endif

    if( !(pa_data.pa_Flags & PAF_NO_OBSERVE) )
    {
        sa.sa_handler = sigalrm;
        if( sigaction(SIGALRM, &sa, NULL) < 0 )
        {
            perror("sigprocmask");
            
            ensure(0);
        }
    }

    /* Catch SIGCHLD so we exit with the child. */
    sa.sa_handler = sigchld;
    if( sigaction(SIGCHLD, &sa, NULL) < 0 )
    {
        perror("sigprocmask");

        ensure(0);
    }
    
    /*
     * Pass SIGINT/SIGTERM onto the child so they can handle them as they
     * choose.  If they exit, we will get the SIGCHLD and exit shortly
     * thereafter.
     */
    sa.sa_handler = sigpass;
    if( (sigaction(SIGINT, &sa, NULL) < 0) ||
        (sigaction(SIGTERM, &sa, NULL) < 0) )
    {
        perror("sigprocmask");

        ensure(0);
    }
    
    itv.it_interval.tv_sec = 0;
    itv.it_interval.tv_usec = pa_data.pa_ChildPeriod;
    itv.it_value.tv_sec = 0;
    itv.it_value.tv_usec = pa_data.pa_ChildPeriod;
    /** @todo Use rk periodic stuff here. */
    if( !(pa_data.pa_Flags & PAF_NO_OBSERVE) &&
        (setitimer(ITIMER_REAL, &itv, NULL) == -1) )
    {
        perror("setitimer");
        retval = EXIT_FAILURE;
    }
    else
    {
        sigset_t empty_sigmask;
        int status;

        sigemptyset(&empty_sigmask);

        pa_data.pa_Advocate = rtt;
        
        /* Keep handling signals until the child has died. */
        while( !(pa_data.pa_Flags & PAF_DONE) )
        {
            INSTR_proc_len(&INSTR_proc_len_point, {
            INSTR_proc_advocate_periodic(&INSTR_proc_advocate_periodic_point, {
                sigsuspend(&empty_sigmask);
            });
            });
        }
        
        pa_data.pa_Advocate = NULL;
        
        /* Cleanup the child's status. */
        if( wait(&status) >= 0 )
        {
            if( WIFEXITED(status) )
            {
                retval = WEXITSTATUS(status);
            }
            else if( WIFSIGNALED(status) )
            {
                retval = EXIT_SUCCESS;
            }
            else
            {
                retval = EXIT_FAILURE;
            }
        }
        else
        {
            perror("wait");
            retval = EXIT_FAILURE;
        }
        
        /* Clear the timer. */
        memset(&itv, 0, sizeof(itv));
        if( setitimer(ITIMER_REAL, &itv, NULL) < 0 )
        {
            perror("setitimer");
        }
    }

    /* We will be dying soon, ignore any signals and */
    signal(SIGALRM, SIG_IGN);
    signal(SIGCHLD, SIG_IGN);
    signal(SIGINT, SIG_IGN);
    signal(SIGTERM, SIG_IGN);

    /* ... restore the old signal mask. */
    if( sigprocmask(SIG_UNBLOCK, &sigmask, NULL) < 0 )
    {
        perror("sigprocmask");
        
        ensure(0);
    }
    
    return( retval );
}

/**
 * The child portion of the fork(2) between the proc_advocate and the monitored
 * utility.  This function will execvp(3) the utility with the given arguments.
 *
 * @param argv The utility to startup and its arguments.
 * @return A failure exit code, otherwise this function will not return because
 * of the execvp(3).
 *
 * @sa paParentPart
 */
static int paChildPart(char *argv[])
{
    int retval = EXIT_SUCCESS;
    sigset_t sigmask;

    /* Reset the signal mask for the child. */
    sigfillset(&sigmask);
    if( sigprocmask(SIG_UNBLOCK, &sigmask, NULL) < 0 )
    {
        perror("sigprocmask");
        
        ensure(0);
    }

    execvp(argv[0], argv);
    
    /* FALLTHROUGH, normal operation will not reach this point. */
    perror(argv[0]);
    switch( errno )
    {
    case ENOENT:
    case EPERM:
        retval = 127;
        break;
    default:
        retval = EXIT_FAILURE;
        break;
    }
    return( retval );
}

int main(int argc, char *argv[])
{
    int lpc, old_argc = argc, retval = EXIT_FAILURE;
    const char *prog_name = argv[0];

    /* Default values. */
    pa_data.pa_ManagerIOR = "file://manager.ior";
    pa_data.pa_ChildPeriod = 33333;

    atexit(iPrintPointsAtExit);
    
    /*
     * Change the argc to not include the wrappee's arguments, otherwise,
     * our call to ORB_init could consume the wrappee's ORB arguments.
     */
    for( lpc = 0; (lpc < argc) && (strcmp(argv[lpc], "--") != 0); lpc++ );

    if( lpc < argc )
    {
        argc = lpc;
    }
    
    try
    {
        CORBA::ORB_var orb = CORBA::ORB_init(argc, argv);
        int rc;

        for( ; lpc < old_argc; argc++, lpc++ )
        {
            argv[argc] = argv[lpc];
        }
        argv[argc] = NULL;

        rc = paProcessOptions(argc, argv);
        
        if( pa_data.pa_TaskIOR == NULL )
        {
            CORBA::Object_var obj = orb->resolve_initial_references("RootPOA");
            
            PortableServer::POA_var root_poa =
                PortableServer::POA::_narrow(obj.in());
            PortableServer::POAManager_var mgr = root_poa->the_POAManager();
            
            PortableServer::ThreadPolicy_var thread = root_poa->
                create_thread_policy(PortableServer::SINGLE_THREAD_MODEL);
            
            CORBA::PolicyList policy_list;
            policy_list.length(1);
            policy_list[0] =
                PortableServer::ThreadPolicy::_duplicate(thread.in());
            
            pa_data.pa_POA = root_poa->
                create_POA("SingleThread", mgr.in(), policy_list);
            
            thread->destroy();
            thread = PortableServer::ThreadPolicy::_nil();
            
            mgr->activate();
        }
        
        if( (rc == 0) && (argc > 0) )
        {
            Broker::RealTimeTask_var rtt;
            Broker::Manager_var manager;
            CORBA::Object_var obj;

            obj = orb->string_to_object(pa_data.pa_ManagerIOR);
            manager = Broker::Manager::_narrow(obj.in());
            if( CORBA::is_nil(manager.in()) )
            {
                cerr << "Invalid manager IOR: "
                     << pa_data.pa_ManagerIOR
                     << endl;
                throw CORBA::BAD_PARAM();
            }

            /*
             * Block signals so we do not die without removing ourself from the
             * manager.
             */
            {
                sigset_t sigmask;
                
                sigaddset(&sigmask, SIGINT);
                sigaddset(&sigmask, SIGTERM);
                if( sigprocmask(SIG_BLOCK, &sigmask, NULL) < 0 )
                {
                    perror("sigprocmask");
                    
                    ensure(0);
                }
            }
            
            rtt = paGetAdvocate(orb.in(),
                                manager.in(),
                                pa_data.pa_TaskIOR,
                                pa_data.pa_TaskName);
            if( (pa_data.pa_ChildPID = fork()) > 0 )
            {
                retval = paParentPart(rtt.in());
            }
            else if( pa_data.pa_ChildPID == 0 )
            {
                return( paChildPart(argv) );
            }
            else
            {
                /* The fork failed. */
                retval = EXIT_FAILURE;
            }
            manager->RemoveTask(rtt.in());
        }
        else
        {
            if( rc == 0 )
            {
                cerr << "No utility to execute\n";
            }
            if( rc >= 0 )
            {
                paUsage(prog_name);
            }
        }
    }
    catch(const CORBA::UNKNOWN &e)
    {
        // Ignore...
        paUsage(prog_name);
    }
    catch(const CORBA::SystemException &e)
    {
        cerr << "Caught Exception: " << e << endl;
    }
    catch(...)
    {
        cerr << "Caught an unhandled exception" << endl;
    }

    if( pa_data.pa_Flags & PAF_CREATED_RESOURCE_SET )
    {
        rk_resource_set_destroy(pa_data.pa_ResourceSet);
        pa_data.pa_ResourceSet = NULL_RESOURCE_SET;
    }
    
    return( retval );
}

---