<commit_before>/*
 * 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 <libgen.h>
#include <malloc.h>

#include <unordered_map>
#include <boost/math/common_factor.hpp>
#include <boost/format.hpp>

#include "safs_file.h"
#include "io_request.h"
#include "io_interface.h"
#include "in_mem_io.h"

#include "EM_vector.h"
#include "matrix_config.h"
#include "mem_worker_thread.h"
#include "local_vec_store.h"
#include "matrix_store.h"
#include "bulk_operate_ext.h"
#include "EM_dense_matrix.h"

namespace fm
{

namespace detail
{

namespace
{
/*
 * When we write data to disks, we need to have something to hold the buffer.
 * This holds the local buffer until the write completes.
 */
class portion_write_complete: public portion_compute
{
	local_buf_vec_store::const_ptr store;
public:
	portion_write_complete(local_buf_vec_store::const_ptr store) {
		this->store = store;
	}

	virtual void run(char *buf, size_t size) {
		assert(store->get_raw_arr() == buf);
		assert(store->get_length() * store->get_entry_size() == size);
	}
};

}

class EM_vec_dispatcher: public task_dispatcher
{
	const EM_vec_store &store;
	off_t portion_idx;
	pthread_spinlock_t lock;
	size_t portion_size;
public:
	EM_vec_dispatcher(const EM_vec_store &_store,
			size_t portion_size = 0): store(_store) {
		pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE);
		portion_idx = 0;
		if (portion_size == 0)
			this->portion_size = store.get_portion_size();
		else
			this->portion_size = portion_size;
	}

	virtual bool issue_task() {
		pthread_spin_lock(&lock);
		off_t global_start = portion_idx * portion_size;
		if ((size_t) global_start >= store.get_length()) {
			pthread_spin_unlock(&lock);
			return false;
		}
		size_t length = std::min(portion_size, store.get_length() - global_start);
		portion_idx++;
		pthread_spin_unlock(&lock);
		create_vec_task(global_start, length, store);
		return true;
	}

	virtual void create_vec_task(off_t global_start, size_t length,
			const EM_vec_store &from_vec) = 0;
};

class seq_writer
{
	size_t local_buf_size;	// In the number of elements
	off_t merge_end;		// In the number of elements
	local_buf_vec_store::ptr buf;
	size_t data_size_in_buf;		// In the number of elements
	EM_vec_store::ptr to_vec;
public:
	seq_writer(EM_vec_store::ptr vec, off_t write_start) {
		this->local_buf_size
			= matrix_conf.get_write_io_buf_size() / vec->get_type().get_size();
		this->to_vec = vec;
		merge_end = write_start;
		buf = local_buf_vec_store::ptr(new local_buf_vec_store(
					-1, local_buf_size, to_vec->get_type(), -1));
		data_size_in_buf = 0;
	}

	~seq_writer() {
		if (data_size_in_buf > 0)
			flush_buffer_data(true);
	}

	void flush_buffer_data(bool last);
	void append(local_vec_store::const_ptr data);
};

void seq_writer::flush_buffer_data(bool last)
{
	if (data_size_in_buf == 0)
		return;

	if (!last)
		assert((data_size_in_buf * buf->get_length()) % PAGE_SIZE == 0);
	else
		data_size_in_buf = ROUNDUP((data_size_in_buf * buf->get_entry_size()),
				PAGE_SIZE) / buf->get_entry_size();
	buf->resize(data_size_in_buf);
	const scalar_type &type = buf->get_type();
	to_vec->write_portion_async(buf, merge_end);
	merge_end += data_size_in_buf;

	if (!last)
		// The data is written asynchronously, we need to allocate
		// a new buffer.
		buf = local_buf_vec_store::ptr(new local_buf_vec_store(
					-1, local_buf_size, type, -1));
	else
		buf = NULL;
	data_size_in_buf = 0;
}

