Store type (point2point or collective) and priority of comms in traces

_STARPU_MPI_FUT_COLLECTIVE_SEND will be used later

mpi/src/mpi/starpu_mpi_mpi.c

@@ -403,7 +403,9 @@ static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
 	_starpu_mpi_simgrid_wait_req(&req->backend->data_request, &req->status_store, &req->queue, &req->done);
 #endif
 
-	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(req->node_tag.node.rank, req->node_tag.data_tag, starpu_data_get_size(req->data_handle), req->pre_sync_jobid, req->data_handle);
+	// this trace event is the start of the communication link:
+	// the last parameter, set to 0, is for communication priority (not used in this MPI backend)
+	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(_STARPU_MPI_FUT_POINT_TO_POINT_SEND, req->node_tag.node.rank, req->node_tag.data_tag, starpu_data_get_size(req->data_handle), req->pre_sync_jobid, req->data_handle, 0);
 
 	/* somebody is perhaps waiting for the MPI request to be posted */
 	STARPU_PTHREAD_MUTEX_LOCK(&req->backend->req_mutex);
@@ -892,6 +894,7 @@ static void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req)
 				_starpu_mpi_datatype_free(req->data_handle, &req->datatype);
 			}
 		}
+		// for recv requests, this event is the end of the communication link:
 		_STARPU_MPI_TRACE_TERMINATED(req, req->node_tag.node.rank, req->node_tag.data_tag);
 	}
 

mpi/src/nmad/starpu_mpi_nmad.c

@@ -111,6 +111,10 @@ static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
 	nm_sr_send_pack_data(req->backend->session, &(req->backend->data_request), &data);
 	nm_sr_send_set_priority(req->backend->session, &req->backend->data_request, req->prio);
 
+	// this trace event is the start of the communication link:
+	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(_STARPU_MPI_FUT_POINT_TO_POINT_SEND, req->node_tag.node.rank, req->node_tag.data_tag,
+			starpu_data_get_size(req->data_handle), req->pre_sync_jobid, req->data_handle, req->prio);
+
 	if (req->sync == 0)
 	{
 		req->ret = nm_sr_send_isend(req->backend->session, &(req->backend->data_request), req->backend->gate, req->node_tag.data_tag);
@@ -122,8 +126,6 @@ static void _starpu_mpi_isend_data_func(struct _starpu_mpi_req *req)
 		STARPU_ASSERT_MSG(req->ret == NM_ESUCCESS, "MPI_Issend returning %d", req->ret);
 	}
 
-	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(req->node_tag.node.rank, req->node_tag.data_tag, starpu_data_get_size(req->data_handle), req->pre_sync_jobid, req->data_handle);
-
 	_starpu_mpi_handle_pending_request(req);
 
 	_STARPU_MPI_LOG_OUT();
@@ -353,7 +355,10 @@ void _starpu_mpi_handle_request_termination(struct _starpu_mpi_req *req,nm_sr_ev
 			_starpu_mpi_datatype_free(req->data_handle, &req->datatype);
 		}
 	}
+
+	// for recv requests, this event is the end of the communication link:
 	_STARPU_MPI_TRACE_TERMINATED(req, req->node_tag.node.rank, req->node_tag.data_tag);
+
 	_starpu_mpi_release_req_data(req);
 
 	/* Execute the specified callback, if any */

mpi/src/nmad/starpu_mpi_nmad_unknown_datatype.c

@@ -105,7 +105,9 @@ void _starpu_mpi_isend_unknown_datatype(struct _starpu_mpi_req *req)
 		STARPU_ASSERT_MSG(req->ret == NM_ESUCCESS, "nm_sr_send_issend returning %d", req->ret);
 	}
 
