Merge branch 'cortex' into 'master'

Cortex

Integration of Cortex in CODES. Cortex can be enabled with --with-cortex=... when configuring CODES. When Cortex is enabled, it will be used in place of DUMPI and automatically translate MPI collectives using algorithms as implemented in Mpich.

This merge request also adds support for MPI_Sendrecv in the MPI replayer, and corrects an issue with CODES expecting the size of a posted MPI_Recv to match with the corresponding MPI_Send's size, which is not a requirement in the MPI standard.

See merge request !14

Makefile.am

@@ -39,6 +39,12 @@ include $(top_srcdir)/src/Makefile.subdir
 include $(top_srcdir)/tests/Makefile.subdir
 include $(top_srcdir)/doc/Makefile.subdir
 
+if USE_DEBUG
+AM_CPPFLAGS += -g
+AM_CFLAGS   += -g
+AM_CXXFLAGS += -g
+endif
+
 if USE_DARSHAN
 AM_CPPFLAGS += ${DARSHAN_CFLAGS} -DUSE_DARSHAN=1
 src_libcodes_la_SOURCES += src/workload/methods/codes-darshan-io-wrkld.c
@@ -55,4 +61,17 @@ if USE_DUMPI
 AM_CPPFLAGS += ${DUMPI_CFLAGS} -DUSE_DUMPI=1
 src_libcodes_la_SOURCES += src/workload/methods/codes-dumpi-trace-nw-wrkld.c
 LDADD += ${DUMPI_LIBS}
+if USE_CORTEX
+if USE_PYTHON
+if USE_CORTEX_PYTHON
+AM_CPPFLAGS += ${CORTEX_PYTHON_CFLAGS} -DENABLE_CORTEX_PYTHON=1
+LDADD += ${CORTEX_PYTHON_LIBS}
+AM_CPPFLAGS += ${PYTHON_CFLAGS}
+LDADD += ${PYTHON_LIBS}
+endif
 endif
+AM_CPPFLAGS += ${CORTEX_CFLAGS} -DENABLE_CORTEX=1
+LDADD += ${CORTEX_LIBS}
+endif
+endif
+

codes/codes-workload.h

@@ -69,6 +69,10 @@ struct recorder_params
 struct dumpi_trace_params {
    char file_name[MAX_NAME_LENGTH_WKLD];
    int num_net_traces;
+#ifdef ENABLE_CORTEX_PYTHON
+   char cortex_script[MAX_NAME_LENGTH_WKLD];
+   char cortex_class[MAX_NAME_LENGTH_WKLD];
+#endif
 };
 
 struct checkpoint_wrkld_params

configure.ac

