<commit_before>#include <assert.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <getopt.h>
#include <netdb.h>
#include <signal.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>

#include "list.h"
#include "rand.h"

static struct {
	char *blob_dir;
	char *node;
	char *service;
	uint32_t num_conns;
	uint32_t cycle_ms;
} config = {
	.num_conns = 100,
	.cycle_ms = 50,
};

struct file {
	char *path;
	char *buf;
	size_t len;
	struct list_head file_list;
};

struct conn {
	int fd;
	enum {
		CONN_CONNECTING,
		CONN_SENDING,
	} state;
	struct file *file;
	size_t offset;
	struct list_head conn_list;
};

static struct list_head file_head = LIST_HEAD_INIT(file_head);
static struct list_head conn_head = LIST_HEAD_INIT(conn_head);

static struct addrinfo *addrs = NULL;
static uint64_t rounds = 0, open_conns = 0;
static bool shutdown_flag = false;

static void do_shutdown(int __attribute__((unused)) signal) {
	shutdown_flag = true;
}

static bool parse_opts(int argc, char *argv[]) {
	static struct option long_options[] = {
		{"blob-dir",  required_argument, 0, 'b'},
		{"host",      required_argument, 0, 'h'},
		{"port",      required_argument, 0, 'p'},
		{"cycle-ms",  required_argument, 0, 'c'},
		{"num-conns", required_argument, 0, 'n'},
		{0,           0,                 0, 0  },
	};

	int opt;
	while ((opt = getopt_long_only(argc, argv, "", long_options, NULL)) != -1) {
		switch (opt) {
			case 'b':
				config.blob_dir = optarg;
				break;

			case 'h':
				config.node = optarg;
				break;

			case 'p':
				config.service = optarg;
				break;

			case 'c':
				config.cycle_ms = (uint32_t) strtoul(optarg, NULL, 10);
				break;

			case 'n':
				config.num_conns = (uint32_t) strtoul(optarg, NULL, 10);
				assert(config.num_conns);
				break;

			default:
				return false;
		}
	}

	if (optind != argc) {
		return false;
	}

	if (!config.blob_dir ||
			!config.node ||
			!config.service) {
		return false;
	}

	return true;
}

static void file_open() {
	size_t dirlen = strlen(config.blob_dir);

	DIR *dir = opendir(config.blob_dir);
	assert(dir);

	uint64_t max_size = 0, min_size = UINT64_MAX, total_size = 0, num_blobs = 0;

	while (true) {
		struct dirent entry, *ret;
		assert(!readdir_r(dir, &entry, &ret));
		if (!ret) {
			break;
		}
		if (entry.d_name[0] == '.') {
			continue;
		}

		struct file *file = malloc(sizeof(*file));
		assert(file);
		list_add(&file->file_list, &file_head);

		size_t max_len = dirlen + strlen(entry.d_name) + 2;
		file->path = malloc(max_len);
		assert(file->path);
		snprintf(file->path, max_len, "%s/%s", config.blob_dir, entry.d_name);

		int fd = open(file->path, O_RDONLY);
		assert(fd >= 0);

		struct stat stat;
		assert(!fstat(fd, &stat));

		file->len = (size_t) stat.st_size;

		max_size = max_size < file->len ? file->len : max_size;
		min_size = min_size > file->len ? file->len : min_size;
		num_blobs++;
		total_size += file->len;

		file->buf = mmap(NULL, file->len, PROT_READ, MAP_SHARED, fd, 0);
		assert(file->buf);

		assert(!close(fd));
	}

	assert(num_blobs);

	fprintf(stderr, "Loaded %ju blobs. Bytes: %ju min, %ju mean, %ju max\n",
			(uintmax_t) num_blobs,
			(uintmax_t) min_size,
			(uintmax_t) (total_size / num_blobs),
			(uintmax_t) max_size);

	assert(!closedir(dir));
}

static void file_del(struct file *file) {
	free(file->path);
	assert(!munmap(file->buf, file->len));
	list_del(&file->file_list);
	free(file);
}

static void file_cleanup() {
	struct file *iter, *next;
	list_for_each_entry_safe(iter, next, &file_head, file_list) {
		file_del(iter);
	}
}

static struct file *file_next() {
	static struct list_head *first = NULL, *iter;

	if (!first) {
		iter = first = file_head.next;
	}

	if (iter->next == &file_head) {
		iter = iter->next->next;
		rounds++;
		if (!(++rounds % 100)) {
			fprintf(stderr, "\rRounds: %ju\r", (uintmax_t) rounds);
		}
	} else {
		iter = iter->next;
	}
	return list_entry(iter, struct file, file_list);
}

static size_t conn_get_split(struct conn *conn) {
	size_t total_len = conn->file->len;
	size_t remaining_len = total_len - conn->offset;
	size_t rnd;
	rand_fill(&rnd, sizeof(rnd));
	rnd %= total_len;
	return rnd > remaining_len ? remaining_len : rnd;
}