-	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(req->node_tag.node.rank, req->node_tag.data_tag, starpu_data_get_size(req->data_handle), req->pre_sync_jobid, req->data_handle);
+	// this trace event is the start of the communication link:
+	_STARPU_MPI_TRACE_ISEND_SUBMIT_END(_STARPU_MPI_FUT_POINT_TO_POINT_SEND, req->node_tag.node.rank, req->node_tag.data_tag,
+			starpu_data_get_size(req->data_handle), req->pre_sync_jobid, req->data_handle, req->prio);
 
 	_starpu_mpi_handle_pending_request(req);
 

mpi/src/starpu_mpi_fxt.h

@@ -29,6 +29,9 @@ extern "C"
 {
 #endif
 
+#define _STARPU_MPI_FUT_POINT_TO_POINT_SEND		0x100
+#define _STARPU_MPI_FUT_COLLECTIVE_SEND			0x101
+
 #define _STARPU_MPI_FUT_START				0x5201
 #define _STARPU_MPI_FUT_STOP				0x5202
 #define _STARPU_MPI_FUT_BARRIER				0x5203
@@ -60,6 +63,7 @@ extern "C"
 #define _STARPU_MPI_FUT_DRIVER_RUN_BEGIN		0x5216
 #define _STARPU_MPI_FUT_DRIVER_RUN_END			0x5217
 #define _STARPU_MPI_FUT_DATA_SET_TAG			0x5218
+// available codes: 0x5219 and > 0x5220
 
 #ifdef STARPU_USE_FXT
 
@@ -73,8 +77,8 @@ extern "C"
 } while (0)
 #define _STARPU_MPI_TRACE_ISEND_SUBMIT_BEGIN(dest, data_tag, size)	\
 	FUT_FULL_PROBE4(_STARPU_FUT_KEYMASK_MPI, _STARPU_MPI_FUT_ISEND_SUBMIT_BEGIN, (dest), (data_tag), (size), _starpu_gettid());
-#define _STARPU_MPI_TRACE_ISEND_SUBMIT_END(dest, data_tag, size, jobid, handle)	\
-	FUT_FULL_PROBE6(_STARPU_FUT_KEYMASK_MPI, _STARPU_MPI_FUT_ISEND_SUBMIT_END, (dest), (data_tag), (size), (jobid), (handle), _starpu_gettid());
+#define _STARPU_MPI_TRACE_ISEND_SUBMIT_END(type, dest, data_tag, size, jobid, handle, prio)	\
+	FUT_FULL_PROBE8(_STARPU_FUT_KEYMASK_MPI, _STARPU_MPI_FUT_ISEND_SUBMIT_END, (type), (dest), (data_tag), (size), (jobid), (handle), (prio), _starpu_gettid());
 #define _STARPU_MPI_TRACE_IRECV_SUBMIT_BEGIN(src, data_tag)	\
 	FUT_FULL_PROBE3(_STARPU_FUT_KEYMASK_MPI, _STARPU_MPI_FUT_IRECV_SUBMIT_BEGIN, (src), (data_tag), _starpu_gettid());
 #define _STARPU_MPI_TRACE_IRECV_SUBMIT_END(src, data_tag)	\
@@ -150,7 +154,7 @@ extern "C"
 #define _STARPU_MPI_TRACE_STOP(a, b)				do {} while(0);
 #define _STARPU_MPI_TRACE_BARRIER(a, b, c)			do {} while(0);
 #define _STARPU_MPI_TRACE_ISEND_SUBMIT_BEGIN(a, b, c)		do {} while(0);
-#define _STARPU_MPI_TRACE_ISEND_SUBMIT_END(a, b, c, d, e)	do {} while(0);
+#define _STARPU_MPI_TRACE_ISEND_SUBMIT_END(a, b, c, d, e, f, g)	do {} while(0);
 #define _STARPU_MPI_TRACE_IRECV_SUBMIT_BEGIN(a, b)		do {} while(0);
 #define _STARPU_MPI_TRACE_IRECV_SUBMIT_END(a, b)		do {} while(0);
 #define _STARPU_MPI_TRACE_ISEND_COMPLETE_BEGIN(a, b, c)		do {} while(0);