@@ -84,6 +84,12 @@ AC_TRY_COMPILE([#include <mpi.h>], [int ret = MPI_Init(0, (void*)0)],
 PKG_CHECK_MODULES_STATIC([ROSS], [ross], [],
                   [AC_MSG_ERROR([Could not find working ross installation via pkg-config])])
 
+# check for enable-g
+AC_ARG_ENABLE([g],[AS_HELP_STRING([--enable-g],
+			[Build with GDB symbols])],
+		[use_debug=yes],[use_debug=no])
+AM_CONDITIONAL(USE_DEBUG, [test "x${use_debug}" = xyes])
+
 # check for Darshan
 AC_ARG_WITH([darshan],[AS_HELP_STRING([--with-darshan],
                         [Build with the darshan workload support])],
@@ -121,6 +127,54 @@ else
 	AM_CONDITIONAL(USE_DUMPI, false)
 fi
 
+# check for Cortex
+AC_ARG_WITH([cortex],[AS_HELP_STRING([--with-cortex@<:@=DIR@:>@],
+			[location of Cortex installation])])
+
+# check for Python
+AC_ARG_WITH([python],[AS_HELP_STRING([--with-python@<:@=DIR@:>@],
+			[location of Python 2.7 installation])])
+
+# check for Boost Python
+AC_ARG_WITH([boost],[AS_HELP_STRING([--with-boost@<:@=DIR@:>@],
+			[location of Boost Python installation])])
+
+if [ test "x${with_python}" != "x" -a "x${with_boost}" != "x"] ; then
+        AC_CHECK_FILES([${with_python}/lib/libpython2.7.a ${with_boost}/lib/libboost_python.a],
+                AM_CONDITIONAL(USE_PYTHON, true),
+                AC_MSG_ERROR(Could not find Python and/or Boost-Python libraries))
+        PYTHON_CFLAGS="-I${with_python}/include -I${with_boost}/include"
+        PYTHON_LIBS="-L${with_boost}/lib -lboost_python -L${with_python}/lib/ -lpython2.7"
+        AC_SUBST(PYTHON_LIBS)
+        AC_SUBST(PYTHON_CFLAGS)
+else
+        AM_CONDITIONAL(USE_PYTHON, false)
+fi
+
+if test "x${with_cortex}" != "x" ; then
+	 AC_CHECK_FILES([${with_cortex}/lib/libcortex.a ${with_cortex}/lib/libcortex-mpich.a],
+		AM_CONDITIONAL(USE_CORTEX, true),
+		AC_MSG_ERROR(Could not find Cortex libraries libcortex.a and/or libcortex-mpich.a))
+	CORTEX_CFLAGS="-I${with_cortex}/include"
+	CORTEX_LIBS="-L${with_cortex}/lib/ -lcortex-mpich -lcortex -lstdc++"
+	AC_SUBST(CORTEX_LIBS)
+	AC_SUBST(CORTEX_CFLAGS)
+else
+	AM_CONDITIONAL(USE_CORTEX, false)
+fi
+
+if [ test "x${with_cortex}" != "x" -a "x${with_python}" != "x" -a "x${with_boost}" != "x"] ; then
+	AC_CHECK_FILE([${with_cortex}/lib/libcortex-python.a],
+		AM_CONDITIONAL(USE_CORTEX_PYTHON, true),
+		AC_MSG_ERROR(Could not find library libcortex-python.a))
+	CORTEX_PYTHON_CFLAGS="-I${with_cortex}/include"
+	CORTEX_PYTHON_LIBS="-L${with_cortex}/lib/ -lcortex-python"
+	AC_SUBST(CORTEX_PYTHON_LIBS)
+	AC_SUBST(CORTEX_PYTHON_CFLAGS)
+else
+	AM_CONDITIONAL(USE_CORTEX_PYTHON, false)
+fi
+
 dnl ======================================================================
 dnl  Try harder to be valgrind safe
 dnl ======================================================================

maint/codes.pc.in

@@ -8,11 +8,17 @@ darshan_libs=@DARSHAN_LIBS@
 darshan_cflags=@DARSHAN_CFLAGS@
 dumpi_cflags=@DUMPI_CFLAGS@
 dumpi_libs=@DUMPI_LIBS@
+cortex_cflags=@CORTEX_CFLAGS@
+cortex_libs=@CORTEX_LIBS@
+python_cflags=@PYTHON_CFLAGS@
+python_libs=@PYTHON_LIBS@
+boost_cflags=@BOOST_CFLAGS@
+boost_libs=@BOOST_LIBS@
 
 Name: codes-base
 Description: Base functionality for CODES storage simulation
 Version: @PACKAGE_VERSION@
 URL: http://trac.mcs.anl.gov/projects/CODES
 Requires:
-Libs: -L${libdir} -lcodes ${ross_libs} ${darshan_libs} ${dumpi_libs} 
-Cflags: -I${includedir} ${ross_cflags} ${darshan_cflags} ${dumpi_cflags}
+Libs: -L${libdir} -lcodes ${ross_libs} ${darshan_libs} ${dumpi_libs} ${cortex_libs}
+Cflags: -I${includedir} ${ross_cflags} ${darshan_cflags} ${dumpi_cflags} ${cortex_cflags}

src/network-workloads/model-net-mpi-replay.c

@@ -51,6 +51,14 @@ struct codes_jobmap_ctx *jobmap_ctx;
 struct codes_jobmap_params_list jobmap_p;
 /* Xu's additions end */
 
+/* Variables for Cortex Support */
+/* Matthieu's additions start */
+#ifdef ENABLE_CORTEX_PYTHON
+static char cortex_file[512];
+static char cortex_class[512];
+#endif
+/* Matthieu's additions end */
+
 typedef struct nw_state nw_state;
 typedef struct nw_message nw_message;
 typedef int32_t dumpi_req_id;
@@ -635,10 +643,12 @@ static int rm_matching_send(nw_state * ns,
     int index = 0;
     qlist_for_each(ent, &ns->arrival_queue){
         qi = qlist_entry(ent, mpi_msgs_queue, ql);
-        if((qi->num_bytes == qitem->num_bytes)
-                && (qi->tag == qitem->tag || qitem->tag == -1)
+        if(//(qi->num_bytes == qitem->num_bytes) // it is not a requirement in MPI that the send and receive sizes match
+                // && 
+		(qi->tag == qitem->tag || qitem->tag == -1)
                 && ((qi->source_rank == qitem->source_rank) || qitem->source_rank == -1))
         {
+            qitem->num_bytes = qi->num_bytes;
             matched = 1;
             break;
         }
@@ -1145,6 +1155,10 @@ void nw_test_init(nw_state* s, tw_lp* lp)
        s->local_rank = lid.rank;
 //       printf("network LP nw id %d app id %d local rank %d generating events, lp gid is %ld \n", s->nw_id, 
 //               s->app_id, s->local_rank, lp->gid);
+#ifdef ENABLE_CORTEX_PYTHON
+	strcpy(params_d.cortex_script, cortex_file);
+	strcpy(params_d.cortex_class, cortex_class);
+#endif
    }
 
    wrkld_id = codes_workload_load("dumpi-trace-workload", params, s->app_id, s->local_rank);
@@ -1515,6 +1529,10 @@ const tw_optdef app_opt [] =
     TWOPT_CHAR("lp-io-dir", lp_io_dir, "Where to place io output (unspecified -> no output"),
     TWOPT_UINT("lp-io-use-suffix", lp_io_use_suffix, "Whether to append uniq suffix to lp-io directory (default 0)"),
 	TWOPT_CHAR("offset_file", offset_file, "offset file name"),
+#ifdef ENABLE_CORTEX_PYTHON
+	TWOPT_CHAR("cortex-file", cortex_file, "Python file (without .py) containing the CoRtEx translation class"),
+	TWOPT_CHAR("cortex-class", cortex_class, "Python class implementing the CoRtEx translator"),
+#endif
 	TWOPT_END()
 };
 
@@ -1555,8 +1573,14 @@ int main( int argc, char** argv )
     {
 	if(tw_ismaster())
 		printf("Usage: mpirun -np n ./modelnet-mpi-replay --sync=1/3"
-                " --workload_type=dumpi --workload_conf_file=prefix-workload-file-name"
-                " --alloc_file=alloc-file-name -- config-file-name\n"
+                " --workload_type=dumpi"
+		" --workload_conf_file=prefix-workload-file-name"
+                " --alloc_file=alloc-file-name"
+#ifdef ENABLE_CORTEX_PYTHON
+		" --cortex-file=cortex-file-name"
+		" --cortex-class=cortex-class-name"
+#endif
+		" -- config-file-name\n"
                 "See model-net/doc/README.dragonfly.txt and model-net/doc/README.torus.txt"
                 " for instructions on how to run the models with network traces ");
 	tw_end();

src/workload/methods/codes-dumpi-trace-nw-wrkld.c

@@ -15,6 +15,24 @@
 #include "codes/codes-workload.h"
 #include "codes/quickhash.h"
 
+#if ENABLE_CORTEX
+#include <cortex/cortex.h>
+#include <cortex/datatype.h>
+#include <cortex/cortex-mpich.h>
+#ifdef ENABLE_CORTEX_PYTHON
+#include <cortex/cortex-python.h>
+#endif
+#define PROFILE_TYPE cortex_dumpi_profile*
+#define UNDUMPI_OPEN cortex_undumpi_open
+#define DUMPI_START_STREAM_READ cortex_dumpi_start_stream_read
+#define UNDUMPI_CLOSE cortex_undumpi_close
+#else
+#define PROFILE_TYPE dumpi_profile*
+#define UNDUMPI_OPEN undumpi_open
+#define DUMPI_START_STREAM_READ dumpi_start_stream_read
+#define UNDUMPI_CLOSE undumpi_close
+#endif
+
 #define MAX_LENGTH 512
 #define MAX_OPERATIONS 32768
 #define DUMPI_IGNORE_DELAY 100
@@ -26,6 +44,7 @@ static int rank_tbl_pop = 0;
 /* context of the MPI workload */
 typedef struct rank_mpi_context
 {
+    PROFILE_TYPE profile;
     int my_app_id;
     // whether we've seen an init op (needed for timing correctness)
     int is_init;
@@ -92,7 +111,7 @@ static int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank
 static void dumpi_trace_nw_workload_get_next(int app_id, int rank, struct codes_workload_op *op);
 
 /* get number of bytes from the workload data type and count */
-static int64_t get_num_bytes(dumpi_datatype dt);
+static int64_t get_num_bytes(rank_mpi_context* my_ctx, dumpi_datatype dt);
 
 /* computes the delay between MPI operations */
 static void update_compute_time(const dumpi_time* time, rank_mpi_context* my_ctx);
@@ -339,7 +358,7 @@ int handleDUMPIISend(const dumpi_isend *prm, uint16_t thread, const dumpi_time *
 	wrkld_per_rank.u.send.tag = prm->tag;
 	wrkld_per_rank.u.send.count = prm->count;
 	wrkld_per_rank.u.send.data_type = prm->datatype;
-    wrkld_per_rank.u.send.num_bytes = prm->count * get_num_bytes(prm->datatype);
+    wrkld_per_rank.u.send.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	assert(wrkld_per_rank.u.send.num_bytes >= 0);
     	wrkld_per_rank.u.send.req_id = prm->request;
         wrkld_per_rank.u.send.dest_rank = prm->dest;
@@ -359,7 +378,7 @@ int handleDUMPIIRecv(const dumpi_irecv *prm, uint16_t thread, const dumpi_time *
 	wrkld_per_rank.u.recv.data_type = prm->datatype;
 	wrkld_per_rank.u.recv.count = prm->count;
 	wrkld_per_rank.u.recv.tag = prm->tag;
-        wrkld_per_rank.u.recv.num_bytes = prm->count * get_num_bytes(prm->datatype);
+        wrkld_per_rank.u.recv.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	    
         assert(wrkld_per_rank.u.recv.num_bytes >= 0);
         wrkld_per_rank.u.recv.source_rank = prm->source;
@@ -381,7 +400,7 @@ int handleDUMPISend(const dumpi_send *prm, uint16_t thread,
 	    wrkld_per_rank.u.send.tag = prm->tag;
         wrkld_per_rank.u.send.count = prm->count;
         wrkld_per_rank.u.send.data_type = prm->datatype;
-        wrkld_per_rank.u.send.num_bytes = prm->count * get_num_bytes(prm->datatype);
+        wrkld_per_rank.u.send.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	    assert(wrkld_per_rank.u.send.num_bytes >= 0);
         wrkld_per_rank.u.send.dest_rank = prm->dest;
         wrkld_per_rank.u.send.source_rank = myctx->my_rank;
@@ -403,7 +422,7 @@ int handleDUMPIRecv(const dumpi_recv *prm, uint16_t thread,
     wrkld_per_rank.u.recv.tag = prm->tag;
     wrkld_per_rank.u.recv.count = prm->count;
     wrkld_per_rank.u.recv.data_type = prm->datatype;
-    wrkld_per_rank.u.recv.num_bytes = prm->count * get_num_bytes(prm->datatype);
+    wrkld_per_rank.u.recv.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	assert(wrkld_per_rank.u.recv.num_bytes >= 0);
         wrkld_per_rank.u.recv.source_rank = prm->source;
         wrkld_per_rank.u.recv.dest_rank = -1;
@@ -413,6 +432,41 @@ int handleDUMPIRecv(const dumpi_recv *prm, uint16_t thread,
 
 }
 
+int handleDUMPISendrecv(const dumpi_sendrecv* prm, uint16_t thread,
+			const dumpi_time *cpu, const dumpi_time *wall,
+			const dumpi_perfinfo *perf, void *uarg)
+{
+	rank_mpi_context* myctx = (rank_mpi_context*)uarg;
+
+	{
+		struct codes_workload_op wrkld_per_rank;
+		wrkld_per_rank.op_type = CODES_WK_SEND;
+		wrkld_per_rank.u.send.tag = prm->sendtag;
+		wrkld_per_rank.u.send.count = prm->sendcount;
+		wrkld_per_rank.u.send.data_type = prm->sendtype;
+		wrkld_per_rank.u.send.num_bytes = prm->sendcount * get_num_bytes(myctx,prm->sendtype);
+		assert(wrkld_per_rank.u.send.num_bytes >= 0);
+		wrkld_per_rank.u.send.dest_rank = prm->dest;
+		wrkld_per_rank.u.send.source_rank = myctx->my_rank;
+		wrkld_per_rank.u.send.req_id = -1;
+		update_times_and_insert(&wrkld_per_rank, wall, myctx);
+	}
+
+	{
+		struct codes_workload_op wrkld_per_rank;
+		wrkld_per_rank.op_type = CODES_WK_RECV;
+		wrkld_per_rank.u.recv.tag = prm->recvtag;
+		wrkld_per_rank.u.recv.count = prm->recvcount;
+		wrkld_per_rank.u.recv.data_type = prm->recvtype;
+		wrkld_per_rank.u.recv.num_bytes = prm->recvcount * get_num_bytes(myctx,prm->recvtype);
+		assert(wrkld_per_rank.u.recv.num_bytes >= 0);
+		wrkld_per_rank.u.recv.source_rank = prm->source;
+		wrkld_per_rank.u.recv.dest_rank = -1;
+		update_times_and_insert(&wrkld_per_rank, wall, myctx);
+	}
+	return 0;
+}
+
 int handleDUMPIBcast(const dumpi_bcast *prm, uint16_t thread,
                        const dumpi_time *cpu, const dumpi_time *wall,
                        const dumpi_perfinfo *perf, void *uarg)
@@ -421,7 +475,7 @@ int handleDUMPIBcast(const dumpi_bcast *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_BCAST;
-        wrkld_per_rank.u.collective.num_bytes = prm->count * get_num_bytes(prm->datatype);
+        wrkld_per_rank.u.collective.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	assert(wrkld_per_rank.u.collective.num_bytes >= 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -436,7 +490,7 @@ int handleDUMPIAllgather(const dumpi_allgather *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_ALLGATHER;
-        wrkld_per_rank.u.collective.num_bytes = prm->sendcount * get_num_bytes(prm->sendtype);
+        wrkld_per_rank.u.collective.num_bytes = prm->sendcount * get_num_bytes(myctx,prm->sendtype);
 	assert(wrkld_per_rank.u.collective.num_bytes > 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -451,7 +505,7 @@ int handleDUMPIAllgatherv(const dumpi_allgatherv *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_ALLGATHERV;
-        wrkld_per_rank.u.collective.num_bytes = prm->sendcount * get_num_bytes(prm->sendtype);
+        wrkld_per_rank.u.collective.num_bytes = prm->sendcount * get_num_bytes(myctx,prm->sendtype);
 	assert(wrkld_per_rank.u.collective.num_bytes > 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -466,7 +520,7 @@ int handleDUMPIAlltoall(const dumpi_alltoall *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_ALLTOALL;
-        wrkld_per_rank.u.collective.num_bytes = prm->sendcount * get_num_bytes(prm->sendtype);
+        wrkld_per_rank.u.collective.num_bytes = prm->sendcount * get_num_bytes(myctx,prm->sendtype);
 	assert(wrkld_per_rank.u.collective.num_bytes > 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -481,7 +535,7 @@ int handleDUMPIAlltoallv(const dumpi_alltoallv *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_ALLTOALLV;
-        wrkld_per_rank.u.collective.num_bytes = prm->sendcounts[0] * get_num_bytes(prm->sendtype);
+        wrkld_per_rank.u.collective.num_bytes = prm->sendcounts[0] * get_num_bytes(myctx,prm->sendtype);
 	assert(wrkld_per_rank.u.collective.num_bytes > 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -496,7 +550,7 @@ int handleDUMPIReduce(const dumpi_reduce *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_REDUCE;
-        wrkld_per_rank.u.collective.num_bytes = prm->count * get_num_bytes(prm->datatype);
+        wrkld_per_rank.u.collective.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	assert(wrkld_per_rank.u.collective.num_bytes > 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -511,7 +565,7 @@ int handleDUMPIAllreduce(const dumpi_allreduce *prm, uint16_t thread,
 	struct codes_workload_op wrkld_per_rank;
 
         wrkld_per_rank.op_type = CODES_WK_ALLREDUCE;
-        wrkld_per_rank.u.collective.num_bytes = prm->count * get_num_bytes(prm->datatype);
+        wrkld_per_rank.u.collective.num_bytes = prm->count * get_num_bytes(myctx,prm->datatype);
 	assert(wrkld_per_rank.u.collective.num_bytes > 0);
 
         update_times_and_insert(&wrkld_per_rank, wall, myctx);
@@ -550,7 +604,6 @@ static int hash_rank_compare(void *key, struct qhash_head *link)
     tmp = qhash_entry(link, rank_mpi_context, hash_link);
     if (tmp->my_rank == in->rank && tmp->my_app_id == in->app)
         return 1;
-
     return 0;
 }
 
@@ -558,7 +611,10 @@ int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank)
 {
 	libundumpi_callbacks callbacks;
 	libundumpi_cbpair callarr[DUMPI_END_OF_STREAM];
-	dumpi_profile* profile;
+#ifdef ENABLE_CORTEX
+	libundumpi_cbpair transarr[DUMPI_END_OF_STREAM];
+#endif
+	PROFILE_TYPE profile;
 	dumpi_trace_params* dumpi_params = (dumpi_trace_params*)params;
 	char file_name[MAX_LENGTH];
 
@@ -589,7 +645,12 @@ int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank)
              sprintf(file_name, "%s0%d.bin", dumpi_params->file_name, rank);
              else
               sprintf(file_name, "%s%d.bin", dumpi_params->file_name, rank);
+#ifdef ENABLE_CORTEX
+	profile = cortex_undumpi_open(file_name, app_id, dumpi_params->num_net_traces, rank);
+#else
 	profile =  undumpi_open(file_name);
+#endif
+        my_ctx->profile = profile;
         if(NULL == profile) {
                 printf("Error: unable to open DUMPI trace: %s", file_name);
                 exit(-1);
@@ -597,6 +658,9 @@ int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank)
 	
 	memset(&callbacks, 0, sizeof(libundumpi_callbacks));
         memset(&callarr, 0, sizeof(libundumpi_cbpair) * DUMPI_END_OF_STREAM);
+#ifdef ENABLE_CORTEX
+	memset(&transarr, 0, sizeof(libundumpi_cbpair) * DUMPI_END_OF_STREAM);
+#endif
 
 	/* handle MPI function calls */	        
         callbacks.on_init = handleDUMPIInit;
@@ -635,7 +699,7 @@ int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank)
 	callbacks.on_recv_init = (dumpi_recv_init_call)handleDUMPIIgnore;
 	callbacks.on_start = (dumpi_start_call)handleDUMPIIgnore;
 	callbacks.on_startall = (dumpi_startall_call)handleDUMPIIgnore;
-	callbacks.on_sendrecv = (dumpi_sendrecv_call)handleDUMPIIgnore;
+	callbacks.on_sendrecv = (dumpi_sendrecv_call)handleDUMPISendrecv;
 	callbacks.on_sendrecv_replace = (dumpi_sendrecv_replace_call)handleDUMPIIgnore;
 	callbacks.on_type_contiguous = (dumpi_type_contiguous_call)handleDUMPIIgnore;
 	callbacks.on_barrier = (dumpi_barrier_call)handleDUMPIIgnore;
@@ -662,19 +726,34 @@ int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank)
 
         libundumpi_populate_callbacks(&callbacks, callarr);
 
-        dumpi_start_stream_read(profile);
+#ifdef ENABLE_CORTEX
+#ifdef ENABLE_CORTEX_PYTHON
+	libundumpi_populate_callbacks(CORTEX_PYTHON_TRANSLATION, transarr);
+#else
+	libundumpi_populate_callbacks(CORTEX_MPICH_TRANSLATION, transarr);
+#endif
+#endif
+        DUMPI_START_STREAM_READ(profile);
         //dumpi_header* trace_header = undumpi_read_header(profile);
         //dumpi_free_header(trace_header);
 
+#ifdef ENABLE_CORTEX_PYTHON
+	cortex_python_set_module(dumpi_params->cortex_script,dumpi_params->cortex_class);
+#endif
+
         int finalize_reached = 0;
         int active = 1;
         int num_calls = 0;
         while(active && !finalize_reached)
         {
            num_calls++;
+#ifdef ENABLE_CORTEX
+	   active = cortex_undumpi_read_single_call(profile, callarr, transarr, (void*)my_ctx, &finalize_reached);
+#else
            active = undumpi_read_single_call(profile, callarr, (void*)my_ctx, &finalize_reached);
+#endif
         }
-	undumpi_close(profile);
+	UNDUMPI_CLOSE(profile);
 	dumpi_finalize_mpi_op_data(my_ctx->dumpi_mpi_array);
 	/* add this rank context to hash table */	
         rank_mpi_compare cmp;
@@ -686,8 +765,11 @@ int dumpi_trace_nw_workload_load(const char* params, int app_id, int rank)
 	return 0;
 }
 
-static int64_t get_num_bytes(dumpi_datatype dt)
+static int64_t get_num_bytes(rank_mpi_context* myctx, dumpi_datatype dt)
 {
+#ifdef ENABLE_CORTEX
+   return cortex_datatype_get_size(myctx->profile,dt);
+#endif
    switch(dt)
    {
 	case DUMPI_DATATYPE_ERROR: