<commit_before>/**
 * Copyright 2013 <NAME>
 *
 * This file is part of SA-GraphLib.
 *
 * SA-GraphLib 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.
 *
 * SA-GraphLib 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 SA-GraphLib.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <algorithm>
#ifndef MEMCHECK
#include <parallel/algorithm>
#endif

#include "io_interface.h"

#include "bitmap.h"
#include "graph_config.h"
#include "graph_engine.h"
#include "messaging.h"
#include "worker_thread.h"
#include "vertex_compute.h"
#include "vertex_request.h"

graph_config graph_conf;

struct prio_compute
{
	user_compute *compute;
	io_request req;

	prio_compute(io_interface *io, user_compute *compute) {
		this->compute = compute;
		bool ret = compute->fetch_request(io, req);
		assert(ret);
	}
};

class forward_comp_prio_compute
{
public:
	bool operator()(const prio_compute &c1, const prio_compute &c2) {
		// We want the priority queue returns requests with
		// the smallest offset first. If we use less, the priority queue
		// will return the request with the greatest offset.
		return c1.req.get_offset() > c2.req.get_offset();
	}
};

class backward_comp_prio_compute
{
public:
	bool operator()(const prio_compute &c1, const prio_compute &c2) {
		// We want the priority queue returns requests with
		// the smallest offset first. If we use less, the priority queue
		// will return the request with the greatest offset.
		return c1.req.get_offset() < c2.req.get_offset();
	}
};

template<class prio_queue_type>
class comp_io_schedule_queue
{
	bool forward;
	// Construct a priority queue on user tasks, ordered by the offset
	// of their next requests.
	prio_queue_type user_computes;
	comp_io_scheduler *scheduler;
public:
	comp_io_schedule_queue(comp_io_scheduler *scheduler, bool forward) {
		this->scheduler = scheduler;
		this->forward = forward;
	}

	size_t get_requests(fifo_queue<io_request> &reqs);

	bool is_empty() const {
		return user_computes.empty();
	}
};

template<class prio_queue_type>
size_t comp_io_schedule_queue<prio_queue_type>::get_requests(
		fifo_queue<io_request> &reqs)
{
	size_t num = 0;

	if (!reqs.is_full()) {
		// we don't have user computes, we can get some from the queue of
		// incomplete user computes in comp_io_scheduler.
		if (user_computes.empty()) {
			comp_io_scheduler::compute_iterator it = scheduler->get_begin();
			comp_io_scheduler::compute_iterator end = scheduler->get_end();
			for (; it != end; ++it) {
				user_compute *compute = *it;
				// Skip the ones without user tasks.
				if (!compute->has_requests())
					continue;

				// We have a reference to the user compute. Let's increase
				// its ref count. User computes should be in the queue of
				// comp_io_scheduler as long as they can generate more
				// requests. It might not be necessary to increase the ref
				// count, but it can work as a sanity check.
				compute->inc_ref();
				((vertex_compute *) compute)->set_scan_dir(forward);
				prio_compute prio_comp(scheduler->get_io(), compute);
				user_computes.push(prio_comp);
			}
		}

		// Add requests to the queue in a sorted order.
		int num = 0;
		while (!reqs.is_full() && !user_computes.empty()) {
			num++;
			prio_compute prio_comp = user_computes.top();
			user_computes.pop();
			reqs.push_back(prio_comp.req);
			num++;
			user_compute *compute = prio_comp.compute;
			if (compute->has_requests()) {
				prio_compute prio_comp(scheduler->get_io(), compute);
				user_computes.push(prio_comp);
			}
			else {
				// This user compute no longer needs to stay in the priority
				// queue. We can decrease its ref count.
				compute->dec_ref();
			}
		}
	}
	return num;
}

/**
 * This I/O scheduler is to favor maximizing throughput.
 * Therefore, it processes all user tasks together to potentially increase
 * the page cache hit rate.
 */
