<commit_before>#include "utils.h"
#include "crypto/sha2.h"
#include "flaggedarrayset.h"
#include "relayprocess.h"

#include <stdio.h>
#include <sys/time.h>
#include <algorithm>
#include <random>
#include <string.h>
#include <unistd.h>

void do_nothing(...) {}

#ifdef BENCH
#define PRINT_TIME do_nothing
#else
#define PRINT_TIME printf
#endif

std::linear_congruential_engine<std::uint_fast32_t, 48271, 0, 2147483647> engine(42);

void fill_txv(std::vector<unsigned char>& block, std::vector<std::shared_ptr<std::vector<unsigned char> > >& txVectors, float includeP) {
	std::vector<unsigned char>::const_iterator readit = block.begin();
	move_forward(readit, sizeof(struct bitcoin_msg_header), block.end());
	move_forward(readit, 80, block.end());
	uint32_t txcount = read_varint(readit, block.end());

	std::uniform_real_distribution<double> distribution(0.0, 1.0);

	for (uint32_t i = 0; i < txcount; i++) {
		std::vector<unsigned char>::const_iterator txstart = readit;

		move_forward(readit, 4, block.end());

		uint32_t txins = read_varint(readit, block.end());
		for (uint32_t j = 0; j < txins; j++) {
			move_forward(readit, 36, block.end());
			uint32_t scriptlen = read_varint(readit, block.end());
			move_forward(readit, scriptlen + 4, block.end());
		}

		uint32_t txouts = read_varint(readit, block.end());
		for (uint32_t j = 0; j < txouts; j++) {
			move_forward(readit, 8, block.end());
			uint32_t scriptlen = read_varint(readit, block.end());
			move_forward(readit, scriptlen, block.end());
		}

		move_forward(readit, 4, block.end());

		if (distribution(engine) < includeP)
			txVectors.push_back(std::make_shared<std::vector<unsigned char> >(txstart, readit));
	}

	std::shuffle(txVectors.begin(), txVectors.end(), engine);
}

int pipefd[2];
uint32_t block_tx_count;
std::shared_ptr<std::vector<unsigned char> > decompressed_block;
RelayNodeCompressor receiver;

void recv_block() {
	struct timeval start, decompressed;
	gettimeofday(&start, NULL);
	auto res = receiver.decompress_relay_block(pipefd[0], block_tx_count);
	gettimeofday(&decompressed, NULL);

	if (std::get<2>(res)) {
		printf("ERROR Decompressing block %s\n", std::get<2>(res));
		exit(2);
	} else
		PRINT_TIME("Decompressed block in %lu ms\n", int64_t(decompressed.tv_sec - start.tv_sec)*1000 + (int64_t(decompressed.tv_usec) - start.tv_usec)/1000);
	decompressed_block = std::get<1>(res);
}

void test_compress_block(std::vector<unsigned char>& data, std::vector<std::shared_ptr<std::vector<unsigned char> > > txVectors) {
	std::vector<unsigned char> fullhash(32);
	getblockhash(fullhash, data, sizeof(struct bitcoin_msg_header));

	RelayNodeCompressor sender, tester;
	receiver.reset();

	for (auto& v : txVectors) {
		unsigned int made = sender.get_relay_transaction(v).use_count();
		if (made)
			receiver.recv_tx(v);
		if (made != tester.get_relay_transaction(v).use_count()) {
			printf("get_relay_transaction behavior not consistent???\n");
			exit(5);
		}
	}

	unsigned int i = 0;
	sender.for_each_sent_tx([&](std::shared_ptr<std::vector<unsigned char> > tx) {
		ssize_t index = std::max((ssize_t)0, (ssize_t)txVectors.size() - 1525) + i++;
		if (tx != txVectors[index]) {
			printf("for_each_sent_tx was not in order!\n");
			exit(6);
		}
		if (tester.send_tx_cache.remove(0) != txVectors[index]) {
			printf("for_each_sent_tx output did not match remove(0)\n");
			exit(7);
		}
	});

	struct timeval start, compressed;
	gettimeofday(&start, NULL);
	auto res = sender.maybe_compress_block(fullhash, data, true);
	gettimeofday(&compressed, NULL);
	PRINT_TIME("Compressed from %lu to %lu in %ld ms with %lu txn pre-relayed\n", data.size(), std::get<0>(res)->size(), int64_t(compressed.tv_sec - start.tv_sec)*1000 + (int64_t(compressed.tv_usec) - start.tv_usec)/1000, txVectors.size());

	struct relay_msg_header header;
	memcpy(&header, &(*std::get<0>(res))[0], sizeof(header));
	block_tx_count = ntohl(header.length);

	if (pipe(pipefd)) {
		printf("Failed to create pipe?\n");
		exit(3);
	}

	std::thread recv(recv_block);
	write(pipefd[1], &(*std::get<0>(res))[sizeof(header)], std::get<0>(res)->size() - sizeof(header));
	recv.join();

	if (*decompressed_block != data) {
		printf("Re-constructed block did not match!\n");
		exit(4);
	}
}