src/common/fxt.h

@@ -656,6 +656,18 @@ do {									\
         FUT_RAW_ALWAYS_PROBE7(FUT_CODE(CODE, 7),P1,P2,P3,P4,P5,P6,P7); \
 } while (0)
 
+#ifndef FUT_RAW_ALWAYS_PROBE8
+#define FUT_RAW_ALWAYS_PROBE8(CODE,P1,P2,P3,P4,P5,P6,P7,P8) do {	\
+		unsigned long *__args __attribute__((unused))=	\
+			fut_getstampedbuffer(CODE,		\
+					     FUT_SIZE(8)); \
+		*(__args++)=(unsigned long)(P1);*(__args++)=(unsigned long)(P2);*(__args++)=(unsigned long)(P3);*(__args++)=(unsigned long)(P4);*(__args++)=(unsigned long)(P5);*(__args++)=(unsigned long)(P6);*(__args++)=(unsigned long)(P7);*(__args++)=(unsigned long)(P8);				\
+		fut_commitstampedbuffer(FUT_SIZE(8)); \
+	} while (0)
+#endif
+#define FUT_DO_ALWAYS_PROBE8(CODE,P1,P2,P3,P4,P5,P6,P7,P8) do { \
+        FUT_RAW_ALWAYS_PROBE8(FUT_CODE(CODE, 8),P1,P2,P3,P4,P5,P6,P7,P8); \
+} while (0)
 
 /* full probes */
 #ifndef FUT_FULL_PROBE0
@@ -722,6 +734,13 @@ do {									\
 } while(0)
 #endif
 
+#ifndef FUT_FULL_PROBE8
+#define FUT_FULL_PROBE8(KEYMASK,CODE,P1,P2,P3,P4,P5,P6,P7,P8) do { \
+        if( KEYMASK & fut_active ) { \
+                FUT_RAW_ALWAYS_PROBE8(FUT_CODE(CODE, 8),P1,P2,P3,P4,P5,P6,P7,P8); \
+        } \
+} while(0)
+#endif
 
 #define _STARPU_TRACE_NEW_MEM_NODE(nodeid)			do {\
 	if (_starpu_fxt_started) \

src/debug/traces/starpu_fxt.c

@@ -2990,11 +2990,13 @@ static void handle_mpi_isend_submit_begin(struct fxt_ev_64 *ev, struct starpu_fx
 static int mpi_warned;
 static void handle_mpi_isend_submit_end(struct fxt_ev_64 *ev, struct starpu_fxt_options *options)
 {
-	int dest = ev->param[0];
-	int mpi_tag = ev->param[1];
-	size_t size = ev->param[2];
-	long jobid = ev->param[3];
-	unsigned long handle = ev->param[4];
+	unsigned type = ev->param[0];
+	int dest = ev->param[1];
+	int mpi_tag = ev->param[2];
+	size_t size = ev->param[3];
+	long jobid = ev->param[4];
+	unsigned long handle = ev->param[5];
+	int prio = ev->param[6];
 	double date = get_event_time_stamp(ev, options);
 
 	if (out_paje_file)
@@ -3011,7 +3013,7 @@ static void handle_mpi_isend_submit_end(struct fxt_ev_64 *ev, struct starpu_fxt_
 		}
 	}
 	else
-		_starpu_fxt_mpi_add_send_transfer(options->file_rank, dest, mpi_tag, size, date, jobid, handle);
+		_starpu_fxt_mpi_add_send_transfer(options->file_rank, dest, mpi_tag, size, date, jobid, handle, type, prio);
 }
 
 static void handle_mpi_irecv_submit_begin(struct fxt_ev_64 *ev, struct starpu_fxt_options *options)

src/debug/traces/starpu_fxt.h

@@ -60,7 +60,7 @@ void _starpu_fxt_dag_add_sync_point(void);
  */
 
 int _starpu_fxt_mpi_find_sync_point(char *filename_in, uint64_t *offset, int *key, int *rank);
