<commit_before>/******************************************************************************
gram_job_manager.c 

Description:
    Resource Allocation Job Manager

CVS Information:

    $Source$
    $Date$
    $Revision$
    $Author$
******************************************************************************/

/******************************************************************************
                             Include header files
******************************************************************************/
#include <stdio.h>
#include <malloc.h>
#include <sys/param.h>
#include <sys/time.h>
#include <string.h> /* for strdup() */
#include <memory.h>
#include <nexus.h>
#include "gram_client.h"
#include "grami_client.h"
#include "gram_rsl.h"
#include "grami_jm.h"

/******************************************************************************
                               Type definitions
******************************************************************************/
typedef struct _gram_job_manager_monitor_t
{
    nexus_mutex_t          mutex;
    nexus_cond_t           cond;
    volatile nexus_bool_t  done;
} gram_job_manager_monitor_t;

/******************************************************************************
                          Module specific prototypes
******************************************************************************/
static void 
graml_cancel_handler(nexus_endpoint_t * endpoint,
                     nexus_buffer_t * buffer,
                     nexus_bool_t is_non_threaded_handler);

static void 
graml_start_time_handler(nexus_endpoint_t * endpoint,
                         nexus_buffer_t * buffer,
                         nexus_bool_t is_non_threaded_handler);

static int 
attach_requested(void * arg,
                 char * url,
                 nexus_startpoint_t * sp);
static int 
status_file_gen(char * job_status);

static void 
tree_free(gram_specification_t * sp);

static char *
genfilename(char * prefix, char * path, char * sufix);

static void
notice(char * s);

/******************************************************************************
                       Define variables for external use
******************************************************************************/
char * grami_jm_libexecdir = GLOBUS_LIBEXECDIR;
int  gram_print_debug = 0;

/******************************************************************************
                       Define module specific variables
******************************************************************************/
static nexus_handler_t handlers[] =
{ 
    {NEXUS_HANDLER_TYPE_NON_THREADED, graml_cancel_handler},
    {NEXUS_HANDLER_TYPE_NON_THREADED, graml_start_time_handler},
};

static char     tmpbuf[1024];
#define notice2(a,b) {sprintf(tmpbuf, a,b); notice(tmpbuf);}
#define notice3(a,b,c) {sprintf(tmpbuf, a,b,c); notice(tmpbuf);}
#define notice4(a,b,c,d) {sprintf(tmpbuf, a,b,c,d); notice(tmpbuf);}

static char * grami_jm_home_dir = NULL;
static char * grami_jm_arg_libexecdir = NULL;
static char * grami_jm_status_dir = NULL;
static gram_job_manager_monitor_t  job_manager_monitor;
static nexus_endpointattr_t        EpAttr;
static nexus_endpoint_t            GlobalEndpoint;
static char                        callback_contact[GRAM_MAX_MSG_SIZE];
static char                        job_contact[GRAM_MAX_MSG_SIZE];
static char                        my_globusid[GRAM_MAX_MSG_SIZE];
static int                         my_count;

static FILE *                      gram_log_fp;

/*
 * NexusExit() (required of all Nexus programs)
 */
void NexusExit(void)
{
} /* end NexusExit() */

/*
 * NexusAcquiredAsNode() (required of all Nexus programs)
 */
int NexusAcquiredAsNode(nexus_startpoint_t *startpoint)
{
    nexus_startpoint_bind(startpoint, &GlobalEndpoint);
    return 0;
} /* end NexusAcquiredAsNode() */

/*
 * NexusBoot() (required of all Nexus programs)
 */
int NexusBoot(nexus_startpoint_t * startpoint)
{
    nexus_endpointattr_init(&EpAttr);
    nexus_endpointattr_set_handler_table(&EpAttr,
                                    handlers,
                                    sizeof(handlers)/sizeof(nexus_handler_t));

    nexus_endpoint_init(&GlobalEndpoint, &EpAttr);
    nexus_startpoint_bind(startpoint, &GlobalEndpoint);

    return NEXUS_SUCCESS;

} /* end NexusBoot() */

