dragonfly.c

/*
 * Copyright (C) 2013 University of Chicago.
 * See COPYRIGHT notice in top-level directory.
 *
 */

// Local router ID: 0 --- total_router-1
// Router LP ID 
// Terminal LP ID

#include <ross.h>

#include "codes/codes_mapping.h"
#include "codes/jenkins-hash.h"
#include "codes/codes.h"
#include "codes/model-net.h"
#include "codes/model-net-method.h"
#include "codes/model-net-lp.h"
#include "codes/net/dragonfly.h"

#define CREDIT_SIZE 8
#define MEAN_PROCESS 1.0

/* collective specific parameters */
#define TREE_DEGREE 4
#define LEVEL_DELAY 1000
#define DRAGONFLY_COLLECTIVE_DEBUG 0
#define NUM_COLLECTIVES  1
#define COLLECTIVE_COMPUTATION_DELAY 5700
#define DRAGONFLY_FAN_OUT_DELAY 20.0
#define WINDOW_LENGTH 0

// debugging parameters
#define TRACK -1
#define PRINT_ROUTER_TABLE 1
#define DEBUG 0

#define LP_CONFIG_NM (model_net_lp_config_names[DRAGONFLY])
#define LP_METHOD_NM (model_net_method_names[DRAGONFLY])

#define DRAGONFLY_DBG 0
#define dprintf(_fmt, ...) \
    do {if (CLIENT_DBG) printf(_fmt, __VA_ARGS__);} while (0)

long term_ecount, router_ecount, term_rev_ecount, router_rev_ecount;

static double maxd(double a, double b) { return a < b ? b : a; }

// arrival rate
static double MEAN_INTERVAL=200.0;
// threshold for adaptive routing
static int adaptive_threshold = 10;

/* minimal and non-minimal packet counts for adaptive routing*/
unsigned int minimal_count=0, nonmin_count=0, completed_packets = 0;

typedef struct dragonfly_param dragonfly_param;
/* annotation-specific parameters (unannotated entry occurs at the 
 * last index) */
static uint64_t                  num_params = 0;
static dragonfly_param         * all_params = NULL;
static const config_anno_map_t * anno_map   = NULL;

/* global variables for codes mapping */
static char lp_group_name[MAX_NAME_LENGTH];
static int mapping_grp_id, mapping_type_id, mapping_rep_id, mapping_offset;

/* router magic number */
int router_magic_num = 0;

/* terminal magic number */
int terminal_magic_num = 0;

/* number of routers in a mapping group */
static int num_routers_per_mgrp = 0;

/* maximum number of terminals and routers */
int max_term_occupancy, max_router_occupancy;

/* noise of 1ns */
double noise = 1.0;

struct dragonfly_param
{
    // configuration parameters
    int num_routers; /*Number of routers in a group*/
    double local_bandwidth;/* bandwidth of the router-router channels within a group */
    double global_bandwidth;/* bandwidth of the inter-group router connections */
    double cn_bandwidth;/* bandwidth of the compute node channels connected to routers */
    int num_vcs; /* number of virtual channels */
    int local_vc_size; /* buffer size of the router-router channels */
    int global_vc_size; /* buffer size of the global channels */
    int cn_vc_size; /* buffer size of the compute node channels */
    int chunk_size; /* full-sized packets are broken into smaller chunks.*/

    // derived parameters
    int num_cn;
    int num_groups;
    int radix;
    int total_routers;
    int total_terminals;
    int num_global_channels;
};

struct pending_router_msgs
{
    struct pending_router_msgs * next;
    struct pending_router_msgs * prev;
    char * event_data;
    terminal_message msg;
    int output_chan;
    int next_stop; 
};
/* handles terminal and router events like packet generate/send/receive/buffer */
typedef enum event_t event_t;

typedef struct terminal_state terminal_state;
typedef struct router_state router_state;

/* dragonfly compute node data structure */
struct terminal_state
{
   unsigned long long packet_counter;

   // Dragonfly specific parameters
   tw_lpid router_id;
   tw_lpid terminal_id;

   // Each terminal will have an input and output channel with the router
   int* vc_occupancy; // NUM_VC
   int* output_vc_state;
   tw_stime terminal_available_time;
   tw_stime next_credit_available_time;
// Terminal generate, sends and arrival T_SEND, T_ARRIVAL, T_GENERATE
// Router-Router Intra-group sends and receives RR_LSEND, RR_LARRIVE
// Router-Router Inter-group sends and receives RR_GSEND, RR_GARRIVE
   struct mn_stats dragonfly_stats_array[CATEGORY_MAX];
  /* collective init time */
  tw_stime collective_init_time;

  /* node ID in the tree */ 
   tw_lpid node_id;

   /* messages sent & received in collectives may get interchanged several times so we have to save the 
     origin server information in the node's state */
   tw_lpid origin_svr; 
  
  /* parent node ID of the current node */
   tw_lpid parent_node_id;
   /* array of children to be allocated in terminal_init*/
   tw_lpid* children;