void run_test(std::vector<unsigned char>& data) {
	std::vector<std::shared_ptr<std::vector<unsigned char> > > txVectors;
	test_compress_block(data, txVectors);

	fill_txv(data, txVectors, 1.0);
	test_compress_block(data, txVectors);

	txVectors.clear();
	fill_txv(data, txVectors, 0.5);
	test_compress_block(data, txVectors);<commit_msg>Close them file descriptors<commit_after>#include "utils.h"
#include "crypto/sha2.h"
#include "flaggedarrayset.h"
#include "relayprocess.h"

#include <stdio.h>
#include <sys/time.h>
#include <algorithm>
#include <random>
#include <string.h>
#include <unistd.h>

void do_nothing(...) {}

#ifdef BENCH
#define PRINT_TIME do_nothing
#else
#define PRINT_TIME printf
#endif

std::linear_congruential_engine<std::uint_fast32_t, 48271, 0, 2147483647> engine(42);

void fill_txv(std::vector<unsigned char>& block, std::vector<std::shared_ptr<std::vector<unsigned char> > >& txVectors, float includeP) {
	std::vector<unsigned char>::const_iterator readit = block.begin();
	move_forward(readit, sizeof(struct bitcoin_msg_header), block.end());
	move_forward(readit, 80, block.end());
	uint32_t txcount = read_varint(readit, block.end());

	std::uniform_real_distribution<double> distribution(0.0, 1.0);

	for (uint32_t i = 0; i < txcount; i++) {
		std::vector<unsigned char>::const_iterator txstart = readit;

		move_forward(readit, 4, block.end());

		uint32_t txins = read_varint(readit, block.end());
		for (uint32_t j = 0; j < txins; j++) {
			move_forward(readit, 36, block.end());
			uint32_t scriptlen = read_varint(readit, block.end());
			move_forward(readit, scriptlen + 4, block.end());
		}

		uint32_t txouts = read_varint(readit, block.end());
		for (uint32_t j = 0; j < txouts; j++) {
			move_forward(readit, 8, block.end());
			uint32_t scriptlen = read_varint(readit, block.end());
			move_forward(readit, scriptlen, block.end());
		}

		move_forward(readit, 4, block.end());

		if (distribution(engine) < includeP)
			txVectors.push_back(std::make_shared<std::vector<unsigned char> >(txstart, readit));
	}

	std::shuffle(txVectors.begin(), txVectors.end(), engine);
}

int pipefd[2];
uint32_t block_tx_count;
std::shared_ptr<std::vector<unsigned char> > decompressed_block;
RelayNodeCompressor receiver;

void recv_block() {
	struct timeval start, decompressed;
	gettimeofday(&start, NULL);
	auto res = receiver.decompress_relay_block(pipefd[0], block_tx_count);
	gettimeofday(&decompressed, NULL);

	if (std::get<2>(res)) {
		printf("ERROR Decompressing block %s\n", std::get<2>(res));
		exit(2);
	} else
		PRINT_TIME("Decompressed block in %lu ms\n", int64_t(decompressed.tv_sec - start.tv_sec)*1000 + (int64_t(decompressed.tv_usec) - start.tv_usec)/1000);
	decompressed_block = std::get<1>(res);
}

void test_compress_block(std::vector<unsigned char>& data, std::vector<std::shared_ptr<std::vector<unsigned char> > > txVectors) {
	std::vector<unsigned char> fullhash(32);
	getblockhash(fullhash, data, sizeof(struct bitcoin_msg_header));

	RelayNodeCompressor sender, tester;
	receiver.reset();

	for (auto& v : txVectors) {
		unsigned int made = sender.get_relay_transaction(v).use_count();
		if (made)
			receiver.recv_tx(v);
		if (made != tester.get_relay_transaction(v).use_count()) {
			printf("get_relay_transaction behavior not consistent???\n");
			exit(5);
		}
	}

	unsigned int i = 0;
	sender.for_each_sent_tx([&](std::shared_ptr<std::vector<unsigned char> > tx) {
		ssize_t index = std::max((ssize_t)0, (ssize_t)txVectors.size() - 1525) + i++;
		if (tx != txVectors[index]) {
			printf("for_each_sent_tx was not in order!\n");
			exit(6);
		}
		if (tester.send_tx_cache.remove(0) != txVectors[index]) {
			printf("for_each_sent_tx output did not match remove(0)\n");
			exit(7);
		}
	});

	struct timeval start, compressed;
	gettimeofday(&start, NULL);
	auto res = sender.maybe_compress_block(fullhash, data, true);
	gettimeofday(&compressed, NULL);
	PRINT_TIME("Compressed from %lu to %lu in %ld ms with %lu txn pre-relayed\n", data.size(), std::get<0>(res)->size(), int64_t(compressed.tv_sec - start.tv_sec)*1000 + (int64_t(compressed.tv_usec) - start.tv_usec)/1000, txVectors.size());

	struct relay_msg_header header;
	memcpy(&header, &(*std::get<0>(res))[0], sizeof(header));
	block_tx_count = ntohl(header.length);

	if (pipe(pipefd)) {
		printf("Failed to create pipe?\n");
		exit(3);
	}

	std::thread recv(recv_block);
	write(pipefd[1], &(*std::get<0>(res))[sizeof(header)], std::get<0>(res)->size() - sizeof(header));
	recv.join();

	if (*decompressed_block != data) {
		printf("Re-constructed block did not match!\n");
		exit(4);
	}

	close(pipefd[0]);
	close(pipefd[1]);
}

void run_test(std::vector<unsigned char>& data) {
	std::vector<std::shared_ptr<std::vector<unsigned char> > > txVectors;
	test_compress_block(data, txVectors);

	fill_txv(data, txVectors, 1.0);
	test_compress_block(data, txVectors);

	txVectors.clear();
	fill_txv(data, txVectors, 0.5);
	test_compress_block(data, txVectors);
{'fraction_non_alphanumeric': 0.16809642740797046, 'fraction_numerical': 0.023889673145835597, 'mean_word_length': 6.0466717674062735, 'pattern_counts': {'":': 0, '<': 83, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2913331', 'n_tokens_mistral': 3766, 'n_tokens_neox': 3473, 'n_words': 662}