sparse-tools/src/bs_recv.c

//
// Created by haraldwolff on 07.08.22.
//

#include <blksync.h>
#include <hash.h>
#include <time.h>

int bs_receiver_tx(bsync_engine_t *engine);

int bs_receiver(bsync_engine_t *engine){
	uint32_t _magic = 0, _version = 0;
	uint64_t _filesize = 0;
	uint32_t _blocksize = 0;

	bs_send( engine->clientSocket, &magic, sizeof(magic));
	bs_send( engine->clientSocket, &version, sizeof(version));

	bs_recv( engine->clientSocket, &_magic, sizeof(_magic));
	bs_recv( engine->clientSocket, &_version, sizeof(_version));
	bs_recv( engine->clientSocket, &_filesize, sizeof(_filesize));
	bs_recv( engine->clientSocket, &_blocksize, sizeof(_blocksize));

	if (magic != _magic){
		fprintf(stderr, "magic missmatch (0x%08x != 0x%08x)\n", magic, _magic);
		exit(EXIT_FAILURE);
	}
	if (version != _version){
		fprintf(stderr, "version missmatch\n");
		exit(EXIT_FAILURE);
	}

	if (_filesize != engine->filesize){
		if (ftruncate( engine->file, _filesize ) == -1){
			if (engine->filesize > _filesize){
				fprintf(stderr, "warning: could not truncate %s\n", engine->filename);
			} else {
				fprintf(stderr, "error: could not extent %s\n", engine->filename);
				exit(EXIT_FAILURE);
			}
		}
		engine->filesize = _filesize;
	}

	if (_blocksize != engine->blocksize)
	{
		if (engine->blocksize == 0)
			engine->blocksize = _blocksize;
		if (_blocksize == 0)
			_blocksize = engine->blocksize;
	}
	if (_blocksize != engine->blocksize)
	{
		fprintf(stderr, "don't agree about blocksize (%d <> %d)", engine->blocksize, _blocksize);
		exit(EXIT_FAILURE);
	}

	if (!engine->blocksize)
		engine->blocksize = defaultBlocksize;

	bs_send( engine->clientSocket, &engine->filesize, sizeof(engine->filesize));
	bs_send( engine->clientSocket, &engine->blocksize, sizeof(engine->blocksize));

	fprintf(stdout, "size: %ld bytes, blocksize: %ld bytes\n", engine->filesize, engine->blocksize);

	char *block = malloc(engine->blocksize);
	if (!block)
	{
		fprintf(stderr, "no memory available for block");
		exit(EXIT_FAILURE);
	}
	long offset = 0;
	long residual;

	pthread_t thread_tx;
	int s = pthread_create(&thread_tx, NULL, (void * (*)(void *))bs_receiver_tx, engine);
	if (s != 0){
		fprintf(stderr, "could not create tx thread\n");
		fflush(stderr);
		exit(EXIT_FAILURE);
	}

	while (-1){
		if (bs_recv( engine->clientSocket, &offset, sizeof(offset))==-1){
			fprintf(stderr, "could not receive offset\n");
			exit(EXIT_FAILURE);
		}
		if (offset == -1ul){
			fprintf(stderr, "received end mark [0x%lx].\n", offset);
			break;
		}

		residual = engine->filesize - offset;
		int32_t size = residual > engine->blocksize ? engine->blocksize : residual;

		if (bs_recv( engine->clientSocket, block, size)==-1){
			fprintf(stderr, "could not receive block at offset 0x%lx\n", offset);
			exit(EXIT_FAILURE);
		}

		engine->stat_blocks++;
		engine->stat_block_bytes += size;

		//fprintf(stderr, "received block with %d bytes @ offset 0x%lx\n", size, offset);

		if (!(engine->ds_flags & DS_SIMULATE)) {
			if (write_block(engine, offset, block, size) != size)
				exit(EXIT_FAILURE);
		}

	}

	fprintf( stderr, "bs_receiver(); finished.");
	fflush(stderr);

	fprintf( stdout, "\n");
}


int bs_receiver_tx(bsync_engine_t *engine){
	char *block = malloc(engine->blocksize);
	if (!block)
	{
		fprintf(stderr, "no memory available for block");
		exit(EXIT_FAILURE);
	}
	long offset = 0;
	long residual = engine->filesize;

	fprintf( stderr, "bs_receiver_tx(); starts");
	fflush(stderr);

	while (residual > 0){
		int32_t size = residual > engine->blocksize ? engine->blocksize : residual;
		uint64_t blockhash, remotehash;

		if (read_block(engine, offset, block, size) != size)
			exit(EXIT_FAILURE);

		blockhash = dhash( block, size);

		int percent = 100 * offset / engine->filesize;

		if (
				(bs_send(engine->clientSocket, &offset, sizeof(offset)) == -1) ||
				(bs_send( engine->clientSocket, &blockhash, sizeof(blockhash))==-1)
			)
		{
			fprintf(stderr, "could not send block hash at offset 0x%lx\n", offset);
			exit(EXIT_FAILURE);
		}

		engine->last_offset = offset;

		time_t t = time(NULL);
		if (engine->t_last_update != t)
			dump_engine_state(engine);

		offset += size;
		residual -= size;
	}

	fprintf( stderr, "bs_receiver_tx(); finished.");
	fflush(stderr);

	offset = -1l;
	if (bs_send(engine->clientSocket, &offset, sizeof(offset)) == -1)
	{
		fprintf(stderr, "could not send end mark\n");
		exit(EXIT_FAILURE);
	}

	sleep(30);

	return 0;
}