<commit_before>#ifndef __SPARSE_MATRIX_H__
#define __SPARSE_MATRIX_H__

/*
 * Copyright 2014 Open Connectome Project (http://openconnecto.me)
 * Written by <NAME> (<EMAIL>)
 *
 * This file is part of FlashMatrix.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <memory>

#include "FGlib.h"
#include "matrix_io.h"
#include "io_interface.h"
#include "mem_vector.h"

namespace fm
{

class compute_task
{
public:
	typedef std::shared_ptr<compute_task> ptr;

	virtual ~compute_task() {
	}

	virtual void run(char *buf, size_t size) = 0;
	virtual safs::io_request get_request() const = 0;
};

class task_creator
{
public:
	typedef std::shared_ptr<task_creator> ptr;

	virtual compute_task::ptr create(const matrix_io &) const = 0;
};

/*
 * This task performs computation on a sparse matrix in the FlashGraph format.
 */
class fg_row_compute_task: public compute_task
{
	matrix_io io;
	off_t off;
	char *buf;
	size_t buf_size;
public:
	fg_row_compute_task(const matrix_io &_io): io(_io) {
		off_t orig_off = io.get_loc().get_offset();
		off = ROUND_PAGE(orig_off);
		buf_size = ROUNDUP_PAGE(orig_off - off + io.get_size());
		buf = (char *) valloc(buf_size);
	}

	~fg_row_compute_task() {
		free(buf);
	}
	virtual void run(char *buf, size_t size);
	virtual void run_on_row(const fg::ext_mem_undirected_vertex &v) = 0;
	virtual safs::io_request get_request() const {
		return safs::io_request(buf, safs::data_loc_t(io.get_loc().get_file_id(),
					off), buf_size, READ);
	}
};

/*
 * This task performs matrix vector multiplication on a sparse matrix
 * in the FlashGraph format.
 */
template<class T>
class fg_row_multiply_task: public fg_row_compute_task
{
	const type_mem_vector<T> &input;
	type_mem_vector<T> &output;
public:
	fg_row_multiply_task(const type_mem_vector<T> &_input, type_mem_vector<T> &_output,
			const matrix_io &_io): fg_row_compute_task(_io), input(
				_input), output(_output) {
	}

	void run_on_row(const fg::ext_mem_undirected_vertex &v);
};

template<class T>
void fg_row_multiply_task<T>::run_on_row(const fg::ext_mem_undirected_vertex &v)
{
	T res = 0;
	for (size_t i = 0; i < v.get_num_edges(); i++) {
		fg::vertex_id_t id = v.get_neighbor(i);
		res += input.get(id);
	}
	output.set(v.get_id(), res);
}

/*
 * This task performs matrix vector multiplication on a sparse matrix in
 * a native format with 2D partitioning.
 */
template<class T>
class block_multiply_task: public compute_task
{
	matrix_io io;
	block_2d_size block_size;
	off_t off;
	char *buf;
	size_t buf_size;

	const type_mem_vector<T> &input;
	type_mem_vector<T> &output;

	void run_on_row_part(const sparse_row_part &rpart, size_t start_row_idx,
			size_t start_col_idx) {
		size_t row_idx = start_row_idx + rpart.get_rel_row_idx();
		size_t num_non_zeros = rpart.get_num_non_zeros();
		T sum = 0;
		for (size_t i = 0; i < num_non_zeros; i++) {
			size_t col_idx = start_col_idx + rpart.get_rel_col_idx(i);
			sum += input.get(col_idx);
		}
		output.set(row_idx, output.get(row_idx) + sum);
	}

	void run_on_block(const sparse_block_2d &block) {
		row_part_iterator it = block.get_iterator();
		size_t start_col_idx
			= block.get_block_col_idx() * block_size.get_num_cols();
		size_t start_row_idx
			= block.get_block_row_idx() * block_size.get_num_rows();
		while (it.has_next())
			run_on_row_part(it.next(), start_row_idx, start_col_idx);
	}
public:
	block_multiply_task(const type_mem_vector<T> &_input,
			type_mem_vector<T> &_output, const matrix_io &_io,
			const block_2d_size &_block_size): io(_io), block_size(
				_block_size), input(_input), output(_output) {
		off_t orig_off = io.get_loc().get_offset();
		off = ROUND_PAGE(orig_off);
		buf_size = ROUNDUP_PAGE(orig_off - off + io.get_size());
		buf = (char *) valloc(buf_size);
	}

	~block_multiply_task() {
		free(buf);
	}

	virtual void run(char *buf, size_t size) {
		block_row_iterator it((const sparse_block_2d *) buf,
				(const sparse_block_2d *) (buf + size));
		while (it.has_next())
			run_on_block(it.next());
	}

	virtual safs::io_request get_request() const {
		return safs::io_request(buf, safs::data_loc_t(io.get_loc().get_file_id(),
					off), buf_size, READ);
	}
};

template<class T>
class fg_row_multiply_creator: public task_creator
{
	const type_mem_vector<T> &input;
	type_mem_vector<T> &output;

