broker_allup.cc

Go to the documentation of this file.

/*
 * broker_allup.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 broker_allup.cc
 *
 * Main file for an edu::utah::pces::Allup server.
 */

#include "config.h"

#include <stdlib.h>
#include <signal.h>
#include <unistd.h>

#include <sys/time.h>

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

#include <ltdl.h>

#if defined(HAVE_LIBRK)
#include <rk/rk.h>
#include <rk/rk_error.h>
#endif

#include <fstream>
#include <iostream>

#include <AllupImpl.hh>

using namespace std;

/**
 * Flag bit numbers for the executable as a whole.
 */
enum {
    BAB_DAEMON, /**< Fork into the background. */
};

/**
 * Flags bits derived from the bit numbers above.
 */
enum {
    BAF_DAEMON = (1L << BAB_DAEMON),
};

/**
 * Global data for the server.
 *
 * ba_ResourceSet - The resource set for this process.
 */
static struct {
    CORBA::ORB_var ba_ORB;
    unsigned long ba_Flags;
    const char *ba_IORFileName;
    const char *ba_PIDFileName;
#if defined(HAVE_LIBRK)
    rk_resource_set_t ba_ResourceSet;
#endif
} ba_data;

/**
 * Signal handler for SIGINT/SIGTERM.
 *
 * @param sig The actual signal number received.
 */
static void sig_exit_handler(int sig)
{
    ba_data.ba_ORB->shutdown(0); /* Break out of orb->run(). */
}

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

    cerr << "Server for CPU broker objects." << endl
         << "Usage: "
         << prog_name
         << " [options]\n"
         << endl
         << "Options:\n"
         << "  -h\t\tThis help message.\n"
         << "  -V\t\tShow the version number.\n"

         << "  -f\t\tFork into the background.\n"
         << "  -p <file>\tThe PID file name.\n"
         << "  -o <file>\tThe IOR file name. (Default: "
         << ba_data.ba_IORFileName << ")" << endl

         << 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 baProcessOptions(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, "hVfp:o:")) != -1) && (retval == 0) )
    {
        switch( ch )
        {
        case 'f':
            ba_data.ba_Flags |= BAF_DAEMON;
            break;
        case 'p':
            if( strlen(optarg) == 0 )
            {
                cerr << "Empty PID file name" << endl;
                retval = 1;
            }
            else
            {
                ba_data.ba_PIDFileName = optarg;
            }
            break;
        case 'o':
            if( strlen(optarg) == 0 )
            {
                cerr << "Empty IOR file name" << endl;
                retval = 1;
            }
            else
            {
                ba_data.ba_IORFileName = optarg;
            }
            break;
        case 'V':
            cerr << PACKAGE_VERSION << endl;
            retval = -1;
            break;
        case 'h':
        case '?':
        default:
            retval = 1;
            break;
        }
    }
    argc_inout -= optind;
    argv_inout += optind;
    return( retval );
}

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

    /* Default values. */
    ba_data.ba_IORFileName = "allup.ior";

    /* Do not let them kill us until we are ready. */
    signal(SIGINT, SIG_IGN);
    signal(SIGTERM, SIG_IGN);

    atexit(iPrintPointsAtExit);
    
#if defined(HAVE_ACE)
    ACE_Log_Msg::instance()->priority_mask(0, ACE_Log_Msg::PROCESS);
#endif

