Skip to content
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

Signal operation type for put with signal #929

Merged
merged 10 commits into from
Feb 24, 2020
Merged

Signal operation type for put with signal #929

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
1b471e5
Signal operation type for put with signal
wrrobin Feb 6, 2020
be9e259
Added changes to portals4 put-signal implementation
wrrobin Feb 6, 2020
67912f5
Correction for instance with len = 0
wrrobin Feb 6, 2020
42a61cf
Correction in API call for instance with len = 0
wrrobin Feb 6, 2020
217c430
Trial hack for the fix related to on-node transport of signal atomic
wrrobin Feb 12, 2020
bada86b
Bug fix for missing function declaration and incorrect macro
wrrobin Feb 12, 2020
32bf17d
Update in configure script to enable shr atomics when only xpmem is e…
wrrobin Feb 12, 2020
3af033e
Bug fix for data race in put_signal_nbi test
wrrobin Feb 12, 2020
e24e99a
Set the enable-shr-atomics to auto by default rather than disabled
wrrobin Feb 12, 2020
432835a
Moved signal op constants to shmemx.h; incorporated git comments
wrrobin Feb 14, 2020
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
3 changes: 3 additions & 0 deletions mpp/shmem-def.h.in
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,9 @@ typedef struct shmem_impl_ctx_t { int dummy; } * shmem_ctx_t;
#define SHMEM_CTX_SERIALIZED (1l<<1)
#define SHMEM_CTX_NOSTORE (1l<<2)

#define SHMEM_SIGNAL_SET 0
#define SHMEM_SIGNAL_ADD 1

