<commit_before>/**
 * Copyright 2013 <NAME>
 *
 * This file is part of SAFSlib.
 *
 * SAFSlib is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * SAFSlib is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with SAFSlib.  If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * This file implements global cache.
 * There are three types of requests:
 * original request: it is a copy of the request passed from access() and 
 *		is allocated in the heap.
 * partial request: it represents part of a request, and it should have 
 *		a point to the original request. This exists in two cases: when
 *		writing an old dirty page; when reading a page with a pending IO.
 * underlying request: it is a request sent to the underlying IO and is
 *		allocated on the stack. It has a point to the original request for
 *		a read request and the partial request for a write request. Only
 *		this type of requests can be multi-buf requests, and it should have
 *		a point to a page if it's a single-buf request.
 *
 * When a request is issued to the cache with access(), we first make a copy
 * of the request as there is no guarantee that the original request from
 * the invoker will be available after access() returns.
 *
 * However, it's often that we need to break a request into smaller ones for
 * different reasons when issuing them to the underlying IO.
 *
 * In some case, a request tries to access a page that another request has
 * been issued to the underlying IO for the page, the request will be added
 * to the page.
 */

#include <vector>
#include <algorithm>

#include "global_cached_private.h"
#include "slab_allocator.h"

const int COMPLETE_QUEUE_SIZE = 10240;
const int REQ_BUF_SIZE = 64;
const int OBJ_ALLOC_INC_SIZE = 1024 * 1024;

comp_io_scheduler::comp_io_scheduler(int node_id): incomplete_computes(node_id,
		COMPLETE_QUEUE_SIZE, true)
{
	io = NULL;
}

void comp_io_scheduler::gc_computes()
{
	int size = incomplete_computes.get_num_entries();
	for (int i = 0; i < size; i++) {
		user_compute *compute = incomplete_computes.front();
		incomplete_computes.pop_front();
		if (compute->has_completed()) {
			delete_compute(compute);
		}
		else {
			// The computation hasn't been completed, we need to add it
			// back to the queue for further processing.
			incomplete_computes.push_back(compute);
		}
	}
}

/**
 * This is a default I/O scheduler for requests generated by user tasks.
 * This scheduler doesn't make much assumption on the requests.
 * So it reads as many requests as possible from a user task, and then
 * read from the next user task until it reaches the limit. The requests
 * will be returned to global_cached_io in a sorted order.
 *
 * This scheduler works to favor latency, so we finish one user task
 * before processing the next task.
 */
class default_comp_io_scheduler: public comp_io_scheduler
{
	const int MAX_REQS_FETCH;
public:
	default_comp_io_scheduler(int node_id,
			int max_fetch = INT_MAX): comp_io_scheduler(
			node_id), MAX_REQS_FETCH(max_fetch) {
	}

	virtual size_t get_requests(fifo_queue<io_request> &reqs);
};

size_t default_comp_io_scheduler::get_requests(fifo_queue<io_request> &requests)
{
	// If the request queue is already full, don't do anything.
	if (requests.is_full())
		return 0;

	bool has_reqs;
	size_t num = 0;
	do {
		has_reqs = false;
		compute_iterator it = this->get_begin();
		compute_iterator end = this->get_end();
		for (; it != end; ++it) {
			user_compute *compute = *it;
			int ret = compute->fetch_requests(get_io(), requests,
					min(MAX_REQS_FETCH, requests.get_num_remaining()));
			num += ret;
			// We should limit the number of pending requests generated by
			// user compute.
			if (requests.is_full()) {
				assert(ret > 0);
				break;
			}
			has_reqs |= compute->has_requests();
		}
	} while (has_reqs && !requests.is_full());
	return num;
}

/**
 * The initial size of the queue for pending IO requests
 */
const int INIT_GCACHE_PENDING_SIZE = 1000;

void thread_safe_page::add_req(original_io_request *req)
{
	req->set_next_req_on_page(this, reqs);
	reqs = req;
}

original_io_request *thread_safe_page::reset_reqs()
{
	original_io_request *ret = reqs;
	reqs = NULL;
	return ret;
}

original_io_request *thread_safe_page::get_io_req() const
{
	return reqs;
}

class req_ext_allocator: public obj_allocator<io_req_extension>
{
	class ext_initiator: public obj_initiator<io_req_extension>
	{
	public:
		void init(io_req_extension *ext) {
			if (ext->is_valid()) {
				ext->init();
			}
			else {
				new (ext) io_req_extension();
			}
		}
	} initiator;
public:
	req_ext_allocator(int node_id, long max_size = params.get_max_obj_alloc_size(
				)): obj_allocator<io_req_extension>(
				std::string("req_ext_allocator-") + itoa(node_id), node_id,
				OBJ_ALLOC_INC_SIZE, max_size, &initiator) {
	}
	
	virtual io_req_extension *alloc_obj() {
		io_req_extension *ext = obj_allocator<io_req_extension>::alloc_obj();
		assert(ext);
		return ext;
	}
};

/**
 * This slab allocator allocates IO requests, and all of them are
 * extended requests.
 */
class request_allocator: public obj_allocator<original_io_request><commit_msg>[SAFS]: destroy io_req_extension.<commit_after>/**
 * Copyright 2013 <NAME>
 *
 * This file is part of SAFSlib.
 *
 * SAFSlib is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * SAFSlib is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with SAFSlib.  If not, see <http://www.gnu.org/licenses/>.
 */

