650 const U& x, std::array<std::int64_t, 2> shape,
651 std::int64_t rank_offset)
653 assert(rank_offset >= 0 or x.empty());
654 using T = std::ranges::range_value_t<U>;
657 assert(shape[1] > 0);
661 assert(x.size() % shape[1] == 0);
662 const std::int64_t shape0_local = x.size() / shape[1];
670 auto [send_dest, num_items_per_send_dest, pos_to_neigh_rank]
671 = impl::postoffice_plan(
size,
rank, shape0_local, shape[0], rank_offset);
676 std::vector<std::tuple<int, std::int64_t, std::int32_t>> src_to_index;
677 for (std::size_t i = 0; i < indices.size(); ++i)
679 std::int64_t idx = indices[i];
680 if (idx >= rank_offset and idx < rank_offset + shape0_local)
683 src_to_index.push_back({src, idx, i});
689 std::vector<std::int32_t> perm(src_to_index.size());
690 std::iota(perm.begin(), perm.end(), 0);
692 {
return std::get<0>(src_to_index[i]); });
693 std::vector<std::tuple<int, std::int64_t, std::int32_t>> sorted(
694 src_to_index.size());
695 for (std::size_t i = 0; i < perm.size(); ++i)
696 sorted[i] = src_to_index[perm[i]];
697 src_to_index = std::move(sorted);
702 std::vector<std::int32_t> num_items_per_src;
703 std::vector<int> src;
705 auto it = src_to_index.begin();
706 while (it != src_to_index.end())
708 src.push_back(std::get<0>(*it));
709 auto it1 = std::ranges::find_if(it, src_to_index.end(),
710 [r = src.back()](
auto& idx)
711 { return std::get<0>(idx) != r; });
712 num_items_per_src.push_back(std::ranges::distance(it, it1));
722 comm, send_dest,
static_cast<int>(tag::consensus_nbx), src,
723 static_cast<int>(tag::consensus_nbx) + 1);
725 "Neighbourhood destination ranks from post office in "
726 "distribute_data (rank, num dests, num dests/mpi_size): {}, {}, {}",
727 rank, dest.size(),
static_cast<double>(dest.size()) /
size);
731 auto [post_indices, post_x] = impl::postoffice_exchange(
732 comm, x, {shape[0], shape[1]}, rank_offset, send_dest,
733 num_items_per_send_dest, pos_to_neigh_rank, post_src);
734 assert(post_indices.size() == post_x.size() / shape[1]);
738 MPI_Comm neigh_comm0;
739 int err = MPI_Dist_graph_create_adjacent(
740 comm, dest.size(), dest.data(), MPI_UNWEIGHTED, src.size(), src.data(),
741 MPI_UNWEIGHTED, MPI_INFO_NULL,
false, &neigh_comm0);
745 std::vector<int> num_items_recv(dest.size());
746 num_items_per_src.reserve(1);
747 num_items_recv.reserve(1);
748 err = MPI_Neighbor_alltoall(num_items_per_src.data(), 1, MPI_INT,
749 num_items_recv.data(), 1, MPI_INT, neigh_comm0);
753 std::vector<std::int32_t> send_disp{0};
754 std::partial_sum(num_items_per_src.begin(), num_items_per_src.end(),
755 std::back_inserter(send_disp));
756 std::vector<std::int32_t> recv_disp = {0};
757 std::partial_sum(num_items_recv.begin(), num_items_recv.end(),
758 std::back_inserter(recv_disp));
762 assert(send_disp.back() ==
static_cast<int>(src_to_index.size()));
763 std::vector<std::int64_t> send_buffer_index(src_to_index.size());
764 std::ranges::transform(src_to_index, send_buffer_index.begin(),
765 [](
auto x) { return std::get<1>(x); });
768 std::vector<std::int64_t> recv_buffer_index(recv_disp.back());
769 err = MPI_Neighbor_alltoallv(
770 send_buffer_index.data(), num_items_per_src.data(), send_disp.data(),
771 MPI_INT64_T, recv_buffer_index.data(), num_items_recv.data(),
772 recv_disp.data(), MPI_INT64_T, neigh_comm0);
775 err = MPI_Comm_free(&neigh_comm0);
784 const std::array<std::int64_t, 2> postoffice_range
786 std::vector<std::int32_t> post_indices_map(
787 postoffice_range[1] - postoffice_range[0], -1);
788 for (std::size_t i = 0; i < post_indices.size(); ++i)
790 assert(post_indices[i] <
static_cast<int>(post_indices_map.size()));
791 post_indices_map[post_indices[i]] = i;
795 std::vector<T> send_buffer_data(shape[1] * recv_disp.back());
796 for (std::int32_t i = 0; i < recv_disp.back(); ++i)
798 std::int64_t index = recv_buffer_index[i];
799 if (index >= rank_offset and index < (rank_offset + shape0_local))
802 std::int32_t local_index = index - rank_offset;
803 std::copy_n(std::next(x.begin(), shape[1] * local_index), shape[1],
804 std::next(send_buffer_data.begin(), shape[1] * i));
809 std::int64_t local_index = index - postoffice_range[0];
810 std::int32_t pos = post_indices_map[local_index];
812 std::copy_n(std::next(post_x.begin(), shape[1] * pos), shape[1],
813 std::next(send_buffer_data.begin(), shape[1] * i));
817 err = MPI_Dist_graph_create_adjacent(
818 comm, src.size(), src.data(), MPI_UNWEIGHTED, dest.size(), dest.data(),
819 MPI_UNWEIGHTED, MPI_INFO_NULL,
false, &neigh_comm0);
822 MPI_Datatype compound_type0;
824 MPI_Type_commit(&compound_type0);
826 std::vector<T> recv_buffer_data(shape[1] * send_disp.back());
827 err = MPI_Neighbor_alltoallv(
828 send_buffer_data.data(), num_items_recv.data(), recv_disp.data(),
829 compound_type0, recv_buffer_data.data(), num_items_per_src.data(),
830 send_disp.data(), compound_type0, neigh_comm0);
833 err = MPI_Type_free(&compound_type0);
835 err = MPI_Comm_free(&neigh_comm0);
838 std::vector<std::int32_t> index_pos_to_buffer(indices.size(), -1);
839 for (std::size_t i = 0; i < src_to_index.size(); ++i)
840 index_pos_to_buffer[std::get<2>(src_to_index[i])] = i;
843 std::vector<T> x_new(shape[1] * indices.size());
844 for (std::size_t i = 0; i < indices.size(); ++i)
846 const std::int64_t index = indices[i];
847 if (index >= rank_offset and index < (rank_offset + shape0_local))
850 std::int64_t local_index = index - rank_offset;
851 std::copy_n(std::next(x.begin(), shape[1] * local_index), shape[1],
852 std::next(x_new.begin(), shape[1] * i));
854 else if (std::int32_t pos = index_pos_to_buffer[i]; pos != -1)
859 std::copy_n(std::next(recv_buffer_data.begin(), shape[1] * pos), shape[1],
860 std::next(x_new.begin(), shape[1] * i));
865 std::int64_t local_index = index - postoffice_range[0];
866 std::int32_t bag_pos = post_indices_map[local_index];
867 assert(bag_pos != -1);
868 std::copy_n(std::next(post_x.begin(), shape[1] * bag_pos), shape[1],
869 std::next(x_new.begin(), shape[1] * i));