#ifdef __cplusplus
} /* extern "C" */
#endif
Expand Down
16 changes: 8 additions & 8 deletions mpp/shmemx.h4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,24 +190,24 @@ SHMEM_CXX_DEFINE_FOR_BITWISE_AMO(`SHMEM_CXX_ATOMIC_FETCH_XOR_NBI')
/* Blocking put with signal */
define(`SHMEM_CXX_PUT_SIGNAL',
`static inline void shmemx_put_signal($2* dest, const $2* source,
size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe) {
shmemx_$1_put_signal(dest, source, nelems, sig_addr, signal, pe);
size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe) {
shmemx_$1_put_signal(dest, source, nelems, sig_addr, signal, sig_op, pe);
}
static inline void shmemx_put_signal(shmem_ctx_t ctx, $2* dest, const $2* source,
size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe) {
shmemx_ctx_$1_put_signal(ctx, dest, source, nelems, sig_addr, signal, pe);
size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe) {
shmemx_ctx_$1_put_signal(ctx, dest, source, nelems, sig_addr, signal, sig_op, pe);
}')dnl
SHMEM_CXX_DEFINE_FOR_RMA(`SHMEM_CXX_PUT_SIGNAL')

/* Non-blocking put with signal */
define(`SHMEM_CXX_PUT_SIGNAL_NBI',
`static inline void shmemx_put_signal_nbi($2* dest, const $2* source,
size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe) {
shmemx_$1_put_signal_nbi(dest, source, nelems, sig_addr, signal, pe);
size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe) {
shmemx_$1_put_signal_nbi(dest, source, nelems, sig_addr, signal, sig_op, pe);
}
static inline void shmemx_put_signal_nbi(shmem_ctx_t ctx, $2* dest, const $2* source,
size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe) {
shmemx_ctx_$1_put_signal_nbi(ctx, dest, source, nelems, sig_addr, signal, pe);
size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe) {
shmemx_ctx_$1_put_signal_nbi(ctx, dest, source, nelems, sig_addr, signal, sig_op, pe);
}')dnl
SHMEM_CXX_DEFINE_FOR_RMA(`SHMEM_CXX_PUT_SIGNAL_NBI')

Expand Down
16 changes: 8 additions & 8 deletions mpp/shmemx_c_func.h4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -155,39 +155,39 @@ SHMEM_DECLARE_FOR_BITWISE_AMO(`SHMEM_C_CTX_FETCH_OR_NBI')

/* Blocking put with signal */
define(`SHMEM_C_PUT_SIGNAL',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_$1_put_signal($2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_$1_put_signal($2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_RMA(`SHMEM_C_PUT_SIGNAL')

define(`SHMEM_C_CTX_PUT_SIGNAL',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_$1_put_signal(shmem_ctx_t ctx, $2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_$1_put_signal(shmem_ctx_t ctx, $2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_RMA(`SHMEM_C_CTX_PUT_SIGNAL')

define(`SHMEM_C_PUT_N_SIGNAL',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_put$1_signal(void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_put$1_signal(void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_SIZES(`SHMEM_C_PUT_N_SIGNAL')
SHMEM_C_PUT_N_SIGNAL(mem,1);

define(`SHMEM_C_CTX_PUT_N_SIGNAL',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_put$1_signal(shmem_ctx_t ctx, void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_put$1_signal(shmem_ctx_t ctx, void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_SIZES(`SHMEM_C_CTX_PUT_N_SIGNAL')
SHMEM_C_CTX_PUT_N_SIGNAL(mem,1);

/* Non-blocking put with signal */
define(`SHMEM_C_PUT_SIGNAL_NBI',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_$1_put_signal_nbi($2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_$1_put_signal_nbi($2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_RMA(`SHMEM_C_PUT_SIGNAL_NBI')

define(`SHMEM_C_CTX_PUT_SIGNAL_NBI',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_$1_put_signal_nbi(shmem_ctx_t ctx, $2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_$1_put_signal_nbi(shmem_ctx_t ctx, $2 *target, const $2 *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_RMA(`SHMEM_C_CTX_PUT_SIGNAL_NBI')

define(`SHMEM_C_PUT_N_SIGNAL_NBI',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_put$1_signal_nbi(void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_put$1_signal_nbi(void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_SIZES(`SHMEM_C_PUT_N_SIGNAL_NBI')
SHMEM_C_PUT_N_SIGNAL_NBI(mem,1);

define(`SHMEM_C_CTX_PUT_N_SIGNAL_NBI',
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_put$1_signal_nbi(shmem_ctx_t ctx, void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int pe)')dnl
`SHMEM_FUNCTION_ATTRIBUTES void SHPRE()shmemx_ctx_put$1_signal_nbi(shmem_ctx_t ctx, void* target, const void *source, size_t nelems, uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)')dnl
SHMEM_DECLARE_FOR_SIZES(`SHMEM_C_CTX_PUT_N_SIGNAL_NBI')
SHMEM_C_CTX_PUT_N_SIGNAL_NBI(mem,1);

Expand Down
40 changes: 30 additions & 10 deletions src/data_c.c4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -473,66 +473,86 @@ SHMEM_PROF_DEF_CTX_PUT_N_SIGNAL_NBI(`mem')
void SHMEM_FUNCTION_ATTRIBUTES \
SHMEMX_FUNC_PROTOTYPE(STYPE##_put_signal, TYPE *target, \
const TYPE *source, size_t nelems, \
uint64_t *sig_addr, uint64_t signal, int pe) \
uint64_t *sig_addr, uint64_t signal, \
int sig_op, int pe) \
long completion = 0; \
SHMEM_ERR_CHECK_INITIALIZED(); \
SHMEM_ERR_CHECK_PE(pe); \
SHMEM_ERR_CHECK_SYMMETRIC(target, sizeof(TYPE) * nelems); \
SHMEM_ERR_CHECK_NULL(source, nelems); \
SHMEM_ERR_CHECK_SIG_OP(sig_op); \
shmem_internal_put_nb(ctx, target, source, \
sizeof(TYPE) * nelems, pe, \
&completion); \
shmem_internal_put_wait(ctx, &completion); \
shmem_internal_fence(ctx); \
shmem_internal_put_scalar(ctx, sig_addr, &signal, \
sizeof(uint64_t), pe); \
if (sig_op == SHMEM_SIGNAL_ADD) \
shmem_internal_atomic(ctx, sig_addr, &signal, sizeof(uint64_t), \
pe, SHM_INTERNAL_SUM, \
SHM_INTERNAL_UINT64); \
else \
shmem_internal_atomic_set(ctx, sig_addr, &signal, \
sizeof(uint64_t), pe, \
SHM_INTERNAL_UINT64); \
}


#define SHMEM_DEF_PUT_N_SIGNAL(NAME,SIZE) \
void SHMEM_FUNCTION_ATTRIBUTES \
SHMEMX_FUNC_PROTOTYPE(put##NAME##_signal, void *target, \
const void *source, size_t nelems, \
uint64_t *sig_addr, uint64_t signal, int pe) \
uint64_t *sig_addr, uint64_t signal, \
int sig_op, int pe) \
long completion = 0; \
SHMEM_ERR_CHECK_INITIALIZED(); \
SHMEM_ERR_CHECK_PE(pe); \
SHMEM_ERR_CHECK_SYMMETRIC(target, (SIZE) * nelems); \
SHMEM_ERR_CHECK_NULL(source, nelems); \
SHMEM_ERR_CHECK_SIG_OP(sig_op); \
shmem_internal_put_nb(ctx, target, source, (SIZE) * nelems, \
pe, &completion); \
shmem_internal_put_wait(ctx, &completion); \
shmem_internal_fence(ctx); \
shmem_internal_put_scalar(ctx, sig_addr, &signal, \
sizeof(uint64_t), pe); \
if (sig_op == SHMEM_SIGNAL_ADD) \
shmem_internal_atomic(ctx, sig_addr, &signal, sizeof(uint64_t), \
pe, SHM_INTERNAL_SUM, \
SHM_INTERNAL_UINT64); \
else \
shmem_internal_atomic_set(ctx, sig_addr, &signal, \
sizeof(uint64_t), pe, \
SHM_INTERNAL_UINT64); \
}

#define SHMEM_DEF_PUT_SIGNAL_NBI(STYPE,TYPE) \
void SHMEM_FUNCTION_ATTRIBUTES \
SHMEMX_FUNC_PROTOTYPE(STYPE##_put_signal_nbi, TYPE *target, \
const TYPE *source, size_t nelems, \
uint64_t *sig_addr, uint64_t signal, int pe) \
uint64_t *sig_addr, uint64_t signal, \
int sig_op, int pe) \
SHMEM_ERR_CHECK_INITIALIZED(); \
SHMEM_ERR_CHECK_PE(pe); \
SHMEM_ERR_CHECK_SYMMETRIC(target, sizeof(TYPE) * nelems); \
SHMEM_ERR_CHECK_NULL(source, nelems); \
SHMEM_ERR_CHECK_SIG_OP(sig_op); \
shmem_internal_put_signal_nbi(ctx, target, source, \
sizeof(TYPE) * nelems, sig_addr, \
signal, pe); \
signal, sig_op, pe); \
}


#define SHMEM_DEF_PUT_N_SIGNAL_NBI(NAME,SIZE) \
void SHMEM_FUNCTION_ATTRIBUTES \
SHMEMX_FUNC_PROTOTYPE(put##NAME##_signal_nbi, void *target, \
const void *source, size_t nelems, \
uint64_t *sig_addr, uint64_t signal, int pe) \
uint64_t *sig_addr, uint64_t signal, \
int sig_op, int pe) \
SHMEM_ERR_CHECK_INITIALIZED(); \
SHMEM_ERR_CHECK_PE(pe); \
SHMEM_ERR_CHECK_SYMMETRIC(target, (SIZE) * nelems); \
SHMEM_ERR_CHECK_NULL(source, nelems); \
SHMEM_ERR_CHECK_SIG_OP(sig_op); \
shmem_internal_put_signal_nbi(ctx, target, source, (SIZE) * nelems, \
sig_addr, signal, pe); \
sig_addr, signal, sig_op, pe); \
}


Expand Down
13 changes: 9 additions & 4 deletions src/shmem_comm.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,17 +61,22 @@ shmem_internal_put_nb(shmem_ctx_t ctx, void *target, const void *source, size_t
static inline
void
shmem_internal_put_signal_nbi(shmem_ctx_t ctx, void *target, const void *source, size_t len,
uint64_t *sig_addr, uint64_t signal, int pe)
uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)
{
if (len == 0) {
shmem_internal_put_scalar(ctx, sig_addr, &signal, sizeof(uint64_t), pe);
if (sig_op == SHMEM_SIGNAL_ADD)
shmem_transport_atomic((shmem_transport_ctx_t *) ctx, sig_addr, &signal, sizeof(uint64_t),
pe, SHM_INTERNAL_SUM, SHM_INTERNAL_UINT64);
else
shmem_transport_atomic_set((shmem_transport_ctx_t *) ctx, sig_addr, &signal,
sizeof(uint64_t), pe, SHM_INTERNAL_UINT64);
return;
}

if (shmem_shr_transport_use_write(ctx, target, source, len, pe)) {
shmem_shr_transport_put_signal(ctx, target, source, len, sig_addr, signal, pe);
shmem_shr_transport_put_signal(ctx, target, source, len, sig_addr, signal, sig_op, pe);
} else {
shmem_transport_put_signal_nbi((shmem_transport_ctx_t *) ctx, target, source, len, sig_addr, signal, pe);
shmem_transport_put_signal_nbi((shmem_transport_ctx_t *) ctx, target, source, len, sig_addr, signal, sig_op, pe);
}
}

Expand Down
15 changes: 15 additions & 0 deletions src/shmem_internal.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,20 @@ extern unsigned int shmem_internal_rand_seed;
} \
} while (0)

