Skip to content

Commit 8e4e553

Browse files
dycz0fxdycz0fx
committed
Add MPI_IN_PLACE support in SOLO module
1 parent f6466bc commit 8e4e553

3 files changed

Lines changed: 42 additions & 12 deletions

File tree

‎ompi/mca/coll/solo/coll_solo.h‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -160,8 +160,8 @@ static inline void mca_coll_solo_pack_to_shared(void *local_buf, void *shared_bu
160160
memcpy((char *) shared_buf, (char *) local_buf, count * extent);
161161
}
162162
else {
163-
MPI_Aint pos = 0;
164-
ompi_datatype_pack_external("external32", local_buf, count, dtype, shared_buf, count * extent, &pos);
163+
int pos = 0;
164+
MPI_Pack(local_buf, count, dtype, shared_buf, count * extent, &pos, MPI_COMM_SELF);
165165
}
166166
}
167167

@@ -171,8 +171,8 @@ static inline void mca_coll_solo_unpack_from_shared(void *local_buf, void *share
171171
memcpy((char *) local_buf, (char *) shared_buf, count * extent);
172172
}
173173
else {
174-
MPI_Aint pos = 0;
175-
ompi_datatype_unpack_external("external32", shared_buf, count * extent, &pos, local_buf, count, dtype);
174+
int pos = 0;
175+
MPI_Unpack(shared_buf, count * extent, &pos, local_buf, count, dtype, MPI_COMM_SELF);
176176
}
177177
}
178178

‎ompi/mca/coll/solo/coll_solo_allreduce.c‎

Lines changed: 18 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,12 @@ int mca_coll_solo_allreduce_intra(const void *sbuf, void *rbuf,
1818
struct ompi_communicator_t *comm,
1919
mca_coll_base_module_t * module)
2020
{
21-
return mca_coll_solo_allreduce_ring_intra_memcpy(sbuf, rbuf, count, dtype, op, comm, module);
21+
if (ompi_op_is_commute(op)) {
22+
return mca_coll_solo_allreduce_ring_intra_memcpy(sbuf, rbuf, count, dtype, op, comm, module);
23+
}
24+
else {
25+
return ompi_coll_base_allreduce_intra_nonoverlapping(sbuf, rbuf, count, dtype, op, comm, module);
26+
}
2227
}
2328

2429

