source: proiecte/PDAD/trunk/nodeslocation/mpi/nodesLocation.c @ 154

// Frequent (>1000) failure cause for different categories of events grouped by duration (short, medium, long)

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include "mpi.h"
#include "hashmap_public.h"

#define LINEMAX 512 
#define NAMEMAX 32 
#define CHUNK 20480
#define ENCODED_NO 33
#define MAX_UNITS 10

#define JOB_TAG 0 
#define RESPONSE_TAG 1
#define QUERY_TAG 2

typedef struct location_unit {
    char location[NAMEMAX]; 
    unsigned long failures; 
} location_unit; 

typedef struct resp_payload {
    int unsigned count; 
    location_unit units[MAX_UNITS]; 
} resp_payload; 

typedef struct node_key {
    int unsigned node_id; 
    int unsigned platform_id; 
} node_key; 

typedef struct query_payload {
    node_key key; 
} query_payload; 

typedef struct node_value {
    char location[0]; 
} node_value; 

uint32_t node_hash(key generic_key) {
    node_key *k = (node_key *) generic_key; 
    return (k->node_id * k->platform_id); 
}

bool node_eq(key a, key b) {
    node_key *na = (node_key *)a; 
    node_key *nb = (node_key *)b; 
    return (na->node_id == nb->node_id) && (na->platform_id == nb->platform_id); 
}

typedef struct loc_key {
    char *name; 
} loc_key; 

typedef struct loc_value {
    int unsigned failures; 
} loc_value; 

uint32_t loc_hash(key generic_key) {
    loc_key *k = (loc_key *) generic_key; 
    return (uint32_t)k->name[0]; 
}

bool loc_eq(key a, key b) {
    loc_key *la = (loc_key *) a; 
    loc_key *lb = (loc_key *) b; 

    return strcmp(la->name, lb->name) == 0; 
}


/* for master */
hashmap *nodes_map;

/* for slaves: */
hashmap *locations; 
hashmap *nodes_cache; 
 
unsigned long* poz;
unsigned long recv[2];
unsigned long poz_size = 0;
unsigned long recv_m[ENCODED_NO];
unsigned long send_m[ENCODED_NO];

int failure_key[] = {-1, 0, 999, 1999, 2999, 3999, 4999, 5999, 6999, 7000, 7001};
char* failure_value[] = {"not reported", "reported as undetermined", "infrastructure", "hardware", "IO", "network", "software", "human error",
                         "user", "end of measurement", "TYPING"};
long duration_key[] = {1000, 100000, 2147483647};
char* duration_value[] = {"short", "medium", "long"};
char* encoded_values[] = {"short-not reported", "medium-not reported", "long-not reported", "short-reported as undetermined",
                          "medium-reported as undetermined", "long-reported as undetermined", "short-infrastructure", "medium-infrastructure",
                          "long-infrastructure", "short-hardware", "medium-hardware", "long-hardware", "short-IO", "medium-IO", "long-IO", "short-network",
                          "medium-network", "long-network", "short-software", "medium-software", "long-software", "short-human error",
                          "medium-human error", "long-human error", "short-user", "medium-user", "long-user", "short-end of measurement",
                          "medium-end of measurement", "long-end of measurement", "short-TYPING", "medium-TYPING", "long-TYPING"};

unsigned long file_length(FILE *f) {
    long pos;
    long end;

    pos = ftell (f);
    fseek (f, 0, SEEK_END);
    end = ftell (f);
    fseek (f, pos, SEEK_SET);

    return end;
}


void write_result (char *filename) {
    FILE *f;
    int i;

    f = fopen(filename, "wt");
    if (f == NULL) {
        fprintf(stderr, "Could not open file %s for write\n", filename);
        MPI_Abort(MPI_COMM_WORLD, 7777);
    }

    for(i = 0 ; i < ENCODED_NO ; i++) {
        fprintf(f, "%-40s", encoded_values[i]);
        fprintf(f, "%lu\n", recv_m[i]);
    }
}

