<commit_before>#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/time.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <assert.h>

#include "cache.h"
#include "associative_cache.h"
#include "global_cached_private.h"
#include "part_global_cached_private.h"
#include "parameters.h"

const int MIN_NUM_PROCESS_REQ = 100;
const int BUF_SIZE = 100;

#if 0
thread_group::~thread_group()
{
	for (size_t i = 0; i < process_request_threads.size(); i++) {
		thread *t = process_request_threads[i];
		t->stop();
		// TODO I need to free memory of the thread here.
	}
	for (size_t i = 0; i < process_reply_threads.size(); i++) {
		thread *t = process_reply_threads[i];
		t->stop();
		// TODO I need to free memory of the thread here.
	}
	delete request_queue;
	delete reply_queue;
}
#endif

/**
 * This wraps global cached IO to handle requests in the cache
 * of the local node.
 */
class node_cached_io: public global_cached_io
{
	// thread id <-> msg sender
	std::tr1::unordered_map<int, msg_sender<io_reply> *> reply_senders;
	const std::tr1::unordered_map<int, struct thread_group> *groups;
	const struct thread_group *local_group;
	long processed_requests;

	pthread_t processing_thread_id;

	const struct thread_group *get_group(int group_id) {
		std::tr1::unordered_map<int, struct thread_group>::const_iterator it
			= groups->find(get_node_id());
		assert(it != groups->end());
		return &it->second;
	}
public:
	node_cached_io(io_interface *underlying, const std::tr1::unordered_map<int,
			struct thread_group> *groups);

	virtual page_cache *get_global_cache() {
		return local_group->cache;
	}

	void init_repliers();

	int process_requests(int max_nreqs);

	int reply(io_request *requests, io_reply *replies, int num);

	void flush_replies();

	virtual void notify_completion(io_request *req);
};

node_cached_io::node_cached_io(io_interface *underlying,
		const std::tr1::unordered_map<int, struct thread_group> *groups):
	global_cached_io(underlying)
{
	this->groups = groups;
	local_group = get_group(underlying->get_node_id());
	assert(local_group);
	processed_requests = 0;
	// This IO instance is created inside the right thread.
	global_cached_io::init();

	processing_thread_id = 0;
}

void node_cached_io::init_repliers()
{
	for (std::tr1::unordered_map<int, struct thread_group>::const_iterator it
			= groups->begin(); it != groups->end(); it++) {
		const struct thread_group *group = &it->second;
		for (int j = 0; j < group->nthreads; j++) {
			// We need a sender for each parted global IO because we need to
			// make sure replies are sent to the right the IO instance that
			// issued the request.
			thread_safe_FIFO_queue<io_reply> *q1[1];
			q1[0] = group->ios[j]->reply_queue;
			reply_senders.insert(std::pair<int, msg_sender<io_reply> *>(
						group->ios[j]->thread_id,
						new msg_sender<io_reply>(NUMA_MSG_CACHE_SIZE, q1, 1, true)));
		}
	}
}

void node_cached_io::flush_replies()
{
	for (std::tr1::unordered_map<int, msg_sender<io_reply> *>::const_iterator it
			= reply_senders.begin(); it != reply_senders.end(); it++) {
		msg_sender<io_reply> *sender = it->second;
		sender->flush_all();
	}
}

/* process the requests sent to this thread */
int node_cached_io::process_requests(int max_nreqs)
{
	if (processing_thread_id == 0)
		processing_thread_id = pthread_self();
	assert(processing_thread_id == pthread_self());
	int num_processed = 0;
	thread_safe_FIFO_queue<io_request> *request_queue
		= local_group->request_queue;
	io_request local_reqs[BUF_SIZE];
	while (num_processed < max_nreqs) {
		int num = request_queue->fetch(local_reqs, BUF_SIZE);
		for (int i = 0; i < num; i++) {
			io_request *req = &local_reqs[i];
			assert(req->get_offset() >= 0);
			assert(req->get_size() > 0);
			// The thread will be blocked by the global cache IO
			// if too many requests flush in.
			int ret = global_cached_io::access(req, 1);
			if (ret < 0) {
				fprintf(stderr, "part global cache can't issue a request\n");
				io_reply rep(req, false, errno);
				reply(req, &rep, 1);
			}
		}
		num_processed += num;
	}
	return num_processed;
}

/**
 * send replies to the thread that sent the requests.
 */
int node_cached_io::reply(io_request *requests, io_reply *replies, int num)
{
	for (int i = 0; i < num; i++) {
		part_global_cached_io *io = (part_global_cached_io *) requests[i].get_io();
		// If the reply is sent to the thread on a different node.
		if (io->get_node_id() != this->get_node_id()) {
			int num_sent = reply_senders[io->thread_id]->send_cached(&replies[i]);
			// We use blocking queues here, so the send must succeed.
			assert(num_sent > 0);
		}
		else {
			io->process_reply(&replies[i]);
			io->processed_replies.inc(1);
		}
	}
	return 0;
}

void node_cached_io::notify_completion(io_request *req)
{
	assert(get_node_id() == 1);
	io_reply rep(req, true, 0);
	reply(req, &rep, 1);
}