#if defined(HAVE_LIBRK)
    {
        char set_name[RK_NAME_LEN];
        rk_resource_set_t rs;

        snprintf(set_name, sizeof(set_name), "allup.%d", getpid());
        if( rk_proc_get_rset(getpid()) != NULL_RESOURCE_SET )
        {
            /* Already have a resource set. */
        }
        else if( (rs = rk_resource_set_create(set_name)) != NULL )
        {
            struct cpu_reserve_attr cra;
            rk_reserve_t cr;
            
            memset(&cra, 0, sizeof(cra));
            cra.compute_time.tv_sec = 0;
            cra.compute_time.tv_nsec = 50000;
            cra.period.tv_sec = 0;
            cra.period.tv_nsec = 10000000;
            cra.deadline = cra.period;
            cra.reserve_type.sch_mode = RSV_SOFT;
            cra.reserve_type.enf_mode = RSV_SOFT;
            cra.reserve_type.rep_mode = RSV_SOFT;
            cra.processor = RK_ANY_CPU;
            if( (rk_cpu_reserve_create(rs, &cr, &cra) == RK_SUCCESS) &&
                (rk_resource_set_attach_process(rs, getpid()) == RK_SUCCESS) )
            {
                ba_data.ba_ResourceSet = rs;
            }
            else
            {
                rk_resource_set_destroy(rs);
            }
        }
        else
        {
            perror("rk_resource_set_create");
        }
    }
#endif
    
    try
    {
        int rc;
        
        /* ORB/POA setup */
        ba_data.ba_ORB = CORBA::ORB_init(argc, argv);

        try
        {
            CORBA::Object_var obj = ba_data.ba_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());
            
            PortableServer::POA_var st_poa = root_poa->
                create_POA("SingleThread", mgr.in(), policy_list);
            
            thread->destroy();
            thread = PortableServer::ThreadPolicy::_nil();
            
            mgr->activate();
            
            rc = baProcessOptions(argc, argv);
            if( rc > 0 )
            {
                baUsage(prog_name);
            }
            else if( rc == 0 )
            {
                PortableServer::ObjectId_var allup_oid;
                edu::utah::pces::Allup_var allup;
                struct sigaction sa;
                sigset_t sigmask;
                AllupImpl *ai;
                
                ai = new AllupImpl(st_poa.in());
                allup_oid = st_poa->activate_object(ai);
                allup = edu::utah::pces::Allup::
                    _narrow(st_poa->id_to_reference(allup_oid.in()));
                
                {
                    ofstream ostr(ba_data.ba_IORFileName);
                    
                    ostr << ba_data.ba_ORB->object_to_string(allup.in())
                         << endl;
                    ostr.close();
                }

                /* Install our signal handlers, */
                sigemptyset(&sigmask);
                sigaddset(&sigmask, SIGINT);
                sigaddset(&sigmask, SIGTERM);

                sa.sa_mask = sigmask;
                sa.sa_flags = 0;
#if defined(SA_RESTART)
                sa.sa_flags |= SA_RESTART;
#endif
                sa.sa_handler = sig_exit_handler;
                sigaction(SIGINT, &sa, NULL);
                sigaction(SIGTERM, &sa, NULL);

                /* ... try to become a daemon, */
                if( ba_data.ba_Flags & BAF_DAEMON )
                {
                    daemon(1, 1);
                    /* Parent exits here. */
                }

                /* ... try to write out our pid, */
                if( ba_data.ba_PIDFileName != NULL )
                {
                    FILE *file;

                    if( (file = fopen(ba_data.ba_PIDFileName, "w")) != NULL )
                    {
                        fprintf(file, "%d", getpid());
                        fclose(file);
                        file = NULL;
                    }
                }
                
                /* ... handle events, and then */
                ba_data.ba_ORB->run();

                /* ... uninstall our signal handlers. */
                signal(SIGINT, SIG_IGN);
                signal(SIGTERM, SIG_IGN);
            }
        }
        catch(const CORBA::SystemException &e)
        {
            cerr << "Caught CORBA exception: " << e << endl;
        }
        catch(...)
        {
            cerr << "Caught unknown exception..." << endl;
        }
        
        ba_data.ba_ORB->destroy();
        ba_data.ba_ORB = CORBA::ORB::_nil();
    }
    catch(const CORBA::SystemException &e)
    {
        cerr << "Caught CORBA exception: " << e << endl;
    }
    catch(...)
    {
        cerr << "Caught unknown exception..." << endl;
    }
    
#if defined(HAVE_LIBRK)
    if( ba_data.ba_ResourceSet != NULL_RESOURCE_SET )
    {
        rk_resource_set_destroy(ba_data.ba_ResourceSet);
    }
#endif
    
    return( retval );
}

---