Skip to content

UCX Notified Communication #20

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
joe-explr wants to merge 21 commits into
base: notified-rma
Choose a base branch
from
Open

UCX Notified Communication #20

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
68aa529
First draft and TODOs for notified comm
devreal Oct 24, 2025
4ed64cc
osc/sm: Add notification support for put/get operations
joe-explr Nov 4, 2025
7d1c62d
osc/sm: Nit picking edits
joe-explr Nov 4, 2025
60edd52
Public APIs for:
Nov 19, 2025
6600f17
Edits for Public APIs:
Dec 22, 2025
1de2197
Notified RMA counters memory allocation in the shared memory segment …
Feb 3, 2026
39764a7
Editing Notified RMA implementation
Feb 4, 2026
32bafd4
Editing Notified RMA implementation with new design
Feb 19, 2026
a8c825d
Implementing Notify Query
Mar 4, 2026
5a14d83
Changes to Notify Query
Mar 4, 2026
782d3d8
Implemented Set Notify Query
Mar 4, 2026
268cb1a
Changes to Notify Query
Mar 4, 2026
f01f16d
Configuring memory barries to propagate the update
Mar 11, 2026
b96b5bf
Reset notify value Implementation Signed-off-by: Joseph Antony <jajo…
Mar 11, 2026
18426d7
Addressing review comments and bug fixes Signed-off-by: Joseph Anton…
Mar 13, 2026
2ec1c98
Removing Duplicate changes Signed-off-by: Joseph Antony <jajoseph.an…
Mar 13, 2026
9f4abbf
Bug fixes
joe-explr Mar 30, 2026
15c4a68
Addressing review comment
joe-explr Mar 31, 2026
ac17159
UCX Notified Communication
joe-explr Mar 31, 2026
487a080
UCX Notified Communication changes
joe-explr Jul 21, 2026
2ac49ff
Get and set Notify changes
joe-explr Jul 24, 2026
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
37 changes: 37 additions & 0 deletions ompi/include/mpi.h.in
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -764,6 +764,7 @@ enum {
#define MPI_ERR_SESSION 78
#define MPI_ERR_VALUE_TOO_LARGE 79
#define MPI_ERR_ERRHANDLER 80
#define MPI_ERR_NOTIFY_IDX 81

/* Per MPI-3 p349 47, MPI_ERR_LASTCODE must be >= the last predefined
MPI_ERR_<foo> code. Set the last code to allow some room for adding
Expand Down Expand Up @@ -1917,6 +1918,14 @@ OMPI_DECLSPEC int MPI_Get_c(void *origin_addr, MPI_Count origin_count,
MPI_Datatype origin_datatype, int target_rank,
MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, MPI_Win win);
OMPI_DECLSPEC int MPI_Get_notify(void *origin_addr, int origin_count,
MPI_Datatype origin_datatype, int target_rank,
MPI_Aint target_disp, int target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int MPI_Get_notify_c(void *origin_addr, MPI_Count origin_count,
MPI_Datatype origin_datatype, int target_rank,
MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int MPI_Get_accumulate(const void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
void *result_addr, int result_count, MPI_Datatype result_datatype,
int target_rank, MPI_Aint target_disp, int target_count,
Expand Down Expand Up @@ -2180,6 +2189,12 @@ OMPI_DECLSPEC int MPI_Put(const void *origin_addr, int origin_count, MPI_Dataty
OMPI_DECLSPEC int MPI_Put_c(const void *origin_addr, MPI_Count origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, MPI_Win win);
OMPI_DECLSPEC int MPI_Put_notify(const void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, int target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int MPI_Put_notify_c(const void *origin_addr, MPI_Count origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int MPI_Query_thread(int *provided);
OMPI_DECLSPEC int MPI_Raccumulate(const void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, int target_count,
Expand Down Expand Up @@ -2604,13 +2619,17 @@ OMPI_DECLSPEC int MPI_Win_get_errhandler(MPI_Win win, MPI_Errhandler *errhandle
OMPI_DECLSPEC int MPI_Win_get_group(MPI_Win win, MPI_Group *group);
OMPI_DECLSPEC int MPI_Win_get_info(MPI_Win win, MPI_Info *info_used);
OMPI_DECLSPEC int MPI_Win_get_name(MPI_Win win, char *win_name, int *resultlen);
OMPI_DECLSPEC int MPI_Win_get_notify_value(MPI_Win win, int notification_idx, MPI_Count *value);
OMPI_DECLSPEC int MPI_Win_reset_notify_value(MPI_Win win, int notification_idx, MPI_Count *value);
OMPI_DECLSPEC int MPI_Win_get_num_notify(MPI_Win win, int target_rank, int *num_notifications);
OMPI_DECLSPEC int MPI_Win_lock(int lock_type, int rank, int mpi_assert, MPI_Win win);
OMPI_DECLSPEC int MPI_Win_lock_all(int mpi_assert, MPI_Win win);
OMPI_DECLSPEC int MPI_Win_post(MPI_Group group, int mpi_assert, MPI_Win win);
OMPI_DECLSPEC int MPI_Win_set_attr(MPI_Win win, int win_keyval, void *attribute_val);
OMPI_DECLSPEC int MPI_Win_set_errhandler(MPI_Win win, MPI_Errhandler errhandler);
OMPI_DECLSPEC int MPI_Win_set_info(MPI_Win win, MPI_Info info);
OMPI_DECLSPEC int MPI_Win_set_name(MPI_Win win, const char *win_name);
OMPI_DECLSPEC int MPI_Win_set_num_notify(MPI_Win win, MPI_Info info, int num_notifications);
OMPI_DECLSPEC int MPI_Win_shared_query(MPI_Win win, int rank, MPI_Aint *size, int *disp_unit, void *baseptr);
OMPI_DECLSPEC int MPI_Win_shared_query_c(MPI_Win win, int rank, MPI_Aint *size, MPI_Aint *disp_unit, void *baseptr);
OMPI_DECLSPEC int MPI_Win_start(MPI_Group group, int mpi_assert, MPI_Win win);
Expand Down Expand Up @@ -3091,6 +3110,14 @@ OMPI_DECLSPEC int PMPI_Get_c(void *origin_addr, MPI_Count origin_count,
MPI_Datatype origin_datatype, int target_rank,
MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, MPI_Win win);
OMPI_DECLSPEC int PMPI_Get_notify(void *origin_addr, int origin_count,
MPI_Datatype origin_datatype, int target_rank,
MPI_Aint target_disp, int target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int PMPI_Get_notify_c(void *origin_addr, MPI_Count origin_count,
MPI_Datatype origin_datatype, int target_rank,
MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int PMPI_Get_accumulate(const void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
void *result_addr, int result_count, MPI_Datatype result_datatype,
int target_rank, MPI_Aint target_disp, int target_count,
Expand Down Expand Up @@ -3354,6 +3381,12 @@ OMPI_DECLSPEC int PMPI_Put(const void *origin_addr, int origin_count, MPI_Datat
OMPI_DECLSPEC int PMPI_Put_c(const void *origin_addr, MPI_Count origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, MPI_Win win);
OMPI_DECLSPEC int PMPI_Put_notify(const void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, int target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int PMPI_Put_notify_c(const void *origin_addr, MPI_Count origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, MPI_Count target_count,
MPI_Datatype target_datatype, int notification_idx, MPI_Win win);
OMPI_DECLSPEC int PMPI_Query_thread(int *provided);
OMPI_DECLSPEC int PMPI_Raccumulate(const void *origin_addr, int origin_count, MPI_Datatype origin_datatype,
int target_rank, MPI_Aint target_disp, int target_count,
Expand Down Expand Up @@ -3778,13 +3811,17 @@ OMPI_DECLSPEC int PMPI_Win_get_errhandler(MPI_Win win, MPI_Errhandler *errhandl
OMPI_DECLSPEC int PMPI_Win_get_group(MPI_Win win, MPI_Group *group);
OMPI_DECLSPEC int PMPI_Win_get_info(MPI_Win win, MPI_Info *info_used);
OMPI_DECLSPEC int PMPI_Win_get_name(MPI_Win win, char *win_name, int *resultlen);
OMPI_DECLSPEC int PMPI_Win_get_notify_value(MPI_Win win, int notification_idx, MPI_Count *value);
OMPI_DECLSPEC int PMPI_Win_reset_notify_value(MPI_Win win, int notification_idx, MPI_Count *value);
OMPI_DECLSPEC int PMPI_Win_get_num_notify(MPI_Win win, int target_rank, int *num_notifications);
OMPI_DECLSPEC int PMPI_Win_lock(int lock_type, int rank, int mpi_assert, MPI_Win win);
OMPI_DECLSPEC int PMPI_Win_lock_all(int mpi_assert, MPI_Win win);
OMPI_DECLSPEC int PMPI_Win_post(MPI_Group group, int mpi_assert, MPI_Win win);
OMPI_DECLSPEC int PMPI_Win_set_attr(MPI_Win win, int win_keyval, void *attribute_val);
OMPI_DECLSPEC int PMPI_Win_set_errhandler(MPI_Win win, MPI_Errhandler errhandler);
OMPI_DECLSPEC int PMPI_Win_set_info(MPI_Win win, MPI_Info info);
OMPI_DECLSPEC int PMPI_Win_set_name(MPI_Win win, const char *win_name);
OMPI_DECLSPEC int PMPI_Win_set_num_notify(MPI_Win win, MPI_Info info, int num_notifications);
OMPI_DECLSPEC int PMPI_Win_shared_query(MPI_Win win, int rank, MPI_Aint *size, int *disp_unit, void *baseptr);
OMPI_DECLSPEC int PMPI_Win_shared_query_c(MPI_Win win, int rank, MPI_Aint *size, MPI_Aint *disp_unit, void *baseptr);
OMPI_DECLSPEC int PMPI_Win_start(MPI_Group group, int mpi_assert, MPI_Win win);
Expand Down
1 change: 1 addition & 0 deletions ompi/include/mpif-values.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -301,6 +301,7 @@
'MPI_ERR_SESSION': 78,
'MPI_ERR_VALUE_TOO_LARGE': 79,
'MPI_ERR_ERRHANDLER': 80,
'MPI_ERR_NOTIFY_IDX': 81,
'MPI_ERR_LASTCODE': 92,
'MPI_IDENT': 0,
'MPI_CONGRUENT': 1,
Expand Down
69 changes: 69 additions & 0 deletions ompi/mca/osc/osc.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -216,6 +216,15 @@ typedef int (*ompi_osc_base_module_put_fn_t)(const void *origin_addr,
struct ompi_datatype_t *target_dt,
struct ompi_win_t *win);

typedef int (*ompi_osc_base_module_put_notify_fn_t)(const void *origin_addr,
size_t origin_count,
struct ompi_datatype_t *origin_dt,
int target,
ptrdiff_t target_disp,
size_t target_count,
struct ompi_datatype_t *target_dt,
int notify,
struct ompi_win_t *win);

typedef int (*ompi_osc_base_module_get_fn_t)(void *origin_addr,
size_t origin_count,
Expand All @@ -226,6 +235,23 @@ typedef int (*ompi_osc_base_module_get_fn_t)(void *origin_addr,
struct ompi_datatype_t *target_dt,
struct ompi_win_t *win);

typedef int (*ompi_osc_base_module_get_notify_fn_t)(void *origin_addr,
size_t origin_count,
struct ompi_datatype_t *origin_dt,
int target,
ptrdiff_t target_disp,
size_t target_count,
struct ompi_datatype_t *target_dt,
int notify,
struct ompi_win_t *win);

typedef int (*ompi_osc_base_module_win_get_notify_value_fn_t)(struct ompi_win_t *win,
int notify,
OMPI_MPI_COUNT_TYPE *value);

typedef int (*ompi_osc_base_module_win_reset_notify_value_fn_t)(struct ompi_win_t *win,
int notify,
OMPI_MPI_COUNT_TYPE *value);

typedef int (*ompi_osc_base_module_accumulate_fn_t)(const void *origin_addr,
size_t origin_count,
Expand Down Expand Up @@ -276,6 +302,17 @@ typedef int (*ompi_osc_base_module_rput_fn_t)(const void *origin_addr,
struct ompi_win_t *win,
struct ompi_request_t **request);

typedef int (*ompi_osc_base_module_rput_notify_fn_t)(const void *origin_addr,
size_t origin_count,
struct ompi_datatype_t *origin_dt,
int target,
ptrdiff_t target_disp,
size_t target_count,
struct ompi_datatype_t *target_dt,
int notify,
struct ompi_win_t *win,
struct ompi_request_t **request);

typedef int (*ompi_osc_base_module_rget_fn_t)(void *origin_addr,
size_t origin_count,
struct ompi_datatype_t *origin_dt,
Expand All @@ -286,6 +323,16 @@ typedef int (*ompi_osc_base_module_rget_fn_t)(void *origin_addr,
struct ompi_win_t *win,
struct ompi_request_t **request);

typedef int (*ompi_osc_base_module_rget_notify_fn_t)(void *origin_addr,
size_t origin_count,
struct ompi_datatype_t *origin_dt,
int target,
ptrdiff_t target_disp,
size_t target_count,
struct ompi_datatype_t *target_dt,
int notify,
struct ompi_win_t *win,
struct ompi_request_t **request);

typedef int (*ompi_osc_base_module_raccumulate_fn_t)(const void *origin_addr,
size_t origin_count,
Expand All @@ -312,6 +359,18 @@ typedef int (*ompi_osc_base_module_rget_accumulate_fn_t)(const void *origin_addr
struct ompi_win_t *win,
struct ompi_request_t **request);

/* Notified communication (MPI Standard Section 12.6). The put_notify,
* get_notify, rput_notify, rget_notify, win_get_notify_value and
* win_reset_notify_value function-pointer types are declared above; only the
* counter-management types are declared here. */
typedef int (*ompi_osc_base_module_win_set_num_notify_fn_t)(struct ompi_win_t *win,
struct opal_info_t *info,
int num_notifications);

typedef int (*ompi_osc_base_module_win_get_num_notify_fn_t)(struct ompi_win_t *win,
int target_rank,
int *num_notifications);

typedef int (*ompi_osc_base_module_fence_fn_t)(int mpi_assert, struct ompi_win_t *win);


Expand Down Expand Up @@ -370,6 +429,7 @@ typedef int (*ompi_osc_base_module_flush_local_all_fn_t)(struct ompi_win_t *win)
* free to create a structure that inherits this one for use as the
* module structure.
*/

struct ompi_osc_base_module_4_0_0_t {
ompi_osc_base_module_win_shared_query_fn_t osc_win_shared_query;

Expand All @@ -389,6 +449,15 @@ struct ompi_osc_base_module_4_0_0_t {
ompi_osc_base_module_raccumulate_fn_t osc_raccumulate;
ompi_osc_base_module_rget_accumulate_fn_t osc_rget_accumulate;

ompi_osc_base_module_put_notify_fn_t osc_put_notify;
ompi_osc_base_module_get_notify_fn_t osc_get_notify;
ompi_osc_base_module_rput_notify_fn_t osc_rput_notify;
ompi_osc_base_module_rget_notify_fn_t osc_rget_notify;
ompi_osc_base_module_win_get_notify_value_fn_t osc_win_get_notify_value;
ompi_osc_base_module_win_reset_notify_value_fn_t osc_win_reset_notify_value;
ompi_osc_base_module_win_set_num_notify_fn_t osc_win_set_num_notify;
ompi_osc_base_module_win_get_num_notify_fn_t osc_win_get_num_notify;

ompi_osc_base_module_fence_fn_t osc_fence;

ompi_osc_base_module_start_fn_t osc_start;
Expand Down
Loading
Loading