   /* children of a node can be less than or equal to the tree degree */
   int num_children;

   short is_root;
   short is_leaf;

   /* to maintain a count of child nodes that have fanned in at the parent during the collective
      fan-in phase*/
   int num_fan_nodes;
   int max_term_vc_occupancy;

   const char * anno;
   const dragonfly_param *params;
};

/* terminal event type (1-4) */
enum event_t
{
  T_GENERATE=1,
  T_ARRIVE,
  T_BUFFER,
  R_FORWARD,
  R_BUFFER,
  D_COLLECTIVE_INIT,
  D_COLLECTIVE_FAN_IN,
  D_COLLECTIVE_FAN_OUT
};
/* status of a virtual channel can be idle, active, allocated or wait for credit */
enum vc_status
{
   VC_IDLE,
   VC_ACTIVE,
   VC_ALLOC,
   VC_CREDIT
};

/* whether the last hop of a packet was global, local or a terminal */
enum last_hop
{
   GLOBAL,
   LOCAL,
   TERMINAL
};

/* three forms of routing algorithms available, adaptive routing is not
 * accurate and fully functional in the current version as the formulas
 * for detecting load on global channels are not very accurate */
enum ROUTING_ALGO
{
    MINIMAL = 0,
    NON_MINIMAL,
    ADAPTIVE,
    PROG_ADAPTIVE
};

struct router_state
{
   unsigned int router_id;
   unsigned int group_id;
   
   tw_stime* next_output_available_time;
   tw_stime* next_credit_available_time;
   tw_stime* cur_hist_start_time;
   
   int* vc_occupancy;
   int* output_vc_state;
   int * global_channel;
   int max_router_vc_occupancy;

   const char * anno;
   const dragonfly_param *params;

   int* prev_hist_num;
   int* cur_hist_num;

   struct pending_router_msgs * head;
   struct pending_router_msgs * tail;
   int num_waiting;
};

static short routing = MINIMAL;

static tw_stime         dragonfly_total_time = 0;
static tw_stime         dragonfly_max_latency = 0;
static tw_stime         max_collective = 0;


static long long       total_hops = 0;
static long long       N_finished_packets = 0;

/* function definitions */
static void router_credit_send(router_state * s, tw_bf * bf, terminal_message * msg, tw_lp * lp);

/* returns the dragonfly router lp type for lp registration */
static const tw_lptype* dragonfly_get_router_lp_type(void);

/* remove head of the queue */
void remove_pending_list_head(struct pending_router_msgs** head,
                              struct pending_router_msgs** tail,
                              terminal_message * msg)
{
    struct pending_router_msgs * elem = *head;

    if(*head == *tail)
        *tail = NULL;

    *head = (*head)->next;
    
    if(*head)
        (*head)->prev = NULL;
    
    free(elem);
}

/* pending router messages */
void add_pending_router_message(struct pending_router_msgs** head,
                                struct pending_router_msgs** tail,
                                terminal_message * msg,
                                int chan,
                                int next_stop)
{
    struct pending_router_msgs * elem = malloc(sizeof(struct pending_router_msgs));
    memcpy(&(elem->msg), msg, sizeof(terminal_message));
    elem->prev = NULL;
    elem->next_stop = next_stop;
    elem->output_chan = chan;

    if(msg->remote_event_size_bytes)
    {
        void * m_data = msg+1;
        elem->event_data = (void*)malloc(msg->remote_event_size_bytes);
        memcpy(elem->event_data, m_data, msg->remote_event_size_bytes);
    }
    elem->next = *head;

    if(*head)
      (*head)->prev = elem;
    
    if(!(*head))
        *tail = elem;
    
    *head = elem;
    return;
}

/* remove message from the queue */
struct pending_router_msgs * remove_pending_router_msgs(struct pending_router_msgs** head, 
                                struct pending_router_msgs** tail,
                                int chan)
{
   struct pending_router_msgs * elem = *head;

   if(!elem)
     return NULL;

   while(elem != NULL)
   {
       if(elem->output_chan == chan)
       {
        /* Remove elemt from the list */
           /* if there is just one element */
           if(elem == *head && elem == *tail)
           {
               *head = NULL;
               *tail = NULL;
           }
           /* if element if at the head */
           if(elem == *head && elem != *tail)
           {
               *head = elem->next;
               elem->next->prev = NULL;
           }

           /* if element is at the tail */
           if(elem == *tail && elem != *head)
           {
               *tail = elem->prev;
               elem->prev->next = NULL;
           }

           /* if element is in the middle */
           if(elem->prev)
               elem->prev->next = elem->next;
           
           if(elem->next)
               elem->next->prev = elem->prev;
          

           //printf("\n Returned element %d %d %d", elem->msg.dest_terminal_id, chan, elem->next_stop);
           return elem;
       }
       elem = elem->next;
   }
   return NULL;
}

/* returns the dragonfly message size */
static int dragonfly_get_msg_sz(void)
{
	   return sizeof(terminal_message);
}

