Skip to content

Transform Reduce E test #1798

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rapids-bot merged 5 commits into from
Sep 15, 2021
Merged

Transform Reduce E test #1798

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
1a86f5b
First commit
kaatish Aug 31, 2021
ce4633f
Fix block reduce of thrust tuples
kaatish Sep 3, 2021
0b7b97b
Undo mg_louvain_helper fix
kaatish Sep 3, 2021
89750f8
Merge branch 'branch-21.10' of github.com:rapidsai/cugraph into raft-…
kaatish Sep 14, 2021
63ce53f
PR review fixes
kaatish Sep 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions cpp/include/cugraph/prims/copy_to_adj_matrix_row_col.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,8 +157,7 @@ void copy_to_matrix_major(raft::handle_t const& handle,
auto rx_tmp_buffer = allocate_dataframe_buffer<
typename std::iterator_traits<VertexValueInputIterator>::value_type>(rx_counts[i],
handle.get_stream());
auto rx_value_first = get_dataframe_buffer_begin<
typename std::iterator_traits<VertexValueInputIterator>::value_type>(rx_tmp_buffer);
auto rx_value_first = get_dataframe_buffer_begin(rx_tmp_buffer);

if (col_comm_rank == i) {
auto vertex_partition =
Expand Down Expand Up @@ -348,8 +347,7 @@ void copy_to_matrix_minor(raft::handle_t const& handle,
auto rx_tmp_buffer = allocate_dataframe_buffer<
typename std::iterator_traits<VertexValueInputIterator>::value_type>(rx_counts[i],
handle.get_stream());
auto rx_value_first = get_dataframe_buffer_begin<
typename std::iterator_traits<VertexValueInputIterator>::value_type>(rx_tmp_buffer);
auto rx_value_first = get_dataframe_buffer_begin(rx_tmp_buffer);

if (row_comm_rank == i) {
auto vertex_partition =
Expand Down
4 changes: 2 additions & 2 deletions cpp/include/cugraph/prims/copy_v_transform_reduce_in_out_nbr.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -428,7 +428,7 @@ void copy_v_transform_reduce_nbr(raft::handle_t const& handle,
: graph_view.get_number_of_local_adj_matrix_partition_cols()
: vertex_t{0};
auto minor_tmp_buffer = allocate_dataframe_buffer<T>(minor_tmp_buffer_size, handle.get_stream());
auto minor_buffer_first = get_dataframe_buffer_begin<T>(minor_tmp_buffer);
auto minor_buffer_first = get_dataframe_buffer_begin(minor_tmp_buffer);

if (in != GraphViewType::is_adj_matrix_transposed) {
auto minor_init = init;
Expand Down Expand Up @@ -463,7 +463,7 @@ void copy_v_transform_reduce_nbr(raft::handle_t const& handle,
GraphViewType::is_multi_gpu && update_major ? matrix_partition.get_major_size() : vertex_t{0};
auto major_tmp_buffer =
allocate_dataframe_buffer<T>(major_tmp_buffer_size, handle.get_stream());
auto major_buffer_first = get_dataframe_buffer_begin<T>(major_tmp_buffer);
auto major_buffer_first = get_dataframe_buffer_begin(major_tmp_buffer);

auto major_init = T{};
if (update_major) {
Expand Down
21 changes: 10 additions & 11 deletions cpp/include/cugraph/prims/copy_v_transform_reduce_key_aggregated_out_nbr.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,7 +420,7 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(
handle.get_stream());
device_bcast(row_comm,
map_value_first,
get_dataframe_buffer_begin<value_t>(map_value_buffer) + map_displacements[i],
get_dataframe_buffer_begin(map_value_buffer) + map_displacements[i],
map_counts[i],
i,
handle.get_stream());
Expand All @@ -432,11 +432,10 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(
map_unique_key_first,
map_unique_key_last,
map_keys.begin() + map_displacements[row_comm_rank]);
thrust::copy(
execution_policy,
map_value_first,
map_value_first + thrust::distance(map_unique_key_first, map_unique_key_last),
get_dataframe_buffer_begin<value_t>(map_value_buffer) + map_displacements[row_comm_rank]);
thrust::copy(execution_policy,
map_value_first,
map_value_first + thrust::distance(map_unique_key_first, map_unique_key_last),
get_dataframe_buffer_begin(map_value_buffer) + map_displacements[row_comm_rank]);

handle.get_stream_view().synchronize(); // cuco::static_map currently does not take stream

Expand All @@ -453,7 +452,7 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(
stream_adapter);

auto pair_first = thrust::make_zip_iterator(
thrust::make_tuple(map_keys.begin(), get_dataframe_buffer_begin<value_t>(map_value_buffer)));
thrust::make_tuple(map_keys.begin(), get_dataframe_buffer_begin(map_value_buffer)));
kv_map_ptr->insert(pair_first, pair_first + map_keys.size());
} else {
handle.get_stream_view().synchronize(); // cuco::static_map currently does not take stream
Expand Down Expand Up @@ -633,7 +632,7 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(

auto tmp_e_op_result_buffer =
allocate_dataframe_buffer<T>(tmp_major_vertices.size(), handle.get_stream());
auto tmp_e_op_result_buffer_first = get_dataframe_buffer_begin<T>(tmp_e_op_result_buffer);
auto tmp_e_op_result_buffer_first = get_dataframe_buffer_begin(tmp_e_op_result_buffer);

auto matrix_partition_row_value_input = adj_matrix_row_value_input;
matrix_partition_row_value_input.add_offset(matrix_partition.get_major_value_start_offset());
Expand Down Expand Up @@ -689,7 +688,7 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(
handle.get_stream());
device_gatherv(col_comm,
tmp_e_op_result_buffer_first,
get_dataframe_buffer_begin<T>(rx_tmp_e_op_result_buffer),
get_dataframe_buffer_begin(rx_tmp_e_op_result_buffer),
tmp_major_vertices.size(),
rx_sizes,
rx_displs,
Expand Down Expand Up @@ -729,7 +728,7 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(
thrust::sort_by_key(execution_policy,
major_vertices.begin(),
major_vertices.end(),
get_dataframe_buffer_begin<T>(e_op_result_buffer));
get_dataframe_buffer_begin(e_op_result_buffer));

auto num_uniques = thrust::count_if(execution_policy,
thrust::make_counting_iterator(size_t{0}),
Expand All @@ -748,7 +747,7 @@ void copy_v_transform_reduce_key_aggregated_out_nbr(
thrust::reduce_by_key(execution_policy,
major_vertices.begin(),
major_vertices.end(),
get_dataframe_buffer_begin<T>(e_op_result_buffer),
get_dataframe_buffer_begin(e_op_result_buffer),
thrust::make_discard_iterator(),
thrust::make_permutation_iterator(
vertex_value_output_first,
Expand Down
4 changes: 2 additions & 2 deletions cpp/include/cugraph/prims/property_op_utils.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,13 +133,13 @@ struct property_add<thrust::tuple<Args...>>

private:
template <typename T, std::size_t... Is>
__device__ constexpr auto sum_impl(T& t1, T& t2, std::index_sequence<Is...>)
__host__ __device__ constexpr auto sum_impl(T& t1, T& t2, std::index_sequence<Is...>)
{
return thrust::make_tuple((thrust::get<Is>(t1) + thrust::get<Is>(t2))...);
}

public:
__device__ constexpr auto operator()(const Type& t1, const Type& t2)
__host__ __device__ constexpr auto operator()(const Type& t1, const Type& t2)
{
return sum_impl(t1, t2, std::make_index_sequence<thrust::tuple_size<Type>::value>());
}
Expand Down
24 changes: 10 additions & 14 deletions cpp/include/cugraph/prims/row_col_properties.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,23 +106,21 @@ class major_properties_t {

void fill(T value, rmm::cuda_stream_view stream)
{
thrust::fill(rmm::exec_policy(stream),
value_data(),
value_data() + size_dataframe_buffer<T>(buffer_),
value);
thrust::fill(
rmm::exec_policy(stream), value_data(), value_data() + size_dataframe_buffer(buffer_), value);
}

auto value_data() { return get_dataframe_buffer_begin<T>(buffer_); }
auto value_data() { return get_dataframe_buffer_begin(buffer_); }

auto device_view() const
{
auto value_first = get_dataframe_buffer_cbegin<T>(buffer_);
auto value_first = get_dataframe_buffer_cbegin(buffer_);
return major_properties_device_view_t<vertex_t, decltype(value_first)>(value_first);
}

auto mutable_device_view()
{
auto value_first = get_dataframe_buffer_begin<T>(buffer_);
auto value_first = get_dataframe_buffer_begin(buffer_);
return major_properties_device_view_t<vertex_t, decltype(value_first)>(value_first);
}

Expand Down Expand Up @@ -159,17 +157,15 @@ class minor_properties_t {

void fill(T value, rmm::cuda_stream_view stream)
{
thrust::fill(rmm::exec_policy(stream),
value_data(),
value_data() + size_dataframe_buffer<T>(buffer_),
value);
thrust::fill(
rmm::exec_policy(stream), value_data(), value_data() + size_dataframe_buffer(buffer_), value);
}

auto value_data() { return get_dataframe_buffer_begin<T>(buffer_); }
auto value_data() { return get_dataframe_buffer_begin(buffer_); }

auto device_view() const
{
auto value_first = get_dataframe_buffer_cbegin<T>(buffer_);
auto value_first = get_dataframe_buffer_cbegin(buffer_);
if (key_first_) {
return minor_properties_device_view_t<vertex_t, decltype(value_first)>(
*key_first_, *key_last_, value_first);
Expand All @@ -180,7 +176,7 @@ class minor_properties_t {

auto mutable_device_view()
{
auto value_first = get_dataframe_buffer_begin<T>(buffer_);
auto value_first = get_dataframe_buffer_begin(buffer_);
if (key_first_) {
return minor_properties_device_view_t<vertex_t, decltype(value_first)>(
*key_first_, *key_last_, value_first);
Expand Down
30 changes: 14 additions & 16 deletions cpp/include/cugraph/prims/transform_reduce_by_adj_matrix_row_col_key_e.cuh
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -318,10 +318,8 @@ template <typename vertex_t, typename value_t, typename BufferType>
std::tuple<rmm::device_uvector<vertex_t>, BufferType> reduce_to_unique_kv_pairs(
rmm::device_uvector<vertex_t>&& keys, BufferType&& value_buffer, cudaStream_t stream)
{
thrust::sort_by_key(rmm::exec_policy(stream),
keys.begin(),
keys.end(),
get_dataframe_buffer_begin<value_t>(value_buffer));
thrust::sort_by_key(
rmm::exec_policy(stream), keys.begin(), keys.end(), get_dataframe_buffer_begin(value_buffer));
auto num_uniques =
thrust::count_if(rmm::exec_policy(stream),
thrust::make_counting_iterator(size_t{0}),
Expand All @@ -335,9 +333,9 @@ std::tuple<rmm::device_uvector<vertex_t>, BufferType> reduce_to_unique_kv_pairs(
thrust::reduce_by_key(rmm::exec_policy(stream),
keys.begin(),
keys.end(),
get_dataframe_buffer_begin<value_t>(value_buffer),
get_dataframe_buffer_begin(value_buffer),
unique_keys.begin(),
get_dataframe_buffer_begin<value_t>(value_for_unique_key_buffer));
get_dataframe_buffer_begin(value_for_unique_key_buffer));

return std::make_tuple(std::move(unique_keys), std::move(value_for_unique_key_buffer));
}
Expand Down Expand Up @@ -428,7 +426,7 @@ transform_reduce_by_adj_matrix_row_col_key_e(
matrix_partition_row_col_key_input,
e_op,
tmp_keys.data(),
get_dataframe_buffer_begin<T>(tmp_value_buffer));
get_dataframe_buffer_begin(tmp_value_buffer));
}
if ((*segment_offsets)[2] - (*segment_offsets)[1] > 0) {
raft::grid_1d_warp_t update_grid(
Expand All @@ -445,7 +443,7 @@ transform_reduce_by_adj_matrix_row_col_key_e(
matrix_partition_row_col_key_input,
e_op,
tmp_keys.data(),
get_dataframe_buffer_begin<T>(tmp_value_buffer));
get_dataframe_buffer_begin(tmp_value_buffer));
}
if ((*segment_offsets)[3] - (*segment_offsets)[2] > 0) {
raft::grid_1d_thread_t update_grid(
Expand All @@ -462,7 +460,7 @@ transform_reduce_by_adj_matrix_row_col_key_e(
matrix_partition_row_col_key_input,
e_op,
tmp_keys.data(),
get_dataframe_buffer_begin<T>(tmp_value_buffer));
get_dataframe_buffer_begin(tmp_value_buffer));
}
if (matrix_partition.get_dcs_nzd_vertex_count() &&
(*(matrix_partition.get_dcs_nzd_vertex_count()) > 0)) {
Expand All @@ -479,7 +477,7 @@ transform_reduce_by_adj_matrix_row_col_key_e(
matrix_partition_row_col_key_input,
e_op,
tmp_keys.data(),
get_dataframe_buffer_begin<T>(tmp_value_buffer));
get_dataframe_buffer_begin(tmp_value_buffer));
}
} else {
raft::grid_1d_thread_t update_grid(
Expand All @@ -497,7 +495,7 @@ transform_reduce_by_adj_matrix_row_col_key_e(
matrix_partition_row_col_key_input,
e_op,
tmp_keys.data(),
get_dataframe_buffer_begin<T>(tmp_value_buffer));
get_dataframe_buffer_begin(tmp_value_buffer));
}
}
std::tie(tmp_keys, tmp_value_buffer) = reduce_to_unique_kv_pairs<vertex_t, T>(
Expand All @@ -514,7 +512,7 @@ transform_reduce_by_adj_matrix_row_col_key_e(
comm,
tmp_keys.begin(),
tmp_keys.end(),
get_dataframe_buffer_begin<T>(tmp_value_buffer),
get_dataframe_buffer_begin(tmp_value_buffer),
[key_func = detail::compute_gpu_id_from_vertex_t<vertex_t>{comm_size}] __device__(
auto val) { return key_func(val); },
handle.get_stream());
Expand All @@ -532,14 +530,14 @@ transform_reduce_by_adj_matrix_row_col_key_e(
// can reserve address space to avoid expensive reallocation.
// https://devblogs.nvidia.com/introducing-low-level-gpu-virtual-memory-management
keys.resize(cur_size + tmp_keys.size(), handle.get_stream());
resize_dataframe_buffer<T>(value_buffer, keys.size(), handle.get_stream());
resize_dataframe_buffer(value_buffer, keys.size(), handle.get_stream());

auto execution_policy = handle.get_thrust_policy();
thrust::copy(execution_policy, tmp_keys.begin(), tmp_keys.end(), keys.begin() + cur_size);
thrust::copy(execution_policy,
get_dataframe_buffer_begin<T>(tmp_value_buffer),
get_dataframe_buffer_begin<T>(tmp_value_buffer) + tmp_keys.size(),
get_dataframe_buffer_begin<T>(value_buffer) + cur_size);
get_dataframe_buffer_begin(tmp_value_buffer),
get_dataframe_buffer_begin(tmp_value_buffer) + tmp_keys.size(),
get_dataframe_buffer_begin(value_buffer) + cur_size);
}
}

Expand Down
Loading