<commit_before>#include <assert.h>
#include <pthread.h>

#include <vector>

#include "RAID_config.h"
#include "io_interface.h"
#include "read_private.h"
#include "direct_private.h"
#include "aio_private.h"
#include "remote_access.h"
#include "global_cached_private.h"
#include "part_global_cached_private.h"

class io_tracker
{
	io_interface *io;
	bool has_init;
	pthread_spinlock_t lock;
	bool taken;
public:
	io_tracker(io_interface *io) {
		this->io = io;
		pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE);
		taken = false;
		has_init = false;
	}

	io_interface *take() {
		pthread_spin_lock(&lock);
		io_interface *ret = NULL;
		if (!taken)
			ret = io;
		taken = true;
		// If the IO is used for the first time, initialize it.
		if (!has_init && ret) {
			ret->init();
			has_init = true;
		}
		pthread_spin_unlock(&lock);
		return ret;
	}

	io_interface *get() {
		assert(has_init && taken);
		return io;
	}

	void release() {
		pthread_spin_lock(&lock);
		taken = false;
		pthread_spin_unlock(&lock);
	}
};

static std::vector<io_tracker *> io_table;

int get_num_ios()
{
	return io_table.size();
}

void register_io(io_interface *io)
{
	// Make sure the index hasn't been set.
	assert(io->get_io_idx() < 0);

	io_table.push_back(new io_tracker(io));
	int idx = io_table.size() - 1;
	io->set_io_idx(idx);
}

io_interface *get_io(int idx)
{
	assert(idx >= 0);
	io_interface *io = io_table[idx]->get();
	assert(io);
	assert(io->get_io_idx() == idx);
	return io;
}

io_interface *allocate_io()
{
	for (unsigned i = 0; i < io_table.size(); i++) {
		io_interface *io = io_table[i]->take();
		if (io)
			return io;
	}
	return NULL;
}

void release_io(io_interface *io)
{
	io_table[io->get_io_idx()]->release();
}

std::vector<io_interface *> create_ios(const RAID_config &raid_conf,
		cache_config *cache_conf, const std::vector<int> &node_id_array,
		int nthreads, int access_option, long size, bool preload)
{
	int num_nodes = node_id_array.size();
	file_mapper *mapper = raid_conf.create_file_mapper();

	std::vector<int> indices;
	int num_files = mapper->get_num_files();
	for (int i = 0; i < num_files; i++)
		indices.push_back(i);
	// The partition contains all files.
	logical_file_partition global_partition(indices, mapper);

	std::vector<disk_read_thread *> read_threads(num_files);
	std::tr1::unordered_map<int, aio_complete_thread *> complete_threads;
	if (access_option == GLOBAL_CACHE_ACCESS
			|| access_option == PART_GLOBAL_ACCESS
			|| access_option == REMOTE_ACCESS) {
		// Create threads for helping process completed AIO requests.
		for (int i = 0; i < num_nodes; i++) {
			int node_id = node_id_array[i];
			complete_threads.insert(std::pair<int, aio_complete_thread *>(node_id,
						new aio_complete_thread(node_id)));
		}
		for (int k = 0; k < num_files; k++) {
			std::vector<int> indices(1, k);
			logical_file_partition partition(indices, mapper);
			// Create disk accessing threads.<commit_msg>check number of threads and number of nodes.<commit_after>#include <assert.h>
#include <pthread.h>

#include <vector>

#include "RAID_config.h"
#include "io_interface.h"
#include "read_private.h"
#include "direct_private.h"
#include "aio_private.h"
#include "remote_access.h"
#include "global_cached_private.h"
#include "part_global_cached_private.h"

class io_tracker
{
	io_interface *io;
	bool has_init;
	pthread_spinlock_t lock;
	bool taken;
public:
	io_tracker(io_interface *io) {
		this->io = io;
		pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE);
		taken = false;
		has_init = false;
	}

	io_interface *take() {
		pthread_spin_lock(&lock);
		io_interface *ret = NULL;
		if (!taken)
			ret = io;
		taken = true;
		// If the IO is used for the first time, initialize it.
		if (!has_init && ret) {
			ret->init();
			has_init = true;
		}
		pthread_spin_unlock(&lock);
		return ret;
	}

	io_interface *get() {
//		assert(has_init && taken);
		return io;
	}

	void release() {
		pthread_spin_lock(&lock);
		taken = false;
		pthread_spin_unlock(&lock);
	}
};

static std::vector<io_tracker *> io_table;

int get_num_ios()
{
	return io_table.size();
}

void register_io(io_interface *io)
{
	// Make sure the index hasn't been set.
	assert(io->get_io_idx() < 0);

	io_table.push_back(new io_tracker(io));
	int idx = io_table.size() - 1;
	io->set_io_idx(idx);
}

io_interface *get_io(int idx)
{
	assert(idx >= 0);
	io_interface *io = io_table[idx]->get();
	assert(io);
	assert(io->get_io_idx() == idx);
	return io;
}

io_interface *allocate_io()
{
	for (unsigned i = 0; i < io_table.size(); i++) {
		io_interface *io = io_table[i]->take();
		if (io)
			return io;
	}
	return NULL;
}

void release_io(io_interface *io)
{
	io_table[io->get_io_idx()]->release();
}

std::vector<io_interface *> create_ios(const RAID_config &raid_conf,
		cache_config *cache_conf, const std::vector<int> &node_id_array,
		int nthreads, int access_option, long size, bool preload)
{
	int num_nodes = node_id_array.size();
	assert(num_nodes <= nthreads && nthreads % num_nodes == 0);
	file_mapper *mapper = raid_conf.create_file_mapper();

	std::vector<int> indices;
	int num_files = mapper->get_num_files();
	for (int i = 0; i < num_files; i++)
		indices.push_back(i);
	// The partition contains all files.
	logical_file_partition global_partition(indices, mapper);

	std::vector<disk_read_thread *> read_threads(num_files);
	std::tr1::unordered_map<int, aio_complete_thread *> complete_threads;
	if (access_option == GLOBAL_CACHE_ACCESS
			|| access_option == PART_GLOBAL_ACCESS
			|| access_option == REMOTE_ACCESS) {
		// Create threads for helping process completed AIO requests.
		for (int i = 0; i < num_nodes; i++) {
			int node_id = node_id_array[i];
			complete_threads.insert(std::pair<int, aio_complete_thread *>(node_id,
						new aio_complete_thread(node_id)));
		}
		for (int k = 0; k < num_files; k++) {
			std::vector<int> indices(1, k);
			logical_file_partition partition(indices, mapper);
			// Create disk accessing threads.
{'fraction_non_alphanumeric': 0.1417283120376597, 'fraction_numerical': 0.003194351042367182, 'mean_word_length': 4.770126091173617, 'pattern_counts': {'":': 0, '<': 36, '<?xml version=': 0, '>': 47, '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': '1650074', 'n_tokens_mistral': 2403, 'n_tokens_neox': 2252, 'n_words': 499}