static void dragonfly_read_config(const char * anno, dragonfly_param *params){
    // shorthand
    dragonfly_param *p = params;

    configuration_get_value_int(&config, "PARAMS", "num_routers", anno,
            &p->num_routers);
    if(p->num_routers <= 0) {
        p->num_routers = 4;
        fprintf(stderr, "Number of dimensions not specified, setting to %d\n",
                p->num_routers);
    }

    configuration_get_value_int(&config, "PARAMS", "num_vcs", anno,
            &p->num_vcs);
    if(!p->num_vcs) {
        p->num_vcs = 1;
        fprintf(stderr, "Number of virtual channels not specified, setting to %d\n", p->num_vcs);
    }

    configuration_get_value_int(&config, "PARAMS", "local_vc_size", anno, &p->local_vc_size);
    if(p->local_vc_size <= 0) {
        p->local_vc_size = 1024;
        fprintf(stderr, "Buffer size of local channels not specified, setting to %d\n", p->local_vc_size);
    }

    configuration_get_value_int(&config, "PARAMS", "global_vc_size", anno, &p->global_vc_size);
    if(p->global_vc_size <= 0) {
        p->global_vc_size = 2048;
        fprintf(stderr, "Buffer size of global channels not specified, setting to %d\n", p->global_vc_size);
    }

    configuration_get_value_int(&config, "PARAMS", "cn_vc_size", anno, &p->cn_vc_size);
    if(p->cn_vc_size <= 0) {
        p->cn_vc_size = 1024;
        fprintf(stderr, "Buffer size of compute node channels not specified, setting to %d\n", p->cn_vc_size);
    }

    configuration_get_value_int(&config, "PARAMS", "chunk_size", anno, &p->chunk_size);
    if(p->chunk_size <= 0) {
        p->chunk_size = 64;
        fprintf(stderr, "Chunk size for packets is not specified, setting to %d\n", p->chunk_size);
    }

    configuration_get_value_double(&config, "PARAMS", "local_bandwidth", anno, &p->local_bandwidth);
    if(p->local_bandwidth <= 0) {
        p->local_bandwidth = 5.25;
        fprintf(stderr, "Bandwidth of local channels not specified, setting to %lf\n", p->local_bandwidth);
    }

    configuration_get_value_double(&config, "PARAMS", "global_bandwidth", anno, &p->global_bandwidth);
    if(p->global_bandwidth <= 0) {
        p->global_bandwidth = 4.7;
        fprintf(stderr, "Bandwidth of global channels not specified, setting to %lf\n", p->global_bandwidth);
    }

    configuration_get_value_double(&config, "PARAMS", "cn_bandwidth", anno, &p->cn_bandwidth);
    if(p->cn_bandwidth <= 0) {
        p->cn_bandwidth = 5.25;
        fprintf(stderr, "Bandwidth of compute node channels not specified, setting to %lf\n", p->cn_bandwidth);
    }



    char routing_str[MAX_NAME_LENGTH];
    configuration_get_value(&config, "PARAMS", "routing", anno, routing_str,
            MAX_NAME_LENGTH);
    if(strcmp(routing_str, "minimal") == 0)
        routing = MINIMAL;
    else if(strcmp(routing_str, "nonminimal")==0 || strcmp(routing_str,"non-minimal")==0)
        routing = NON_MINIMAL;
    else if (strcmp(routing_str, "adaptive") == 0)
        routing = ADAPTIVE;
    else if (strcmp(routing_str, "prog-adaptive") == 0)
	routing = PROG_ADAPTIVE;
    else
    {
        fprintf(stderr, 
                "No routing protocol specified, setting to minimal routing\n");
        routing = -1;
    }

    // set the derived parameters
    p->num_cn = p->num_routers/2;
    p->num_global_channels = p->num_routers/2;
    p->num_groups = p->num_routers * p->num_cn + 1;
    p->radix = p->num_vcs *
        (p->num_routers + p->num_global_channels + p->num_cn);
    p->total_routers = p->num_groups * p->num_routers;
    p->total_terminals = p->total_routers * p->num_cn;
    
    if(!g_tw_mynode)
    	printf("\n Total nodes %d routers %d groups %d radix %d num_vc %d ", p->num_cn * p->total_routers,
								p->total_routers,
								p->num_groups,
								p->radix,
                                                                p->num_vcs);
}

/* convert GiB/s and bytes to ns */
static tw_stime bytes_to_ns(uint64_t bytes, double GB_p_s)
{
    tw_stime time;

    /* bytes to GB */
    time = ((double)bytes)/(1024.0*1024.0*1024.0);
    /* MB to s */
    time = time / GB_p_s;
    /* s to ns */
    time = time * 1000.0 * 1000.0 * 1000.0;

    return(time);
}