@@ -88,7 +93,11 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
8893
int num_segments = (count + seg_count - 1) / seg_count;
8994
int last_count = count - seg_count * (num_segments - 1);
9095
for (int i = 0; i < num_segments; i++) {
91-
char *temp_sbuf = (char *)sbuf + seg_count * extent * i;
96+
char *temp_sbuf;
97+
if (sbuf == MPI_IN_PLACE)
98+
temp_sbuf = MPI_IN_PLACE;
99+
else
100+
temp_sbuf = (char *)sbuf + seg_count * extent * i;
92101
char *temp_rbuf = (char *)rbuf + seg_count * extent * i;
93102
int temp_count = seg_count;
94103
if (i == num_segments - 1) {
@@ -100,6 +109,11 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
100109
return MPI_SUCCESS;
101110
}
102111

112+
char *sbuf_temp = (char *)sbuf;
113+
if( sbuf == MPI_IN_PLACE ) {
114+
sbuf_temp = (char *)rbuf;
115+
}
116+
103117
*(int *) (solo_module->ctrl_bufs[rank]) = rank;
104118
mac_coll_solo_barrier_intra(comm, module);
105119

@@ -112,13 +126,13 @@ int mca_coll_solo_allreduce_ring_intra_memcpy(const void *sbuf, void *rbuf, int
112126
}
113127
/* At first iteration, copy local data to the solo data buffer */
114128
if (cur == rank) {
115-
mca_coll_solo_copy((void *) ((char *) sbuf + cur * l_seg_count * extent), (void *) data_bufs[cur], dtype, seg_count, extent);
129+
mca_coll_solo_copy((void *) ((char *) sbuf_temp + cur * l_seg_count * extent), (void *) data_bufs[cur], dtype, seg_count, extent);
116130
mac_coll_solo_barrier_intra(comm, module);
117131

118132
}
119133
/* For other iterations, do operations on the solo data buffer */
120134
else {
121-
ompi_op_reduce(op, (char *) sbuf + cur * l_seg_count * extent,
135+
ompi_op_reduce(op, (char *) sbuf_temp + cur * l_seg_count * extent,
122136
data_bufs[cur], seg_count, dtype);
123137
mac_coll_solo_barrier_intra(comm, module);
124138
}

‎ompi/mca/coll/solo/coll_solo_reduce.c‎

Lines changed: 20 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,12 @@ int mca_coll_solo_reduce_intra(const void *sbuf, void *rbuf, int count,
1818
struct ompi_communicator_t *comm,
1919
mca_coll_base_module_t * module)
2020
{
21-
return mca_coll_solo_reduce_ring_intra_memcpy(sbuf, rbuf, count, dtype, op, root, comm, module);
21+
if (ompi_op_is_commute(op)) {
22+
return mca_coll_solo_reduce_ring_intra_memcpy(sbuf, rbuf, count, dtype, op, root, comm, module);
23+
}
24+
else {
25+
return ompi_coll_base_reduce_intra_basic_linear(sbuf, rbuf, count, dtype, op, root, comm, module);
26+
}
2227
}
2328

2429
/**
@@ -46,6 +51,7 @@ int mca_coll_solo_reduce_ring_intra_memcpy(const void *sbuf, void *rbuf,
4651
int size = ompi_comm_size(comm);
4752
int rank = ompi_comm_rank(comm);
4853
int i;
54+
4955
ptrdiff_t extent, lower_bound;
5056
ompi_datatype_get_extent(dtype, &lower_bound, &extent);
5157

@@ -89,7 +95,11 @@ int mca_coll_solo_reduce_ring_intra_memcpy(const void *sbuf, void *rbuf,
8995
int num_segments = (count + seg_count - 1) / seg_count;
9096
int last_count = count - seg_count * (num_segments - 1);
9197
for (int i = 0; i < num_segments; i++) {
92-
char *temp_sbuf = (char *)sbuf + seg_count * extent * i;
98+
char *temp_sbuf;
99+
if (sbuf == MPI_IN_PLACE)
100+
temp_sbuf = MPI_IN_PLACE;
101+
else
102+
temp_sbuf = (char *)sbuf + seg_count * extent * i;
93103
char *temp_rbuf = (char *)rbuf + seg_count * extent * i;
94104
int temp_count = seg_count;
95105
if (i == num_segments - 1) {
@@ -101,6 +111,11 @@ int mca_coll_solo_reduce_ring_intra_memcpy(const void *sbuf, void *rbuf,
101111
return MPI_SUCCESS;
102112
}
103113

114+
char *sbuf_temp = (char *)sbuf;
115+
if( sbuf == MPI_IN_PLACE ) {
116+
sbuf_temp = (char *)rbuf;
117+
}
118+
104119
*(int *) (solo_module->ctrl_bufs[rank]) = rank;
105120
mac_coll_solo_barrier_intra(comm, module);
106121

@@ -113,12 +128,13 @@ int mca_coll_solo_reduce_ring_intra_memcpy(const void *sbuf, void *rbuf,
113128
}
114129
/* At first iteration, copy local data to the shared data buffer */
115130
if (cur == rank) {
116-
mca_coll_solo_copy((void *) ((char *) sbuf + cur * l_seg_count * extent), (void *) data_bufs[cur], dtype, seg_count, extent);
131+
mca_coll_solo_copy((void *) ((char *) sbuf_temp + cur * l_seg_count * extent),
132+
(void *) data_bufs[cur], dtype, seg_count, extent);
117133
mac_coll_solo_barrier_intra(comm, module);
118134
}
119135
/* For other iterations, do operations on the shared data buffer */
120136
else {
121-
ompi_op_reduce(op, (char *) sbuf + cur * l_seg_count * extent,
137+
ompi_op_reduce(op, (char *) sbuf_temp + cur * l_seg_count * extent,
122138
data_bufs[cur], seg_count, dtype);
123139
mac_coll_solo_barrier_intra(comm, module);
124140
}

0 commit comments

Comments
 (0)