dfs / src / meta-data.c

#include <stdio.h>
#include <stdlib.h>
#include "lib/enchufe.h"
#include "lib/inode.h"
#include "lib/dnode.h"
#include "lib/block.h"
#include "lib/log.h"
#define BUF_LEN 0x1000

void add(DNode node) {
	insert_into_dnode(node.address, node.port);
}

void list(Enchufe cliente) {
	ListINode inodes = INode_get_db();

	Buffer inside_buf = ListINode_serialize(inodes);
	zumba(cliente, inside_buf);
	free(inside_buf.buf);

	for (ListINode iptr = inodes; iptr != NULL;) {
		free((void*)iptr->head.fname.str);
		ListINode kod = iptr;
		iptr = iptr->rest;
		free(kod);
	}
}

/// Registers a data node
void reg(IPv4 address, Port port) {
	// Create data node tuple
	DNode dnode = {
		.nid = new_uuid(),
		.address = address,
		.port = port,
	};

	// Insert it in the data base
	insert_into_dnode(dnode.address, dnode.port);

	log(INFO, "Finished registering data node on: %d.%d.%d.%d:%d\n", dnode.address.bytes[0], dnode.address.bytes[1], dnode.address.bytes[2], dnode.address.bytes[3], dnode.port);
}

/// Puts file data on the server
void put(Enchufe cliente, size_t fsize, SafeStr fname) {
	// Send dnodes :: ListDNode to the client
	log(INFO, "Sending the following data nodes:\n");
	ListDNode dnodes = select_from_dnode(NULL, NULL, NULL);
	for (ListDNode dptr = dnodes; dptr != NULL; dptr = dptr->rest) {
		DNode d = dptr->head;
		log(INFO, "\tdnode on %d.%d.%d.%d:%d\n", d.address.bytes[0], d.address.bytes[1], d.address.bytes[2], d.address.bytes[3], d.port);
	}
	Buffer inside_buf = ListDNode_serialize(dnodes);
	zumba(cliente, inside_buf);
	free(inside_buf.buf);

	// Create an inode for the new file
	insert_into_inode(fname, fsize);
	// Get new file UUID
	ListINode inodes = select_from_inode(NULL, &fname, &fsize);
	exists(inodes);
	UUID fid = inodes->head.fid;
	// Free the memory
	ListINode_deinit(inodes);
	free((void*)fname.str);


	// Receive dinfos :: ListDNodeInfo from client
	Byte buf[BUF_LEN] = {0};
	Buffer outside_buf = { .buf = buf, .len = BUF_LEN };
	size_t len = recibe(cliente, outside_buf);
	outside_buf.len = len;

	ListDNodeInfo dinfo = ListDNodeInfo_deserialize(outside_buf);
	exists(dinfo);

	// Create tuples in Block table
	for (ListDNodeInfo lptr = dinfo; lptr != NULL; lptr = lptr->rest) {
		DNodeInfo d = lptr->head;

		insert_into_block(&fid, &d.dnode.nid, &d.cid, &d.seq);

		log(INFO, "Received the following Block: (cid: %s, seq: %zu, dnode: %d.%d.%d.%d:%d)\n", utoa(d.cid).uuid, d.seq, d.dnode.address.bytes[0], d.dnode.address.bytes[1], d.dnode.address.bytes[2], d.dnode.address.bytes[3], d.dnode.port);
	}
}

/// Get file data from server
void get(Enchufe cliente, SafeStr fname) {
	// Get the inode associated to that file
	ListINode inodes = select_from_inode(NULL, &fname, NULL);
	exists(inodes);
	INode inode = inodes->head;
	//
	// Get the blocks associated to that inode
	ListBlock blocks = select_from_block(NULL, &inode.fid, NULL, NULL, NULL);
	exists(blocks);

	// Construct list of dnode info for client
	ListDNodeInfo dinfos = NULL;
	for (ListBlock bptr = blocks; bptr != NULL; bptr = bptr->rest) {
		// Get datanode associated to th node UUID of the block
		ListDNode dnodes = select_from_dnode(&bptr->head.nid, NULL, NULL);
		exists(dnodes);

		// construct the data node info
		for (ListDNode dptr = dnodes; dptr != NULL; dptr = dptr->rest) {
			log(INFO, "Sending DNodeInfo(dnode: ");
			print_dnode(dptr->head);
			printf(", cid: %s, seq: %zu)\n", utoa(bptr->head.cid).uuid, bptr->head.seq);
			dinfos = ListDNodeInfo_cons((DNodeInfo){
					.dnode = dptr->head,
					.cid = bptr->head.cid,
					.seq = bptr->head.seq,
					}, dinfos);
		}
	}

	// Send dinfos :: ListDNodeInfo to client
	Buffer dinfo_buf = ListDNodeInfo_serialize(dinfos);
	zumba(cliente, dinfo_buf);
	free(dinfo_buf.buf);
}

/// Handles the connections when they come in
bool handler(Enchufe enchufe) {
	log(INFO, "Waiting for a connection\n");
	Enchufe cliente = acepta(enchufe);
	log(INFO, "Accepted a connection\n");

	// Receive Command from client
	Byte buf[BUF_LEN] = {0};
	Buffer outside_buf = { .buf = buf, .len = BUF_LEN };
	size_t len = recibe(cliente, outside_buf);
	outside_buf.len = len;

	// Call the correct sub-handler depending on the command
	Command cmd = Command_deserialize(outside_buf);
	switch (cmd.tag) {
		case REG: {
			log(INFO, "Received REG\n");
			reg(cmd.as.reg.address, cmd.as.reg.port);
			log(INFO, "Finished REG\n");
		} break;
		case PUT: {
			log(INFO, "Received PUT(fsize: %zu, fname: \"%s\")\n", cmd.as.put.fsize, cmd.as.put.fname.str);
			put(cliente, cmd.as.put.fsize, cmd.as.put.fname);
			// Free memory allocated for file name
			log(INFO, "Finished PUT\n");
		} break;
		case LIST: {
			log(INFO, "Received LIST\n");
			list(cliente);
			log(INFO, "Finished LIST\n");
		} break;
		case GET: {
			log(INFO, "Received GET(fname: \"%s\")\n", cmd.as.get.fname.str);
			get(cliente, cmd.as.get.fname);
			// Free memory allocated for file name
			log(INFO, "Finished LIST\n");
		} break;
		default: {
			log(ERROR, "That command was incorrect, proceeding to ignore\n");
		} break;
	}
	desenchufa(cliente);

	return true;
}

/// This is the main metadata server
int main(int argc, char** argv) {
	// Parse command line arguments
	Port port;
	switch (argc) {
		case 1: {
			port = 8000;
		} break;
		case 2: {
			Str str = argv[1];
			port = (Port)atoi(str);
		} break;
		default: {
			log(ERROR, "USAGE: %s [PORT]\n", argv[0]);
			log(ERROR, "When no port is issued, metadata server will listen on port 8000\n");
			exit(EXIT_FAILURE);
		} break;
	}

	// Wait for connections
	log(INFO, "Creating socket for metadata server on 127.0.0.1:%d\n", port);
	Enchufe enchufe = enchufa((IPv4){ .bytes = {127,0,0,1}}, port);
	amarra(enchufe);
	log(INFO, "Waiting for up to 10 connections\n");
	escucha(enchufe, 10);

	// Call the handler for ench connection
	log(INFO, "Waiting for connections\n");
	while (true) {
		if (!handler(enchufe)) break;
	}

	desenchufa(enchufe);
	return EXIT_SUCCESS;
}