void load_nodes_info(char *nodes_filename) {
    FILE *nodes_file;
    char buffer[LINEMAX]; 
    char node_name[NAMEMAX], node_ip[NAMEMAX], node_location[NAMEMAX]; 
    node_key *node; 
    node_value *value; 

    /* initialize hashmap */
    nodes_map = mk_hmap(node_hash, node_eq); 

    nodes_file = fopen(nodes_filename, "rt"); 
    if (nodes_file == NULL) { 
        fprintf(stderr, "Could not open nodes file %s \n", nodes_filename); 
        MPI_Abort(MPI_COMM_WORLD, 7777);
    }

    while (fgets(buffer, LINEMAX, nodes_file) != NULL) { 
        /* skip comments */
        if (buffer[0] == '#') 
            continue; 
        
        node = malloc(sizeof(node_key)); 
        if (node == NULL) {
            fprintf(stderr, "Not enough memory \n"); 
            MPI_Abort(MPI_COMM_WORLD, 7777); 
        }
        
        sscanf(buffer, "%u %u %s %s %s", &node->node_id, &node->platform_id, node_name, node_ip, node_location);
        value = malloc(sizeof(node_value) + strlen(node_location) + 1); 
        if (value == NULL) { 
            fprintf(stderr, "Not enough memory \n"); 
            MPI_Abort(MPI_COMM_WORLD, 7777); 
        }

        strcpy(value->location, node_location); 
        if (hmap_add(nodes_map, node, value) == false) {
            fprintf(stderr, "Hashmap out of memory \n"); 
            MPI_Abort(MPI_COMM_WORLD, 7777); 
        } 
    }

    fclose(nodes_file); 

    /* 
    node_key k;
    k.node_id = 160; 
    k.platform_id = 3; 
    value = hmap_get(nodes_map, &k); 
    if (value == NULL) {
        printf("Error \n"); 
    } else {
        printf("Search result: %s \n", value->location); 
    }
    */
}

int init(char *filename) {
    unsigned long file_len = 0;
    unsigned long i, j;
    FILE *f;

    f = fopen(filename, "rt");
    if (f == NULL) {
        fprintf(stderr, "Could not open file init.\n");
        MPI_Abort(MPI_COMM_WORLD, 7777);
    }
    file_len = file_length(f);
    poz_size = file_len / CHUNK + 1;

    poz = (unsigned long*) malloc(poz_size * sizeof(unsigned long));
    poz[0] = 0;
    for (i = CHUNK, j = 1; j < poz_size - 1; i += CHUNK, j++) {
        poz[j] = i;
    }
    poz[j] = file_len;

    return 1;
}