static void conn_new() {
	if (!addrs) {
		struct addrinfo hints = {<commit_msg>Stop double-counting rounds.<commit_after>#include <assert.h>
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <getopt.h>
#include <netdb.h>
#include <signal.h>
#include <stdbool.h>
#include <stdio.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <sys/mman.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>

#include "list.h"
#include "rand.h"

static struct {
	char *blob_dir;
	char *node;
	char *service;
	uint32_t num_conns;
	uint32_t cycle_ms;
} config = {
	.num_conns = 100,
	.cycle_ms = 50,
};

struct file {
	char *path;
	char *buf;
	size_t len;
	struct list_head file_list;
};

struct conn {
	int fd;
	enum {
		CONN_CONNECTING,
		CONN_SENDING,
	} state;
	struct file *file;
	size_t offset;
	struct list_head conn_list;
};

static struct list_head file_head = LIST_HEAD_INIT(file_head);
static struct list_head conn_head = LIST_HEAD_INIT(conn_head);

static struct addrinfo *addrs = NULL;
static uint64_t rounds = 0, open_conns = 0;
static bool shutdown_flag = false;

static void do_shutdown(int __attribute__((unused)) signal) {
	shutdown_flag = true;
}

static bool parse_opts(int argc, char *argv[]) {
	static struct option long_options[] = {
		{"blob-dir",  required_argument, 0, 'b'},
		{"host",      required_argument, 0, 'h'},
		{"port",      required_argument, 0, 'p'},
		{"cycle-ms",  required_argument, 0, 'c'},
		{"num-conns", required_argument, 0, 'n'},
		{0,           0,                 0, 0  },
	};

	int opt;
	while ((opt = getopt_long_only(argc, argv, "", long_options, NULL)) != -1) {
		switch (opt) {
			case 'b':
				config.blob_dir = optarg;
				break;

			case 'h':
				config.node = optarg;
				break;

			case 'p':
				config.service = optarg;
				break;

			case 'c':
				config.cycle_ms = (uint32_t) strtoul(optarg, NULL, 10);
				break;

			case 'n':
				config.num_conns = (uint32_t) strtoul(optarg, NULL, 10);
				assert(config.num_conns);
				break;

			default:
				return false;
		}
	}

	if (optind != argc) {
		return false;
	}

	if (!config.blob_dir ||
			!config.node ||
			!config.service) {
		return false;
	}

	return true;
}

static void file_open() {
	size_t dirlen = strlen(config.blob_dir);

	DIR *dir = opendir(config.blob_dir);
	assert(dir);

	uint64_t max_size = 0, min_size = UINT64_MAX, total_size = 0, num_blobs = 0;

	while (true) {
		struct dirent entry, *ret;
		assert(!readdir_r(dir, &entry, &ret));
		if (!ret) {
			break;
		}
		if (entry.d_name[0] == '.') {
			continue;
		}

		struct file *file = malloc(sizeof(*file));
		assert(file);
		list_add(&file->file_list, &file_head);

		size_t max_len = dirlen + strlen(entry.d_name) + 2;
		file->path = malloc(max_len);
		assert(file->path);
		snprintf(file->path, max_len, "%s/%s", config.blob_dir, entry.d_name);

		int fd = open(file->path, O_RDONLY);
		assert(fd >= 0);

		struct stat stat;
		assert(!fstat(fd, &stat));

		file->len = (size_t) stat.st_size;

		max_size = max_size < file->len ? file->len : max_size;
		min_size = min_size > file->len ? file->len : min_size;
		num_blobs++;
		total_size += file->len;

		file->buf = mmap(NULL, file->len, PROT_READ, MAP_SHARED, fd, 0);
		assert(file->buf);

		assert(!close(fd));
	}

	assert(num_blobs);

	fprintf(stderr, "Loaded %ju blobs. Bytes: %ju min, %ju mean, %ju max\n",
			(uintmax_t) num_blobs,
			(uintmax_t) min_size,
			(uintmax_t) (total_size / num_blobs),
			(uintmax_t) max_size);

	assert(!closedir(dir));
}

static void file_del(struct file *file) {
	free(file->path);
	assert(!munmap(file->buf, file->len));
	list_del(&file->file_list);
	free(file);
}

static void file_cleanup() {
	struct file *iter, *next;
	list_for_each_entry_safe(iter, next, &file_head, file_list) {
		file_del(iter);
	}
}

static struct file *file_next() {
	static struct list_head *first = NULL, *iter;

	if (!first) {
		iter = first = file_head.next;
	}

	if (iter->next == &file_head) {
		iter = iter->next->next;
		if (!(++rounds % 100)) {
			fprintf(stderr, "\rRounds: %ju\r", (uintmax_t) rounds);
		}
	} else {
		iter = iter->next;
	}
	return list_entry(iter, struct file, file_list);
}

static size_t conn_get_split(struct conn *conn) {
	size_t total_len = conn->file->len;
	size_t remaining_len = total_len - conn->offset;
	size_t rnd;
	rand_fill(&rnd, sizeof(rnd));
	rnd %= total_len;
	return rnd > remaining_len ? remaining_len : rnd;
}

static void conn_new() {
	if (!addrs) {
		struct addrinfo hints = {
{'fraction_non_alphanumeric': 0.16105204252937885, 'fraction_numerical': 0.010072747621712367, 'mean_word_length': 3.814655172413793, 'pattern_counts': {'":': 0, '<': 41, '<?xml version=': 0, '>': 95, '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': '3545939', 'n_tokens_mistral': 3900, 'n_tokens_neox': 3559, 'n_words': 823}