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

#include <tr1/unordered_set>
#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"
#include "cache_config.h"
#include "disk_read_thread.h"
#include "debugger.h"
#include "mem_tracker.h"

#define DEBUG

/**
 * This global data collection is very static.
 * Once the data is initialized, no data needs to be changed.
 * The mutex is to used only at the initialization.
 * As long as all threads call init_io_system() first before using
 * the global data, they will all see the complete global data.
 */
struct global_data_collection
{
	RAID_config raid_conf;
	std::vector<disk_read_thread *> read_threads;
	pthread_mutex_t mutex;

#ifdef DEBUG
	std::tr1::unordered_set<io_interface *> ios;
#endif

	global_data_collection() {
		pthread_mutex_init(&mutex, NULL);
	}

#ifdef DEBUG
	void register_io(io_interface *io) {
		ios.insert(io);
	}
#endif
};

static global_data_collection global_data;

#if 0
/**
 * We flush notifications of IO completion buffered by I/O threads
 * periodically. By default, it's every 200ms.
 */
static void handler(int sig, siginfo_t *si, void *uc)
{
	// We can't call the flush requests to wake up I/O threads.
	// There might be a deadlock if the lock has been held by the thread
	// interrupted by the signal.
#if 0
	for (unsigned i = 0; i < global_data.read_threads.size(); i++) {
		global_data.read_threads[i]->flush_requests();
	}
#endif
}

static void set_completion_flush_timer()
{
	/**
	 * The code here is copied from the example code in the manual of
	 * timer_create.
	 */
	timer_t timerid;
	struct sigevent sev;
	struct itimerspec its;
	sigset_t mask;
	struct sigaction sa;

	/* Establish handler for timer signal */

	sa.sa_flags = SA_SIGINFO;
	sa.sa_sigaction = handler;
	sigemptyset(&sa.sa_mask);
	if (sigaction(SIGRTMIN, &sa, NULL) == -1) {
		perror("sigaction");
		exit(1);
	}

	/* Block timer signal temporarily */

	sigemptyset(&mask);
	sigaddset(&mask, SIGRTMIN);
	if (sigprocmask(SIG_SETMASK, &mask, NULL) == -1) {
		perror("sigprocmask");
		exit(1);
	}

	/* Create the timer */

	sev.sigev_notify = SIGEV_SIGNAL;
	sev.sigev_signo = SIGRTMIN;
	sev.sigev_value.sival_ptr = &timerid;
	if (timer_create(CLOCK_REALTIME, &sev, &timerid) == -1) {
		perror("timer_create");
		exit(1);
	}

	/* Start the timer */

	its.it_value.tv_sec = 0;
	its.it_value.tv_nsec = COMPLETION_FLUSH_INTERVAL;
	its.it_interval.tv_sec = its.it_value.tv_sec;
	its.it_interval.tv_nsec = its.it_value.tv_nsec;

	if (timer_settime(timerid, 0, &its, NULL) == -1) {
		perror("timer_settime");
		exit(1);
	}

	/* Unlock the timer signal, so that timer notification
	 * can be delivered */

	if (sigprocmask(SIG_UNBLOCK, &mask, NULL) == -1) {
		perror("sigprocmask");
		exit(1);
	}
}
#endif

class debug_global_data: public debug_task
{
public:
	void run();
};

void debug_global_data::run()
{
	for (unsigned i = 0; i < global_data.read_threads.size(); i++)
		global_data.read_threads[i]->print_state();
#ifdef DEBUG
	for (std::tr1::unordered_set<io_interface *>::iterator it
			= global_data.ios.begin(); it != global_data.ios.end(); it++)
		(*it)->print_state();
#endif
}

