master_element_info_per_processor.cc
No OneTemporary
Actions

Subscribers

None

File Metadata

Created: Tue, Jun 25, 19:43

master_element_info_per_processor.cc
View Options

	/**
	* @file master_element_info_per_processor.cc
	*
	* @author Nicolas Richart <nicolas.richart@epfl.ch>
	*
	* @date Fri Mar 11 14:57:13 2016
	*
	* @brief
	*
	* @section LICENSE
	*
	* Copyright (©) 2010-2011 EPFL (Ecole Polytechnique Fédérale de Lausanne)
	* Laboratory (LSMS - Laboratoire de Simulation en Mécanique des Solides)
	*
	* Akantu is free software: you can redistribute it and/or modify it under the
	* terms of the GNU Lesser General Public License as published by the Free
	* Software Foundation, either version 3 of the License, or (at your option) any
	* later version.
	*
	* Akantu 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 Lesser General Public License for more
	* details.
	*
	* You should have received a copy of the GNU Lesser General Public License
	* along with Akantu. If not, see <http://www.gnu.org/licenses/>.
	*
	*/

	/* -------------------------------------------------------------------------- */
	#include "element_group.hh"
	#include "element_info_per_processor.hh"
	#include "element_synchronizer.hh"
	#include "mesh_utils.hh"
	#include "static_communicator.hh"
	/* -------------------------------------------------------------------------- */
	#include <algorithm>
	#include <iostream>
	#include <map>
	/* -------------------------------------------------------------------------- */

	__BEGIN_AKANTU__

	/* -------------------------------------------------------------------------- */
	MasterElementInfoPerProc::MasterElementInfoPerProc(
	ElementSynchronizer & synchronizer, UInt message_cnt, UInt root,
	ElementType type, const MeshPartition & partition)
	: ElementInfoPerProc(synchronizer, message_cnt, root, type),
	partition(partition), all_nb_local_element(nb_proc, 0),
	all_nb_ghost_element(nb_proc, 0), all_nb_element_to_send(nb_proc, 0) {
	Vector<UInt> size(5);
	size(0) = (UInt)type;

	if (type != _not_defined) {
	nb_nodes_per_element = Mesh::getNbNodesPerElement(type);
	nb_element = mesh.getNbElement(type);
	const Array<UInt> & partition_num =
	this->partition.getPartition(this->type, _not_ghost);
	const CSR<UInt> & ghost_partition =
	this->partition.getGhostPartitionCSR()(this->type, _not_ghost);

	for (UInt el = 0; el < nb_element; ++el) {
	this->all_nb_local_element[partition_num(el)]++;
	for (CSR<UInt>::const_iterator part = ghost_partition.begin(el);
	part != ghost_partition.end(el); ++part) {
	this->all_nb_ghost_element[*part]++;
	}
	this->all_nb_element_to_send[partition_num(el)] +=
	ghost_partition.getNbCols(el) + 1;
	}

	/// tag info
	std::vector<std::string> tag_names;
	this->getMeshData().getTagNames(tag_names, type);
	this->nb_tags = tag_names.size();

	size(4) = nb_tags;

	for (UInt p = 0; p < nb_proc; ++p) {
	if (p != root) {
	size(1) = this->all_nb_local_element[p];
	size(2) = this->all_nb_ghost_element[p];
	size(3) = this->all_nb_element_to_send[p];
	AKANTU_DEBUG_INFO(
	"Sending connectivities informations to proc "
	<< p << " TAG("
	<< Tag::genTag(this->rank, this->message_count, Tag::_SIZES)
	<< ")");
	comm.send(size, p,
	Tag::genTag(this->rank, this->message_count, Tag::_SIZES));
	} else {
	this->nb_local_element = this->all_nb_local_element[p];
	this->nb_ghost_element = this->all_nb_ghost_element[p];
	}
	}
	} else {
	for (UInt p = 0; p < this->nb_proc; ++p) {
	if (p != this->root) {
	AKANTU_DEBUG_INFO(
	"Sending empty connectivities informations to proc "
	<< p << " TAG("
	<< Tag::genTag(this->rank, this->message_count, Tag::_SIZES)
	<< ")");
	comm.send(size, p,
	Tag::genTag(this->rank, this->message_count, Tag::_SIZES));
	}
	}
	}
	}

	/* ------------------------------------------------------------------------ */
	void MasterElementInfoPerProc::synchronizeConnectivities() {
	const Array<UInt> & partition_num =
	this->partition.getPartition(this->type, _not_ghost);
	const CSR<UInt> & ghost_partition =
	this->partition.getGhostPartitionCSR()(this->type, _not_ghost);

	std::vector<Array<UInt>> buffers(this->nb_proc);

	auto conn_it = this->mesh.getConnectivity(this->type, _not_ghost)
	.begin(this->nb_nodes_per_element);
	auto conn_end = this->mesh.getConnectivity(this->type, _not_ghost)
	.end(this->nb_nodes_per_element);

	/// copying the local connectivity
	auto part_it = partition_num.begin();
	for (; conn_it != conn_end; ++conn_it, ++part_it) {
	const auto & conn = *conn_it;
	for (UInt i = 0; i < conn.size(); ++i) {
	buffers[*part_it].push_back(conn[i]);
	}
	}

	/// copying the connectivity of ghost element
	conn_it = this->mesh.getConnectivity(this->type, _not_ghost)
	.begin(this->nb_nodes_per_element);
	for (UInt el = 0; conn_it != conn_end; ++el, ++conn_it) {
	for (auto part = ghost_partition.begin(el); part != ghost_partition.end(el);
	++part) {
	UInt proc = *part;
	const Vector<UInt> & conn = *conn_it;
	for (UInt i = 0; i < conn.size(); ++i) {
	buffers[proc].push_back(conn[i]);
	}
	}
	}

	#ifndef AKANTU_NDEBUG
	for (UInt p = 0; p < this->nb_proc; ++p) {
	UInt size = this->nb_nodes_per_element *
	(this->all_nb_local_element[p] + this->all_nb_ghost_element[p]);
	AKANTU_DEBUG_ASSERT(
	buffers[p].getSize() == size,
	"The connectivity data packed in the buffer are not correct");
	}
	#endif

	/// send all connectivity and ghost information to all processors
	std::vector<CommunicationRequest> requests;
	for (UInt p = 0; p < this->nb_proc; ++p) {
	if (p != this->root) {
	AKANTU_DEBUG_INFO(
	"Sending connectivities to proc "
	<< p << " TAG("
	<< Tag::genTag(this->rank, this->message_count, Tag::_CONNECTIVITY)
	<< ")");
	requests.push_back(comm.asyncSend(
	buffers[p], p,
	Tag::genTag(this->rank, this->message_count, Tag::_CONNECTIVITY)));
	}
	}

	Array<UInt> & old_nodes = this->getNodesGlobalIds();

	/// create the renumbered connectivity
	AKANTU_DEBUG_INFO("Renumbering local connectivities");
	MeshUtils::renumberMeshNodes(mesh, buffers[root], all_nb_local_element[root],
	all_nb_ghost_element[root], type, old_nodes);

	comm.waitAll(requests);
	comm.freeCommunicationRequest(requests);
	}

	/* ------------------------------------------------------------------------ */
	void MasterElementInfoPerProc::synchronizePartitions() {
	const Array<UInt> & partition_num =
	this->partition.getPartition(this->type, _not_ghost);
	const CSR<UInt> & ghost_partition =
	this->partition.getGhostPartitionCSR()(this->type, _not_ghost);

	std::vector<Array<UInt>> buffers(this->partition.getNbPartition());

	/// splitting the partition information to send them to processors
	Vector<UInt> count_by_proc(nb_proc, 0);
	for (UInt el = 0; el < nb_element; ++el) {
	UInt proc = partition_num(el);
	buffers[proc].push_back(ghost_partition.getNbCols(el));

	UInt i(0);
	for (CSR<UInt>::const_iterator part = ghost_partition.begin(el);
	part != ghost_partition.end(el); ++part, ++i) {
	buffers[proc].push_back(*part);
	}
	}

	for (UInt el = 0; el < nb_element; ++el) {
	UInt i(0);
	for (CSR<UInt>::const_iterator part = ghost_partition.begin(el);
	part != ghost_partition.end(el); ++part, ++i) {
	buffers[*part].push_back(partition_num(el));
	}
	}

	#ifndef AKANTU_NDEBUG
	for (UInt p = 0; p < this->nb_proc; ++p) {
	AKANTU_DEBUG_ASSERT(
	buffers[p].getSize() ==
	(this->all_nb_ghost_element[p] + this->all_nb_element_to_send[p]),
	"Data stored in the buffer are most probably wrong");
	}
	#endif

	std::vector<CommunicationRequest> requests;
	/// last data to compute the communication scheme
	for (UInt p = 0; p < this->nb_proc; ++p) {
	if (p != this->root) {
	AKANTU_DEBUG_INFO(
	"Sending partition informations to proc "
	<< p << " TAG("
	<< Tag::genTag(this->rank, this->message_count, Tag::_PARTITIONS)
	<< ")");
	requests.push_back(comm.asyncSend(
	buffers[p], p,
	Tag::genTag(this->rank, this->message_count, Tag::_PARTITIONS)));
	}
	}

	if (Mesh::getSpatialDimension(this->type) ==
	this->mesh.getSpatialDimension()) {
	AKANTU_DEBUG_INFO("Creating communications scheme");
	this->fillCommunicationScheme(buffers[this->rank]);
	}

	comm.waitAll(requests);
	comm.freeCommunicationRequest(requests);
	}

	/* -------------------------------------------------------------------------- */
	void MasterElementInfoPerProc::synchronizeTags() {
	AKANTU_DEBUG_IN();

	if (this->nb_tags == 0) {
	AKANTU_DEBUG_OUT();
	return;
	}

	UInt mesh_data_sizes_buffer_length;
	MeshData & mesh_data = this->getMeshData();

	/// tag info
	std::vector<std::string> tag_names;
	mesh_data.getTagNames(tag_names, type);
	// Make sure the tags are sorted (or at least not in random order),
	// because they come from a map !!
	std::sort(tag_names.begin(), tag_names.end());

	// Sending information about the tags in mesh_data: name, data type and
	// number of components of the underlying array associated to the current
	// type
	DynamicCommunicationBuffer mesh_data_sizes_buffer;
	std::vector<std::string>::const_iterator names_it = tag_names.begin();
	std::vector<std::string>::const_iterator names_end = tag_names.end();
	for (; names_it != names_end; ++names_it) {
	mesh_data_sizes_buffer << *names_it;
	mesh_data_sizes_buffer << mesh_data.getTypeCode(*names_it);
	mesh_data_sizes_buffer << mesh_data.getNbComponent(*names_it, type);
	}

	mesh_data_sizes_buffer_length = mesh_data_sizes_buffer.getSize();
	AKANTU_DEBUG_INFO(
	"Broadcasting the size of the information about the mesh data tags: ("
	<< mesh_data_sizes_buffer_length << ").");
	comm.broadcast(mesh_data_sizes_buffer_length, root);
	AKANTU_DEBUG_INFO(
	"Broadcasting the information about the mesh data tags, addr "
	<< (void *)mesh_data_sizes_buffer.storage());

	if (mesh_data_sizes_buffer_length != 0)
	comm.broadcast(mesh_data_sizes_buffer, root);

	if (mesh_data_sizes_buffer_length != 0) {
	// Sending the actual data to each processor
	DynamicCommunicationBuffer * buffers =
	new DynamicCommunicationBuffer[nb_proc];
	std::vector<std::string>::const_iterator names_it = tag_names.begin();
	std::vector<std::string>::const_iterator names_end = tag_names.end();

	// Loop over each tag for the current type
	for (; names_it != names_end; ++names_it) {
	// Type code of the current tag (i.e. the tag named *names_it)
	this->fillTagBuffer(buffers, *names_it);
	}

	std::vector<CommunicationRequest> requests;
	for (UInt p = 0; p < nb_proc; ++p) {
	if (p != root) {
	AKANTU_DEBUG_INFO("Sending "
	<< buffers[p].getSize()
	<< " bytes of mesh data to proc " << p << " TAG("
	<< Tag::genTag(this->rank, this->message_count,
	Tag::_MESH_DATA)
	<< ")");

	requests.push_back(comm.asyncSend(
	buffers[p], p,
	Tag::genTag(this->rank, this->message_count, Tag::_MESH_DATA)));
	}
	}

	names_it = tag_names.begin();
	// Loop over each tag for the current type
	for (; names_it != names_end; ++names_it) {
	// Reinitializing the mesh data on the master
	this->fillMeshData(buffers[root], *names_it,
	mesh_data.getTypeCode(*names_it),
	mesh_data.getNbComponent(*names_it, type));
	}

	comm.waitAll(requests);
	comm.freeCommunicationRequest(requests);
	requests.clear();
	delete[] buffers;
	}

	AKANTU_DEBUG_OUT();
	}

	/* -------------------------------------------------------------------------- */
	template <typename T>
	void MasterElementInfoPerProc::fillTagBufferTemplated(
	DynamicCommunicationBuffer * buffers, const std::string & tag_name) {
	MeshData & mesh_data = this->getMeshData();

	const Array<T> & data = mesh_data.getElementalDataArray<T>(tag_name, type);
	const Array<UInt> & partition_num =
	this->partition.getPartition(this->type, _not_ghost);
	const CSR<UInt> & ghost_partition =
	this->partition.getGhostPartitionCSR()(this->type, _not_ghost);

	// Not possible to use the iterator because it potentially triggers the
	// creation of complex
	// type templates (such as akantu::Vector< std::vector<Element> > which don't
	// implement the right interface
	// (e.g. operator<< in that case).
	// typename Array<T>::template const_iterator< Vector<T> > data_it =
	// data.begin(data.getNbComponent());
	// typename Array<T>::template const_iterator< Vector<T> > data_end =
	// data.end(data.getNbComponent());

	const T * data_it = data.storage();
	const T * data_end = data.storage() + data.getSize() * data.getNbComponent();
	const UInt * part = partition_num.storage();

	/// copying the data, element by element
	for (; data_it != data_end; ++part) {
	for (UInt j(0); j < data.getNbComponent(); ++j, ++data_it) {
	buffers[part] << data_it;
	}
	}

	data_it = data.storage();
	/// copying the data for the ghost element
	for (UInt el(0); data_it != data_end;
	data_it += data.getNbComponent(), ++el) {
	CSR<UInt>::const_iterator it = ghost_partition.begin(el);
	CSR<UInt>::const_iterator end = ghost_partition.end(el);
	for (; it != end; ++it) {
	UInt proc = *it;
	for (UInt j(0); j < data.getNbComponent(); ++j) {
	buffers[proc] << data_it[j];
	}
	}
	}
	}

	/* -------------------------------------------------------------------------- */
	void MasterElementInfoPerProc::fillTagBuffer(
	DynamicCommunicationBuffer * buffers, const std::string & tag_name) {
	MeshData & mesh_data = this->getMeshData();

	#define AKANTU_DISTRIBUTED_SYNHRONIZER_TAG_DATA(r, extra_param, elem) \
	case BOOST_PP_TUPLE_ELEM(2, 0, elem): { \
	this->fillTagBufferTemplated<BOOST_PP_TUPLE_ELEM(2, 1, elem)>(buffers, \
	tag_name); \
	break; \
	}

	MeshDataTypeCode data_type_code = mesh_data.getTypeCode(tag_name);
	switch (data_type_code) {
	BOOST_PP_SEQ_FOR_EACH(AKANTU_DISTRIBUTED_SYNHRONIZER_TAG_DATA, ,
	AKANTU_MESH_DATA_TYPES)
	default:
	AKANTU_DEBUG_ERROR("Could not obtain the type of tag" << tag_name << "!");
	break;
	}
	#undef AKANTU_DISTRIBUTED_SYNHRONIZER_TAG_DATA
	}

	/* -------------------------------------------------------------------------- */
	void MasterElementInfoPerProc::synchronizeGroups() {
	AKANTU_DEBUG_IN();

	DynamicCommunicationBuffer * buffers =
	new DynamicCommunicationBuffer[nb_proc];

	typedef std::vector<std::vector<std::string>> ElementToGroup;
	ElementToGroup element_to_group;
	element_to_group.resize(nb_element);

	GroupManager::const_element_group_iterator egi = mesh.element_group_begin();
	GroupManager::const_element_group_iterator ege = mesh.element_group_end();
	for (; egi != ege; ++egi) {
	ElementGroup & eg = *(egi->second);

	std::string name = egi->first;

	ElementGroup::const_element_iterator eit =
	eg.element_begin(type, _not_ghost);
	ElementGroup::const_element_iterator eend =
	eg.element_end(type, _not_ghost);
	for (; eit != eend; ++eit) {
	element_to_group[*eit].push_back(name);
	}

	eit = eg.element_begin(type, _not_ghost);
	if (eit != eend)
	const_cast<Array<UInt> &>(eg.getElements(type)).empty();
	}

	const Array<UInt> & partition_num =
	this->partition.getPartition(this->type, _not_ghost);
	const CSR<UInt> & ghost_partition =
	this->partition.getGhostPartitionCSR()(this->type, _not_ghost);

	/// preparing the buffers
	const UInt * part = partition_num.storage();

	/// copying the data, element by element
	ElementToGroup::const_iterator data_it = element_to_group.begin();
	ElementToGroup::const_iterator data_end = element_to_group.end();
	for (; data_it != data_end; ++part, ++data_it) {
	buffers[part] << data_it;
	}

	data_it = element_to_group.begin();
	/// copying the data for the ghost element
	for (UInt el(0); data_it != data_end; ++data_it, ++el) {
	CSR<UInt>::const_iterator it = ghost_partition.begin(el);
	CSR<UInt>::const_iterator end = ghost_partition.end(el);
	for (; it != end; ++it) {
	UInt proc = *it;
	buffers[proc] << *data_it;
	}
	}

	std::vector<CommunicationRequest> requests;
	for (UInt p = 0; p < this->nb_proc; ++p) {
	if (p == this->rank)
	continue;
	AKANTU_DEBUG_INFO("Sending element groups to proc "
	<< p << " TAG("
	<< Tag::genTag(this->rank, p, Tag::_ELEMENT_GROUP)
	<< ")");
	requests.push_back(comm.asyncSend(
	buffers[p], p, Tag::genTag(this->rank, p, Tag::_ELEMENT_GROUP)));
	}

	this->fillElementGroupsFromBuffer(buffers[this->rank]);

	comm.waitAll(requests);
	comm.freeCommunicationRequest(requests);
	requests.clear();
	delete[] buffers;

	AKANTU_DEBUG_OUT();
	}

	/* -------------------------------------------------------------------------- */

	__END_AKANTU__

master_element_info_per_processor.ccNo OneTemporaryActions

File Metadata

master_element_info_per_processor.ccView Options

Event Timeline

master_element_info_per_processor.cc
No OneTemporary
Actions

master_element_info_per_processor.cc
View Options