/**
 * This thread runs on the node with cache.
 * It is dedicated to processing requests.
 * The application threads send requests to the node with cache<commit_msg>Remove unnecessary assert in parted global cache.<commit_after>#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <sys/time.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <assert.h>

#include "cache.h"
#include "associative_cache.h"
#include "global_cached_private.h"
#include "part_global_cached_private.h"
#include "parameters.h"

const int MIN_NUM_PROCESS_REQ = 100;
const int BUF_SIZE = 100;

#if 0
thread_group::~thread_group()
{
	for (size_t i = 0; i < process_request_threads.size(); i++) {
		thread *t = process_request_threads[i];
		t->stop();
		// TODO I need to free memory of the thread here.
	}
	for (size_t i = 0; i < process_reply_threads.size(); i++) {
		thread *t = process_reply_threads[i];
		t->stop();
		// TODO I need to free memory of the thread here.
	}
	delete request_queue;
	delete reply_queue;
}
#endif

/**
 * This wraps global cached IO to handle requests in the cache
 * of the local node.
 */
class node_cached_io: public global_cached_io
{
	// thread id <-> msg sender
	std::tr1::unordered_map<int, msg_sender<io_reply> *> reply_senders;
	const std::tr1::unordered_map<int, struct thread_group> *groups;
	const struct thread_group *local_group;
	long processed_requests;

	pthread_t processing_thread_id;

	const struct thread_group *get_group(int group_id) {
		std::tr1::unordered_map<int, struct thread_group>::const_iterator it
			= groups->find(get_node_id());
		assert(it != groups->end());
		return &it->second;
	}
public:
	node_cached_io(io_interface *underlying, const std::tr1::unordered_map<int,
			struct thread_group> *groups);

	virtual page_cache *get_global_cache() {
		return local_group->cache;
	}

	void init_repliers();

	int process_requests(int max_nreqs);

	int reply(io_request *requests, io_reply *replies, int num);

	void flush_replies();

	virtual void notify_completion(io_request *req);
};

node_cached_io::node_cached_io(io_interface *underlying,
		const std::tr1::unordered_map<int, struct thread_group> *groups):
	global_cached_io(underlying)
{
	this->groups = groups;
	local_group = get_group(underlying->get_node_id());
	assert(local_group);
	processed_requests = 0;
	// This IO instance is created inside the right thread.
	global_cached_io::init();

	processing_thread_id = 0;
}

void node_cached_io::init_repliers()
{
	for (std::tr1::unordered_map<int, struct thread_group>::const_iterator it
			= groups->begin(); it != groups->end(); it++) {
		const struct thread_group *group = &it->second;
		for (int j = 0; j < group->nthreads; j++) {
			// We need a sender for each parted global IO because we need to
			// make sure replies are sent to the right the IO instance that
			// issued the request.
			thread_safe_FIFO_queue<io_reply> *q1[1];
			q1[0] = group->ios[j]->reply_queue;
			reply_senders.insert(std::pair<int, msg_sender<io_reply> *>(
						group->ios[j]->thread_id,
						new msg_sender<io_reply>(NUMA_MSG_CACHE_SIZE, q1, 1, true)));
		}
	}
}

void node_cached_io::flush_replies()
{
	for (std::tr1::unordered_map<int, msg_sender<io_reply> *>::const_iterator it
			= reply_senders.begin(); it != reply_senders.end(); it++) {
		msg_sender<io_reply> *sender = it->second;
		sender->flush_all();
	}
}

/* process the requests sent to this thread */
int node_cached_io::process_requests(int max_nreqs)
{
	if (processing_thread_id == 0)
		processing_thread_id = pthread_self();
	assert(processing_thread_id == pthread_self());
	int num_processed = 0;
	thread_safe_FIFO_queue<io_request> *request_queue
		= local_group->request_queue;
	io_request local_reqs[BUF_SIZE];
	while (num_processed < max_nreqs) {
		int num = request_queue->fetch(local_reqs, BUF_SIZE);
		for (int i = 0; i < num; i++) {
			io_request *req = &local_reqs[i];
			assert(req->get_offset() >= 0);
			assert(req->get_size() > 0);
			// The thread will be blocked by the global cache IO
			// if too many requests flush in.
			int ret = global_cached_io::access(req, 1);
			if (ret < 0) {
				fprintf(stderr, "part global cache can't issue a request\n");
				io_reply rep(req, false, errno);
				reply(req, &rep, 1);
			}
		}
		num_processed += num;
	}
	return num_processed;
}

/**
 * send replies to the thread that sent the requests.
 */
int node_cached_io::reply(io_request *requests, io_reply *replies, int num)
{
	for (int i = 0; i < num; i++) {
		part_global_cached_io *io = (part_global_cached_io *) requests[i].get_io();
		// If the reply is sent to the thread on a different node.
		if (io->get_node_id() != this->get_node_id()) {
			int num_sent = reply_senders[io->thread_id]->send_cached(&replies[i]);
			// We use blocking queues here, so the send must succeed.
			assert(num_sent > 0);
		}
		else {
			io->process_reply(&replies[i]);
			io->processed_replies.inc(1);
		}
	}
	return 0;
}

void node_cached_io::notify_completion(io_request *req)
{
	io_reply rep(req, true, 0);
	reply(req, &rep, 1);
}

/**
 * This thread runs on the node with cache.
 * It is dedicated to processing requests.
 * The application threads send requests to the node with cache
{'fraction_non_alphanumeric': 0.136421134572346, 'fraction_numerical': 0.007687068210567286, 'mean_word_length': 4.672185430463577, 'pattern_counts': {'":': 0, '<': 73, '<?xml version=': 0, '>': 121, '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': '6621398', 'n_tokens_mistral': 3934, 'n_tokens_neox': 3665, 'n_words': 937}