Mycorrhiza/src/myco-ipc.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-ipc.h"

int myco_error(char *error_message, int message_queue_id) {
    fprintf(stderr, "ERROR: ");
    fprintf(stderr, error_message);
    fprintf(stderr, "=> %s\n", strerror(errno));
    myco_remove_message_queue(message_queue_id);
    exit(1);
}

int myco_create_private_message_queue() {
    int message_queue_id;

    message_queue_id = msgget(IPC_PRIVATE, PERM | IPC_CREAT);

    if (message_queue_id == -1) {
        fprintf(stderr, "ERROR: myco_create_private_message_queue() => %s\n", strerror(errno));
        error(1);
    }

    return message_queue_id;
}

int myco_create_global_message_queue() {
    int message_queue_id;

    message_queue_id = msgget(KEY, PERM | IPC_CREAT);
    if (message_queue_id == -1) {
        fprintf(stderr, "ERROR: myco_create_global_message_queue() => %s\n", strerror(errno));
        error(1);
    }

    return message_queue_id;
}

int myco_get_global_message_queue() {
    int message_queue_id;

    message_queue_id = msgget(KEY, 0);
    if (message_queue_id == -1) {
        fprintf(stderr, "ERROR: myco_get_global_message_queue() => %s\n", strerror(errno));
        error(1);
    }

    return message_queue_id;
}

int myco_remove_message_queue(int message_queue_id) {
    if (msgctl(message_queue_id, IPC_RMID, NULL) == -1) {
        fprintf(stderr, "ERROR: myco_remove_message_queue() %d => %s\n", message_queue_id, strerror(errno));
        exit(1);
    }
    
    return 0;
}

int myco_send(int message_queue_id, message msg) {
    if (msgsnd(message_queue_id, &msg, TOTAL_LENGTH, 0) < 0) {
        myco_error("myco_send()", message_queue_id);
    }

    return 0;
}

message myco_receive(int message_queue_id) {
    message msg;

    if (msgrcv(message_queue_id, &msg, TOTAL_LENGTH, 0, 0) == -1) {
        myco_error("myco_receive()", message_queue_id);
    }

    if (msg.message == NULL) {
        myco_error("myco_receive()", message_queue_id);
    }

    return msg;
}

message myco_send_and_receive(message msg, int send_message_queue_id, int receive_message_queue_id) {
    myco_send(send_message_queue_id, msg);

    msg = myco_receive(receive_message_queue_id);

    return msg;
}

int myco_read_transactional(pid_t source_pid, void *source_pointer, int source_length, void *target_pointer, int target_length) {
    struct iovec local[1];
    struct iovec remote[1];
    ssize_t nread;

    pid_t pid = source_pid;
    local[0].iov_base = target_pointer;
    local[0].iov_len = target_length;
    remote[0].iov_base = source_pointer;
    remote[0].iov_len = source_length;

    nread = process_vm_readv(pid, local, 1, remote, 1, 0);
    if (nread != target_length) {
        return -1;
    }

    return 0;
}