class throughput_comp_io_scheduler: public comp_io_scheduler
{
	// The scheduler process user computes in batches. It reads all
	// available user computes in the beginning of a batch and then processes
	// them. A batch ends if the scheduler processes all of the user computes.
	size_t batch_num;
	comp_io_schedule_queue<std::priority_queue<prio_compute,
		std::vector<prio_compute>, forward_comp_prio_compute> > forward_queue;
	comp_io_schedule_queue<std::priority_queue<prio_compute,
		std::vector<prio_compute>, backward_comp_prio_compute> > backward_queue;
public:
	throughput_comp_io_scheduler(int node_id): comp_io_scheduler(
			node_id), forward_queue(this, true), backward_queue(this, false) {
		batch_num = 0;
	}

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

class throughput_comp_io_sched_creater: public comp_io_sched_creater
{
public:
	comp_io_scheduler *create(int node_id) const {
		return new throughput_comp_io_scheduler(node_id);
	}
};

size_t throughput_comp_io_scheduler::get_requests(fifo_queue<io_request> &reqs)
{
	size_t ret;
	if (batch_num % 2 == 0) {
		ret = forward_queue.get_requests(reqs);
		if (forward_queue.is_empty())
			batch_num++;
	}
	else {
		ret = backward_queue.get_requests(reqs);
		if (backward_queue.is_empty())
			batch_num++;
	}<commit_msg>[Graph]: allow to enable elevator alg for requesting neighbors.<commit_after>/**
 * Copyright 2013 <NAME>
 *
 * This file is part of SA-GraphLib.
 *
 * SA-GraphLib 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.
 *
 * SA-GraphLib 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 SA-GraphLib.  If not, see <http://www.gnu.org/licenses/>.
 */

#include <algorithm>
#ifndef MEMCHECK
#include <parallel/algorithm>
#endif

#include "io_interface.h"

#include "bitmap.h"
#include "graph_config.h"
#include "graph_engine.h"
#include "messaging.h"
#include "worker_thread.h"
#include "vertex_compute.h"
#include "vertex_request.h"

graph_config graph_conf;

struct prio_compute
{
	user_compute *compute;
	io_request req;

	prio_compute(io_interface *io, user_compute *compute) {
		this->compute = compute;
		bool ret = compute->fetch_request(io, req);
		assert(ret);
	}
};

class forward_comp_prio_compute
{
public:
	bool operator()(const prio_compute &c1, const prio_compute &c2) {
		// We want the priority queue returns requests with
		// the smallest offset first. If we use less, the priority queue
		// will return the request with the greatest offset.
		return c1.req.get_offset() > c2.req.get_offset();
	}
};

class backward_comp_prio_compute
{
public:
	bool operator()(const prio_compute &c1, const prio_compute &c2) {
		// We want the priority queue returns requests with
		// the smallest offset first. If we use less, the priority queue
		// will return the request with the greatest offset.
		return c1.req.get_offset() < c2.req.get_offset();
	}
};

template<class prio_queue_type>
class comp_io_schedule_queue
{
	bool forward;
	// Construct a priority queue on user tasks, ordered by the offset
	// of their next requests.
	prio_queue_type user_computes;
	comp_io_scheduler *scheduler;
public:
	comp_io_schedule_queue(comp_io_scheduler *scheduler, bool forward) {
		this->scheduler = scheduler;
		this->forward = forward;
	}

	size_t get_requests(fifo_queue<io_request> &reqs);

	bool is_empty() const {
		return user_computes.empty();
	}
};

template<class prio_queue_type>
size_t comp_io_schedule_queue<prio_queue_type>::get_requests(
		fifo_queue<io_request> &reqs)
{
	size_t num = 0;

	if (!reqs.is_full()) {
		// we don't have user computes, we can get some from the queue of
		// incomplete user computes in comp_io_scheduler.
		if (user_computes.empty()) {
			comp_io_scheduler::compute_iterator it = scheduler->get_begin();
			comp_io_scheduler::compute_iterator end = scheduler->get_end();
			for (; it != end; ++it) {
				user_compute *compute = *it;
				// Skip the ones without user tasks.
				if (!compute->has_requests())
					continue;

				// We have a reference to the user compute. Let's increase
				// its ref count. User computes should be in the queue of
				// comp_io_scheduler as long as they can generate more
				// requests. It might not be necessary to increase the ref
				// count, but it can work as a sanity check.
				compute->inc_ref();
				((vertex_compute *) compute)->set_scan_dir(forward);
				prio_compute prio_comp(scheduler->get_io(), compute);
				user_computes.push(prio_comp);
			}
		}

		// Add requests to the queue in a sorted order.
		int num = 0;
		while (!reqs.is_full() && !user_computes.empty()) {
			num++;
			prio_compute prio_comp = user_computes.top();
			user_computes.pop();
			reqs.push_back(prio_comp.req);
			num++;
			user_compute *compute = prio_comp.compute;
			if (compute->has_requests()) {
				prio_compute prio_comp(scheduler->get_io(), compute);
				user_computes.push(prio_comp);
			}
			else {
				// This user compute no longer needs to stay in the priority
				// queue. We can decrease its ref count.
				compute->dec_ref();
			}
		}
	}
	return num;
}

/**
 * This I/O scheduler is to favor maximizing throughput.
 * Therefore, it processes all user tasks together to potentially increase
 * the page cache hit rate.
 */
class throughput_comp_io_scheduler: public comp_io_scheduler
{
	// The scheduler process user computes in batches. It reads all
	// available user computes in the beginning of a batch and then processes
	// them. A batch ends if the scheduler processes all of the user computes.
	size_t batch_num;
	comp_io_schedule_queue<std::priority_queue<prio_compute,
		std::vector<prio_compute>, forward_comp_prio_compute> > forward_queue;
	comp_io_schedule_queue<std::priority_queue<prio_compute,
		std::vector<prio_compute>, backward_comp_prio_compute> > backward_queue;
public:
	throughput_comp_io_scheduler(int node_id): comp_io_scheduler(
			node_id), forward_queue(this, true), backward_queue(this, false) {
		batch_num = 0;
	}

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

class throughput_comp_io_sched_creater: public comp_io_sched_creater
{
public:
	comp_io_scheduler *create(int node_id) const {
		return new throughput_comp_io_scheduler(node_id);
	}
};

size_t throughput_comp_io_scheduler::get_requests(fifo_queue<io_request> &reqs)
{
	size_t ret;
	if (graph_conf.get_elevator_enabled()) {
		if (batch_num % 2 == 0) {
			ret = forward_queue.get_requests(reqs);
			if (forward_queue.is_empty())
				batch_num++;
		}
		else {
			ret = backward_queue.get_requests(reqs);
			if (backward_queue.is_empty())
				batch_num++;
		}
	}
	else
		ret = forward_queue.get_requests(reqs);
{'fraction_non_alphanumeric': 0.1005911938586429, 'fraction_numerical': 0.0031765640165887232, 'mean_word_length': 4.269177126917713, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 65, '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': '4400541', 'n_tokens_mistral': 3995, 'n_tokens_neox': 3664, 'n_words': 1111}