/******************************************************************************
Function:       main()
Description:
Parameters:
Returns:
******************************************************************************/
int 
main(int argc,
     char **argv)
{
    int                    i;
    int                    size;
    int                    rc;
    int                    count;
    int                    n_nodes;
    int                    format;
    int                    job_status;
    int                    job_state_mask;
    int                    message_handled;
    char                   description[GRAM_MAX_MSG_SIZE];
    char                   test_dat_file[GRAM_MAX_MSG_SIZE];
    char                   gram_logfile[GRAM_MAX_MSG_SIZE];
    char *                 tmp_ptr;
    char *                 my_host;
    char *                 globusid_ptr;
    unsigned short         my_port;
    FILE *                 args_fp;
    nexus_byte_t           type;
    nexus_byte_t *         ptr;
    nexus_byte_t           bformat;
    nexus_byte_t           buffer[GRAM_MAX_MSG_SIZE];
    nexus_buffer_t         reply_buffer;
    nexus_startpoint_t     reply_sp;
    gram_specification_t * description_tree;


    rc = nexus_init(&argc,
		    &argv,
		    "NEXUS_ARGS", /* conf info env variable          */
		    "nx",         /* package designator              */
		    NULL);        /* additional modules              */

    if (rc != NEXUS_SUCCESS)
    {
	fprintf(stderr, "nexus_init() failed with rc=%d\n", rc);
	exit(1);
    }

    *test_dat_file = '\0';

    /*
     * Parse the command line arguments
     */
    for (i = 1; i < argc; i++)
    {
        if ((strcmp(argv[i], "-t") == 0)
                 && (i + 1 < argc))
        {
            strcpy(test_dat_file, argv[i+1]);
            i++;
        }
        else if (strcmp(argv[i], "-d") == 0)
        {
            gram_print_debug = 1;
        }
        else if ((strcmp(argv[i], "-home") == 0)
                 && (i + 1 < argc))
        {
            grami_jm_home_dir = argv[i+1];
            i++;
        }
        else if ((strcmp(argv[i], "-e") == 0)
                 && (i + 1 < argc))
        {
            grami_jm_arg_libexecdir = argv[i+1];
            i++;
        }
        else
        {
            fprintf(stderr, "Usage: %s [-h deploy home dir ] [-e lib exe dir] [-d debug print] [-t test dat file]\n", argv[0]);
            exit(1);
        }
    }

    if (gram_print_debug)
    {
        /*
         * Open the gram logfile just for testing!
         */
        sprintf(gram_logfile, "gram_job_mgr_%lu.log",
                (unsigned long) getpid());

        if ((gram_log_fp = fopen(gram_logfile, "a")) == NULL)
        {
            sprintf(gram_logfile, "/tmp/gram_job_mgr_%lu.log",
                   (unsigned long) getpid());

            if ((gram_log_fp = fopen(gram_logfile, "a")) == NULL)
            {
                fprintf(stderr, "Cannot open gram logfile.\n");
                exit(1);
            }
        }

        setbuf(gram_log_fp, NULL);
    }

    notice("-------------------------------------------------");
    notice("entering gram_job_manager");

    if (grami_jm_home_dir)
    {
        grami_jm_status_dir = genfilename(grami_jm_home_dir, "tmp", NULL);

        if (grami_jm_arg_libexecdir)
        {
            grami_jm_libexecdir = genfilename(grami_jm_home_dir, grami_jm_arg_libexecdir, NULL);
        }
        else
        {
            grami_jm_libexecdir = genfilename(grami_jm_home_dir, "libexec", NULL);
        }
    }

    notice2("grami_jm_libexecdir = %s", grami_jm_libexecdir);

    if ((globusid_ptr = (char*) getenv("GLOBUSID")) == NULL)
    {
        strcpy(my_globusid, "unknown_globusid");
    }
    else
    {
        strcpy(my_globusid, globusid_ptr);
    }
 
    /*
     *  if a test_dat_file has been defined, read data from the file 
     *  instead of from stdin.
     */
    if (strlen(test_dat_file) > 0)
    {
        if ((args_fp = fopen(test_dat_file, "r")) == NULL)
        {
            notice2("Cannot open test file %s.", test_dat_file);
            exit(1);
        }
    }
    else
    {
         args_fp = stdin;
    }

    /*
     * Read the format incomming message.
     */
    if (fread(buffer, 1, 1, args_fp) <= 0)
    {
        notice("fread() failed.");
    }
    format = (int)buffer[0];

    /*
     * Read the size incomming message.
    if (fread(buffer, 1, 4, args_fp) <= 0)
     */
    if (fread(buffer, 1, nexus_dc_sizeof_remote_int(1, format), args_fp) <= 0)
    {
        notice("fread() failed.");
    }
    ptr = buffer;
    nexus_dc_get_int(&ptr, &count, 1, format);

    /*
     * Read the remainder of the incomming message.
     */
    if (fread(buffer, 1, count - nexus_dc_sizeof_remote_int(1, format) + 1,
        args_fp) <= 0)
    {
        notice("fread() failed.");
    }

    ptr = buffer;
    nexus_user_get_int(&ptr, &count, 1, format);
    nexus_user_get_char(&ptr, description, count, format);
    *(description+count)= '\0';
    nexus_user_get_int(&ptr, &job_state_mask, 1, format);
    nexus_user_get_int(&ptr, &count, 1, format);
    nexus_user_get_char(&ptr, callback_contact, count, format);
    *(callback_contact+count)= '\0';
    nexus_user_get_startpoint(&ptr, &reply_sp, 1, format);

    notice2("description = %s", description);
    notice2("job state mask = %i",job_state_mask);
    notice2("callback contact = %s", callback_contact);

    /* Initialize termination monitor */
    nexus_mutex_init(&job_manager_monitor.mutex, (nexus_mutexattr_t *) NULL);
    nexus_cond_init(&job_manager_monitor.cond, (nexus_condattr_t *) NULL);
    job_manager_monitor.done = NEXUS_FALSE;

    /* allow other Nexus programs to attach to us */
    my_port = 0;
    rc = nexus_allow_attach(&my_port,      /* port            */
                            &my_host,     /* host            */
                            attach_requested, /* approval_func() */
                            NULL);
    if (rc != 0)
    {
       return(GRAM_ERROR_INVALID_REQUEST);
    } 
    else
    {
        sprintf(job_contact, "x-nexus://%s:%hu/%lu/%lu/", 
                              my_host,
                              my_port,
                              (unsigned long) getpid(),
                              (unsigned long) time(0));
    }
    description_tree = gram_specification_parse(description);

    /*
     * Start the job.  If successful reply with job_contact else
     * send error status.
     */
    job_status = grami_jm_job_request(job_contact, description_tree);

    if (job_status == 0)
    {
        count= strlen(job_contact);
	size = nexus_sizeof_int(1);
	size += nexus_sizeof_int(1);
	size += nexus_sizeof_char(count);
	nexus_buffer_init(&reply_buffer, size, 0);
        nexus_put_int(&reply_buffer, &job_status, 1);
        nexus_put_int(&reply_buffer, &count, 1);
	nexus_put_char(&reply_buffer, job_contact, count);
    }
    else
    {
	size = nexus_sizeof_int(1);
	nexus_buffer_init(&reply_buffer, size, 0);
        nexus_put_int(&reply_buffer, &job_status, 1);
    }
 
    nexus_send_rsr(&reply_buffer,
                   &reply_sp,
                   GRAMI_CLIENT_REPLY_HANDLER_ID,
                   NEXUS_TRUE,
                   NEXUS_FALSE);
/*
    nexus_mutex_lock(&job_manager_monitor.mutex);
*/
    notice2("job status = %d", job_status);

    if (job_status == 0)
    {
        while (!job_manager_monitor.done)
        {
            /*
            nexus_cond_wait(&job_manager_monitor.cond, 
                            &job_manager_monitor.mutex);
            */
	    nexus_usleep(1000000);
    	    nexus_fd_handle_events(NEXUS_FD_POLL_NONBLOCKING_ALL, 
                                   &message_handled);
	    grami_jm_poll(); 
        } /* endwhile */
/*
        nexus_mutex_unlock(&job_manager_monitor.mutex);
*/
    }

    tree_free(description_tree);

    nexus_disallow_attach(my_port);

    nexus_mutex_destroy(&job_manager_monitor.mutex);
    nexus_cond_destroy(&job_manager_monitor.cond);

    notice("exiting gram_job_request");

    nexus_shutdown();

    return(0);

} /* main() */