/**
 * This file implements global cache.
 * There are three types of requests:
 * original request: it is a copy of the request passed from access() and 
 *		is allocated in the heap.
 * partial request: it represents part of a request, and it should have 
 *		a point to the original request. This exists in two cases: when
 *		writing an old dirty page; when reading a page with a pending IO.
 * underlying request: it is a request sent to the underlying IO and is
 *		allocated on the stack. It has a point to the original request for
 *		a read request and the partial request for a write request. Only
 *		this type of requests can be multi-buf requests, and it should have
 *		a point to a page if it's a single-buf request.
 *
 * When a request is issued to the cache with access(), we first make a copy
 * of the request as there is no guarantee that the original request from
 * the invoker will be available after access() returns.
 *
 * However, it's often that we need to break a request into smaller ones for
 * different reasons when issuing them to the underlying IO.
 *
 * In some case, a request tries to access a page that another request has
 * been issued to the underlying IO for the page, the request will be added
 * to the page.
 */

#include <vector>
#include <algorithm>

#include "global_cached_private.h"
#include "slab_allocator.h"

const int COMPLETE_QUEUE_SIZE = 10240;
const int REQ_BUF_SIZE = 64;
const int OBJ_ALLOC_INC_SIZE = 1024 * 1024;

comp_io_scheduler::comp_io_scheduler(int node_id): incomplete_computes(node_id,
		COMPLETE_QUEUE_SIZE, true)
{
	io = NULL;
}

void comp_io_scheduler::gc_computes()
{
	int size = incomplete_computes.get_num_entries();
	for (int i = 0; i < size; i++) {
		user_compute *compute = incomplete_computes.front();
		incomplete_computes.pop_front();
		if (compute->has_completed()) {
			delete_compute(compute);
		}
		else {
			// The computation hasn't been completed, we need to add it
			// back to the queue for further processing.
			incomplete_computes.push_back(compute);
		}
	}
}

/**
 * This is a default I/O scheduler for requests generated by user tasks.
 * This scheduler doesn't make much assumption on the requests.
 * So it reads as many requests as possible from a user task, and then
 * read from the next user task until it reaches the limit. The requests
 * will be returned to global_cached_io in a sorted order.
 *
 * This scheduler works to favor latency, so we finish one user task
 * before processing the next task.
 */
class default_comp_io_scheduler: public comp_io_scheduler
{
	const int MAX_REQS_FETCH;
public:
	default_comp_io_scheduler(int node_id,
			int max_fetch = INT_MAX): comp_io_scheduler(
			node_id), MAX_REQS_FETCH(max_fetch) {
	}

	virtual size_t get_requests(fifo_queue<io_request> &reqs);
};

size_t default_comp_io_scheduler::get_requests(fifo_queue<io_request> &requests)
{
	// If the request queue is already full, don't do anything.
	if (requests.is_full())
		return 0;

	bool has_reqs;
	size_t num = 0;
	do {
		has_reqs = false;
		compute_iterator it = this->get_begin();
		compute_iterator end = this->get_end();
		for (; it != end; ++it) {
			user_compute *compute = *it;
			int ret = compute->fetch_requests(get_io(), requests,
					min(MAX_REQS_FETCH, requests.get_num_remaining()));
			num += ret;
			// We should limit the number of pending requests generated by
			// user compute.
			if (requests.is_full()) {
				assert(ret > 0);
				break;
			}
			has_reqs |= compute->has_requests();
		}
	} while (has_reqs && !requests.is_full());
	return num;
}

/**
 * The initial size of the queue for pending IO requests
 */
const int INIT_GCACHE_PENDING_SIZE = 1000;

void thread_safe_page::add_req(original_io_request *req)
{
	req->set_next_req_on_page(this, reqs);
	reqs = req;
}

original_io_request *thread_safe_page::reset_reqs()
{
	original_io_request *ret = reqs;
	reqs = NULL;
	return ret;
}

original_io_request *thread_safe_page::get_io_req() const
{
	return reqs;
}

class req_ext_allocator: public obj_allocator<io_req_extension>
{
	class ext_initiator: public obj_initiator<io_req_extension>
	{
	public:
		void init(io_req_extension *ext) {
			new (ext) io_req_extension();
		}
	} initiator;

	class ext_destructor: public obj_destructor<io_req_extension>
	{
	public:
		void destroy(io_req_extension *ext) {
			ext->~io_req_extension();
		}
	} destructor;
public:
	req_ext_allocator(int node_id, long max_size = params.get_max_obj_alloc_size(
				)): obj_allocator<io_req_extension>(
				std::string("req_ext_allocator-") + itoa(node_id), node_id,
				OBJ_ALLOC_INC_SIZE, max_size, &initiator, &destructor) {
	}
	
	virtual io_req_extension *alloc_obj() {
		io_req_extension *ext = obj_allocator<io_req_extension>::alloc_obj();
		assert(ext);
		return ext;
	}
};

/**
 * This slab allocator allocates IO requests, and all of them are
 * extended requests.
 */
class request_allocator: public obj_allocator<original_io_request>
{'fraction_non_alphanumeric': 0.08823268599804912, 'fraction_numerical': 0.004965859714463066, 'mean_word_length': 4.152124257651896, 'pattern_counts': {'":': 0, '<': 28, '<?xml version=': 0, '>': 43, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '1665819', 'n_tokens_mistral': 3817, 'n_tokens_neox': 3557, 'n_words': 1300}