	fg_row_multiply_creator(const type_mem_vector<T> &_input,
			type_mem_vector<T> &_output): input(_input), output(_output) {
	}
public:
	static task_creator::ptr create(const type_mem_vector<T> &_input,
			type_mem_vector<T> &_output) {
		return task_creator::ptr(new fg_row_multiply_creator<T>(_input, _output));
	}
<commit_msg>[Matrix]: fix a bug in SpMV on 2D-partitioned matrix.<commit_after>#ifndef __SPARSE_MATRIX_H__
#define __SPARSE_MATRIX_H__

/*
 * Copyright 2014 Open Connectome Project (http://openconnecto.me)
 * Written by <NAME> (<EMAIL>)
 *
 * This file is part of FlashMatrix.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <memory>

#include "FGlib.h"
#include "matrix_io.h"
#include "io_interface.h"
#include "mem_vector.h"

namespace fm
{

class compute_task
{
public:
	typedef std::shared_ptr<compute_task> ptr;

	virtual ~compute_task() {
	}

	virtual void run(char *buf, size_t size) = 0;
	virtual safs::io_request get_request() const = 0;
};

class task_creator
{
public:
	typedef std::shared_ptr<task_creator> ptr;

	virtual compute_task::ptr create(const matrix_io &) const = 0;
};

/*
 * This task performs computation on a sparse matrix in the FlashGraph format.
 */
class fg_row_compute_task: public compute_task
{
	matrix_io io;
	off_t off;
	char *buf;
	size_t buf_size;
public:
	fg_row_compute_task(const matrix_io &_io): io(_io) {
		off_t orig_off = io.get_loc().get_offset();
		off = ROUND_PAGE(orig_off);
		buf_size = ROUNDUP_PAGE(orig_off - off + io.get_size());
		buf = (char *) valloc(buf_size);
	}

	~fg_row_compute_task() {
		free(buf);
	}
	virtual void run(char *buf, size_t size);
	virtual void run_on_row(const fg::ext_mem_undirected_vertex &v) = 0;
	virtual safs::io_request get_request() const {
		return safs::io_request(buf, safs::data_loc_t(io.get_loc().get_file_id(),
					off), buf_size, READ);
	}
};

/*
 * This task performs matrix vector multiplication on a sparse matrix
 * in the FlashGraph format.
 */
template<class T>
class fg_row_multiply_task: public fg_row_compute_task
{
	const type_mem_vector<T> &input;
	type_mem_vector<T> &output;
public:
	fg_row_multiply_task(const type_mem_vector<T> &_input, type_mem_vector<T> &_output,
			const matrix_io &_io): fg_row_compute_task(_io), input(
				_input), output(_output) {
	}

	void run_on_row(const fg::ext_mem_undirected_vertex &v);
};

template<class T>
void fg_row_multiply_task<T>::run_on_row(const fg::ext_mem_undirected_vertex &v)
{
	T res = 0;
	for (size_t i = 0; i < v.get_num_edges(); i++) {
		fg::vertex_id_t id = v.get_neighbor(i);
		res += input.get(id);
	}
	output.set(v.get_id(), res);
}

/*
 * This task performs matrix vector multiplication on a sparse matrix in
 * a native format with 2D partitioning.
 */
template<class T>
class block_multiply_task: public compute_task
{
	matrix_io io;
	block_2d_size block_size;
	off_t off;
	char *buf;
	size_t buf_size;

	const type_mem_vector<T> &input;
	type_mem_vector<T> &output;

	void run_on_row_part(const sparse_row_part &rpart, size_t start_row_idx,
			size_t start_col_idx) {
		size_t row_idx = start_row_idx + rpart.get_rel_row_idx();
		size_t num_non_zeros = rpart.get_num_non_zeros();
		T sum = 0;
		for (size_t i = 0; i < num_non_zeros; i++) {
			size_t col_idx = start_col_idx + rpart.get_rel_col_idx(i);
			sum += input.get(col_idx);
		}
		output.set(row_idx, output.get(row_idx) + sum);
	}

	void run_on_block(const sparse_block_2d &block) {
		row_part_iterator it = block.get_iterator();
		size_t start_col_idx
			= block.get_block_col_idx() * block_size.get_num_cols();
		size_t start_row_idx
			= block.get_block_row_idx() * block_size.get_num_rows();
		while (it.has_next())
			run_on_row_part(it.next(), start_row_idx, start_col_idx);
	}
public:
	block_multiply_task(const type_mem_vector<T> &_input,
			type_mem_vector<T> &_output, const matrix_io &_io,
			const block_2d_size &_block_size): io(_io), block_size(
				_block_size), input(_input), output(_output) {
		off_t orig_off = io.get_loc().get_offset();
		off = ROUND_PAGE(orig_off);
		buf_size = ROUNDUP_PAGE(orig_off - off + io.get_size());
		buf = (char *) valloc(buf_size);
	}

	~block_multiply_task() {
		free(buf);
	}

	virtual void run(char *buf, size_t size) {
		off_t orig_off = io.get_loc().get_offset();
		off_t local_off = orig_off - ROUND_PAGE(orig_off);
		assert(local_off + io.get_size() <= size);
		const sparse_block_2d *block = (const sparse_block_2d *) (buf + local_off);
		block_row_iterator it(block,
				(const sparse_block_2d *) (buf + local_off + io.get_size()));
		while (it.has_next())
			run_on_block(it.next());
	}

	virtual safs::io_request get_request() const {
		return safs::io_request(buf, safs::data_loc_t(io.get_loc().get_file_id(),
					off), buf_size, READ);
	}
};

template<class T>
class fg_row_multiply_creator: public task_creator
{
	const type_mem_vector<T> &input;
	type_mem_vector<T> &output;

	fg_row_multiply_creator(const type_mem_vector<T> &_input,
			type_mem_vector<T> &_output): input(_input), output(_output) {
	}
public:
	static task_creator::ptr create(const type_mem_vector<T> &_input,
			type_mem_vector<T> &_output) {
		return task_creator::ptr(new fg_row_multiply_creator<T>(_input, _output));
	}

{'fraction_non_alphanumeric': 0.11964268744654566, 'fraction_numerical': 0.004371376983749882, 'mean_word_length': 5.041331802525832, 'pattern_counts': {'":': 0, '<': 58, '<?xml version=': 0, '>': 53, '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': '2061420', 'n_tokens_mistral': 4209, 'n_tokens_neox': 3970, 'n_words': 928}