dfs / src / copy.c

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

/// Copies from the server to the client.
void server_to_client(Enchufe dnode, UUID cid, FILE* f) {
	conecta(dnode);

	// Send READ(cid :: UID) to dnode
	Command read = (Command){
		.tag = READ,
		.as.read.cid = cid,
	};
	Buffer read_buf = Command_serialize(read);
	zumba(dnode, read_buf);
	free(read_buf.buf);

	// Receive nbytes :: size_t from dnode
	Byte fsize_bytes[sizeof(size_t)] = {0};
	Buffer fsize_buf = (Buffer){ .buf = fsize_bytes, .len = sizeof(size_t) };
	size_t len = recibe(dnode, fsize_buf);

	// Crash if metadata server sent something wrong
	if (len != sizeof(size_t)) {
		log(FATAL, "Metadata server sent wrong amount of bytes. Expected %zu, got %zu\n", sizeof(size_t), len);
		exit(EXIT_FAILURE);
	}

	size_t nbytes = size_t_deserialize(fsize_buf);

	// Receive all bytes from dnode
	for (size_t bytes_received = 0; bytes_received < nbytes;) {
		Byte buf[CHUNK_SIZE] = {0};
		Buffer outside_buf = (Buffer){ .buf = buf, .len = CHUNK_SIZE };
		len = recibe(dnode, outside_buf);
		outside_buf.len = len;

		Buffer_write(outside_buf, f);

		bytes_received += len;
		printf("\r");
		log(INFO, "Received %zu bytes out of %zu", bytes_received, nbytes);
		fflush(stdout);
	}
	printf("\n");
}

/// Copy from the client to the server
UUID client_to_server(Enchufe dnode, LazyBuffer* b, size_t nbytes, size_t chunks_per_dnode) {
	exists(b);
	conecta(dnode);

	// Send WRITE(nbytes :: size_t) to dnode
	Command write = (Command){
		.tag = WRITE,
		.as.write = {
			.nbytes = nbytes,
		},
	};
	Buffer write_buf = Command_serialize(write);
	zumba(dnode, write_buf);
	free(write_buf.buf);

	// Receive cid :: UUID from dnode
	Buffer cid_buf = UUID_serialize((UUID){0});
	recibe(dnode, cid_buf);
	UUID cid = UUID_deserialize(cid_buf);

	size_t bytes_sent = 0;
	// Send all bytes to dnode
	for (size_t sent_chunks = 0; sent_chunks < chunks_per_dnode; sent_chunks++) {
		// Get the next chunk of bytes
		*b = next_chunk(*b);
		if (b->len == 0) break;

		// Send buf :: Buffer to dnode
		zumba(dnode, (Buffer){ .buf = b->buf, .len = b->len });

		bytes_sent += b->len;
		printf("\r");
		log(INFO, "Sent %zu bytes out of %zu", bytes_sent, nbytes);
		fflush(stdout);
	}
	printf("\n");
	desenchufa(dnode);

	return cid;
}

