648 const U& x, std::array<std::int64_t, 2> shape,
649 std::int64_t rank_offset)
651 assert(rank_offset >= 0 or x.empty());
652 using T = std::ranges::range_value_t<U>;
655 assert(shape[1] > 0);
659 assert(x.size() % shape[1] == 0);
660 const std::int64_t shape0_local = x.size() / shape[1];
668 auto [send_dest, num_items_per_send_dest, pos_to_neigh_rank]
669 = impl::postoffice_plan(
size,
rank, shape0_local, shape[0], rank_offset);
674 std::vector<std::tuple<int, std::int64_t, std::int32_t>> src_to_index;
675 for (std::size_t i = 0; i < indices.size(); ++i)
677 std::int64_t idx = indices[i];
678 if (idx >= rank_offset and idx < rank_offset + shape0_local)
681 src_to_index.push_back({src, idx, i});
687 std::vector<std::int32_t> perm(src_to_index.size());
688 std::iota(perm.begin(), perm.end(), 0);
690 {
return std::get<0>(src_to_index[i]); });
691 std::vector<std::tuple<int, std::int64_t, std::int32_t>> sorted(
692 src_to_index.size());
693 for (std::size_t i = 0; i < perm.size(); ++i)
694 sorted[i] = src_to_index[perm[i]];
695 src_to_index = std::move(sorted);
700 std::vector<std::int32_t> num_items_per_src;
701 std::vector<int> src;
703 auto it = src_to_index.begin();
704 while (it != src_to_index.end())
706 src.push_back(std::get<0>(*it));
707 auto it1 = std::ranges::find_if(it, src_to_index.end(),
708 [r = src.back()](
auto& idx)
709 { return std::get<0>(idx) != r; });
710 num_items_per_src.push_back(std::ranges::distance(it, it1));
720 comm, send_dest,
static_cast<int>(tag::consensus_nbx), src,
721 static_cast<int>(tag::consensus_nbx) + 1);
723 "Neighbourhood destination ranks from post office in "
724 "distribute_data (rank, num dests, num dests/mpi_size): {}, {}, {}",
725 rank, dest.size(),
static_cast<double>(dest.size()) /
size);
729 auto [post_indices, post_x] = impl::postoffice_exchange(
730 comm, x, {shape[0], shape[1]}, rank_offset, send_dest,
731 num_items_per_send_dest, pos_to_neigh_rank, post_src);
732 assert(post_indices.size() == post_x.size() / shape[1]);
736 MPI_Comm neigh_comm0;
737 int err = MPI_Dist_graph_create_adjacent(
738 comm, dest.size(), dest.data(), MPI_UNWEIGHTED, src.size(), src.data(),
739 MPI_UNWEIGHTED, MPI_INFO_NULL,
false, &neigh_comm0);
743 std::vector<int> num_items_recv(dest.size());
744 num_items_per_src.reserve(1);
745 num_items_recv.reserve(1);
746 err = MPI_Neighbor_alltoall(num_items_per_src.data(), 1, MPI_INT,
747 num_items_recv.data(), 1, MPI_INT, neigh_comm0);
751 std::vector<std::int32_t> send_disp{0};
752 std::partial_sum(num_items_per_src.begin(), num_items_per_src.end(),
753 std::back_inserter(send_disp));
754 std::vector<std::int32_t> recv_disp = {0};
755 std::partial_sum(num_items_recv.begin(), num_items_recv.end(),
756 std::back_inserter(recv_disp));
760 assert(send_disp.back() ==
static_cast<int>(src_to_index.size()));
761 std::vector<std::int64_t> send_buffer_index(src_to_index.size());
762 std::ranges::transform(src_to_index, send_buffer_index.begin(),
763 [](
auto x) { return std::get<1>(x); });
766 std::vector<std::int64_t> recv_buffer_index(recv_disp.back());
767 err = MPI_Neighbor_alltoallv(
768 send_buffer_index.data(), num_items_per_src.data(), send_disp.data(),
769 MPI_INT64_T, recv_buffer_index.data(), num_items_recv.data(),
770 recv_disp.data(), MPI_INT64_T, neigh_comm0);
773 err = MPI_Comm_free(&neigh_comm0);
782 const std::array<std::int64_t, 2> postoffice_range
784 std::vector<std::int32_t> post_indices_map(
785 postoffice_range[1] - postoffice_range[0], -1);
786 for (std::size_t i = 0; i < post_indices.size(); ++i)
788 assert(post_indices[i] <
static_cast<int>(post_indices_map.size()));
789 post_indices_map[post_indices[i]] = i;
793 std::vector<T> send_buffer_data(shape[1] * recv_disp.back());
794 for (std::int32_t i = 0; i < recv_disp.back(); ++i)
796 std::int64_t index = recv_buffer_index[i];
797 if (index >= rank_offset and index < (rank_offset + shape0_local))
800 std::int32_t local_index = index - rank_offset;
801 std::copy_n(std::next(x.begin(), shape[1] * local_index), shape[1],
802 std::next(send_buffer_data.begin(), shape[1] * i));
807 std::int64_t local_index = index - postoffice_range[0];
808 std::int32_t pos = post_indices_map[local_index];
810 std::copy_n(std::next(post_x.begin(), shape[1] * pos), shape[1],
811 std::next(send_buffer_data.begin(), shape[1] * i));
815 err = MPI_Dist_graph_create_adjacent(
816 comm, src.size(), src.data(), MPI_UNWEIGHTED, dest.size(), dest.data(),
817 MPI_UNWEIGHTED, MPI_INFO_NULL,
false, &neigh_comm0);
820 MPI_Datatype compound_type0;
822 MPI_Type_commit(&compound_type0);
824 std::vector<T> recv_buffer_data(shape[1] * send_disp.back());
825 err = MPI_Neighbor_alltoallv(
826 send_buffer_data.data(), num_items_recv.data(), recv_disp.data(),
827 compound_type0, recv_buffer_data.data(), num_items_per_src.data(),
828 send_disp.data(), compound_type0, neigh_comm0);
831 err = MPI_Type_free(&compound_type0);
833 err = MPI_Comm_free(&neigh_comm0);
836 std::vector<std::int32_t> index_pos_to_buffer(indices.size(), -1);
837 for (std::size_t i = 0; i < src_to_index.size(); ++i)
838 index_pos_to_buffer[std::get<2>(src_to_index[i])] = i;
841 std::vector<T> x_new(shape[1] * indices.size());
842 for (std::size_t i = 0; i < indices.size(); ++i)
844 const std::int64_t index = indices[i];
845 if (index >= rank_offset and index < (rank_offset + shape0_local))
848 std::int64_t local_index = index - rank_offset;
849 std::copy_n(std::next(x.begin(), shape[1] * local_index), shape[1],
850 std::next(x_new.begin(), shape[1] * i));
852 else if (std::int32_t pos = index_pos_to_buffer[i]; pos != -1)
857 std::copy_n(std::next(recv_buffer_data.begin(), shape[1] * pos), shape[1],
858 std::next(x_new.begin(), shape[1] * i));
863 std::int64_t local_index = index - postoffice_range[0];
864 std::int32_t bag_pos = post_indices_map[local_index];
865 assert(bag_pos != -1);
866 std::copy_n(std::next(post_x.begin(), shape[1] * bag_pos), shape[1],
867 std::next(x_new.begin(), shape[1] * i));