-void _starpu_fxt_mpi_add_send_transfer(int src, int dst, long mpi_tag, size_t size, float date, long jobid, unsigned long handle);
+void _starpu_fxt_mpi_add_send_transfer(int src, int dst, long mpi_tag, size_t size, float date, long jobid, unsigned long handle, unsigned type, int prio);
 void _starpu_fxt_mpi_add_recv_transfer(int src, int dst, long mpi_tag, float date, long jobid, unsigned long handle);
 void _starpu_fxt_display_mpi_transfers(struct starpu_fxt_options *options, int *ranks, FILE *out_paje_file, FILE* out_comms_file);
 

src/debug/traces/starpu_fxt_mpi.c

@@ -36,6 +36,8 @@ LIST_TYPE(mpi_transfer,
 	long jobid;
 	double bandwidth;
 	unsigned long handle;
+	unsigned type;
+	int prio;
 );
 
 /* Returns 0 if a barrier is found, -1 otherwise. In case of success, offset is
@@ -120,7 +122,7 @@ unsigned mpi_recvs_used[STARPU_FXT_MAX_FILES] = {0};
 unsigned mpi_recvs_matched[STARPU_FXT_MAX_FILES][STARPU_FXT_MAX_FILES] = { {0} };
 unsigned mpi_sends_matched[STARPU_FXT_MAX_FILES][STARPU_FXT_MAX_FILES] = { {0} };
 
-void _starpu_fxt_mpi_add_send_transfer(int src, int dst STARPU_ATTRIBUTE_UNUSED, long mpi_tag, size_t size, float date, long jobid, unsigned long handle)
+void _starpu_fxt_mpi_add_send_transfer(int src, int dst STARPU_ATTRIBUTE_UNUSED, long mpi_tag, size_t size, float date, long jobid, unsigned long handle, unsigned type, int prio)
 {
 	STARPU_ASSERT(src >= 0);
 	if (src >= STARPU_FXT_MAX_FILES)
@@ -149,6 +151,8 @@ void _starpu_fxt_mpi_add_send_transfer(int src, int dst STARPU_ATTRIBUTE_UNUSED,
 	mpi_sends[src][slot].date = date;
 	mpi_sends[src][slot].jobid = jobid;
 	mpi_sends[src][slot].handle = handle;
+	mpi_sends[src][slot].type = type;
+	mpi_sends[src][slot].prio = prio;
 }
 
 void _starpu_fxt_mpi_add_recv_transfer(int src STARPU_ATTRIBUTE_UNUSED, int dst, long mpi_tag, float date, long jobid, unsigned long handle)
@@ -218,6 +222,19 @@ struct mpi_transfer *try_to_match_send_transfer(int src STARPU_ATTRIBUTE_UNUSED,
 
 static unsigned long mpi_com_id = 0;
 
+static const char* get_mpi_type_str(unsigned mpi_type)
+{
+	switch (mpi_type)
+	{
+		case _STARPU_MPI_FUT_POINT_TO_POINT_SEND:
+			return "PointToPoint";
+		case _STARPU_MPI_FUT_COLLECTIVE_SEND:
+			return "Collective";
+		default:
+			return "Unknown";
+	}
+}
+
 static void display_all_transfers_from_trace(FILE *out_paje_file, FILE *out_comms_file, unsigned n)
 {
 	unsigned slot[STARPU_FXT_MAX_FILES] = { 0 }, node;
@@ -331,9 +348,11 @@ static void display_all_transfers_from_trace(FILE *out_paje_file, FILE *out_comm
 
 			char str_mpi_tag[STARPU_POTI_STR_LEN];
 			snprintf(str_mpi_tag, sizeof(str_mpi_tag), "%ld", mpi_tag);
+			char str_priority[STARPU_POTI_STR_LEN];
+			snprintf(str_priority, sizeof(str_priority), "%d", cur->prio);
 			char str_handle[STARPU_POTI_STR_LEN];
 			snprintf(str_handle, sizeof(str_handle), "%lx", send_handle);
-			poti_user_StartLink(_starpu_poti_MpiLinkStart, start_date, "MPIroot", "MPIL", mpi_container, paje_value, paje_key, 2, str_mpi_tag, str_handle);
+			poti_user_StartLink(_starpu_poti_MpiLinkStart, start_date, "MPIroot", "MPIL", mpi_container, paje_value, paje_key, 3, str_mpi_tag, get_mpi_type_str(cur->type), str_priority, str_handle);
 
 			poti_SetVariable(start_date, mpi_container, "bwo_mpi", current_out_bandwidth[src]);
 			snprintf(mpi_container, sizeof(mpi_container), "%d_mpict", dst);
@@ -342,7 +361,7 @@ static void display_all_transfers_from_trace(FILE *out_paje_file, FILE *out_comm
 #else
 			fprintf(out_paje_file, "13	%.9f	%d_mpict	bwo_mpi	%f\n", start_date, src, current_out_bandwidth[src]);
 			fprintf(out_paje_file, "13	%.9f	%d_mpict	bwi_mpi	%f\n", start_date, dst, current_in_bandwidth[dst]);
-			fprintf(out_paje_file, "23	%.9f	MPIL	MPIroot	%lu	%d_mpict	mpicom_%lu	%ld	%lx\n", start_date, (unsigned long)size, src, id, mpi_tag, send_handle);
+			fprintf(out_paje_file, "23	%.9f	MPIL	MPIroot	%lu	%d_mpict	mpicom_%lu	%ld	%s	%d	%lx\n", start_date, (unsigned long)size, src, id, mpi_tag, get_mpi_type_str(cur->type), cur->prio, send_handle);
 			fprintf(out_paje_file, "19	%.9f	MPIL	MPIroot	%lu	%d_mpict	mpicom_%lu\n", end_date, (unsigned long)size, dst, id);
 #endif
 
@@ -360,6 +379,8 @@ static void display_all_transfers_from_trace(FILE *out_paje_file, FILE *out_comm
 				if (match->jobid != -1)
 					fprintf(out_comms_file, "RecvJobId: %d_%ld\n", dst, match->jobid);
 				fprintf(out_comms_file, "Size: %lu\n", (unsigned long)size);
+				fprintf(out_comms_file, "Priority: %d\n", cur->prio);
+				fprintf(out_comms_file, "Type: %s\n", get_mpi_type_str(cur->type));
 				fprintf(out_comms_file, "\n");
 			}
 		}

src/debug/traces/starpu_paje.c

@@ -73,7 +73,7 @@ void _starpu_fxt_write_paje_header(FILE *file STARPU_ATTRIBUTE_UNUSED, struct st
 							     "Size string",
 							     "Dest string");
 	}
-	_starpu_poti_MpiLinkStart = poti_header_DeclareEvent(PAJE_StartLink, 2, "MPITAG string", "Handle string");
+	_starpu_poti_MpiLinkStart = poti_header_DeclareEvent(PAJE_StartLink, 4, "MPITAG string", "MPIType string", "Priority string", "Handle string");
 #endif
 #else
 	poti_header(1,1);
@@ -232,6 +232,8 @@ void _starpu_fxt_write_paje_header(FILE *file STARPU_ATTRIBUTE_UNUSED, struct st
 	fprintf(file, "%%	StartContainer	string\n");
 	fprintf(file, "%%	Key	string\n");
 	fprintf(file, "%%	MPITAG	string\n");
+	fprintf(file, "%%	MPIType	string\n");
+	fprintf(file, "%%	Priority	string\n");
 	fprintf(file, "%%	Handle	string\n");
 	fprintf(file, "%%EndEventDef\n");
 	fprintf(file, "%%EventDef	PajeStartLink	24\n");