/// This is the copy client implementation
int main(int argc, char** argv) {
	// Parse command line arguments
	if (argc != 6) {
		for (int i = 0; i < argc; ++i) {
			fprintf(stderr, "%s ", argv[i]);
		}
		fprintf(stderr, "\n");
		fprintf(stderr, "USAGE: %s IPv4 Port [-s] path/to/source_file.ext [-s] path/to/dest_file.ext\n", argv[0]);
		fprintf(stderr, "USAGE: use the -s flag to indicate which file is in the server\n");
		exit(1);
	}

	// Depending on this flag, we can determine whether the file is being copied
	// from the server or not.
	bool from_server = Str_eq(argv[3], "-s");

	IPv4 ip = parse_address(argv[1]);
	Port port = (Port)atoi(argv[2]);
	Str dest = argv[5];
	Str source = from_server ? argv[4] : argv[3];

	if (from_server) {
		// Connect to metadata server
		Enchufe metadata = enchufa(ip, port);
		conecta(metadata);

		// Send GET(fname :: SafeStr) to metadata server
		Command get = (Command){
			.tag = GET,
			.as.get.fname = atoss(source), // INFO: This is safe because source is a command line argument
		};
		Buffer get_buf = Command_serialize(get);
		zumba(metadata, get_buf);
		free(get_buf.buf);

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

		// Sort the ListDNodeInfo depending on the sequence number.
		dinfos = sort(dinfos);

		desenchufa(metadata);

		FILE* f = fopen(dest, "w");
		exists(f);

		// Get data from each dnode and copy it to the client.
		for (ListDNodeInfo dptr = dinfos; dptr != NULL; dptr = dptr->rest) {
			DNodeInfo dinfo = dptr->head;

			// Do exchange with data node
			Enchufe dnode = enchufa(dinfo.dnode.address, dinfo.dnode.port);
			server_to_client(dnode, dinfo.cid, f);
			desenchufa(dnode);
		}
		fclose(f);
	} else {
		// Check if file exists
		if (access(source, F_OK) == 0) {
			// Connect to metadata server
			Enchufe metadata = enchufa(ip, port);
			conecta(metadata);

			// Open lazyBuffer for reading
			LazyBuffer b = open(source);
			exists(b.fptr);

			// Send PUT(fsize :: size_t, fname :: SafeStr) to metadata server
			Command put = {
				.tag = PUT,
				.as.put = {
					.fsize = b.size,
					.fname = atoss(dest), // INFO: This is safe because dest is a command line argument
				},
			};
			Buffer put_buf = Command_serialize(put);
			zumba(metadata, put_buf);
			free(put_buf.buf);

			// Receive dnodes :: ListDNode from metadata server
			Byte buf1[BUF_LEN] = {0};
			Buffer outside_buf = { .buf = buf1, .len = BUF_LEN };
			size_t len = recibe(metadata, outside_buf);
			outside_buf.len = len;

			if (len == 0) {
				log(FATAL, "Metadata server closed the connection right after sending a PUT request\n");
				exit(EXIT_FAILURE);
			}

			ListDNode dnodes = ListDNode_deserialize(outside_buf);

			// INFO: This will calculate how many chunks will have to be sent. In
			// the case where the amount of bytes that have to be sent aren't the a
			// multiple of CHUNK_SIZE, this will have an extra 1 added. Hence, the slightly
			// funny looking line:
			// \code
			//	!!(b.size % CHUNK_SIZE)
			// \endcode
			size_t chunk_count = (b.size / CHUNK_SIZE) + !!(b.size % CHUNK_SIZE);

			// INFO: Additionally, the chunks_per_dnode line looks like that because I
			// wanted a fast ceiling integer division.
			// from https://stackoverflow.com/questions/2745074/fast-ceiling-of-an-integer-division-in-c-c
			size_t dnode_count = ListDNode_length(dnodes);
			size_t chunks_per_dnode = 1 + (chunk_count - 1) / dnode_count;

			// Connect to each data node and do the exchange
			ListDNodeInfo dinfos = NULL;
			size_t seq = 1;
			for (ListDNode dptr = dnodes; dptr != NULL; dptr = dptr ->rest) {
				DNode d = dptr->head;

				// Connect to data node
				Enchufe dnode = enchufa(d.address, d.port);

				// Calculate remaining bytes
				size_t bytes_remaining = b.size - b.position;
				size_t nbytes = min(bytes_remaining, chunks_per_dnode * CHUNK_SIZE);

				// Do the exchange
				UUID cid = client_to_server(dnode, &b, nbytes, chunks_per_dnode);
				dinfos = ListDNodeInfo_cons((DNodeInfo){
						.dnode = d,
						.cid = cid,
						.seq = seq,
					}, dinfos);
				seq += 1;
				desenchufa(dnode);
			}
			fclose(b.fptr);

			// Send dinfos :: ListDNodeInfo to metadata server
			Buffer dinfo_buf = ListDNodeInfo_serialize(dinfos);
			zumba(metadata, dinfo_buf);
			desenchufa(metadata);
		} else {
			log(FATAL, "'%s': No such file or directory\n", source);
			exit(EXIT_FAILURE);
		}
	}
}