void seq_writer::append(local_vec_store::const_ptr data)
{
	assert(buf);
	// In the number of elements.
	off_t off_in_new_data = 0;
	size_t new_data_size = data->get_length();

	while (new_data_size > 0) {
		size_t copy_data_size = std::min(new_data_size,
				// The size of the available space in the buffer.
				buf->get_length() - data_size_in_buf);
		// We always copy the data to the local buffer and write
		// the local buffer to disks. The reason of doing so is to
		// make sure the size and offset of data written to disks
		// are aligned to the I/O block size.
		// TODO maybe we should remove this extra memory copy.
		memcpy(buf->get(data_size_in_buf), data->get(off_in_new_data),
				copy_data_size * buf->get_entry_size());
		data_size_in_buf += copy_data_size;
		// Update the amount of data in the incoming data buffer.
		off_in_new_data += copy_data_size;
		new_data_size -= copy_data_size;

		// If the buffer is full, we need to write it out first.
		if (data_size_in_buf == buf->get_length())
			flush_buffer_data(false);<commit_msg>[Matrix]: avoid mem copy in EM vec append.<commit_after>/*
 * 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 <libgen.h>
#include <malloc.h>

#include <unordered_map>
#include <boost/math/common_factor.hpp>
#include <boost/format.hpp>

#include "safs_file.h"
#include "io_request.h"
#include "io_interface.h"
#include "in_mem_io.h"

#include "EM_vector.h"
#include "matrix_config.h"
#include "mem_worker_thread.h"
#include "local_vec_store.h"
#include "matrix_store.h"
#include "bulk_operate_ext.h"
#include "EM_dense_matrix.h"

namespace fm
{

namespace detail
{

namespace
{
/*
 * When we write data to disks, we need to have something to hold the buffer.
 * This holds the local buffer until the write completes.
 */
class portion_write_complete: public portion_compute
{
	local_buf_vec_store::const_ptr store;
public:
	portion_write_complete(local_buf_vec_store::const_ptr store) {
		this->store = store;
	}

	virtual void run(char *buf, size_t size) {
		assert(store->get_raw_arr() == buf);
		assert(store->get_length() * store->get_entry_size() == size);
	}
};

}

class EM_vec_dispatcher: public task_dispatcher
{
	const EM_vec_store &store;
	off_t portion_idx;
	pthread_spinlock_t lock;
	size_t portion_size;
public:
	EM_vec_dispatcher(const EM_vec_store &_store,
			size_t portion_size = 0): store(_store) {
		pthread_spin_init(&lock, PTHREAD_PROCESS_PRIVATE);
		portion_idx = 0;
		if (portion_size == 0)
			this->portion_size = store.get_portion_size();
		else
			this->portion_size = portion_size;
	}

	virtual bool issue_task() {
		pthread_spin_lock(&lock);
		off_t global_start = portion_idx * portion_size;
		if ((size_t) global_start >= store.get_length()) {
			pthread_spin_unlock(&lock);
			return false;
		}
		size_t length = std::min(portion_size, store.get_length() - global_start);
		portion_idx++;
		pthread_spin_unlock(&lock);
		create_vec_task(global_start, length, store);
		return true;
	}

	virtual void create_vec_task(off_t global_start, size_t length,
			const EM_vec_store &from_vec) = 0;
};

class seq_writer
{
	size_t local_buf_size;	// In the number of elements
	off_t merge_end;		// In the number of elements
	local_buf_vec_store::ptr buf;
	size_t data_size_in_buf;		// In the number of elements
	EM_vec_store::ptr to_vec;
public:
	seq_writer(EM_vec_store::ptr vec, off_t write_start) {
		this->local_buf_size
			= matrix_conf.get_write_io_buf_size() / vec->get_type().get_size();
		this->to_vec = vec;
		merge_end = write_start;
		buf = local_buf_vec_store::ptr(new local_buf_vec_store(
					-1, local_buf_size, to_vec->get_type(), -1));
		data_size_in_buf = 0;
	}

	~seq_writer() {
		if (data_size_in_buf > 0)
			flush_buffer_data(true);
	}

	void flush_buffer_data(bool last);
	void append(local_vec_store::const_ptr data);
};

void seq_writer::flush_buffer_data(bool last)
{
	if (data_size_in_buf == 0)
		return;

	if (!last)
		assert((data_size_in_buf * buf->get_length()) % PAGE_SIZE == 0);
	else
		data_size_in_buf = ROUNDUP((data_size_in_buf * buf->get_entry_size()),
				PAGE_SIZE) / buf->get_entry_size();
	buf->resize(data_size_in_buf);
	const scalar_type &type = buf->get_type();
	to_vec->write_portion_async(buf, merge_end);
	merge_end += data_size_in_buf;

	if (!last)
		// The data is written asynchronously, we need to allocate
		// a new buffer.
		buf = local_buf_vec_store::ptr(new local_buf_vec_store(
					-1, local_buf_size, type, -1));
	else
		buf = NULL;
	data_size_in_buf = 0;
}

void seq_writer::append(local_vec_store::const_ptr data)
{
	assert(buf);
	// In the number of elements.
	off_t off_in_new_data = 0;
	size_t new_data_size = data->get_length();

	// If possible, we want to write data to disks directly
	// to avoid extra memory copy as below.
	size_t entry_size = data->get_type().get_size();
	long addr = (long) data->get_raw_arr();
	if ((data->get_length() * entry_size) % PAGE_SIZE == 0
			&& (buf->get_length() * entry_size) % PAGE_SIZE == 0
			&& addr % PAGE_SIZE == 0) {
		flush_buffer_data(false);
		to_vec->write_portion_async(data, merge_end);
		merge_end += data->get_length();
		return;
	}

	while (new_data_size > 0) {
		size_t copy_data_size = std::min(new_data_size,
				// The size of the available space in the buffer.
				buf->get_length() - data_size_in_buf);
		// We always copy the data to the local buffer and write
		// the local buffer to disks. The reason of doing so is to
		// make sure the size and offset of data written to disks
		// are aligned to the I/O block size.
		memcpy(buf->get(data_size_in_buf), data->get(off_in_new_data),
				copy_data_size * buf->get_entry_size());
		data_size_in_buf += copy_data_size;
		// Update the amount of data in the incoming data buffer.
		off_in_new_data += copy_data_size;
		new_data_size -= copy_data_size;

		// If the buffer is full, we need to write it out first.
		if (data_size_in_buf == buf->get_length())
			flush_buffer_data(false);
{'fraction_non_alphanumeric': 0.11208110391439031, 'fraction_numerical': 0.0045996432929691165, 'mean_word_length': 4.52020725388601, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 73, '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': '7208027', 'n_tokens_mistral': 4021, 'n_tokens_neox': 3760, 'n_words': 997}