Skip to content

GitLab

Commit 6071b26b authored by Misbah Mubarak's avatar Misbah Mubarak
Browse files

Updating MPI Sim Layer to use quicklists

parent db6998f0
Hide whitespace changes
Inline Side-by-side
Showing with 1261 additions and 114 deletions
src/models/Makefile.subdir
View file @ 6071b26b
......@@ -40,25 +40,18 @@ src_libcodes_net_a_SOURCES = \
src/models/networks/model-net/model-net-sched-impl.h \
src/models/networks/model-net/model-net-sched-impl.c \
src/models/network-workloads/model-net-mpi-wrklds.c \
src/models/network-workloads/model-net-mpi-replay.c \
src/models/network-workloads/model-net-synthetic.c \
src/models/network-workloads/model-net-dumpi-traces-dump.c
bin_PROGRAMS += src/models/network-workloads/model-net-mpi-replay
bin_PROGRAMS += src/models/network-workloads/model-net-mpi-wrklds
bin_PROGRAMS += src/models/network-workloads/model-net-dumpi-traces-dump
bin_PROGRAMS += src/models/network-workloads/model-net-synthetic
src_models_network_workloads_model_net_mpi_wrklds_SOURCES = src/models/network-workloads/model-net-mpi-wrklds.c
#src_models_network_workloads_model_net_mpi_wrklds_LDADD = $(testlib) $(CODES_BASE_LIBS)
#src_models_network_workloads_model_net_mpi_wrklds_LDFLAGS = $(CODES_BASE_LDFLAGS)
#src_models_network_workloads_model_net_mpi_wrklds_CFLAGS = ${CODES_BASE_CFLAGS}
src_models_network_workloads_model_net_mpi_replay_SOURCES = src/models/network-workloads/model-net-mpi-replay.c
src_models_network_workloads_model_net_mpi_wrklds_SOURCES = src/models/network-workloads/model-net-mpi-wrklds.c
src_models_network_workloads_model_net_synthetic_SOURCES = src/models/network-workloads/model-net-synthetic.c
#src_models_network_workloads_model_net_synthetic_LDADD = $(testlib) $(CODES_BASE_LIBS)
#src_models_network_workloads_model_net_synthetic_LDFLAGS = $(CODES_BASE_LDFLAGS)
#src_models_network_workloads_model_net_synthetic_CFLAGS = ${CODES_BASE_CFLAGS}
src_models_network_workloads_model_net_dumpi_traces_dump_SOURCES = src/models/network-workloads/model-net-dumpi-traces-dump.c
#src_models_network_workloads_model_net_dumpi_traces_dump_LDADD = $(testlib) $(CODES_BASE_LIBS)
#src_models_network_workloads_model_net_dumpi_traces_dump_LDFLAGS = $(CODES_BASE_LDFLAGS)
#src_models_network_workloads_model_net_dumpi_traces_dump_CFLAGS = ${CODES_BASE_CFLAGS}
src/models/network-workloads/conf/modelnet-mpi-test-dragonfly.conf
View file @ 6071b26b
......@@ -14,13 +14,13 @@ PARAMS
modelnet_order=( "dragonfly" );
# scheduler options
modelnet_scheduler="fcfs";
chunk_size="32";
chunk_size="64";
# modelnet_scheduler="round-robin";
num_vcs="1";
num_routers="4";
local_vc_size="65535";
global_vc_size="128000";
cn_vc_size="65536";
local_vc_size="32768";
global_vc_size="65536";
cn_vc_size="32768";
local_bandwidth="5.25";
global_bandwidth="4.7";
cn_bandwidth="5.25";
......
src/models/network-workloads/model-net-mpi-replay.c 0 → 100644
View file @ 6071b26b
/*
* Copyright (C) 2014 University of Chicago.
* See COPYRIGHT notice in top-level directory.
*
*/
#include <ross.h>
#include <inttypes.h>
#include "codes/codes-workload.h"
#include "codes/codes.h"
#include "codes/configuration.h"
#include "codes/codes_mapping.h"
#include "codes/model-net.h"
#include "codes/rc-stack.h"
#include "codes/quicklist.h"
#define TRACE -1
#define MAX_WAIT_REQS 200
char workload_type[128];
char workload_file[8192];
char offset_file[8192];
static int wrkld_id;
static int num_net_traces = 0;
/* Doing LP IO*/
static char lp_io_dir[256] = {'\0'};
static lp_io_handle io_handle;
static unsigned int lp_io_use_suffix = 0;
static int do_lp_io = 0;
typedef struct nw_state nw_state;
typedef struct nw_message nw_message;
typedef int16_t dumpi_req_id;
static int net_id = 0;
static float noise = 5.0;
static int num_net_lps, num_nw_lps;
long long num_bytes_sent=0;
long long num_bytes_recvd=0;
double max_time = 0, max_comm_time = 0, max_wait_time = 0, max_send_time = 0, max_recv_time = 0;
double avg_time = 0, avg_comm_time = 0, avg_wait_time = 0, avg_send_time = 0, avg_recv_time = 0;
/* global variables for codes mapping */
static char lp_group_name[MAX_NAME_LENGTH], lp_type_name[MAX_NAME_LENGTH], annotation[MAX_NAME_LENGTH];
static int mapping_grp_id, mapping_type_id, mapping_rep_id, mapping_offset;
/* runtime option for disabling computation time simulation */
static int disable_delay = 0;
/* MPI_OP_GET_NEXT is for getting next MPI operation when the previous operation completes.
* MPI_SEND_ARRIVED is issued when a MPI message arrives at its destination (the message is transported by model-net and an event is invoked when it arrives.
* MPI_SEND_POSTED is issued when a MPI message has left the source LP (message is transported via model-net). */
enum MPI_NW_EVENTS
{
MPI_OP_GET_NEXT=1,
MPI_SEND_ARRIVED,
MPI_SEND_ARRIVED_CB, // for tracking message times on sender
MPI_SEND_POSTED,
};
/* stores pointers of pending MPI operations to be matched with their respective sends/receives. */
struct mpi_msgs_queue
{
int op_type;
int tag;
int source_rank;
int dest_rank;
int num_bytes;
tw_stime req_init_time;
dumpi_req_id req_id;
struct qlist_head ql;
};
/* stores request IDs of completed MPI operations (Isends or Irecvs) */
struct completed_requests
{
dumpi_req_id req_id;
struct qlist_head ql;
};
/* for wait operations, store the pending operation and number of completed waits so far. */
struct pending_waits
{
int op_type;
int req_ids[MAX_WAIT_REQS];
int num_completed;
tw_stime start_time;
struct qlist_head ql;
};
typedef struct mpi_msgs_queue mpi_msgs_queue;
typedef struct completed_requests completed_requests;
typedef struct pending_waits pending_waits;
/* state of the network LP. It contains the pointers to send/receive lists */
struct nw_state
{
long num_events_per_lp;
tw_lpid nw_id;
short wrkld_end;
struct rc_stack * processed_ops;
struct rc_stack * matched_qitems;
/* count of sends, receives, collectives and delays */
unsigned long num_sends;
unsigned long num_recvs;
unsigned long num_cols;
unsigned long num_delays;
unsigned long num_wait;
unsigned long num_waitall;
unsigned long num_waitsome;
/* time spent by the LP in executing the app trace*/
double start_time;
double elapsed_time;
/* time spent in compute operations */
double compute_time;
/* time spent in message send/isend */
double send_time;
/* time spent in message receive */
double recv_time;
/* time spent in wait operation */
double wait_time;
/* FIFO for isend messages arrived on destination */
struct qlist_head arrival_queue;
/* FIFO for irecv messages posted but not yet matched with send operations */
struct qlist_head pending_recvs_queue;
/* List of completed send/receive requests */
struct qlist_head completed_reqs;
};
/* data for handling reverse computation.
* saved_matched_req holds the request ID of matched receives/sends for wait operations.
* ptr_match_op holds the matched MPI operation which are removed from the queues when a send is matched with the receive in forward event handler.
* network event being sent. op is the MPI operation issued by the network workloads API. rv_data holds the data for reverse computation (TODO: Fill this data structure only when the simulation runs in optimistic mode). */
struct nw_message
{
int msg_type;
struct
{
/* forward event handler */
struct
{
int op_type;
tw_lpid src_rank;
tw_lpid dest_rank;
int num_bytes;
int data_type;
double sim_start_time;
// for callbacks - time message was received
double msg_send_time;
int16_t req_id;
int tag;
} msg_info;
/* required for reverse computation*/
struct
{
int found_match;
short matched_op;
dumpi_req_id saved_matched_req;
struct codes_workload_op* ptr_match_op;
struct pending_waits* saved_pending_wait;
double saved_send_time;
double saved_recv_time;
double saved_wait_time;
} rc;
} u;
};
/* executes MPI isend and send operations */
static void codes_exec_mpi_send(
nw_state* s, tw_lp* lp, struct codes_workload_op * mpi_op);
/* execute MPI irecv operation */
static void codes_exec_mpi_recv(
nw_state* s, tw_lp* lp, nw_message * m, struct codes_workload_op * mpi_op);
/* reverse of mpi recv function. */
static void codes_exec_mpi_recv_rc(
nw_state* s, nw_message* m, tw_lp* lp, struct codes_workload_op * mpi_op);
/* execute the computational delay */
static void codes_exec_comp_delay(
nw_state* s, tw_lp* lp, struct codes_workload_op * mpi_op);
/* gets the next MPI operation from the network-workloads API. */
static void get_next_mpi_operation(
nw_state* s, tw_bf * bf, nw_message * m, tw_lp * lp);
/* reverse handler of get next mpi operation. */
static void get_next_mpi_operation_rc(
nw_state* s, tw_bf * bf, nw_message * m, tw_lp * lp);
/* Makes a call to get_next_mpi_operation. */
static void codes_issue_next_event(tw_lp* lp);
/* reverse handler of next operation */
static void codes_issue_next_event_rc(tw_lp* lp);
///////////////////// HELPER FUNCTIONS FOR MPI MESSAGE QUEUE HANDLING ///////////////
/* upon arrival of local completion message, inserts operation in completed send queue */
/* upon arrival of an isend operation, updates the arrival queue of the network */
static void update_arrival_queue(
nw_state*s, tw_bf* bf, nw_message* m, tw_lp * lp);
/* reverse of the above function */
static void update_arrival_queue_rc(
nw_state*s, tw_bf* bf, nw_message* m, tw_lp * lp);
/* callback to a message sender for computing message time */
static void update_message_time(
nw_state*s, tw_bf* bf, nw_message* m, tw_lp * lp);
/* reverse for computing message time */
static void update_message_time_rc(
nw_state*s, tw_bf* bf, nw_message* m, tw_lp * lp);
/* conversion from seconds to eanaoseconds */
static tw_stime s_to_ns(tw_stime ns);
/* helper function - maps an MPI rank to an LP id */
static tw_lpid rank_to_lpid(int rank)
{
return codes_mapping_get_lpid_from_relative(rank, NULL, "nw-lp", NULL, 0);
}
/* reverse handler of notify_waits function. */
/*static void notify_waits_rc(nw_state* s, tw_bf* bf, tw_lp* lp, nw_message* m, dumpi_req_id completed_req)
{
int i;
*//*if(bf->c1)
{*/
/* if pending wait is still present and is of type MPI_WAIT then do nothing*/
/* s->wait_time = s->saved_wait_time;
mpi_completed_queue_insert_op(&s->completed_reqs, completed_req);
s->pending_waits = wait_elem;
s->saved_pending_wait = NULL;
}
*/
/* if(lp->gid == TRACE)
printf("\n %lf reverse -- notify waits req id %d ", tw_now(lp), completed_req);
printCompletedQueue(s, lp);
if(m->u.rc.matched_op == 1)
s->pending_waits->num_completed--;
*//* if a wait-elem exists, it means the request ID has been matched*/
/* if(m->u.rc.matched_op == 2)
{
if(lp->gid == TRACE)
{
printf("\n %lf matched req id %d ", tw_now(lp), completed_req);
printCompletedQueue(s, lp);
}
struct pending_waits* wait_elem = m->u.rc.saved_pending_wait;
s->wait_time = m->u.rc.saved_wait_time;
int count = wait_elem->mpi_op->u.waits.count;
for( i = 0; i < count; i++ )
mpi_completed_queue_insert_op(&s->completed_reqs, wait_elem->mpi_op->u.waits.req_ids[i]);
wait_elem->num_completed--;
s->pending_waits = wait_elem;
tw_rand_reverse_unif(lp->rng);
}
}*/
/* notify the completed send/receive request to the wait operation. */
/*static int notify_waits(nw_state* s, tw_bf* bf, tw_lp* lp, nw_message* m, dumpi_req_id completed_req)
{
int i;
*//* traverse the pending waits list and look what type of wait operations are
there. If its just a single wait and the request ID has just been completed,
then the network node LP can go on with fetching the next operation from the log.
If its waitall then wait for all pending requests to complete and then proceed. */
/*struct pending_waits* wait_elem = s->pending_waits;
m->u.rc.matched_op = 0;
if(lp->gid == TRACE)
printf("\n %lf notify waits req id %d ", tw_now(lp), completed_req);
if(!wait_elem)
return 0;
int op_type = wait_elem->mpi_op->op_type;
if(op_type == CODES_WK_WAIT)
{
if(wait_elem->mpi_op->u.wait.req_id == completed_req)
{
m->u.rc.saved_wait_time = s->wait_time;
s->wait_time += (tw_now(lp) - wait_elem->start_time);
remove_req_id(&s->completed_reqs, completed_req);
m->u.rc.saved_pending_wait = wait_elem;
s->pending_waits = NULL;
codes_issue_next_event(lp);
return 0;
}
}
else if(op_type == CODES_WK_WAITALL)
{
int required_count = wait_elem->mpi_op->u.waits.count;
for(i = 0; i < required_count; i++)
{
if(wait_elem->mpi_op->u.waits.req_ids[i] == completed_req)
{
if(lp->gid == TRACE)
printCompletedQueue(s, lp);
m->u.rc.matched_op = 1;
wait_elem->num_completed++;
}
}
if(wait_elem->num_completed == required_count)
{
if(lp->gid == TRACE)
{
printf("\n %lf req %d completed %d", tw_now(lp), completed_req, wait_elem->num_completed);
printCompletedQueue(s, lp);
}
m->u.rc.matched_op = 2;
m->u.rc.saved_wait_time = s->wait_time;
s->wait_time += (tw_now(lp) - wait_elem->start_time);
m->u.rc.saved_pending_wait = wait_elem;
s->pending_waits = NULL;
for(i = 0; i < required_count; i++)
remove_req_id(&s->completed_reqs, wait_elem->mpi_op->u.waits.req_ids[i]);
codes_issue_next_event(lp); //wait completed
}
}
return 0;
}
*/
/* reverse handler of MPI wait operation */
/*static void codes_exec_mpi_wait_rc(nw_state* s, nw_message* m, tw_lp* lp, struct codes_workload_op * mpi_op)
{
if(s->pending_waits)
{
s->pending_waits = NULL;
return;
}
else
{
codes_issue_next_event_rc(lp);
mpi_completed_queue_insert_op(&s->completed_reqs, mpi_op->u.wait.req_id);
rc_stack_pop(s->st);
}
}
*/
/* execute MPI wait operation */
/*static void codes_exec_mpi_wait(nw_state* s, tw_lp* lp, nw_message * m, struct codes_workload_op * mpi_op)
{
*/ /* check in the completed receives queue if the request ID has already been completed.*/
/* assert(!s->pending_waits);
dumpi_req_id req_id = mpi_op->u.wait.req_id;
struct completed_requests* current = s->completed_reqs;
while(current) {
if(current->req_id == req_id) {
remove_req_id(&s->completed_reqs, req_id);
m->u.rc.saved_wait_time = s->wait_time;
codes_issue_next_event(lp);
return;
}
current = current->next;
}
*/ /* If not, add the wait operation in the pending 'waits' list. */
/*struct pending_waits* wait_op = malloc(sizeof(struct pending_waits));
wait_op->mpi_op = mpi_op;
wait_op->num_completed = 0;
wait_op->start_time = tw_now(lp);
s->pending_waits = wait_op;
// rc_stack_push(lp, wait_op, free, s->st);
}
static void codes_exec_mpi_wait_all_rc(nw_state* s, nw_message* m, tw_lp* lp, struct codes_workload_op * mpi_op)
{
if(lp->gid == TRACE)
{
printf("\n %lf codes exec mpi waitall reverse %d ", tw_now(lp), m->u.rc.found_match);
printCompletedQueue(s, lp);
}
if(m->u.rc.found_match)
{
int i;
int count = mpi_op->u.waits.count;
dumpi_req_id req_id[count];
for( i = 0; i < count; i++)
{
req_id[i] = mpi_op->u.waits.req_ids[i];
mpi_completed_queue_insert_op(&s->completed_reqs, req_id[i]);
}
codes_issue_next_event_rc(lp);
}
else
{
struct pending_waits* wait_op = s->pending_waits;
rc_stack_pop(s->st);
s->pending_waits = NULL;
assert(!s->pending_waits);
if(lp->gid == TRACE)
printf("\n %lf Nullifying codes waitall ", tw_now(lp));
}
}
static void codes_exec_mpi_wait_all(
nw_state* s, tw_lp* lp, nw_message * m, struct codes_workload_op * mpi_op)
{
//assert(!s->pending_waits);
int count = mpi_op->u.waits.count;
*//* If the count is not less than max wait reqs then stop */
/*assert(count < MAX_WAIT_REQS);
int i, num_completed = 0;
dumpi_req_id req_id[count];
struct completed_requests* current = s->completed_reqs;
*//* check number of completed irecvs in the completion queue */
/*if(lp->gid == TRACE)
{
printf(" \n (%lf) MPI waitall posted %d count", tw_now(lp), mpi_op->u.waits.count);
for(i = 0; i < count; i++)
printf(" %d ", (int)mpi_op->u.waits.req_ids[i]);
printCompletedQueue(s, lp);
}
while(current)
{
for(i = 0; i < count; i++)
{
req_id[i] = mpi_op->u.waits.req_ids[i];
if(req_id[i] == current->req_id)
num_completed++;
}
current = current->next;
}
if(TRACE== lp->gid)
printf("\n %lf Num completed %d count %d ", tw_now(lp), num_completed, count);
m->u.rc.found_match = 0;
if(count == num_completed)
{
m->u.rc.found_match = 1;
for( i = 0; i < count; i++)
remove_req_id(&s->completed_reqs, req_id[i]);
codes_issue_next_event(lp);
}
else
{*/
/* If not, add the wait operation in the pending 'waits' list. */
/*struct pending_waits* wait_op = malloc(sizeof(struct pending_waits));
wait_op->mpi_op = mpi_op;
wait_op->num_completed = num_completed;
wait_op->start_time = tw_now(lp);
rc_stack_push(lp, wait_op, free, s->st);
s->pending_waits = wait_op;
}
}*/
/* search for a matching mpi operation and remove it from the list.
* Record the index in the list from where the element got deleted.
* Index is used for inserting the element once again in the queue for reverse computation. */
static int rm_matching_rcv(nw_state * ns, tw_lp * lp, mpi_msgs_queue * qitem)
{
if(!qlist_count(&ns->pending_recvs_queue))
return -1;
int matched = 0;
struct qlist_head *ent = NULL;
mpi_msgs_queue * qi = NULL;
qlist_for_each(ent, &ns->pending_recvs_queue){
qi = qlist_entry(ent, mpi_msgs_queue, ql);
if((qi->num_bytes >= qitem->num_bytes)
&& (qi->tag == qitem->tag || qi->tag == -1)
&& (qi->source_rank == qitem->source_rank || qi->source_rank == -1))
{
matched = 1;
break;
}
}
if(matched)
{
ns->recv_time += (tw_now(lp) - qi->req_init_time);
qlist_del(&qi->ql);
rc_stack_push(lp, qi, free, ns->matched_qitems);
return 1;
}
return -1;
}
static int rm_matching_send(nw_state * ns, tw_lp * lp, mpi_msgs_queue * qitem)
{
int matched = 0;
struct qlist_head *ent = NULL;
mpi_msgs_queue * qi = NULL;
if(!qlist_count(&ns->arrival_queue))
return -1;
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)
&& ((qi->source_rank == qitem->source_rank) || qitem->source_rank == -1))
{
matched = 1;
break;
}
}
if(matched)
{
qlist_del(&qi->ql);
rc_stack_push(lp, qi, free, ns->matched_qitems);
return 1;
}
return -1;
}
static void codes_issue_next_event_rc(tw_lp * lp)
{
tw_rand_reverse_unif(lp->rng);
}
/* Trigger getting next event at LP */
static void codes_issue_next_event(tw_lp* lp)
{
tw_event *e;
nw_message* msg;
tw_stime ts;
ts = g_tw_lookahead + 0.1 + tw_rand_exponential(lp->rng, noise);
e = tw_event_new( lp->gid, ts, lp );
msg = tw_event_data(e);
msg->msg_type = MPI_OP_GET_NEXT;
tw_event_send(e);
}
/* Simulate delays between MPI operations */
static void codes_exec_comp_delay(
nw_state* s, tw_lp* lp, struct codes_workload_op * mpi_op)
{
tw_event* e;
tw_stime ts;
nw_message* msg;
if (disable_delay) {
ts = 0.0; // no compute time sim
}
else {
s->compute_time += s_to_ns(mpi_op->u.delay.seconds);
ts = s_to_ns(mpi_op->u.delay.seconds);
}
ts += g_tw_lookahead + 0.1 + tw_rand_exponential(lp->rng, noise);