int main(int argc, char** argv) {

    int rank, size;
    MPI_Status status;
    MPI_Request req;
    int i, flag;
    void *buffer = malloc(sizeof(resp_payload));
    resp_payload *resp; 
    query_payload *query; 

    MPI_Init(&argc, &argv);
    MPI_Comm_rank(MPI_COMM_WORLD, &rank);
    MPI_Comm_size(MPI_COMM_WORLD, &size);

    if (rank == 0) {
        // master inits
        init(argv[1]);
        load_nodes_info(argv[2]); 
    }

    MPI_Barrier(MPI_COMM_WORLD);

    // master sends indexes, receives results
    if(rank == 0) {

        char null_location = 0;
        for(i = 0 ; i < ENCODED_NO ; i++) {
            recv_m[i] = 0;
            send_m[i] = 0;
        }

        // sends indexes
        for (i = 0; i < poz_size - 1; i++) {
            MPI_Isend(poz+i, 2, MPI_UNSIGNED_LONG, i%(size-1)+1, JOB_TAG, MPI_COMM_WORLD, &req);
        }

        // receive results
        unsigned long recv_temp[ENCODED_NO];
        while(1) {
            
            for(i = 0 ; i < ENCODED_NO ; i++)
                recv_temp[i] = 0;
            
            MPI_Irecv(buffer, ENCODED_NO, MPI_UNSIGNED_LONG, MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &req);
            
            MPI_Test(&req, &flag, &status);
            int times = 0;
            while(!flag && times < 10) {
                times++;
                sleep(1);
                MPI_Test(&req, &flag, &status);
            }
            if(!flag)
                break;
            
            if (status.MPI_TAG == RESPONSE_TAG) { 
                for(i = 0 ; i < ENCODED_NO ; i++) {
                    recv_m[i] += recv_temp[i];
                }
                printf("Response received \n"); 

            } else if (status.MPI_TAG == QUERY_TAG) { 
                query = (query_payload *)buffer;
                node_value *value = hmap_get(nodes_map, query); 
                if (value != NULL) { 
                    MPI_Send(value->location, NAMEMAX, MPI_CHAR, status.MPI_SOURCE, QUERY_TAG, MPI_COMM_WORLD); 
                } else {
                    MPI_Send(&null_location, 1, MPI_CHAR, status.MPI_SOURCE, QUERY_TAG, MPI_COMM_WORLD); 
                }
            }
        }

        // write results
        write_result(argv[3]);
        free(poz);

        free_hmap(nodes_map); 
    }

    // slaves receive indexes, do computation
    else {

        locations = mk_hmap(loc_hash, loc_eq); 
        nodes_cache = mk_hmap(node_hash, node_eq); 
        while(1) {
            
            // receive until there's nothing left
            MPI_Irecv(recv, 2, MPI_UNSIGNED_LONG, 0, JOB_TAG, MPI_COMM_WORLD, &req);
            
            MPI_Test(&req, &flag, &status);
            int times = 0;
            while(!flag && times < 10) {
                times++;
                sleep(1);
                MPI_Test(&req, &flag, &status);
            }
            if(!flag)
                break;

            // read file and do computation
            FILE *f;

            f = fopen(argv[1], "rt");
            if (f == NULL) {
                fprintf(stderr, "Could not open file.\n");
                MPI_Abort(MPI_COMM_WORLD, 7777);
            }
            
            fseek (f, recv[0], SEEK_SET);

            // go to the end of the line
            char * line = NULL;
            size_t len = 0;
            ssize_t read;
            unsigned long count = 0;

            for(i = 0 ; i < ENCODED_NO ; i++) {
                send_m[i] = 0;
                recv_m[i] = 0;
            }

            while ((read = getline(&line, &len, f)) != -1) {

                // skip the comment line and the half lines
                count++;
                if(count == 1)
                    continue;

                // compute
                //char response;

                node_key node; 
                char delims[] = " ";
                char *result = NULL;
                result = strtok( line, delims );
                int token_no = 0;
                while( result != NULL ) {
                    token_no++;
                    if(token_no == 2) {
                        node.node_id = atoi(result);
                    }
                    if(token_no == 3) {
                        node.platform_id = atof(result);
                    }
                    result = strtok( NULL, delims );
                }
                
                node_value *value = hmap_get(nodes_cache, &node); 
                
                if (value == NULL) { 
                    char location[NAMEMAX]; 
                    MPI_Send(&node, sizeof(node_key), MPI_CHAR, 0, QUERY_TAG, MPI_COMM_WORLD); 
                    MPI_Recv(location, NAMEMAX, MPI_CHAR, 0, QUERY_TAG, MPI_COMM_WORLD, &status); 
                    value = malloc(sizeof(node_value) * strlen(location) + 1); 
                    node_key *new_node = malloc(sizeof(node_key)); 
                    new_node->node_id = node.node_id; 
                    new_node->platform_id = node.platform_id; 
                    strcpy(value->location, location); 
                    hmap_add(nodes_cache, new_node, value); 
                    if (strlen(location) > 0) { 
                        /* add to the locations set */
                        loc_key *loc = malloc(sizeof(loc_key)); 
                        loc->name = value->location; 
                        loc_value *locval = malloc(sizeof(loc_value)); 
                        locval->failures = 1; 
                        hmap_add(locations, loc, locval); 
                        printf("Location is %s \n", location); 
                    }
                } else {
                    printf("Found in local cache %s \n", value->location); 
                    if (strlen(value->location) > 0) { 
                        loc_key *loc = malloc(sizeof(loc_key)); 
                        loc->name = value->location; 
                        loc_value *loc_data = hmap_get(locations, loc); 
                        if (loc_data != NULL) {
                            loc_data->failures++; 
                        }
                    }
                }
                // finished reading the chunk?
                if(ftell(f) > recv[1])
                    break;
                
            }

            // send result to master
            MPI_Isend(send_m, ENCODED_NO, MPI_UNSIGNED_LONG, 0, RESPONSE_TAG, MPI_COMM_WORLD, &req);
        }
    }

    // block and finish
    MPI_Barrier(MPI_COMM_WORLD);

    
    MPI_Finalize();
    return 0; 
    }