/******************************************************************************
Function:       attach_requested()
Description:
Parameters:
Returns:
******************************************************************************/
static int 
attach_requested(void * arg,
                 char * url,
                 nexus_startpoint_t * sp)
{
    notice("in attach_requested callback");

    nexus_startpoint_bind(sp, &GlobalEndpoint);

    return(0);
} /* attach_requested() */


/******************************************************************************
Function:       grami_jm_callback()
Description:
Parameters:
Returns:
******************************************************************************/
void 
grami_jm_callback(int state, int errorcode)
{
    int                size;
    int                count;
    int                rc;
    nexus_startpoint_t sp;
    nexus_buffer_t     reply_buffer;
    
    notice("in grami_jm_callback");

    if (grami_jm_home_dir)
    {
        if (state == GRAM_JOB_STATE_ACTIVE)
        {
           status_file_gen("INIT");
           status_file_gen("ACTIVE");
        }
        else if (state == GRAM_JOB_STATE_PENDING)
        {
           status_file_gen("PENDING");
        }
        else if (state == GRAM_JOB_STATE_DONE)
        {
           status_file_gen("DONE");
        }
        else if (state == GRAM_JOB_STATE_FAILED)
        {
           status_file_gen("FAILED");
        }
        else
        {
           status_file_gen("UNKNOWN");
        }
    }
 
    rc = nexus_attach(callback_contact, &sp);
    
    if (rc == 0)
    {
        size  = nexus_sizeof_int(1);<commit_msg>removed status_file_gen("INIT");<commit_after>/******************************************************************************
gram_job_manager.c 

Description:
    Resource Allocation Job Manager

CVS Information:

    $Source$
    $Date$
    $Revision$
    $Author$
******************************************************************************/

/******************************************************************************
                             Include header files
******************************************************************************/
#include <stdio.h>
#include <malloc.h>
#include <sys/param.h>
#include <sys/time.h>
#include <string.h> /* for strdup() */
#include <memory.h>
#include <nexus.h>
#include "gram_client.h"
#include "grami_client.h"
#include "gram_rsl.h"
#include "grami_jm.h"

/******************************************************************************
                               Type definitions
******************************************************************************/
typedef struct _gram_job_manager_monitor_t
{
    nexus_mutex_t          mutex;
    nexus_cond_t           cond;
    volatile nexus_bool_t  done;
} gram_job_manager_monitor_t;

/******************************************************************************
                          Module specific prototypes
******************************************************************************/
static void 
graml_cancel_handler(nexus_endpoint_t * endpoint,
                     nexus_buffer_t * buffer,
                     nexus_bool_t is_non_threaded_handler);

static void 
graml_start_time_handler(nexus_endpoint_t * endpoint,
                         nexus_buffer_t * buffer,
                         nexus_bool_t is_non_threaded_handler);

static int 
attach_requested(void * arg,
                 char * url,
                 nexus_startpoint_t * sp);
static int 
status_file_gen(char * job_status);

static void 
tree_free(gram_specification_t * sp);

static char *
genfilename(char * prefix, char * path, char * sufix);

static void
notice(char * s);

/******************************************************************************
                       Define variables for external use
******************************************************************************/
char * grami_jm_libexecdir = GLOBUS_LIBEXECDIR;
int  gram_print_debug = 0;

/******************************************************************************
                       Define module specific variables
******************************************************************************/
static nexus_handler_t handlers[] =
{ 
    {NEXUS_HANDLER_TYPE_NON_THREADED, graml_cancel_handler},
    {NEXUS_HANDLER_TYPE_NON_THREADED, graml_start_time_handler},
};

static char     tmpbuf[1024];
#define notice2(a,b) {sprintf(tmpbuf, a,b); notice(tmpbuf);}
#define notice3(a,b,c) {sprintf(tmpbuf, a,b,c); notice(tmpbuf);}
#define notice4(a,b,c,d) {sprintf(tmpbuf, a,b,c,d); notice(tmpbuf);}

static char * grami_jm_home_dir = NULL;
static char * grami_jm_arg_libexecdir = NULL;
static char * grami_jm_status_dir = NULL;
static gram_job_manager_monitor_t  job_manager_monitor;
static nexus_endpointattr_t        EpAttr;
static nexus_endpoint_t            GlobalEndpoint;
static char                        callback_contact[GRAM_MAX_MSG_SIZE];
static char                        job_contact[GRAM_MAX_MSG_SIZE];
static char                        my_globusid[GRAM_MAX_MSG_SIZE];
static int                         my_count;

static FILE *                      gram_log_fp;

/*
 * NexusExit() (required of all Nexus programs)
 */
void NexusExit(void)
{
} /* end NexusExit() */

/*
 * NexusAcquiredAsNode() (required of all Nexus programs)
 */
int NexusAcquiredAsNode(nexus_startpoint_t *startpoint)
{
    nexus_startpoint_bind(startpoint, &GlobalEndpoint);
    return 0;
} /* end NexusAcquiredAsNode() */

/*
 * NexusBoot() (required of all Nexus programs)
 */
int NexusBoot(nexus_startpoint_t * startpoint)
{
    nexus_endpointattr_init(&EpAttr);
    nexus_endpointattr_set_handler_table(&EpAttr,
                                    handlers,
                                    sizeof(handlers)/sizeof(nexus_handler_t));

    nexus_endpoint_init(&GlobalEndpoint, &EpAttr);
    nexus_startpoint_bind(startpoint, &GlobalEndpoint);

    return NEXUS_SUCCESS;

} /* end NexusBoot() */

/******************************************************************************
Function:       main()
Description:
Parameters:
Returns:
******************************************************************************/
int 
main(int argc,
     char **argv)
{
    int                    i;
    int                    size;
    int                    rc;
    int                    count;
    int                    n_nodes;
    int                    format;
    int                    job_status;
    int                    job_state_mask;
    int                    message_handled;
    char                   description[GRAM_MAX_MSG_SIZE];
    char                   test_dat_file[GRAM_MAX_MSG_SIZE];
    char                   gram_logfile[GRAM_MAX_MSG_SIZE];
    char *                 tmp_ptr;
    char *                 my_host;
    char *                 globusid_ptr;
    unsigned short         my_port;
    FILE *                 args_fp;
    nexus_byte_t           type;
    nexus_byte_t *         ptr;
    nexus_byte_t           bformat;
    nexus_byte_t           buffer[GRAM_MAX_MSG_SIZE];
    nexus_buffer_t         reply_buffer;
    nexus_startpoint_t     reply_sp;
    gram_specification_t * description_tree;


    rc = nexus_init(&argc,
		    &argv,
		    "NEXUS_ARGS", /* conf info env variable          */
		    "nx",         /* package designator              */
		    NULL);        /* additional modules              */

    if (rc != NEXUS_SUCCESS)
    {
	fprintf(stderr, "nexus_init() failed with rc=%d\n", rc);
	exit(1);
    }

    *test_dat_file = '\0';

    /*
     * Parse the command line arguments
     */
    for (i = 1; i < argc; i++)
    {
        if ((strcmp(argv[i], "-t") == 0)
                 && (i + 1 < argc))
        {
            strcpy(test_dat_file, argv[i+1]);
            i++;
        }
        else if (strcmp(argv[i], "-d") == 0)
        {
            gram_print_debug = 1;
        }
        else if ((strcmp(argv[i], "-home") == 0)
                 && (i + 1 < argc))
        {
            grami_jm_home_dir = argv[i+1];
            i++;
        }
        else if ((strcmp(argv[i], "-e") == 0)
                 && (i + 1 < argc))
        {
            grami_jm_arg_libexecdir = argv[i+1];
            i++;
        }
        else
        {
            fprintf(stderr, "Usage: %s [-h deploy home dir ] [-e lib exe dir] [-d debug print] [-t test dat file]\n", argv[0]);
            exit(1);
        }
    }

    if (gram_print_debug)
    {
        /*
         * Open the gram logfile just for testing!
         */
        sprintf(gram_logfile, "gram_job_mgr_%lu.log",
                (unsigned long) getpid());

        if ((gram_log_fp = fopen(gram_logfile, "a")) == NULL)
        {
            sprintf(gram_logfile, "/tmp/gram_job_mgr_%lu.log",
                   (unsigned long) getpid());

            if ((gram_log_fp = fopen(gram_logfile, "a")) == NULL)
            {
                fprintf(stderr, "Cannot open gram logfile.\n");
                exit(1);
            }
        }

        setbuf(gram_log_fp, NULL);
    }

    notice("-------------------------------------------------");
    notice("entering gram_job_manager");

    if (grami_jm_home_dir)
    {
        grami_jm_status_dir = genfilename(grami_jm_home_dir, "tmp", NULL);

        if (grami_jm_arg_libexecdir)
        {
            grami_jm_libexecdir = genfilename(grami_jm_home_dir, grami_jm_arg_libexecdir, NULL);
        }
        else
        {
            grami_jm_libexecdir = genfilename(grami_jm_home_dir, "libexec", NULL);
        }
    }

    notice2("grami_jm_libexecdir = %s", grami_jm_libexecdir);

    if ((globusid_ptr = (char*) getenv("GLOBUSID")) == NULL)
    {
        strcpy(my_globusid, "unknown_globusid");
    }
    else
    {
        strcpy(my_globusid, globusid_ptr);
    }
 
    /*
     *  if a test_dat_file has been defined, read data from the file 
     *  instead of from stdin.
     */
    if (strlen(test_dat_file) > 0)
    {
        if ((args_fp = fopen(test_dat_file, "r")) == NULL)
        {
            notice2("Cannot open test file %s.", test_dat_file);
            exit(1);
        }
    }
    else
    {
         args_fp = stdin;
    }

    /*
     * Read the format incomming message.
     */
    if (fread(buffer, 1, 1, args_fp) <= 0)
    {
        notice("fread() failed.");
    }
    format = (int)buffer[0];

    /*
     * Read the size incomming message.
    if (fread(buffer, 1, 4, args_fp) <= 0)
     */
    if (fread(buffer, 1, nexus_dc_sizeof_remote_int(1, format), args_fp) <= 0)
    {
        notice("fread() failed.");
    }
    ptr = buffer;
    nexus_dc_get_int(&ptr, &count, 1, format);

    /*
     * Read the remainder of the incomming message.
     */
    if (fread(buffer, 1, count - nexus_dc_sizeof_remote_int(1, format) + 1,
        args_fp) <= 0)
    {
        notice("fread() failed.");
    }

    ptr = buffer;
    nexus_user_get_int(&ptr, &count, 1, format);
    nexus_user_get_char(&ptr, description, count, format);
    *(description+count)= '\0';
    nexus_user_get_int(&ptr, &job_state_mask, 1, format);
    nexus_user_get_int(&ptr, &count, 1, format);
    nexus_user_get_char(&ptr, callback_contact, count, format);
    *(callback_contact+count)= '\0';
    nexus_user_get_startpoint(&ptr, &reply_sp, 1, format);

    notice2("description = %s", description);
    notice2("job state mask = %i",job_state_mask);
    notice2("callback contact = %s", callback_contact);

    /* Initialize termination monitor */
    nexus_mutex_init(&job_manager_monitor.mutex, (nexus_mutexattr_t *) NULL);
    nexus_cond_init(&job_manager_monitor.cond, (nexus_condattr_t *) NULL);
    job_manager_monitor.done = NEXUS_FALSE;

    /* allow other Nexus programs to attach to us */
    my_port = 0;
    rc = nexus_allow_attach(&my_port,      /* port            */
                            &my_host,     /* host            */
                            attach_requested, /* approval_func() */
                            NULL);
    if (rc != 0)
    {
       return(GRAM_ERROR_INVALID_REQUEST);
    } 
    else
    {
        sprintf(job_contact, "x-nexus://%s:%hu/%lu/%lu/", 
                              my_host,
                              my_port,
                              (unsigned long) getpid(),
                              (unsigned long) time(0));
    }
    description_tree = gram_specification_parse(description);

    /*
     * Start the job.  If successful reply with job_contact else
     * send error status.
     */
    job_status = grami_jm_job_request(job_contact, description_tree);

    if (job_status == 0)
    {
        count= strlen(job_contact);
	size = nexus_sizeof_int(1);
	size += nexus_sizeof_int(1);
	size += nexus_sizeof_char(count);
	nexus_buffer_init(&reply_buffer, size, 0);
        nexus_put_int(&reply_buffer, &job_status, 1);
        nexus_put_int(&reply_buffer, &count, 1);
	nexus_put_char(&reply_buffer, job_contact, count);
    }
    else
    {
	size = nexus_sizeof_int(1);
	nexus_buffer_init(&reply_buffer, size, 0);
        nexus_put_int(&reply_buffer, &job_status, 1);
    }
 
    nexus_send_rsr(&reply_buffer,
                   &reply_sp,
                   GRAMI_CLIENT_REPLY_HANDLER_ID,
                   NEXUS_TRUE,
                   NEXUS_FALSE);
/*
    nexus_mutex_lock(&job_manager_monitor.mutex);
*/
    notice2("job status = %d", job_status);

    if (job_status == 0)
    {
        while (!job_manager_monitor.done)
        {
            /*
            nexus_cond_wait(&job_manager_monitor.cond, 
                            &job_manager_monitor.mutex);
            */
	    nexus_usleep(1000000);
    	    nexus_fd_handle_events(NEXUS_FD_POLL_NONBLOCKING_ALL, 
                                   &message_handled);
	    grami_jm_poll(); 
        } /* endwhile */
/*
        nexus_mutex_unlock(&job_manager_monitor.mutex);
*/
    }

    tree_free(description_tree);

    nexus_disallow_attach(my_port);

    nexus_mutex_destroy(&job_manager_monitor.mutex);
    nexus_cond_destroy(&job_manager_monitor.cond);

    notice("exiting gram_job_request");

    nexus_shutdown();

    return(0);

} /* main() */

/******************************************************************************
Function:       attach_requested()
Description:
Parameters:
Returns:
******************************************************************************/
static int 
attach_requested(void * arg,
                 char * url,
                 nexus_startpoint_t * sp)
{
    notice("in attach_requested callback");

    nexus_startpoint_bind(sp, &GlobalEndpoint);

    return(0);
} /* attach_requested() */


/******************************************************************************
Function:       grami_jm_callback()
Description:
Parameters:
Returns:
******************************************************************************/
void 
grami_jm_callback(int state, int errorcode)
{
    int                size;
    int                count;
    int                rc;
    nexus_startpoint_t sp;
    nexus_buffer_t     reply_buffer;
    
    notice("in grami_jm_callback");

    if (grami_jm_home_dir)
    {
        if (state == GRAM_JOB_STATE_ACTIVE)
        {
           status_file_gen("ACTIVE");
        }
        else if (state == GRAM_JOB_STATE_PENDING)
        {
           status_file_gen("PENDING");
        }
        else if (state == GRAM_JOB_STATE_DONE)
        {
           status_file_gen("DONE");
        }
        else if (state == GRAM_JOB_STATE_FAILED)
        {
           status_file_gen("FAILED");
        }
        else
        {
           status_file_gen("UNKNOWN");
        }
    }
 
    rc = nexus_attach(callback_contact, &sp);
    
    if (rc == 0)
    {
        size  = nexus_sizeof_int(1);
{'fraction_non_alphanumeric': 0.20162233006269703, 'fraction_numerical': 0.005596684495767065, 'mean_word_length': 2.227989938257489, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4615991', 'n_tokens_mistral': 8885, 'n_tokens_neox': 8271, 'n_words': 1561}