/* reverse computation for msg ready event */
static void dragonfly_configure(){
    anno_map = codes_mapping_get_lp_anno_map(LP_CONFIG_NM);
    assert(anno_map);
    num_params = anno_map->num_annos + (anno_map->has_unanno_lp > 0);
    all_params = calloc(num_params, sizeof(*all_params));

    for (uint64_t i = 0; i < anno_map->num_annos; i++){
        const char * anno = anno_map->annotations[i].ptr;
        dragonfly_read_config(anno, &all_params[i]);
    }
    if (anno_map->has_unanno_lp > 0){
        dragonfly_read_config(NULL, &all_params[anno_map->num_annos]);
    }
}

/* report dragonfly statistics like average and maximum packet latency, average number of hops traversed */
static void dragonfly_report_stats()
{
   long long avg_hops, total_finished_packets;
   tw_stime avg_time, max_time;
   long total_term_events, total_router_events;
   int total_minimal_packets, total_nonmin_packets, total_completed_packets;

   MPI_Reduce( &total_hops, &avg_hops, 1, MPI_LONG_LONG, MPI_SUM, 0, MPI_COMM_WORLD);
   MPI_Reduce( &N_finished_packets, &total_finished_packets, 1, MPI_LONG_LONG, MPI_SUM, 0, MPI_COMM_WORLD);
   MPI_Reduce( &dragonfly_total_time, &avg_time, 1,MPI_DOUBLE, MPI_SUM, 0, MPI_COMM_WORLD);
   MPI_Reduce( &dragonfly_max_latency, &max_time, 1, MPI_DOUBLE, MPI_MAX, 0, MPI_COMM_WORLD);
   MPI_Reduce( &term_ecount, &total_term_events, 1, MPI_LONG, MPI_SUM, 0, MPI_COMM_WORLD);
   MPI_Reduce( &router_ecount, &total_router_events, 1, MPI_LONG, MPI_SUM, 0, MPI_COMM_WORLD);

   if(routing == ADAPTIVE || routing == PROG_ADAPTIVE)
    {
	MPI_Reduce(&minimal_count, &total_minimal_packets, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
 	MPI_Reduce(&nonmin_count, &total_nonmin_packets, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
 	MPI_Reduce(&completed_packets, &total_completed_packets, 1, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);
    }
   /* print statistics */
   if(!g_tw_mynode)
   {	
      printf("\n total finished packets %lld ", total_finished_packets);
      printf(" Average number of hops traversed %f average message latency %lf us maximum message latency %lf us\n", (float)avg_hops/total_finished_packets, avg_time/(total_finished_packets*1000), max_time/1000);
     if(routing == ADAPTIVE || routing == PROG_ADAPTIVE)
              printf("\n ADAPTIVE ROUTING STATS: %d packets routed minimally %d packets routed non-minimally completed packets %d ", total_minimal_packets, total_nonmin_packets, total_completed_packets);
 
     printf("\n Max terminal occupancy %d max router occupancy %d ", max_term_occupancy, max_router_occupancy);
     printf("\n Event population: total committed terminal events:%ld router events: %ld ", total_term_events, total_router_events);
  }
   return;
}

void dragonfly_collective_init(terminal_state * s,
           		   tw_lp * lp)
{
    // TODO: be annotation-aware
    codes_mapping_get_lp_info(lp->gid, lp_group_name, &mapping_grp_id, NULL,
            &mapping_type_id, NULL, &mapping_rep_id, &mapping_offset);
    int num_lps = codes_mapping_get_lp_count(lp_group_name, 1, LP_CONFIG_NM,
            NULL, 1);
    int num_reps = codes_mapping_get_group_reps(lp_group_name);
    s->node_id = (mapping_rep_id * num_lps) + mapping_offset;

    int i;
   /* handle collective operations by forming a tree of all the LPs */
   /* special condition for root of the tree */
   if( s->node_id == 0)
    {
        s->parent_node_id = -1;
        s->is_root = 1;
   }
   else
   {
       s->parent_node_id = (s->node_id - ((s->node_id - 1) % TREE_DEGREE)) / TREE_DEGREE;
       s->is_root = 0;
   }
   s->children = (tw_lpid*)malloc(TREE_DEGREE * sizeof(tw_lpid));

   /* set the isleaf to zero by default */
   s->is_leaf = 1;
   s->num_children = 0;

   /* calculate the children of the current node. If its a leaf, no need to set children,
      only set isleaf and break the loop*/

   for( i = 0; i < TREE_DEGREE; i++ )
    {
        tw_lpid next_child = (TREE_DEGREE * s->node_id) + i + 1;
        if(next_child < (num_lps * num_reps))
        {
            s->num_children++;
            s->is_leaf = 0;
            s->children[i] = next_child;
        }
        else
           s->children[i] = -1;
    }

#if DRAGONFLY_COLLECTIVE_DEBUG == 1
   printf("\n LP %ld parent node id ", s->node_id);

   for( i = 0; i < TREE_DEGREE; i++ )
        printf(" child node ID %ld ", s->children[i]);
   printf("\n");

   if(s->is_leaf)
        printf("\n LP %ld is leaf ", s->node_id);
#endif
}

/* dragonfly packet event , generates a dragonfly packet on the compute node */
static tw_stime dragonfly_packet_event(char const * category, tw_lpid final_dest_lp, tw_lpid dest_mn_lp, uint64_t packet_size, int is_pull, uint64_t pull_size, tw_stime offset, const mn_sched_params *sched_params, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lpid src_lp, tw_lp *sender, int is_last_pckt)
{
    tw_event * e_new;
    tw_stime xfer_to_nic_time;
    terminal_message * msg;
    char* tmp_ptr;

    xfer_to_nic_time = codes_local_latency(sender); /* Throws an error of found last KP time > current event time otherwise when LPs of one type are placed together*/
    e_new = model_net_method_event_new(sender->gid, xfer_to_nic_time+offset,
            sender, DRAGONFLY, (void**)&msg, (void**)&tmp_ptr);
    strcpy(msg->category, category);
    msg->final_dest_gid = final_dest_lp;
    msg->dest_terminal_id = dest_mn_lp;
    msg->sender_lp=src_lp;
    msg->sender_mn_lp = sender->gid;
    msg->packet_size = packet_size;
    msg->remote_event_size_bytes = 0;
    msg->local_event_size_bytes = 0;
    msg->type = T_GENERATE;
    msg->magic = terminal_magic_num;
    msg->is_pull = is_pull;
    msg->pull_size = pull_size;
    msg->chunk_id = 0;
    msg->packet_ID = 0;

    if(is_last_pckt) /* Its the last packet so pass in remote and local event information*/
      {
	if(remote_event_size > 0)
	 {
		msg->remote_event_size_bytes = remote_event_size;
		memcpy(tmp_ptr, remote_event, remote_event_size);
		tmp_ptr += remote_event_size;
	}
	if(self_event_size > 0)
	{
		msg->local_event_size_bytes = self_event_size;
		memcpy(tmp_ptr, self_event, self_event_size);
		tmp_ptr += self_event_size;
	}
     }
    tw_event_send(e_new);
    return xfer_to_nic_time;
}

/* dragonfly packet event reverse handler */
static void dragonfly_packet_event_rc(tw_lp *sender)
{
	  codes_local_latency_reverse(sender);
	    return;
}

void send_packet_from_router(router_state * s,
                             tw_bf * bf,
                             terminal_message * msg,
                             tw_lp * lp,
                             int output_chan,
                             int next_stop,
                             char * event_data)
{
       router_credit_send(s, bf, msg, lp); 

       tw_stime ts;
       tw_event *e;
       terminal_message *m;
      
       int global=0;
       int output_port = output_chan / s->params->num_vcs;
       double bandwidth = s->params->local_bandwidth;
       // Allocate output Virtual Channel
      if(output_port >= s->params->num_routers && 
              output_port < s->params->num_routers + s->params->num_global_channels)
      {
         global = 1;
         bandwidth = s->params->global_bandwidth;
      }

      // If source router doesn't have global channel and buffer space is available, then assign to appropriate intra-group virtual channel 
	  msg->saved_available_time = s->next_output_available_time[output_port];
	  ts = g_tw_lookahead + bytes_to_ns(s->params->chunk_size, bandwidth) + tw_rand_exponential(lp->rng, noise);

	  s->next_output_available_time[output_port] = maxd(s->next_output_available_time[output_port], tw_now(lp));
	  s->next_output_available_time[output_port] += ts;
	  // dest can be a router or a terminal, so we must check
	  void * m_data;
	  if (next_stop == msg->dest_terminal_id){
	      e = model_net_method_event_new(next_stop, 
		      s->next_output_available_time[output_port] - tw_now(lp), lp,
		      DRAGONFLY, (void**)&m, &m_data);
	  }
	  else{
	      e = tw_event_new(next_stop, s->next_output_available_time[output_port] - tw_now(lp), lp);
	      m = tw_event_data(e);
	      m_data = m+1;
	  }
	  memcpy(m, msg, sizeof(terminal_message));
	  if (msg->remote_event_size_bytes){
          if(!event_data)
	        memcpy(m_data, msg+1, msg->remote_event_size_bytes);
	      else
              memcpy(m_data, event_data, msg->remote_event_size_bytes);
      }
	  if(global)
	    m->last_hop=GLOBAL;
	  else
	    m->last_hop = LOCAL;

	  m->local_id = s->router_id;
	  /* for reverse computation */
	  msg->new_vc = output_chan;
	  /* for sending back credit in forward event handler */
	  m->saved_vc = output_chan;
	  m->intm_lp_id = lp->gid;
	  s->vc_occupancy[output_chan]++;
	 
      if(s->vc_occupancy[output_chan] > s->max_router_vc_occupancy)
	   {
		bf->c3 = 1;
		msg->saved_occupancy = s->max_router_vc_occupancy;
		s->max_router_vc_occupancy = s->vc_occupancy[output_chan];
 	   }
	 if(routing == PROG_ADAPTIVE)
	  {
		  if(tw_now(lp) - s->cur_hist_start_time[output_chan] >= WINDOW_LENGTH)
		  {
			bf->c2 = 1;
			msg->saved_hist_num = s->prev_hist_num[output_chan];
			msg->saved_hist_start_time = s->cur_hist_start_time[output_chan];

			s->prev_hist_num[output_chan] = s->cur_hist_num[output_chan];

			s->cur_hist_start_time[output_chan] = tw_now(lp);
			s->cur_hist_num[output_chan] = 1;
		  }
		  else
		  {
			s->cur_hist_num[output_chan]++;
		  }
	  }
	  /* Determine the event type. If the packet has arrived at the final destination
	     router then it should arrive at the destination terminal next. */
	  if(next_stop == msg->dest_terminal_id)
	  {
	    m->type = T_ARRIVE;
	    m->magic = terminal_magic_num;

	    if(s->vc_occupancy[output_chan] >= s->params->cn_vc_size)
	      s->output_vc_state[output_chan] = VC_CREDIT;
	  }
	  else
	  {
	    /* The packet has to be sent to another router */
	    m->type = R_FORWARD;
	    m->magic = router_magic_num;
	   /* If this is a global channel then the buffer space is different */
	   if( global )
	   {
	     if(s->vc_occupancy[output_chan] >= s->params->global_vc_size)
	       s->output_vc_state[output_chan] = VC_CREDIT;
	   }
	  else
	    {
	     /* buffer space is less for local channels */
	     if( s->vc_occupancy[output_chan] >= s->params->local_vc_size)
		    s->output_vc_state[output_chan] = VC_CREDIT;
	    }
	  }
	  tw_event_send(e);
	  return;

}
/* given two group IDs, find the router of the src_gid that connects to the dest_gid*/
tw_lpid getRouterFromGroupID(int dest_gid, 
		    int src_gid,
		    int num_routers)
{
  int group_begin = src_gid * num_routers;
  int group_end = (src_gid * num_routers) + num_routers-1;
  int offset = (dest_gid * num_routers - group_begin) / num_routers;
  
  if((dest_gid * num_routers) < group_begin)
    offset = (group_begin - dest_gid * num_routers) / num_routers; // take absolute value
  
  int half_channel = num_routers / 4;
  int index = (offset - 1)/(half_channel * num_routers);
  
  offset=(offset - 1) % (half_channel * num_routers);

  // If the destination router is in the same group
  tw_lpid router_id;

  if(index % 2 != 0)
    router_id = group_end - (offset / half_channel); // start from the end
  else
    router_id = group_begin + (offset / half_channel);

  return router_id;
}	

/*When a packet is sent from the current router and a buffer slot becomes available, a credit is sent back to schedule another packet event*/
static void router_credit_send(router_state * s, tw_bf * bf, terminal_message * msg, tw_lp * lp)
{
  tw_event * buf_e;
  tw_stime ts;
  terminal_message * buf_msg;

  int dest=0, type = R_BUFFER;
  int is_terminal = 0;
  int found_magic = router_magic_num;
  tw_stime credit_delay = 0.0;

  const dragonfly_param *p = s->params;
  int sender_radix;
 // Notify sender terminal about available buffer space
  if(msg->last_hop == TERMINAL)
  {
   dest = msg->src_terminal_id;
   sender_radix = msg->local_id % p->num_cn;  
   //determine the time in ns to transfer the credit
   credit_delay = bytes_to_ns(CREDIT_SIZE, p->cn_bandwidth);
   type = T_BUFFER;
   is_terminal = 1;
   found_magic = terminal_magic_num;
  }
   else if(msg->last_hop == GLOBAL)
   {
     dest = msg->intm_lp_id;
     sender_radix = p->num_cn + (msg->local_id % p->num_global_channels);
     credit_delay = bytes_to_ns(CREDIT_SIZE, p->global_bandwidth);
   }
    else if(msg->last_hop == LOCAL)
     {
        dest = msg->intm_lp_id;
        sender_radix = p->num_cn + p->num_global_channels + (msg->local_id % p->num_routers);
     	credit_delay = bytes_to_ns(CREDIT_SIZE, p->local_bandwidth) * CREDIT_SIZE;
     }
    else
      printf("\n Invalid message type");

    msg->sender_radix = sender_radix;

    assert(sender_radix < s->params->radix );

    msg->saved_credit_time = s->next_credit_available_time[sender_radix];
    s->next_credit_available_time[sender_radix] = maxd(tw_now(lp), s->next_credit_available_time[sender_radix]);
    ts = credit_delay + 0.1 + tw_rand_exponential(lp->rng, (double)credit_delay/1000);
	
    s->next_credit_available_time[sender_radix]+=ts;
    if (is_terminal){
        buf_e = model_net_method_event_new(dest, 
                s->next_credit_available_time[sender_radix] - tw_now(lp), lp,
                DRAGONFLY, (void**)&buf_msg, NULL);
    }
    else{
        buf_e = tw_event_new(dest, s->next_credit_available_time[sender_radix] - tw_now(lp) , lp);
        buf_msg = tw_event_data(buf_e);
    }
    buf_msg->origin_router_id = s->router_id;
    buf_msg->vc_index = msg->saved_vc;
    buf_msg->type=type;
    buf_msg->magic = found_magic;
    buf_msg->last_hop = msg->last_hop;
    buf_msg->packet_ID=msg->packet_ID;

    tw_event_send(buf_e);

    return;
}

static void packet_generate_send_rc(terminal_state * s, 
			    tw_bf * bf, 
			    terminal_message * msg, 
			    tw_lp * lp)
{
   term_rev_ecount++;
   term_ecount--;

   tw_rand_reverse_unif(lp->rng);
	 
   s->terminal_available_time = msg->saved_available_time;
   tw_rand_reverse_unif(lp->rng);
   int vc = msg->saved_vc;
   s->vc_occupancy[vc]--;
   s->packet_counter--;
   s->output_vc_state[vc] = VC_IDLE;

   //if(bf->c2)
   //  s->max_term_vc_occupancy = msg->saved_occupancy;

   if (msg->chunk_id == (msg->num_chunks-1)){
     codes_local_latency_reverse(lp);
   }

    if(bf->c1)
        codes_local_latency_reverse(lp);

     struct mn_stats* stat;
     stat = model_net_find_stats(msg->category, s->dragonfly_stats_array);
     stat->send_count--;
     stat->send_bytes -= msg->packet_size;
     stat->send_time -= (1/s->params->cn_bandwidth) * msg->packet_size;
}

/* generates packet at the current dragonfly compute node */
static void packet_generate_send(terminal_state * s, 
			    tw_bf * bf, 
			    terminal_message * msg, 
			    tw_lp * lp)
{
	bf->c1 = 0;
	term_ecount++;
	term_rev_ecount++;

        const dragonfly_param *p = s->params;

	tw_stime ts, travel_start_time;
	tw_event *e;
        tw_lpid router_id;
	terminal_message *m;
	int i, total_event_size;

	uint64_t num_chunks = msg->packet_size / p->chunk_size;
	if (msg->packet_size % s->params->chunk_size)
	  num_chunks++;

	if(!num_chunks)
	   num_chunks = 1;

	if(!msg->packet_ID)
	    msg->packet_ID = lp->gid + g_tw_nlp * s->packet_counter + tw_rand_integer(lp->rng, 0, lp->gid + g_tw_nlp * s->packet_counter);

	//  Each packet is broken into chunks and then sent over the channel
	msg->saved_available_time = s->terminal_available_time;
	tw_stime head_delay = bytes_to_ns(s->params->chunk_size, s->params->cn_bandwidth);
	ts = head_delay + tw_rand_exponential(lp->rng, noise);
	//printf("\n ts %f calculated %f ", ts, s->params->chunk_size * (1/s->params->cn_bandwidth));
	s->terminal_available_time = maxd(s->terminal_available_time, tw_now(lp));
	s->terminal_available_time += ts;

	
	int chan = -1, j;
	for(j = 0; j < p->num_vcs; j++)
	 {
	     if(s->vc_occupancy[j] < p->cn_vc_size)
	      {
	       chan=j;
	       break;
	      }
	 }
	  
	/* for reverse computation */
	msg->saved_vc = chan;
  
    /* simulation should exit */
	if(chan == -1)
	    tw_error(TW_LOC, "\n No terminal buffers available, increase buffer size");
	   
        //TODO: be annotation-aware
        codes_mapping_get_lp_info(lp->gid, lp_group_name, &mapping_grp_id, NULL,
		   &mapping_type_id, NULL, &mapping_rep_id, &mapping_offset);
	   
	codes_mapping_get_lp_id(lp_group_name, "dragonfly_router", NULL, 1,
		   s->router_id/num_routers_per_mgrp, 
		   s->router_id % num_routers_per_mgrp, &router_id);
   
    e = tw_event_new(router_id, s->terminal_available_time - tw_now(lp), lp);
	m = tw_event_data(e);
	memcpy(m, msg, sizeof(terminal_message));
        
	if(msg->chunk_id == 0)
            travel_start_time = tw_now(lp);
	else
	    travel_start_time = msg->travel_start_time;

	m->num_chunks = num_chunks;	
	m->magic = router_magic_num;
	m->origin_router_id = s->router_id;
        m->type = R_FORWARD;
        m->src_terminal_id = lp->gid;
        m->chunk_id = msg->chunk_id;
        m->last_hop = TERMINAL;
        m->intm_group_id = -1;
        m->travel_start_time = travel_start_time;
	m->path_type = -1;
        m->local_event_size_bytes = 0;
        m->local_id = s->terminal_id;

	 if (msg->remote_event_size_bytes){
		memcpy(m+1, model_net_method_get_edata(DRAGONFLY, msg),
			msg->remote_event_size_bytes);
	   }
   
	tw_event_send(e);

	if(msg->packet_ID == TRACK && msg->chunk_id == num_chunks - 1)
	   printf("\n packet %d generated chunk id %d reach at time %lf ", 
			msg->packet_ID, 
			msg->chunk_id, 
			s->terminal_available_time - tw_now(lp));	

         if(msg->chunk_id == num_chunks - 1) 
	{
		// now that message is sent, issue an "idle" event to tell the scheduler
		// when I'm next available
		model_net_method_idle_event(codes_local_latency(lp) +
		      s->terminal_available_time - tw_now(lp), 0, lp);

		/* local completion message */
		if(msg->local_event_size_bytes > 0)
		 {
		   tw_event* e_new;
		   terminal_message* m_new;
		   void* local_event = 
		       (char*)model_net_method_get_edata(DRAGONFLY, msg) + 
		       msg->remote_event_size_bytes;
		   ts = g_tw_lookahead + (1/s->params->cn_bandwidth) * msg->local_event_size_bytes;
		   e_new = tw_event_new(msg->sender_lp, ts, lp);
		   m_new = tw_event_data(e_new);
		   memcpy(m_new, local_event, msg->local_event_size_bytes);
		   tw_event_send(e_new);
		}
	}
	s->packet_counter++;
	s->vc_occupancy[chan]++;

	if(s->vc_occupancy[chan] > s->max_term_vc_occupancy)
	 {
		bf->c2 = 1;
		//msg->saved_occupancy = s->max_term_vc_occupancy;
		s->max_term_vc_occupancy = s->vc_occupancy[chan];
	 }

	if(s->vc_occupancy[chan] >= s->params->cn_vc_size)
	    s->output_vc_state[chan] = VC_CREDIT;

	/* calculate statistics */
	total_event_size = model_net_get_msg_sz(DRAGONFLY) + 
	msg->remote_event_size_bytes + msg->local_event_size_bytes;
	struct mn_stats* stat;
	stat = model_net_find_stats(msg->category, s->dragonfly_stats_array);
	stat->send_count++;
	stat->send_bytes += msg->packet_size;
	stat->send_time += (s->terminal_available_time - tw_now(lp));
	if(stat->max_event_size < total_event_size)
	  stat->max_event_size = total_event_size;

        /* Now schedule another packet generate event */
	if(msg->chunk_id < num_chunks - 1)
	{
	     bf->c1 = 1;
	     /* Issue another packet generate event */
	     tw_event * e_gen;
	     terminal_message * m_gen;
	     void * m_gen_data;

	     /* Keep the packet generate event a little behind packet send */
	     e_gen = model_net_method_event_new(lp->gid, codes_local_latency(lp), lp, DRAGONFLY, (void**)&m_gen,(void**)&m_gen_data); 

	     void *m_gen_data_src = model_net_method_get_edata(DRAGONFLY, msg);
	     memcpy(m_gen, msg, sizeof(terminal_message));

	     m_gen->chunk_id = msg->chunk_id + 1;
         m_gen->type = T_GENERATE;
	     m_gen->travel_start_time = travel_start_time;
     		
	      if (msg->remote_event_size_bytes){
		memcpy(m_gen_data, m_gen_data_src,
			msg->remote_event_size_bytes);
		m_gen_data = (char*)m_gen_data + msg->remote_event_size_bytes;
		m_gen_data_src = (char*)m_gen_data_src + msg->remote_event_size_bytes;
	      }
	    if (msg->local_event_size_bytes)
		memcpy(m_gen_data, m_gen_data_src, msg->local_event_size_bytes);
     
           tw_event_send(e_gen); 
       }
	
  return;
}

static void packet_arrive_rc(terminal_state * s, 
		   tw_bf * bf, 
                   terminal_message * msg, 
                   tw_lp * lp)
{
    term_ecount--;
    term_rev_ecount++;
 
    uint64_t num_chunks = msg->num_chunks;
   
    if(msg->chunk_id == num_chunks - 1)
      completed_packets--;

   if(msg->path_type == MINIMAL && msg->chunk_id == num_chunks - 1)
	minimal_count--;

   if(msg->path_type == NON_MINIMAL && msg->chunk_id == num_chunks - 1)
	nonmin_count--;

   tw_rand_reverse_unif(lp->rng);
   s->next_credit_available_time = msg->saved_credit_time;
   if(msg->chunk_id == num_chunks-1)
   {
    mn_stats* stat;
    stat = model_net_find_stats(msg->category, s->dragonfly_stats_array);
    stat->recv_count--;
    stat->recv_bytes -= msg->packet_size;
    stat->recv_time -= tw_now(lp) - msg->travel_start_time;
    
    N_finished_packets--;
    
    dragonfly_total_time -= (tw_now(lp) - msg->travel_start_time);
    if(bf->c3)
	 dragonfly_max_latency = msg->saved_available_time;
   }
    
   if (msg->chunk_id == num_chunks-1 && 
		msg->remote_event_size_bytes && 
		msg->is_pull)
    {
	    int net_id = model_net_get_id(LP_METHOD_NM);
	    model_net_event_rc(net_id, lp, msg->pull_size);
    }

}