void init_io_system(const config_map &configs)
{
#ifdef ENABLE_MEM_TRACE
	init_mem_tracker();
#endif

	params.init(configs.get_options());
	params.print();

	numa_set_bind_policy(1);
	thread::thread_class_init();

	std::string root_conf_file = configs.get_option("root_conf");
	printf("The root conf file: %s\n", root_conf_file.c_str());
	RAID_config raid_conf(root_conf_file, params.get_RAID_mapping_option(),
			params.get_RAID_block_size());
	int num_files = raid_conf.get_num_disks();
	global_data.raid_conf = raid_conf;<commit_msg>[SAFS]: fix a bug in io_interface.<commit_after>#include <assert.h>
#include <signal.h>
#include <time.h>
#include <pthread.h>

#include <tr1/unordered_set>
#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"
#include "cache_config.h"
#include "disk_read_thread.h"
#include "debugger.h"
#include "mem_tracker.h"

#define DEBUG

/**
 * This global data collection is very static.
 * Once the data is initialized, no data needs to be changed.
 * The mutex is to used only at the initialization.
 * As long as all threads call init_io_system() first before using
 * the global data, they will all see the complete global data.
 */
struct global_data_collection
{
	RAID_config raid_conf;
	std::vector<disk_read_thread *> read_threads;
	pthread_mutex_t mutex;

#ifdef DEBUG
	std::tr1::unordered_set<io_interface *> ios;
	pthread_spinlock_t ios_lock;
#endif

	global_data_collection() {
		pthread_mutex_init(&mutex, NULL);
#ifdef DEBUG
		pthread_spin_init(&ios_lock, PTHREAD_PROCESS_PRIVATE);
#endif
	}

#ifdef DEBUG
	void register_io(io_interface *io) {
		pthread_spin_lock(&ios_lock);
		ios.insert(io);
		pthread_spin_unlock(&ios_lock);
	}
#endif
};

static global_data_collection global_data;

#if 0
/**
 * We flush notifications of IO completion buffered by I/O threads
 * periodically. By default, it's every 200ms.
 */
static void handler(int sig, siginfo_t *si, void *uc)
{
	// We can't call the flush requests to wake up I/O threads.
	// There might be a deadlock if the lock has been held by the thread
	// interrupted by the signal.
#if 0
	for (unsigned i = 0; i < global_data.read_threads.size(); i++) {
		global_data.read_threads[i]->flush_requests();
	}
#endif
}

static void set_completion_flush_timer()
{
	/**
	 * The code here is copied from the example code in the manual of
	 * timer_create.
	 */
	timer_t timerid;
	struct sigevent sev;
	struct itimerspec its;
	sigset_t mask;
	struct sigaction sa;

	/* Establish handler for timer signal */

	sa.sa_flags = SA_SIGINFO;
	sa.sa_sigaction = handler;
	sigemptyset(&sa.sa_mask);
	if (sigaction(SIGRTMIN, &sa, NULL) == -1) {
		perror("sigaction");
		exit(1);
	}

	/* Block timer signal temporarily */

	sigemptyset(&mask);
	sigaddset(&mask, SIGRTMIN);
	if (sigprocmask(SIG_SETMASK, &mask, NULL) == -1) {
		perror("sigprocmask");
		exit(1);
	}

	/* Create the timer */

	sev.sigev_notify = SIGEV_SIGNAL;
	sev.sigev_signo = SIGRTMIN;
	sev.sigev_value.sival_ptr = &timerid;
	if (timer_create(CLOCK_REALTIME, &sev, &timerid) == -1) {
		perror("timer_create");
		exit(1);
	}

	/* Start the timer */

	its.it_value.tv_sec = 0;
	its.it_value.tv_nsec = COMPLETION_FLUSH_INTERVAL;
	its.it_interval.tv_sec = its.it_value.tv_sec;
	its.it_interval.tv_nsec = its.it_value.tv_nsec;

	if (timer_settime(timerid, 0, &its, NULL) == -1) {
		perror("timer_settime");
		exit(1);
	}

	/* Unlock the timer signal, so that timer notification
	 * can be delivered */

	if (sigprocmask(SIG_UNBLOCK, &mask, NULL) == -1) {
		perror("sigprocmask");
		exit(1);
	}
}
#endif

class debug_global_data: public debug_task
{
public:
	void run();
};

void debug_global_data::run()
{
	for (unsigned i = 0; i < global_data.read_threads.size(); i++)
		global_data.read_threads[i]->print_state();
#ifdef DEBUG
	pthread_spin_lock(&global_data.ios_lock);
	std::tr1::unordered_set<io_interface *> ios = global_data.ios;
	pthread_spin_unlock(&global_data.ios_lock);

	for (std::tr1::unordered_set<io_interface *>::iterator it
			= ios.begin(); it != ios.end(); it++)
		(*it)->print_state();
#endif
}

void init_io_system(const config_map &configs)
{
#ifdef ENABLE_MEM_TRACE
	init_mem_tracker();
#endif

	params.init(configs.get_options());
	params.print();

	numa_set_bind_policy(1);
	thread::thread_class_init();

	std::string root_conf_file = configs.get_option("root_conf");
	printf("The root conf file: %s\n", root_conf_file.c_str());
	RAID_config raid_conf(root_conf_file, params.get_RAID_mapping_option(),
			params.get_RAID_block_size());
	int num_files = raid_conf.get_num_disks();
	global_data.raid_conf = raid_conf;
{'fraction_non_alphanumeric': 0.13249907986750092, 'fraction_numerical': 0.005766163660900503, 'mean_word_length': 4.937363437727604, 'pattern_counts': {'":': 0, '<': 26, '<?xml version=': 0, '>': 28, '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': '2195510', 'n_tokens_mistral': 3138, 'n_tokens_neox': 3015, 'n_words': 722}