Mycorrhiza/src/myco-agent.c

/* Copyright (C) 2016 Max Riechelmann <max.riechelmann@student.kit.edu>
   (Karlsruhe Institute of Technology)
   This library is free software; you can redistribute it and/or modify it
   under the terms of the GNU Lesser General Public License as published by the
   Free Software Foundation; either version 2.1 of the License, or (at your
   option) any later version.
   This library is distributed in the hope that it will be useful, but WITHOUT
   ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
   FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
   details.
   You should have received a copy of the GNU Lesser General Public License along
   with this library; if not, write to the Free Software Foundation, Inc., 51
   Franklin St, Fifth Floor, Boston, MA 02110, USA
   */

#include "myco-agent.h"
#include "../src/myco-memory.c"

#define DEBUG 1

int myco_agent_register(const char *agent_name) {
    int agent_message_queue_id;
    int global_message_queue_id;
    message msg;
    sprintf(msg.agent_name, "%s", agent_name);

    agent_message_queue_id = myco_create_private_message_queue();

    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.message, "REGISTER AGENT\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if (DEBUG) {
        printf("%s", msg.message);
    }
    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        myco_remove_message_queue(agent_message_queue_id);
        return -1;
    }

    return agent_message_queue_id;
}

int myco_agent_unregister (const char *agent_name, int agent_message_queue_id) {
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.agent_name, "%s", agent_name);
    sprintf(msg.message, "UNREGISTER AGENT\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if (DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) == 0) {
        myco_remove_message_queue(agent_message_queue_id);
    } else {
        return -1;
    }

    return 0;
}

void *myco_agent_register_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name, int resource_transactional, pid_t pid, int size) {
    int file_descriptor = shm_open(resource_name, O_CREAT | O_TRUNC | O_RDWR, 0666);
    int r = ftruncate(file_descriptor, size);
    void *pointer = mmap(0, size, PROT_READ | PROT_WRITE, MAP_SHARED, file_descriptor, 0);

    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.agent_name, "%s", agent_name);
    sprintf(msg.resource_name, "%s", resource_name);
    msg.resource_transactional = resource_transactional;
    msg.sender_pid = pid;
    msg.resource_size = size;
    msg.resource_pointer = pointer;
    sprintf(msg.message, "REGISTER RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return NULL;
    }

    return pointer;
}

int myco_agent_unregister_resource (int agent_message_queue_id, const char *resource_name) {
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.message, "UNREGISTER RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return -1;
    }

    int r = shm_unlink(resource_name);

    return 0;
}

int myco_agent_lock_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name) {
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.agent_name, "%s", agent_name);
    sprintf(msg.message, "LOCK RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return -1;
    }

    return 0;
}

int myco_agent_release_resource (int agent_message_queue_id, const char *resource_name, void *resource_pointer, int resource_size) {
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.message, "RELEASE RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return -1;
    }

    int r = munmap(resource_pointer, resource_size);

    return 0;
}

int myco_agent_unlock_resource (int agent_message_queue_id, const char *agent_name, const char *resource_name) {
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.agent_name, "%s", agent_name);
    sprintf(msg.message, "UNLOCK RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return -1;
    }

    return 0;
}

struct resource myco_agent_request_resource(int agent_message_queue_id, char *resource_name) {
    struct resource myresource;
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.message, "REQUEST RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        myresource.size = -1;
        return myresource;
    }

    myresource.pointer = myco_malloc(msg.resource_size);
    myresource.size = msg.resource_size;

    if (myco_copy_memory(msg.sender_pid, msg.resource_pointer, msg.resource_size, myresource.pointer, myresource.size) == -1) {
        fprintf(stderr, "FATAL ERROR in myco_copy_memory() %s\n", strerror(errno));
        myresource.size = -1;
        return myresource;
    }

    sprintf(msg.message, "RESOURCE GRANTED\n");
    myco_send(msg.agent_message_queue_id, msg);

    return myresource;
}

int myco_agent_request_resource_list(int agent_message_queue_id) {
    char *resource_pointer;
    char *split;

    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.message, "REQUEST LIST\n");
    msg.resource_size = 0;

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        //printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return -1;
    }

    resource_pointer = myco_malloc(msg.resource_size);

    if (myco_copy_memory(msg.sender_pid, msg.resource_pointer, msg.resource_size, (void *)resource_pointer, msg.resource_size) == -1) {
        printf("ERROR: %s\n", strerror(errno));
        return -1;
    }

    sprintf(msg.message, "RESOURCE LIST GRANTED\n");
    myco_send(agent_message_queue_id, msg);

    printf("%s\n", resource_pointer);
    /*
    split = strtok(resource_pointer, ";");
    while (split != NULL) {
        printf("%s\n", split);
        split = strtok(NULL, ";");
    }
    */
    myco_free(resource_pointer);

    return 0;
}

int myco_agent_write_remote_resource(int agent_message_queue_id, char *resource_name) {
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.message, "WRITE REMOTE RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        return -1;
    }

    return 0;
}

struct resource myco_agent_read_remote_resource(int agent_message_queue_id, char *resource_name) {
    struct resource myresource;
    message msg;
    msg.agent_message_queue_id = agent_message_queue_id;
    sprintf(msg.resource_name, "%s", resource_name);
    sprintf(msg.message, "READ REMOTE RESOURCE\n");

    msg = myco_send_and_receive(msg, myco_get_global_message_queue(), agent_message_queue_id);

    if(DEBUG) {
        printf("%s", msg.message);
    }

    if (strncmp(msg.message, "SUCCESS:", 8) != 0) {
        myresource.size = -1;
        return myresource;
    }

    myresource.pointer = myco_malloc(msg.resource_size);
    myresource.size = msg.resource_size;

    if (myco_copy_memory(msg.sender_pid, msg.resource_pointer, msg.resource_size, myresource.pointer, myresource.size) == -1) {
        fprintf(stderr, "FATAL ERROR in myco_copy_memory() %s\n", strerror(errno));
        myresource.size = -1;
        return myresource;
    }

    sprintf(msg.message, "RESOURCE READ\n");
    myco_send(msg.agent_message_queue_id, msg);

    return myresource;
}