proc_advocate.cc

Go to the documentation of this file.

/*
 * proc_advocate.c
 *
 * 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 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 <iostream>

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

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

enum {
    PAB_DONE,
};

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

/*
 * 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 {
    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;
} pa_data;

/**
 * Handle a SIGCHLD signal.  This function will be called when the child exits.
 *
 * @param 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);

    if( (rc = rk_resource_set_get_usage(pa_data.pa_ResourceSet, &rsu)) == 0 )
    {
        CORBA::ULong status;
        
        try
        {
            status = (CORBA::ULong)(rsu.cpu_usage - pa_data.pa_LastUsage);
            pa_data.pa_LastUsage = rsu.cpu_usage;
            pa_data.pa_Advocate->ReportCPU(pa_data.pa_Advocate,
                                           status,
                                           status);
        }
        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-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:")) != -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 '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;
        }
    }
    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();
        }
    }
    else if( task_name != NULL )
    {
        try
        {
            Broker::RealTimeTask_var rtt;
            MaxDecayTaskAdvocate *mdta;
            Broker::TaskParameters tp;
            CORBA::Any value;
            RKTask *rkt;
            
            tp.length(1);
            tp[0].name = "name";
            tp[0].value <<= task_name;
            rkt = new RKTask(tp);
            rtt = rkt->_this();
            mdta = new MaxDecayTaskAdvocate();
            value <<= rtt;
            mdta->SetDelegateAttribute("remote-object", value);
            retval = mdta->_this();
        }
        catch(const Broker::DuplicateTaskParameter &e)
        {
            ensure(0);
        }
        catch(const Broker::InvalidTaskParameter &e)
        {
            ensure(0);
        }
        catch(const Broker::MissingTaskParameter &e)
        {
            ensure(0);
        }
    }
    else
    {
        ensure(0);
    }
    
    try
    {
        struct rk_resource_set_usage rsu;
        Broker::ScheduleParameters sp;
        
        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);
        pa_data.pa_ResourceSet = rk_proc_get_rset(getpid());
        if( rk_resource_set_get_usage(pa_data.pa_ResourceSet, &rsu) == 0 )
        {
            pa_data.pa_LastUsage = rsu.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

    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;
    /* XXX Use rk periodic stuff here. */
    if( setitimer(ITIMER_REAL, &itv, NULL) == 0 )
    {
        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) )
        {
            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");
        }
    }
    else
    {
        perror("setitimer");
        retval = EXIT_FAILURE;
    }

    /* 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];

    pa_data.pa_ManagerIOR = "file://manager.ior";
    pa_data.pa_ChildPeriod = 10000;

    /*
     * 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;

        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();
        mgr->activate();
        
        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());
        
        PortableServer::POA_var st_poa = root_poa->
            create_POA("SingleThread",
                       PortableServer::POAManager::_nil(),
                       policy_list);
        
        thread->destroy();

        argc += (old_argc - argc);
        rc = paProcessOptions(argc, argv);
        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;
    }
    
    return( retval );
}

---