Skip to content

Refactor atomics to enforce API-based access with relaxed load/store #4

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

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Refactor atomics to enforce API-based access with relaxed load/store #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion ompi/communicator/comm_cid.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@
#include "pmix.h"

/* for use when we don't have a PMIx that supports CID generation */
opal_atomic_int64_t ompi_comm_next_base_cid = 1;
opal_atomic_int64_t ompi_comm_next_base_cid = OPAL_ATOMIC_VAR_INIT(1);

/* A macro comparing two CIDs */
#define OMPI_COMM_CID_IS_LOWER(comm1,comm2) ( ((comm1)->c_index < (comm2)->c_index)? 1:0)
Expand Down
4 changes: 2 additions & 2 deletions ompi/communicator/comm_request.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ int ompi_comm_request_schedule_append_w_flags(ompi_comm_request_t *request, ompi
static int ompi_comm_request_progress (void)
{
ompi_comm_request_t *request, *next;
static opal_atomic_int32_t progressing = 0;
static opal_atomic_int32_t progressing = OPAL_ATOMIC_VAR_INIT(0);
int completed = 0;

/* don't allow re-entry */
Expand Down Expand Up @@ -175,7 +175,7 @@ static int ompi_comm_request_progress (void)
}

opal_mutex_unlock (&ompi_comm_request_mutex);
progressing = 0;
opal_atomic_store(&progressing, 0);

return completed;
}
Expand Down
16 changes: 8 additions & 8 deletions ompi/communicator/ft/comm_ft_detector.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ static opal_event_base_t* fd_event_base = NULL;
static void fd_event_cb(int fd, short flags, void* pdetector);

static bool comm_detector_use_thread = false;
static opal_atomic_int32_t fd_thread_active = 0;
static opal_atomic_int32_t fd_thread_active = OPAL_ATOMIC_VAR_INIT(0);
static opal_thread_t fd_thread;
static void* fd_progress(opal_object_t* obj);

Expand Down Expand Up @@ -168,8 +168,8 @@ int ompi_comm_failure_detector_init(void) {
fd_thread.t_arg = NULL;
ret = opal_thread_start(&fd_thread);
if( OPAL_SUCCESS != ret ) goto cleanup;
while( 0 == fd_thread_active ); /* wait for the fd thread initialization */
if( 0 > fd_thread_active ) goto cleanup;
while( 0 == opal_atomic_load(&fd_thread_active) ); /* wait for the fd thread initialization */
if( 0 > opal_atomic_load(&fd_thread_active) ) goto cleanup;
}

return OMPI_SUCCESS;
Expand Down Expand Up @@ -218,18 +218,18 @@ int ompi_comm_failure_detector_finalize(void) {
#endif
while( observing == detector->hb_observing ) {
/* If observed process changed, recheck if local*/
if( !(0 < fd_thread_active) )
if( !(0 < opal_atomic_load(&fd_thread_active)) )
{
opal_progress();
}
}
}

