gs_seq_dsi.c

Go to the documentation of this file.

/********************************************************************/
/*                            gs_seq_dsi.h                          */
/*      the source code for GridSolve request sequencing DSI API    */
/*                       Yinan Li, 06/17/2008                       */
/********************************************************************/


#include "portability.h"
#include "gs_seq_dsi.h"


/* the default paths to put temporary data files 
   that hold the data to be transferred between
   servers; one for sender and one for receiver */
/* sender: the service that produces the data */
/* receiver: the service that consumes the data */
char *DATA_SEND_FILE_PATH;
char *DATA_RECV_FILE_PATH; 


int gs_seq_set_lfs_dsi_data_storage_path() {
    char *ds_root;
    char *dsend = "/data_storage/send/";
    char *drecv = "/data_storage/recv/";

    
    ds_root = getenv("GRIDSOLVE_LDS_ROOT");
    if (!ds_root) {
        fprintf(stderr, 
        "error getting environment variable GRIDSOLVE_LDS_ROOT\n");
        return -1;
    }
    
    DATA_SEND_FILE_PATH = (char *) malloc(sizeof(char) * 
    (strlen(ds_root) + strlen(dsend) + 1));
    DATA_RECV_FILE_PATH = (char *) malloc(sizeof(char) *
    (strlen(ds_root) + strlen(drecv) + 1));
        
    if (!DATA_SEND_FILE_PATH || !DATA_RECV_FILE_PATH) {
        perror("malloc");
        return -1;
    }

    /* combine the root directory path of GridSolve and 
       relative path for senders and receivers */
    strcpy(DATA_SEND_FILE_PATH, ds_root);
    DATA_SEND_FILE_PATH[strlen(ds_root)] = '\0';
    strcat(DATA_SEND_FILE_PATH, dsend);
    DATA_SEND_FILE_PATH[strlen(ds_root) + strlen(dsend)] = '\0';

    strcpy(DATA_RECV_FILE_PATH, ds_root);
    DATA_RECV_FILE_PATH[strlen(ds_root)] = '\0';
    strcat(DATA_RECV_FILE_PATH, drecv);
    DATA_RECV_FILE_PATH[strlen(ds_root) + strlen(drecv)] = '\0';
        
    return 0;
}


/* 
 * decode a data handle (string) into 
 * a local-files-ystem-based dsi object 
 */
LFS_DSI_OBJECT *gs_seq_decode_lfs_dsi_object(char *data_handle) {
    LFS_DSI_OBJECT *obj;
    char *tmp;
    int nchars;


    if (!data_handle) {
        fprintf(stderr, "bad data handle\n");
        return NULL;
    }

    obj = (LFS_DSI_OBJECT *) malloc(sizeof(LFS_DSI_OBJECT));
    if (!obj) {
        perror("malloc");
        return NULL;
    }

    tmp = (char *) malloc(sizeof(char) * (strlen(data_handle) + 1));
    if (!tmp) {
        perror("malloc");
        return NULL;
    }

    sscanf(data_handle, "<lfs_dsi_object>\n\t<type> %d </type>\n%n",
        &obj->data_type, &nchars);
    data_handle += nchars;

    sscanf(data_handle, "\t<major> %d </major>\n%n", &obj->major, &nchars);
    data_handle += nchars;

    sscanf(data_handle, "\t<size> %d </size>\n%n", &obj->data_size, &nchars);
    data_handle += nchars;

    sscanf(data_handle, "\t<index> %d </index>\n%n", &obj->index, &nchars);
    data_handle += nchars;
    
    sscanf(data_handle, "\t<function> %s </function>\n%n", tmp, &nchars);
    obj->func_name = strdup(tmp);
    data_handle += nchars;
    
    sscanf(data_handle, "\t<source> %s </source>\n%n", tmp, &nchars);
    obj->source = strdup(tmp);
    data_handle += nchars;

    sscanf(data_handle, "\t<path> %s </path>\n%n", tmp, &nchars);
    obj->path = strdup(tmp);
    data_handle += nchars;

    sscanf(data_handle, "\t<fname> %s </fname>\n</lfs_dsi_object>\n", tmp);
    obj->file_name = strdup(tmp);

    return obj;
}


/*
 * encode an local-filesystem-based 
 * dsi object to a data handle (string)
 */
char *gs_seq_encode_lfs_dsi_object(LFS_DSI_OBJECT *obj) {
    char *msg, *tmp;


    msg = dstring_sprintf("<lfs_dsi_object>\n");

    tmp = dstring_sprintf("\t<type> %d </type>\n", obj->data_type);
    msg = dstring_append_free(msg, tmp);

    tmp = dstring_sprintf("\t<major> %d </major>\n", obj->major);
    msg = dstring_append_free(msg, tmp);

    tmp = dstring_sprintf("\t<size> %d </size>\n", obj->data_size);
    msg = dstring_append_free(msg, tmp);

    tmp = dstring_sprintf("\t<index> %d </index>\n", obj->index);
    msg = dstring_append_free(msg, tmp);
    
    tmp = dstring_sprintf("\t<function> %s </function>\n", obj->func_name);
    msg = dstring_append_free(msg, tmp);
    
    tmp = dstring_sprintf("\t<source> %s </source>\n", obj->source);
    msg = dstring_append_free(msg, tmp);

    tmp = dstring_sprintf("\t<path> %s </path>\n", obj->path);
    msg = dstring_append_free(msg, tmp);

    tmp = dstring_sprintf("\t<fname> %s </fname>\n", obj->file_name);
    msg = dstring_append_free(msg, tmp);

    tmp = dstring_sprintf("</lfs_dsi_object>\n");
    msg = dstring_append_free(msg, tmp);

    return msg;
}


/*
 * create a data handle
 */
LFS_DSI_OBJECT *gs_seq_create_data_handle(
gs_argument_t *argptr, char *path, char *fname, int my_major) {
    LFS_DSI_OBJECT *obj;

    if (!argptr || !path || !fname) {
        fprintf(stderr, "bad input arguments\n");
        return NULL;
    }

    
    obj = (LFS_DSI_OBJECT *) malloc(sizeof(LFS_DSI_OBJECT));
    if (!obj) {
        perror("malloc");
        return NULL;
    }

    obj->data_type = argptr->datatype;
    obj->major = my_major;
    obj->data_size = argptr->rows * argptr->cols;
    obj->index = argptr->index;
    obj->func_name = argptr->prob->name;
    obj->source = argptr->hostname;
    obj->path = path;
    obj->file_name = fname;
    
    return obj;
}


char *gs_seq_create_unique_file_name(gs_argument_t *argptr) {
    char *file_name;
    char *tmp;
    struct timeval   tv;
/*  struct timezone  tz; */
    double cur_time;
      
    
    if (!argptr) {
        fprintf(stderr, "bad argument pointer\n");
        return NULL;
    }
    
    /* the file name is like the form */
    /* data_hostname_funcname_objecttype_datatype_inout_index.dat */
    file_name = dstring_sprintf("data_");
    tmp = dstring_sprintf("%s_", argptr->hostname);
    file_name = dstring_append_free(file_name, tmp);
    tmp = dstring_sprintf("%s_", argptr->prob->name);
    file_name = dstring_append_free(file_name, tmp);
    tmp = dstring_sprintf("%d_", argptr->objecttype);
    file_name = dstring_append_free(file_name, tmp);
    tmp = dstring_sprintf("%d_", argptr->datatype);
    file_name = dstring_append_free(file_name, tmp);
    tmp = dstring_sprintf("%d_", argptr->inout);
    file_name = dstring_append_free(file_name, tmp);
    tmp = dstring_sprintf("%d_", argptr->index);
    file_name = dstring_append_free(file_name, tmp);
    gettimeofday(&tv, NULL);
    cur_time = tv.tv_sec + tv.tv_usec / 1000000.0;
    tmp = dstring_sprintf("%f.dat", cur_time);
    file_name = dstring_append_free(file_name, tmp);
    return file_name;
}


int gs_seq_save_data_into_file(
char *fname, gs_argument_t *argptr, int my_major, int my_dsig) {
    int fd, status;

    fd = open(fname, O_WRONLY | O_CREAT | O_TRUNC, 0666);
    if(fd < 0) {
        fprintf(stderr, "file '%s' could not be opened\n", fname);
        return -1;
    }


    if ((status = gs_send_tag(fd, my_major)) < 0) {
        ERRPRINTF("error writing major to file\n");
        return -1;
    }

    if ((status = gs_send_int(fd, my_dsig)) < 0) {
        ERRPRINTF("error writing data signature to file\n");
        return -1;
    }
    
    if ((status = gs_send_arg(fd, argptr, my_dsig)) < 0) {
        ERRPRINTF("error writing argument to file\n");
        return -1;
    }
    
    close(fd);
    
    return 0;
}


int gs_seq_restore_data_from_file(
char *fname, gs_argument_t *argptr, int my_dsig) {
    int fd, status, sender_dsig, sender_major;
    
    fd = open(fname, O_RDONLY, 0666);
    if (fd < 0) {
        ERRPRINTF("file '%s' could not be opened\n", fname);
        return -1;
    }

    if ((status = gs_recv_tag(fd, &sender_major)) < 0) {
        ERRPRINTF("error receiving major from file\n");
        return -1;
    }

    if ((status = gs_recv_int(fd, &sender_dsig)) < 0) {
        ERRPRINTF("error receiving sender data signature from file\n");
        return -1;
    }
    
    if((status = gs_recv_arg(fd, argptr, sender_major, sender_dsig, my_dsig,
        GS_SERVER_SIDE)) < 0) {
        ERRPRINTF("error reading argument from file\n");
        return -1;
    }
    
    close(fd);

    return 0;
}