<commit_before>/*
 * Copyright 2016 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 <unordered_set>
#include <unordered_map>

#include "common.h"
#include "thread.h"

#include "local_mem_buffer.h"
#include "matrix_config.h"
#include "matrix_stats.h"
#include "local_matrix_store.h"
#include "materialize.h"
#include "dense_matrix.h"
#include "EM_object.h"
#include "EM_dense_matrix.h"
#include "mapply_matrix_store.h"
#include "sink_matrix.h"

typedef std::vector<size_t> mat_id_set;

namespace std
{

template<>
struct hash<mat_id_set>
{
	size_t operator() (const mat_id_set &set) const {
		size_t id = 0;
		for (size_t i = 0; i < set.size(); i++)
			id = id * 10 + set[i];
		return id;
	}
};

template<>
struct equal_to<mat_id_set>
{
	bool operator()(const mat_id_set &s1, const mat_id_set &s2) const {
		if (s1.size() != s2.size())
			return false;

		for (size_t i = 0; i < s1.size(); i++)
			if (s1[i] != s2[i])
				return false;
		return true;
	}
};

}

namespace fm
{

namespace detail
{

namespace
{

class mapply_task: public thread_task
{
	const std::vector<matrix_store::const_ptr> &mats;
	const std::vector<matrix_store::ptr> &out_mats;
	size_t portion_idx;
	const portion_mapply_op &op;
	bool one_portion;
public:
	mapply_task(const std::vector<matrix_store::const_ptr> &_mats,
			size_t portion_idx, const portion_mapply_op &_op,
			const std::vector<matrix_store::ptr> &_out_mats): mats(
				_mats), out_mats(_out_mats), op(_op) {
		this->portion_idx = portion_idx;
		this->one_portion = mats.front()->get_num_portions() == 1;
	}

	void run();
};

void mapply_task::run()
{
	std::vector<detail::local_matrix_store::const_ptr> local_stores(
			mats.size());
	std::vector<detail::local_matrix_store::ptr> local_out_stores(
			out_mats.size());
	int node_id = thread::get_curr_thread()->get_node_id();
	for (size_t j = 0; j < mats.size(); j++) {
		local_stores[j] = mats[j]->get_portion(portion_idx);
		if (local_stores[j] && local_stores[j]->get_node_id() >= 0
				&& !one_portion)
			assert(node_id == local_stores[j]->get_node_id());
	}
	for (size_t j = 0; j < out_mats.size(); j++) {
		local_out_stores[j] = out_mats[j]->get_portion(portion_idx);
		if (local_out_stores[j] && local_out_stores[j]->get_node_id() >= 0
				&& !one_portion)
			assert(node_id == local_out_stores[j]->get_node_id());
	}

	if (local_out_stores.empty())
		op.run(local_stores);
	else if (local_out_stores.size() == 1)
		op.run(local_stores, *local_out_stores[0]);
	else
		op.run(local_stores, local_out_stores);
}

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

	static const size_t INVALID_TASK = std::numeric_limits<size_t>::max();

	virtual size_t get_portion_idx() = 0;
	virtual int get_node_id() const = 0;
};

/*
 * This dispatches the computation tasks for all threads in a machine,
 * so it is shared by all threads.
 */
class smp_task_queue: public mem_task_queue
{
	std::atomic<size_t> curr_portion;
	size_t num_portions;
public:<commit_msg>[Matrix]: keep some assert() for debugging only.<commit_after>/*
 * Copyright 2016 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 <unordered_set>
#include <unordered_map>

#include "common.h"
#include "thread.h"

#include "local_mem_buffer.h"
#include "matrix_config.h"
#include "matrix_stats.h"
#include "local_matrix_store.h"
#include "materialize.h"
#include "dense_matrix.h"
#include "EM_object.h"
#include "EM_dense_matrix.h"
#include "mapply_matrix_store.h"
#include "sink_matrix.h"

typedef std::vector<size_t> mat_id_set;

namespace std
{

template<>
struct hash<mat_id_set>
{
	size_t operator() (const mat_id_set &set) const {
		size_t id = 0;
		for (size_t i = 0; i < set.size(); i++)
			id = id * 10 + set[i];
		return id;
	}
};

template<>
struct equal_to<mat_id_set>
{
	bool operator()(const mat_id_set &s1, const mat_id_set &s2) const {
		if (s1.size() != s2.size())
			return false;

		for (size_t i = 0; i < s1.size(); i++)
			if (s1[i] != s2[i])
				return false;
		return true;
	}
};

}

namespace fm
{

namespace detail
{

namespace
{

class mapply_task: public thread_task
{
	const std::vector<matrix_store::const_ptr> &mats;
	const std::vector<matrix_store::ptr> &out_mats;
	size_t portion_idx;
	const portion_mapply_op &op;
	bool one_portion;
public:
	mapply_task(const std::vector<matrix_store::const_ptr> &_mats,
			size_t portion_idx, const portion_mapply_op &_op,
			const std::vector<matrix_store::ptr> &_out_mats): mats(
				_mats), out_mats(_out_mats), op(_op) {
		this->portion_idx = portion_idx;
		this->one_portion = mats.front()->get_num_portions() == 1;
	}

	void run();
};

void mapply_task::run()
{
	std::vector<detail::local_matrix_store::const_ptr> local_stores(
			mats.size());
	std::vector<detail::local_matrix_store::ptr> local_out_stores(
			out_mats.size());
	int node_id = thread::get_curr_thread()->get_node_id();
	for (size_t j = 0; j < mats.size(); j++) {
		local_stores[j] = mats[j]->get_portion(portion_idx);
#ifdef MATRIX_DEBUG
		if (local_stores[j] && local_stores[j]->get_node_id() >= 0
				&& !one_portion)
			assert(node_id == local_stores[j]->get_node_id());
#endif
	}
	for (size_t j = 0; j < out_mats.size(); j++) {
		local_out_stores[j] = out_mats[j]->get_portion(portion_idx);
#ifdef MATRIX_DEBUG
		if (local_out_stores[j] && local_out_stores[j]->get_node_id() >= 0
				&& !one_portion)
			assert(node_id == local_out_stores[j]->get_node_id());
#endif
	}

	if (local_out_stores.empty())
		op.run(local_stores);
	else if (local_out_stores.size() == 1)
		op.run(local_stores, *local_out_stores[0]);
	else
		op.run(local_stores, local_out_stores);
}

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

	static const size_t INVALID_TASK = std::numeric_limits<size_t>::max();

	virtual size_t get_portion_idx() = 0;
	virtual int get_node_id() const = 0;
};

/*
 * This dispatches the computation tasks for all threads in a machine,
 * so it is shared by all threads.
 */
class smp_task_queue: public mem_task_queue
{
	std::atomic<size_t> curr_portion;
	size_t num_portions;
public:
{'fraction_non_alphanumeric': 0.1381569965870307, 'fraction_numerical': 0.007918088737201364, 'mean_word_length': 4.93679092382496, 'pattern_counts': {'":': 0, '<': 47, '<?xml version=': 0, '>': 63, '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': '733378', 'n_tokens_mistral': 2907, 'n_tokens_neox': 2734, 'n_words': 671}