if( 0 < fd_thread_active ) {
if( 0 < opal_atomic_load(&fd_thread_active) ) {
void* tret;
/* this is not a race condition. Accesses are serialized, we use the
* atomic for the mfence part of it. */
OPAL_THREAD_ADD_FETCH32(&fd_thread_active, -fd_thread_active);
OPAL_THREAD_ADD_FETCH32(&fd_thread_active, -opal_atomic_load(&fd_thread_active));
opal_event_base_loopbreak(fd_event_base);
opal_thread_join(&fd_thread, &tret);
}
Expand Down Expand Up @@ -587,9 +587,9 @@ void* fd_progress(opal_object_t* obj) {
return OPAL_THREAD_CANCELLED;
}
OPAL_THREAD_ADD_FETCH32(&fd_thread_active, 1);
while( 1 == fd_thread_active ); /* wait for init stage 2: start_detector */
while( 1 == opal_atomic_load(&fd_thread_active) ); /* wait for init stage 2: start_detector */
ret = MCA_PML_CALL(irecv(NULL, 0, MPI_BYTE, 0, MCA_COLL_BASE_TAG_FT_END, &ompi_mpi_comm_self.comm, &req));
while( fd_thread_active ) {
while( opal_atomic_load(&fd_thread_active) ) {
opal_event_loop(fd_event_base, OPAL_EVLOOP_ONCE);
#if 0
/* This test disabled because rdma emulation over TCP would not work without
Expand Down
2 changes: 1 addition & 1 deletion ompi/errhandler/errhandler.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -213,7 +213,7 @@ extern opal_atomic_int32_t ompi_instance_count;
*/
#define OMPI_ERR_INIT_FINALIZE(name) \
{ \
if (OPAL_UNLIKELY(0 == ompi_instance_count)) { \
if (OPAL_UNLIKELY(0 == opal_atomic_load(&ompi_instance_count))) { \
ompi_errhandler_invoke(NULL, NULL, -1, \
ompi_errcode_get_mpi_code(MPI_ERR_ARG), \
name); \
Expand Down
2 changes: 1 addition & 1 deletion ompi/instance/instance.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,7 +90,7 @@ enum {
OMPI_INSTANCE_FINALIZING = -2,
};

opal_atomic_int32_t ompi_instance_count = 0;
opal_atomic_int32_t ompi_instance_count = OPAL_ATOMIC_VAR_INIT(0);

static const char *ompi_instance_builtin_psets[] = {
"mpi://WORLD",
Expand Down
10 changes: 6 additions & 4 deletions ompi/mca/coll/xhc/coll_xhc_intrinsic.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,11 +43,13 @@ typedef size_t __attribute__((aligned(SIZEOF_SIZE_T))) xf_size_t;
// https://github.com/open-mpi/ompi/issues/9722

#if OPAL_USE_GCC_BUILTIN_ATOMICS || OPAL_USE_C11_ATOMICS
#define xhc_atomic_load_int(addr) __atomic_load_n(addr, __ATOMIC_RELAXED)
#define xhc_atomic_store_int(addr, val) __atomic_store_n(addr, val, __ATOMIC_RELAXED)
#define xhc_atomic_load_int(addr) opal_atomic_load((opal_atomic_int_t *)(addr))
#define xhc_atomic_store_int(addr, val) \
opal_atomic_store((opal_atomic_int_t *)(addr), (val))

#define xhc_atomic_load_size_t(addr) __atomic_load_n(addr, __ATOMIC_RELAXED)
#define xhc_atomic_store_size_t(addr, val) __atomic_store_n(addr, val, __ATOMIC_RELAXED)
#define xhc_atomic_load_size_t(addr) opal_atomic_load((opal_atomic_size_t *)(addr))
#define xhc_atomic_store_size_t(addr, val) \
opal_atomic_store((opal_atomic_size_t *)(addr), (val))
#else
#define xhc_atomic_load_int(addr) (*(addr))
#define xhc_atomic_store_int(addr, val) (*(addr) = (val))
Expand Down
2 changes: 1 addition & 1 deletion ompi/mca/common/monitoring/common_monitoring.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@

/*** Monitoring specific variables ***/
/* Keep tracks of how many components are currently using the common part */
static opal_atomic_int32_t mca_common_monitoring_hold = 0;
static opal_atomic_int32_t mca_common_monitoring_hold = OPAL_ATOMIC_VAR_INIT(0);
/* Output parameters */
int mca_common_monitoring_output_stream_id = -1;
static opal_output_stream_t mca_common_monitoring_output_stream_obj = {
Expand Down
6 changes: 3 additions & 3 deletions ompi/mca/common/ompio/common_ompio_buffer.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,7 @@ static opal_mutex_t mca_common_ompio_buffer_mutex; /* lock for thread s
static mca_allocator_base_component_t* mca_common_ompio_allocator_component=NULL;
static mca_allocator_base_module_t* mca_common_ompio_allocator=NULL;

static opal_atomic_int32_t mca_common_ompio_buffer_init = 0;
static opal_atomic_int32_t mca_common_ompio_buffer_init = OPAL_ATOMIC_VAR_INIT(0);
static int32_t mca_common_ompio_pagesize=4096;
static void* mca_common_ompio_buffer_alloc_seg ( void *ctx, size_t *size );
static void mca_common_ompio_buffer_free_seg ( void *ctx, void *buf );
Expand Down Expand Up @@ -145,7 +145,7 @@ void *mca_common_ompio_alloc_buf ( ompio_file_t *fh, size_t bufsize )
{
char *tmp=NULL;

if ( !mca_common_ompio_buffer_init ){
if ( !opal_atomic_load(&mca_common_ompio_buffer_init) ){
mca_common_ompio_buffer_alloc_init ();
}

Expand All @@ -159,7 +159,7 @@ void *mca_common_ompio_alloc_buf ( ompio_file_t *fh, size_t bufsize )
void mca_common_ompio_release_buf ( ompio_file_t *fh, void *buf )
{

if ( !mca_common_ompio_buffer_init ){
if ( !opal_atomic_load(&mca_common_ompio_buffer_init) ){
/* Should not happen. You can not release a buf without
** having it allocated first.
*/
Expand Down
2 changes: 1 addition & 1 deletion ompi/mca/osc/monitoring/osc_monitoring_module.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@
OSC_MONITORING_SET_TEMPLATE_FCT_NAME(template) (ompi_osc_base_module_t*module) \
{ \
/* Define the ompi_osc_monitoring_module_## template ##_init_done variable */ \
opal_atomic_int32_t init_done = 0; \
opal_atomic_int32_t init_done = OPAL_ATOMIC_VAR_INIT(0); \
/* Define and set the ompi_osc_monitoring_## template \
* ##_template variable. The functions recorded here are \
* linked to the original functions of the original \
Expand Down
2 changes: 1 addition & 1 deletion ompi/mca/pml/base/pml_base_bsend.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ static size_t mca_pml_bsend_size; /* adjusted size of user buffe
static size_t mca_pml_bsend_count; /* number of outstanding requests */
static size_t mca_pml_bsend_pagesz; /* mmap page size */
static int mca_pml_bsend_pagebits; /* number of bits in pagesz */
static opal_atomic_int32_t mca_pml_bsend_init = 0;
static opal_atomic_int32_t mca_pml_bsend_init = OPAL_ATOMIC_VAR_INIT(0);

/* defined in pml_base_open.c */
extern char *ompi_pml_base_bsend_allocator_name;
Expand Down
2 changes: 1 addition & 1 deletion ompi/mca/pml/base/pml_base_sendreq.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ static void mca_pml_base_send_request_destruct(mca_pml_base_send_request_t* req)
#if MPI_VERSION >= 4
int mca_pml_cancel_send_callback(struct ompi_request_t *request, int flag)
{
static opal_atomic_int32_t send_deprecate_count = 0;
static opal_atomic_int32_t send_deprecate_count = OPAL_ATOMIC_VAR_INIT(0);
int32_t val;

val = opal_atomic_add_fetch_32(&send_deprecate_count, 1);
Expand Down
2 changes: 1 addition & 1 deletion ompi/mca/pml/ob1/pml_ob1_progress.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ static inline int mca_pml_ob1_process_pending_accelerator_async_copies(void)
return count;
}

static opal_atomic_int32_t mca_pml_ob1_progress_needed = 0;
static opal_atomic_int32_t mca_pml_ob1_progress_needed = OPAL_ATOMIC_VAR_INIT(0);
int mca_pml_ob1_enable_progress(int32_t count)
{
int32_t progress_count = OPAL_ATOMIC_ADD_FETCH32(&mca_pml_ob1_progress_needed, count);
Expand Down
2 changes: 1 addition & 1 deletion ompi/request/grequestx.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@

static bool requests_initialized = false;
static opal_list_t requests;
static opal_atomic_int32_t active_requests = 0;
static opal_atomic_int32_t active_requests = OPAL_ATOMIC_VAR_INIT(0);
static bool in_progress = false;
static opal_mutex_t lock = OPAL_MUTEX_STATIC_INIT;

Expand Down
4 changes: 2 additions & 2 deletions ompi/runtime/ompi_mpi_init.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ const char ompi_version_string[] = OMPI_IDENT_STRING;
* Global variables and symbols for the MPI layer
*/

opal_atomic_int32_t ompi_mpi_state = OMPI_MPI_STATE_NOT_INITIALIZED;
opal_atomic_int32_t ompi_mpi_state = OPAL_ATOMIC_VAR_INIT(OMPI_MPI_STATE_NOT_INITIALIZED);
volatile bool ompi_rte_initialized = false;

bool ompi_mpi_thread_multiple = false;
Expand Down Expand Up @@ -371,7 +371,7 @@ int ompi_mpi_init(int argc, char **argv, int requested, int *provided,
// silently return successfully once the initializing
// thread has completed.
if (reinit_ok) {
while (ompi_mpi_state < OMPI_MPI_STATE_INIT_COMPLETED) {
while (opal_atomic_load(&ompi_mpi_state) < OMPI_MPI_STATE_INIT_COMPLETED) {
usleep(1);
}
return MPI_SUCCESS;
Expand Down
14 changes: 14 additions & 0 deletions opal/include/opal/sys/atomic.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,20 @@
#include "opal/opal_portable_platform.h"
#include "opal_stdatomic.h"

#if OPAL_USE_C11_ATOMICS
# define opal_atomic_load(addr) atomic_load_explicit(&((addr)->v), memory_order_relaxed)
# define opal_atomic_store(addr, value) \
atomic_store_explicit(&((addr)->v), (value), memory_order_relaxed)
# define opal_atomic_load_explicit(addr, order) atomic_load_explicit(&((addr)->v), (order))
# define opal_atomic_store_explicit(addr, value, order) \
atomic_store_explicit(&((addr)->v), (value), (order))
#else
# define opal_atomic_load(addr) (*(addr))
# define opal_atomic_store(addr, value) (*(addr) = (value))
# define opal_atomic_load_explicit(addr, order) opal_atomic_load(addr)
# define opal_atomic_store_explicit(addr, value, order) opal_atomic_store(addr, value)
#endif

BEGIN_C_DECLS

/**********************************************************************
Expand Down
57 changes: 31 additions & 26 deletions opal/include/opal/sys/atomic_stdc.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@
# include <stdatomic.h>
# include <stdint.h>

# define opal_atomic_address(addr) (&((addr)->v))

/**********************************************************************
*
* Memory Barriers
Expand Down Expand Up @@ -72,39 +74,41 @@ static inline void opal_atomic_rmb(void)
*********************************************************************/

# define opal_atomic_compare_exchange_strong_32(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_relaxed, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_relaxed, memory_order_relaxed)
# define opal_atomic_compare_exchange_strong_acq_32(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_acquire, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_acquire, memory_order_relaxed)
# define opal_atomic_compare_exchange_strong_rel_32(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_release, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_release, memory_order_relaxed)

# define opal_atomic_compare_exchange_strong_64(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_relaxed, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_relaxed, memory_order_relaxed)
# define opal_atomic_compare_exchange_strong_acq_64(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_acquire, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_acquire, memory_order_relaxed)
# define opal_atomic_compare_exchange_strong_rel_64(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_release, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_release, memory_order_relaxed)

# define opal_atomic_compare_exchange_strong_ptr(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_relaxed, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_relaxed, memory_order_relaxed)
# define opal_atomic_compare_exchange_strong_acq_ptr(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_acquire, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_acquire, memory_order_relaxed)
# define opal_atomic_compare_exchange_strong_rel_ptr(addr, compare, value) \
atomic_compare_exchange_strong_explicit(addr, compare, value, memory_order_release, \
memory_order_relaxed)
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_release, memory_order_relaxed)

# if OPAL_HAVE_C11_CSWAP_INT128

/* the C11 atomic compare-exchange is lock free so use it */
# define opal_atomic_compare_exchange_strong_128 atomic_compare_exchange_strong
# define opal_atomic_compare_exchange_strong_128(addr, compare, value) \
atomic_compare_exchange_strong_explicit(opal_atomic_address(addr), compare, value, \
memory_order_relaxed, memory_order_relaxed)

# define OPAL_HAVE_ATOMIC_COMPARE_EXCHANGE_128 1

Expand All @@ -116,7 +120,7 @@ __opal_attribute_always_inline__ static inline bool
opal_atomic_compare_exchange_strong_128(opal_atomic_int128_t *addr, opal_int128_t *oldval,
opal_int128_t newval)
{
opal_int128_t prev = __sync_val_compare_and_swap(addr, *oldval, newval);
opal_int128_t prev = __sync_val_compare_and_swap(&((addr)->v), *oldval, newval);
bool ret = prev == *oldval;
*oldval = prev;
return ret;
Expand All @@ -138,11 +142,11 @@ opal_atomic_compare_exchange_strong_128(opal_atomic_int128_t *addr, opal_int128_
*********************************************************************/

# define opal_atomic_swap_32(addr, value) \
atomic_exchange_explicit((_Atomic unsigned int *) addr, value, memory_order_relaxed)
atomic_exchange_explicit((_Atomic unsigned int *) opal_atomic_address(addr), value, memory_order_relaxed)
# define opal_atomic_swap_64(addr, value) \
atomic_exchange_explicit((_Atomic unsigned long *) addr, value, memory_order_relaxed)
atomic_exchange_explicit((_Atomic unsigned long *) opal_atomic_address(addr), value, memory_order_relaxed)
# define opal_atomic_swap_ptr(addr, value) \
atomic_exchange_explicit((_Atomic unsigned long *) addr, value, memory_order_relaxed)
atomic_exchange_explicit((_Atomic unsigned long *) opal_atomic_address(addr), value, memory_order_relaxed)


/**********************************************************************
Expand Down Expand Up @@ -184,12 +188,13 @@ static inline void opal_atomic_unlock(opal_atomic_lock_t *lock)
# define OPAL_ATOMIC_STDC_DEFINE_FETCH_OP(op, bits, type, operator) \
static inline type opal_atomic_fetch_##op##_##bits(opal_atomic_##type *addr, type value) \
{ \
return atomic_fetch_##op##_explicit(addr, value, memory_order_relaxed); \
return atomic_fetch_##op##_explicit(&((addr)->v), value, memory_order_relaxed); \
} \
\
static inline type opal_atomic_##op##_fetch_##bits(opal_atomic_##type *addr, type value) \
{ \
return atomic_fetch_##op##_explicit(addr, value, memory_order_relaxed) operator value; \
return atomic_fetch_##op##_explicit(&((addr)->v), value, memory_order_relaxed) \
operator value; \
}

OPAL_ATOMIC_STDC_DEFINE_FETCH_OP(add, 32, int32_t, +)
Expand All @@ -208,7 +213,7 @@ OPAL_ATOMIC_STDC_DEFINE_FETCH_OP(add, size_t, size_t, +)
OPAL_ATOMIC_STDC_DEFINE_FETCH_OP(sub, size_t, size_t, -)

# define opal_atomic_add(addr, value) \
(void) atomic_fetch_add_explicit(addr, value, memory_order_relaxed)
(void) atomic_fetch_add_explicit(opal_atomic_address(addr), value, memory_order_relaxed)

#include "opal/sys/atomic_impl_minmax_math.h"

Expand Down
Loading
Loading