#define SHMEM_ERR_CHECK_SIG_OP(op) \
do { \
switch(op) { \
case SHMEM_SIGNAL_SET: \
case SHMEM_SIGNAL_ADD: \
break; \
default: \
fprintf(stderr, "ERROR: %s(): Argument \"%s\", " \
"invalid atomic operation for signal (%d)\n", \
__func__, #op, (int) (op)); \
jdinan marked this conversation as resolved.
Show resolved Hide resolved
shmem_runtime_abort(100, PACKAGE_NAME " exited in error"); \
} \
} while (0)

#else
#define SHMEM_ERR_CHECK_INITIALIZED()
#define SHMEM_ERR_CHECK_POSITIVE(arg)
Expand All @@ -314,6 +328,7 @@ extern unsigned int shmem_internal_rand_seed;
#define SHMEM_ERR_CHECK_SYMMETRIC_HEAP(ptr)
#define SHMEM_ERR_CHECK_NULL(ptr, nelems)
#define SHMEM_ERR_CHECK_CMP_OP(op)
#define SHMEM_ERR_CHECK_SIG_OP(op) \

#endif /* ENABLE_ERROR_CHECKING */

Expand Down
17 changes: 15 additions & 2 deletions src/shr_transport.h4
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -175,23 +175,36 @@ shmem_shr_transport_put(shmem_ctx_t ctx, void *target, const void *source,
static inline void
shmem_shr_transport_put_signal(shmem_ctx_t ctx, void *target,
const void *source, size_t len,
uint64_t *sig_addr, uint64_t signal, int pe)
uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)
{
#if USE_MEMCPY
memcpy(target, source, len);
*sig_addr = signal;
if (sig_op == SHMEM_SIGNAL_ADD) *sig_addr += signal;
else *sig_addr = signal;
#elif USE_XPMEM
shmem_transport_xpmem_put(target, source, len, pe,
shmem_internal_get_shr_rank(pe));
shmem_internal_membar_acq_rel(); /* Memory fence to ensure target PE observes
stores in the correct order */
uint64_t old_signal_val = 0;
if (sig_op == SHMEM_SIGNAL_ADD) {
shmem_transport_xpmem_get(&old_signal_val, sig_addr, sizeof(uint64_t), pe,
shmem_internal_get_shr_rank(pe));
signal += old_signal_val;
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This needs to be an atomic operation, use shmem_shr_transport_atomic.

Copy link
Collaborator Author

@wrrobin wrrobin Feb 7, 2020 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jdinan shmem_shr_transport_atomic is enabled through USE_SHR_ATOMICS. Unless we use --enable-shr-atomics flag, we will not be able to use this API, right? Please let me know if I am missing something.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the target is waiting for several messages using the same signal variable, this will not work, right?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Right, you would need to write code to support builds with and without shared memory atomics.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @jdinan. One question, from the current code, there seems to be no relation between the flags USE_SHR_ATOMICS and USE_XPMEM/USE_CMA. Was there any particular reason behind this? I am thinking shouldn't USE_SHR_ATOMICS be defined when either of these on-node transports is defined?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For nearly all builds, we can't enable shared memory atomics because the networking layer is not coherent with processor atomics. By default, we only enable shared memory copy to implement put/get. This is why shared memory atomics are a separate option, and disabled by default. CMA is essentially memory copy performed by the Linux kernel and it cannot support atomics.

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For CMA, we should directly use the transport layer atomics then, as you also suggested later. But, for XPMEM, can we say that enabling USE_XPMEM should enable USE_SHR_ATOMICS as well? We can use the shared memory atomics if shmem_internal_get_shr_rank returns something other than -1, right? But, currently, shmem_shr_transport_use_atomic also seems to be guarded by USE_SHR_ATOMICS.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Correct, it is always safe to use shared memory put/get because OpenSHMEM says overlapping operations (e.g. between local shared memory and remote network transport accesses) leads to undefined behavior. Atomics do allow this overlapping. Therefore, it is only safe to use shared memory atomics when you know the transport layer is coherent with processor atomics. Because of this, enabling XPMEM only enables shared memory put/get and there is an additional flag to enable shared memory atomics.

shmem_transport_xpmem_put(sig_addr, &signal, sizeof(uint64_t), pe,
shmem_internal_get_shr_rank(pe));
#elif USE_CMA
shmem_transport_cma_put(target, source, len, pe,
shmem_internal_get_shr_rank(pe));
shmem_internal_membar_acq_rel(); /* Memory fence to ensure target PE observes
stores in the correct order */
uint64_t old_signal_val = 0;
if (sig_op == SHMEM_SIGNAL_ADD) {
shmem_transport_cma_get(&old_signal_val, sig_addr, sizeof(uint64_t), pe,
shmem_internal_get_shr_rank(pe));
signal += old_signal_val;
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This needs to be an atomic. CMA does not support atomics; use shmem_transport_atomic.

shmem_transport_cma_put(sig_addr, &signal, sizeof(uint64_t), pe,
shmem_internal_get_shr_rank(pe));
#else
Expand Down
2 changes: 1 addition & 1 deletion src/transport_none.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -128,7 +128,7 @@ shmem_transport_put_nb(shmem_transport_ctx_t* ctx, void *target, const void *sou
static inline
void
shmem_transport_put_signal_nbi(shmem_transport_ctx_t* ctx, void *target, const void *source, size_t len,
uint64_t *sig_addr, uint64_t signal, int pe)
uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)
{
RAISE_ERROR_STR("No path to peer");
}
Expand Down
5 changes: 3 additions & 2 deletions src/transport_ofi.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -658,7 +658,7 @@ void shmem_transport_put_nb(shmem_transport_ctx_t* ctx, void *target, const void

static inline
void shmem_transport_put_signal_nbi(shmem_transport_ctx_t* ctx, void *target, const void *source, size_t len,
uint64_t *sig_addr, uint64_t signal, int pe)
uint64_t *sig_addr, uint64_t signal, int sig_op, int pe)
{
int ret = 0;
uint64_t dst = (uint64_t) pe;
Expand Down Expand Up @@ -765,6 +765,7 @@ void shmem_transport_put_signal_nbi(shmem_transport_ctx_t* ctx, void *target, co
shmem_transport_ofi_get_mr(sig_addr, pe, &addr, &key);
polled = 0;
ret = 0;
int atomic_op = (sig_op == SHMEM_SIGNAL_ADD) ? FI_SUM : FI_ATOMIC_WRITE;

SHMEM_TRANSPORT_OFI_CTX_LOCK(ctx);
SHMEM_TRANSPORT_OFI_CNTR_INC(&ctx->pending_put_cntr);
Expand All @@ -786,7 +787,7 @@ void shmem_transport_put_signal_nbi(shmem_transport_ctx_t* ctx, void *target, co
.rma_iov = &rma_iov_signal,
.rma_iov_count = 1,
.datatype = FI_UINT64,
.op = FI_ATOMIC_WRITE,
.op = atomic_op,
.context = (uint8_t *) &signal,
.data = 0
};
Expand Down
Loading