Mycorrhiza/src/myco-daemon.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-daemon.h"
#include "../src/myco-memory.c"
#include "../src/myco-modules.h"

#define DEBUG 0

myco_agent *first_agent = NULL;
myco_resource *first_resource = NULL;
in_addr_t myco_daemon_node_ip;

myco_agent *myco_daemon_find_agent(const char *agent_name) {
	myco_agent *current_agent;
	current_agent = first_agent;
	while (current_agent != NULL) {
		if (strcmp(current_agent->name, agent_name) == 0) {
			return current_agent;
		}
		current_agent = current_agent->next;
	}
	return NULL;
}

int myco_daemon_register_agent(message msg) {
	myco_agent *current_agent;

	if (myco_daemon_find_agent(msg.agent_name) != NULL) {
		sprintf(msg.message, "ERROR: agent %s already exists\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return -1;
	}

	if (first_agent == NULL) {
		first_agent = malloc(sizeof(myco_agent));
		first_agent->next = NULL;
		first_agent->prev = NULL;
		sprintf(first_agent->name, "%s", msg.agent_name);
		first_agent->message_queue_id = msg.agent_message_queue_id;
	} else {
		current_agent = first_agent;
		while (current_agent->next != NULL) {
			current_agent = current_agent->next;
		}
		current_agent->next = malloc(sizeof(myco_agent));
		current_agent->next->prev = current_agent;
		current_agent->next->next = NULL;
		sprintf(current_agent->next->name, "%s", msg.agent_name);
		current_agent->next->message_queue_id = msg.agent_message_queue_id;
	}

	sprintf(msg.message, "SUCCESS: agent %s registered\n", msg.agent_name);
	myco_send(msg.agent_message_queue_id, msg);

	return 0;

}

int myco_daemon_unregister_agent(message msg) {
	myco_agent *current_agent;
	myco_resource *current_resource;

	current_agent = myco_daemon_find_agent(msg.agent_name);

	if ((current_resource = myco_daemon_find_resource_by_agent(msg.agent_name)) != NULL) {
		sprintf(msg.message, "ERROR: agent %s still has resource %s\n", msg.agent_name, current_resource->name);
		myco_send(msg.agent_message_queue_id, msg);

		return -1;
	}

	if (current_agent == NULL) {
		sprintf(msg.message, "ERROR: agent %s could not be unregistered\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return -1;
	}

	// Element in the middle
	if (current_agent->next != NULL && current_agent->prev != NULL) {
		current_agent->prev->next = current_agent->next;
		current_agent->next->prev = current_agent->prev;
		free(current_agent);
		sprintf(msg.message, "SUCCESS: agent %s unregistered\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	// First element
	if (current_agent->next != NULL && current_agent->prev == NULL) {
		first_agent = current_agent->next;
		current_agent->next->prev = NULL;
		free(current_agent);
		sprintf(msg.message, "SUCCESS: agent %s unregistered\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	// Last element
	if (current_agent->next == NULL && current_agent->prev != NULL) {
		current_agent->prev->next = NULL;
		free(current_agent);
		sprintf(msg.message, "SUCCESS: agent %s unregistered\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	// Only remaining
	if (current_agent->next == NULL && current_agent->prev == NULL) {
		first_agent = NULL;
		free(current_agent);
		sprintf(msg.message, "SUCCESS: agent %s unregistered\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	sprintf(msg.message, "ERROR: agent %s could not be unregistered - data structure seems to be damaged!\n", msg.agent_name);
	myco_send(msg.agent_message_queue_id, msg);

	return -1;
}

myco_resource *myco_daemon_find_resource_by_agent(const char *agent_name) {
	myco_resource *current_resource;
	current_resource = first_resource;
	while (current_resource != NULL) {
		if (strcmp(current_resource->agent, agent_name) == 0) {
			return current_resource;
		}
		current_resource = current_resource->next;
	}
	return NULL;
}

myco_resource *myco_daemon_find_resource(const char *resource_name) {
	myco_resource *current_resource;
	current_resource = first_resource;
	while (current_resource != NULL) {
		if (strcmp(current_resource->name, resource_name) == 0) {
			return current_resource;
		}
		current_resource = current_resource->next;
	}
	return NULL;
}

int myco_daemon_register_resource(message msg) {
	myco_resource *current_resource;

	if (myco_daemon_find_agent(msg.agent_name) == NULL) {
		sprintf(msg.message, "ERROR: agent %s does not exist\n", msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);

		return -1;
	}

	if (myco_daemon_find_resource(msg.resource_name) != NULL) {
		sprintf(msg.message, "ERROR: resource %s already exists\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);

		return -1;
	}

	// Insert as first element
	if (first_resource == NULL) {
		first_resource = malloc(sizeof(myco_resource));
		first_resource->next = NULL;
		first_resource->prev = NULL;
		sprintf(first_resource->name, "%s", msg.resource_name);
		sprintf(first_resource->agent, "%s", msg.agent_name);
		first_resource->pid = msg.sender_pid;
		first_resource->pointer = msg.resource_pointer;
		first_resource->size = msg.resource_size;
		first_resource->read_locked = 0;
		first_resource->transfer_locked = 1;
		first_resource->transactional = msg.resource_transactional;
		first_resource->version = 0;
		first_resource->node_ip = myco_daemon_node_ip;
	} else {
		// Insert as last element
		current_resource = first_resource;
		while (current_resource->next != NULL) {
			current_resource = current_resource->next;
		}
		current_resource->next = malloc(sizeof(myco_resource));
		current_resource->next->prev = current_resource;
		current_resource->next->next = NULL;
		sprintf(current_resource->next->name, "%s", msg.resource_name);
		sprintf(current_resource->next->agent, "%s", msg.agent_name);
		current_resource->next->pid = msg.sender_pid;
		current_resource->next->pointer = msg.resource_pointer;
		current_resource->next->size = msg.resource_size;
		current_resource->next->transfer_locked = 1;
		current_resource->next->read_locked = 0;
		current_resource->next->transactional = msg.resource_transactional;
		current_resource->next->version = 0;
		current_resource->next->node_ip = myco_daemon_node_ip;
	}

	sprintf(msg.message, "SUCCESS: resource %s registered\n", msg.resource_name);
	myco_send(msg.agent_message_queue_id, msg);

	return 0;
}

int myco_daemon_unregister_resource(message msg) {
	myco_resource *current_resource;

	current_resource = myco_daemon_find_resource(msg.resource_name);

	if (current_resource == NULL) {
		sprintf(msg.message, "ERROR: resource %s could not be unregistered\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);

		return -1;
	}

	// Element in the middle
	if (current_resource->next != NULL && current_resource->prev != NULL) {
		current_resource->prev->next = current_resource->next;
		current_resource->next->prev = current_resource->prev;
		free(current_resource);
		sprintf(msg.message, "SUCCESS: resource %s unregistered\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	// First element
	if (current_resource->next != NULL && current_resource->prev == NULL) {
		first_resource = current_resource->next;
		current_resource->next->prev = NULL;
		free(current_resource);
		sprintf(msg.message, "SUCCESS: resource %s unregistered\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	// Last element
	if (current_resource->next == NULL && current_resource->prev != NULL) {
		current_resource->prev->next = NULL;
		free(current_resource);
		sprintf(msg.message, "SUCCESS: resource %s unregistered\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	// Only remaining
	if (current_resource->next == NULL && current_resource->prev == NULL) {
		first_resource = NULL;
		free(current_resource);
		sprintf(msg.message, "SUCCESS: resource %s unregistered\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);

		return 0;
	}

	sprintf(msg.message, "FATAL ERROR: resource %s could not be unregistered - data structure seems to be damaged!\n", msg.resource_name);
	myco_send(msg.agent_message_queue_id, msg);

	return -1;
}

int myco_daemon_request_resource(message msg) {
	myco_resource *current_resource;

  // TODO: Uncommented for performance tests...
  current_resource = myco_daemon_find_resource(msg.resource_name);
	// Check if resource exists on this node
	/*if ((current_resource = myco_daemon_find_resource(msg.resource_name)) != NULL) {
		// Check if agent that requests already owns the resource
		if (strcmp(current_resource->agent, msg.agent_name) == 0) {
			sprintf(msg.message, "ERROR: resource %s already belongs to agent %s\n", msg.resource_name, msg.agent_name);
			myco_send(msg.agent_message_queue_id, msg);
			return -1;
		}
		// Check if resource is transactional
		if (current_resource->transactional != RESOURCE_TRANSACTIONAL) {
			sprintf(msg.message, "ERROR: resource %s is not transactional\n", msg.resource_name);
			myco_send(msg.agent_message_queue_id, msg);
			return -1;
		}
*/
	
		// Check if resource resides on same node
		// printf("\n The adress is: %s\n", inet_ntoa(*(struct in_addr *)&current_resource->node_ip));
		// if (current_resource->node_ip == myco_daemon_node_ip) {
	

		// If on same node, send information
		msg.sender_pid = current_resource->pid; 
		msg.resource_size = current_resource->size;
		msg.resource_pointer = current_resource->pointer;
		sprintf(msg.message, "SUCCESS: resource found on same node, sending information\n");
		myco_send(msg.agent_message_queue_id, msg);

		msg = myco_receive(msg.agent_message_queue_id);
		if (strcmp(msg.message, "RESOURCE GRANTED") == 0) {
			// Transfer ownership of resource to agent
			sprintf(current_resource->agent, "%s", msg.agent_name);
			current_resource->pid = msg.sender_pid;
			current_resource->pointer = msg.resource_pointer;
		} else {
			fprintf(stderr, "ERROR: myco_daemon_request: %s\n", strerror(errno));
			return -1;
		}
	/*} else {
		// Handle case where resource is on another node
		// TODO: Request Resource information from indexer

		// Fetch resource
		myco_network_fetch_resource(msg.resource_pointer, msg.sender_pid, msg.resource_size);

		// if no resource registered with indexer:
		if (0 == 1) {
			sprintf(msg.message, "ERROR: resource %s does not exist\n", msg.resource_name);
			myco_send(msg.agent_message_queue_id, msg);
			return -1;
		}
	}
  */

	return 0;
}

int myco_daemon_request_list(message msg, pid_t pid) {
	myco_resource *current_resource;
	current_resource = first_resource;
	char *resource_pointer;
	char tmp[64];
	int resource_size = 0;
	if (current_resource == NULL) {
		sprintf(msg.message, "ERROR: no resources\n");
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}
	while (current_resource != NULL) {
		resource_size += strlen(current_resource->name) + strlen(current_resource->agent) + strlen("1");
		resource_size += strlen("(,,,);");
		resource_size += 22;
		current_resource = current_resource->next;
	}

	resource_pointer = malloc(resource_size);
	memset(resource_pointer, 0, resource_size);

	current_resource = first_resource;
	while (current_resource != NULL) {
		strcat(resource_pointer, current_resource->name);
		strcat(resource_pointer, "(");
		strcat(resource_pointer, current_resource->agent);
		strcat(resource_pointer, ",");
		sprintf(tmp, "%d", current_resource->size);
		strcat(resource_pointer, tmp);
		strcat(resource_pointer, ",");
		sprintf(tmp, "%p", current_resource->pointer);
		strcat(resource_pointer, tmp);
		strcat(resource_pointer, ",");
		sprintf(tmp, "%d", current_resource->read_locked);
		strcat(resource_pointer, tmp);
		strcat(resource_pointer, ",");
		sprintf(tmp, "%d", current_resource->transfer_locked);
		strcat(resource_pointer, tmp);
		strcat(resource_pointer, ",");
		if (current_resource->transactional) {
			strcat(resource_pointer, "1");
		} else {
			strcat(resource_pointer, "0");
		}
		strcat(resource_pointer, ");");
		current_resource = current_resource->next;
	}

	msg.sender_pid = pid;
	msg.resource_size = resource_size;
	msg.resource_pointer = (void *)resource_pointer;
	sprintf(msg.message, "SUCCESS: sending resource list\n");
	myco_send(msg.agent_message_queue_id, msg);

	msg = myco_receive(msg.agent_message_queue_id);
	if (strcmp(msg.message, "RESOURCE LIST GRANTED") == 0) {
		free(resource_pointer);
	} else {
		fprintf(stderr, "ERROR: resource list was not granted: %s\n", strerror(errno));
		free(resource_pointer);
		return -1;
	}

	return 0;
}

int myco_daemon_write_remote_resource(message msg, int force) {
	myco_resource *current_resource;

	// Check if resource exists
	if ((current_resource = myco_daemon_find_resource(msg.resource_name)) == NULL) {
		sprintf(msg.message, "ERROR: resource %s does not exist\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}

	if (force == 1) {
		sprintf(msg.message, "SUCCESS: forcing to overwrite resource %s, sending information\n", msg.resource_name);
		msg.resource_pointer = current_resource->pointer;
		msg.resource_size = current_resource->size;
		msg.sender_pid = current_resource->pid;
		myco_send(msg.agent_message_queue_id, msg);
		return 0;
	} else {
		if (current_resource->version > msg.version) {
			sprintf(msg.message, "ERROR: resource %s is newer than your copied version\n", msg.resource_name);
			myco_send(msg.agent_message_queue_id, msg);
			return -1;
		} else {
			sprintf(msg.message, "SUCCESS: version updated, sending information for resource %s\n", msg.resource_name);
			msg.resource_pointer = current_resource->pointer;
			msg.resource_size = current_resource->size;
			msg.sender_pid = current_resource->pid;
			myco_send(msg.agent_message_queue_id, msg);
			current_resource->version += 1;
			myco_send(msg.agent_message_queue_id, msg);
			return 0;
		}
	}
}

int myco_daemon_read_remote_resource(message msg) {
	myco_resource *current_resource;

	// Check if resource exists
	if ((current_resource = myco_daemon_find_resource(msg.resource_name)) == NULL) {
		sprintf(msg.message, "ERROR: resource %s does not exist\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}
	// Check if agent that requests already owns the resource
	if (strcmp(current_resource->agent, msg.agent_name) == 0) {
		sprintf(msg.message, "ERROR: resource %s already belongs to agent %s\n", msg.resource_name, msg.agent_name);
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}

	// TODO: Handle case where resource is on another node (another agent)

	// If on same node, send information
	msg.sender_pid = current_resource->pid; 
	msg.resource_size = current_resource->size;
	msg.resource_pointer = current_resource->pointer;
	sprintf(msg.message, "SUCCESS: resource found on same node, sending information\n");
	myco_send(msg.agent_message_queue_id, msg);

	msg = myco_receive(msg.agent_message_queue_id);
	if (strcmp(msg.message, "RESOURCE READ") == 0) {

	} else {
		fprintf(stderr, "ERROR: myco_daemon_reead_remote: %s\n", strerror(errno));
		return -1;
	}

	return 0;
}

int myco_daemon_lock_resource(message msg) {
	myco_resource *current_resource;
	current_resource = myco_daemon_find_resource(msg.resource_name);

	if (current_resource != NULL && current_resource->transfer_locked == 1 && strcmp(current_resource->agent, msg.agent_name) == 0) {
		current_resource->read_locked = 1;
		sprintf(msg.message, "SUCCESS: resource %s locked\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return 0;
	} else {
		sprintf(msg.message, "ERROR: resource %s could not be locked\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}
}

int myco_daemon_release_resource(message msg) {
	myco_resource *current_resource;
	current_resource = myco_daemon_find_resource(msg.resource_name);

	if (current_resource != NULL && strcmp(current_resource->agent, msg.agent_name) == 0) {
		current_resource->transfer_locked = 0;
		sprintf(msg.message, "SUCCESS: resource %s released\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return 0;
	} else {
		sprintf(msg.message, "ERROR: resource %s could not be released\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}

}

int myco_daemon_unlock_resource(message msg) {
	myco_resource *current_resource;
	current_resource = myco_daemon_find_resource(msg.resource_name);

	if (current_resource != NULL && current_resource->transfer_locked == 1 && strcmp(current_resource->agent, msg.agent_name) == 0) {
		current_resource->read_locked = 0;
		sprintf(msg.message, "SUCCESS: resource %s unlocked\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return 0;
	} else {
		sprintf(msg.message, "ERROR: resource %s could not be locked\n", msg.resource_name);
		myco_send(msg.agent_message_queue_id, msg);
		return -1;
	}
}

int myco_daemon_connect(const char* ip, int port) {
	struct sockaddr_in address;
	int mysocket = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK, 0);
	address.sin_family = AF_INET;
	address.sin_port = htons (port);
	inet_aton (ip, &address.sin_addr);
	connect (mysocket, (struct sockaddr *) &address, sizeof (address));

	return mysocket;
}

int myco_daemon_start(pid_t pid, const char *node_ip, const char *indexer_ip) {
	myco_daemon_node_ip = inet_network(node_ip);
	char *buffer = malloc (1024);
	int daemon_message_queue_id;
	message msg = {0};

	// Connect to indexer
	int socket = myco_daemon_connect(indexer_ip, 15000);

	// Create message queue
	daemon_message_queue_id = myco_create_global_message_queue();
	if (daemon_message_queue_id < 0) {
		fprintf(stderr, "ERROR: Message queue could not be created. %s\n", strerror(errno));
		return -1;
	}

	// Receive messages
	while (1) {
		// Receive messages from indexer
		int size = recv (socket, buffer, 1023, 0);
		if( size > 0)
			buffer[size] = '\0';
		printf ("%s\n", buffer);

		// Receive messages from agents
		msg = myco_receive(daemon_message_queue_id);
		if (DEBUG) {
			printf("%d, %s, %s, %s, %p, %d, %d, %d, %d", msg.agent_message_queue_id, msg.agent_name, msg.message, \
					msg.resource_name, msg.resource_pointer, msg.resource_size, msg.resource_transactional, msg.sender_pid, msg.version);
		}
		if (msg.message == NULL) {
			fprintf(stderr, "FATAL ERROR: No message could be received. %s\n", strerror(errno));
			return -1;
		} else {
			// Handle indexer
			char tcp_message[MESSAGE_LENGTH + RESOURCE_NAME_LENGTH + AGENT_NAME_LENGTH];
			strcpy(tcp_message, msg.message);
			strcat(tcp_message, ";");
			if (msg.agent_name[0] != '\0') {
				strcat(tcp_message, msg.agent_name);
			} else {
				strcat(tcp_message, "-");
			}
			strcat(tcp_message, ";");
			if (msg.resource_name[0] != '\0') {
				strcat(tcp_message, msg.resource_name);
			} else {
				strcat(tcp_message, "-");
			}
			send(socket, tcp_message, strlen(tcp_message), 0);
			// Handle agents
			if (strcmp(msg.message, "REGISTER AGENT") == 0) {
				myco_daemon_register_agent(msg);
			}
			if (strcmp(msg.message, "UNREGISTER AGENT") == 0) {
				myco_daemon_unregister_agent(msg);
			}
			if (strcmp(msg.message, "REGISTER RESOURCE") == 0) {
				myco_daemon_register_resource(msg);
			}
			if (strcmp(msg.message, "UNREGISTER RESOURCE") == 0) {
				myco_daemon_unregister_resource(msg);
			}
			if (strcmp(msg.message, "REQUEST RESOURCE") == 0) {
				myco_daemon_request_resource(msg);
			}
			if (strcmp(msg.message, "REQUEST LIST") == 0) {
				myco_daemon_request_list(msg, pid);
			}
			if (strcmp(msg.message, "WRITE REMOTE RESOURCE") == 0) {
				myco_daemon_write_remote_resource(msg, 0);
			}
			if (strcmp(msg.message, "FORCE WRITE REMOTE RESOURCE") == 0) {
				myco_daemon_write_remote_resource(msg, 1);
			}
			if (strcmp(msg.message, "READ REMOTE RESOURCE") == 0) {
				myco_daemon_read_remote_resource(msg);
			}
			if (strcmp(msg.message, "LOCK RESOURCE") == 0) {
				myco_daemon_lock_resource(msg);
			}
			if (strcmp(msg.message, "UNLOCK RESOURCE") == 0) {
				myco_daemon_unlock_resource(msg);
			}
			if (strcmp(msg.message, "RELEASE RESOURCE") == 0) {
				myco_daemon_release_resource(msg);
			}
		} 
	}

	close (socket);

	if (myco_remove_message_queue(daemon_message_queue_id) == -1) {
		fprintf(stderr, "FATAL ERROR: could not remove message queue %d\n", daemon_message_queue_id);
		return -1;
	}


	return EXIT_SUCCESS;
}