自学教程：C++ tw_now函数代码示例

static void node_collective_fan_out(nodes_state * s,                                    tw_bf * bf,                                    nodes_message * msg,                                    tw_lp * lp){    int i;    //TODO: be annotation-aware    int num_lps = codes_mapping_get_lp_count(grp_name, 1, LP_CONFIG_NM,                  NULL, 1);    bf->c1 = 0;    bf->c2 = 0;    send_remote_event(s, bf, msg, lp);    if(!s->is_leaf)    {        bf->c1 = 1;        tw_event* e_new;        nodes_message * msg_new;        tw_stime xfer_to_nic_time;        for( i = 0; i < s->num_children; i++ )        {            xfer_to_nic_time = g_tw_lookahead + TORUS_FAN_OUT_DELAY + tw_rand_exponential(lp->rng, (double)TORUS_FAN_OUT_DELAY/10);            if(s->children[i] > 0)            {                tw_lpid child_nic_id;                /* get global LP ID of the child node */                codes_mapping_get_lp_id(grp_name, LP_CONFIG_NM, NULL, 1,                                        s->children[i]/num_lps,                                        (s->children[i] % num_lps), &child_nic_id);                //e_new = codes_event_new(child_nic_id, xfer_to_nic_time, lp);                //msg_new = tw_event_data(e_new);                //memcpy(msg_new, msg, sizeof(nodes_message) + msg->remote_event_size_bytes);                void* m_data;                e_new = model_net_method_event_new(child_nic_id,                                                   xfer_to_nic_time,                                                   lp, TORUS, (void**)&msg_new, &m_data);                memcpy(msg_new, msg, sizeof(nodes_message));                if (msg->remote_event_size_bytes) {                    memcpy(m_data, model_net_method_get_edata(TORUS, msg),                           msg->remote_event_size_bytes);                }                msg_new->type = T_COLLECTIVE_FAN_OUT;                msg_new->sender_node = s->node_id;                tw_event_send(e_new);            }        }    }    //printf("/n Fan out phase completed %ld ", lp->gid);    if(max_collective < tw_now(lp) - s->collective_init_time )    {        bf->c2 = 1;        max_collective = tw_now(lp) - s->collective_init_time;    }}

static void handle_pong_event(node_state *ns, tw_bf *b, node_msg *m, tw_lp *lp) {  int i;  if(ns->msg_sent_count < num_reqs*num_messages) {    node_msg m_remote;        m_remote.node_event_type = PING;    m_remote.src = lp->gid;    tw_stime tmp = tw_now(lp);    ns->total_ts += tmp - ns->last_ts;    ns->last_ts = tmp;        for (i=0; i<num_messages; i++) {      model_net_event(net_id, "ping", m->src, payload_sz, 0.0, sizeof(node_msg),		      (const void *)&m_remote, 0, NULL, lp);      ns->msg_sent_count++;    }    m->incremented_flag = 1;  }  else {    m->incremented_flag = 0;    ns->end_ts = tw_now(lp);  }  return;}

int ForwardPacket (tw_lp * srcLp, tw_lp * dstLp, Packet * p, tw_stime t){  tw_event * e;  EventMsg * em;  double now = tw_now (dstLp);    if (t + now >= g_tw_ts_end) return 0;  /* Otherwise the program will exit when                                          * creating event */  #ifdef DEBUG_PKT_FWD  printf ("%lf : pkt(%d:%lu) from %d to %d (in-port %d) at %lf/n",           tw_now (srcLp), _addrToIdx[p->srcAddr_],          p->od_.sqn_, srcLp->id, dstLp->id, p->inPort_, now + t);  fflush (stdout);#endif      if ((e = tw_event_new (dstLp, t, dstLp)) == NULL) {    printf ("Failed to create event for packet./n");    tw_exit (-1);  }    (em = (EventMsg *) tw_event_data (e))->type_ = DATA_PKT;    memcpy ((void *) &(em->content_.pkt_), (void *) p, sizeof (Packet));    tw_event_send (e);    return 1;}

	/*	 * epi_agent_add_back: helper routine that adds an agent back to the	 *                     queue from which it was just removed	 *        Must call num_add if appropriate, since we are skipping	 *        the EPI_ADD event	 */voidepi_agent_add_back(epi_state * state, tw_lp * lp, tw_memory * buf){	tw_memory	*b;	epi_agent	*a;	epi_pathogen	*p;	a = tw_memory_data(buf);	if(5 == a->id && 0)		printf("%lld: added back %d at %lf /n", lp->gid, a->id && 0, tw_now(lp));	// Need to move curr but not change location.	if(++a->curr >= a->nloc)		a->curr = 0;	a->ts_next = a->ts_remove = tw_now(lp) + a->dur[a->curr];	for(b = a->pathogens; b; b = b->next)	{		p = tw_memory_data(b);		if (p->ts_stage_tran <= a->ts_remove)			a->ts_next = p->ts_stage_tran;	}	pq_enqueue(g_epi_pq[lp->id], buf);	//epi_num_add(state, (tw_bf *) NULL, a, lp);}

void fcfs_add (        model_net_request     * req,        const mn_sched_params * sched_params,        int                     remote_event_size,        void                  * remote_event,        int                     local_event_size,        void                  * local_event,        void                  * sched,        model_net_sched_rc    * rc,        tw_lp                 * lp){    mn_sched_qitem *q = malloc(sizeof(mn_sched_qitem));    q->entry_time = tw_now(lp);    q->req = *req;    q->sched_params = *sched_params;    q->rem = req->is_pull ? PULL_MSG_SIZE : req->msg_size;    if (remote_event_size > 0){        q->remote_event = malloc(remote_event_size);        memcpy(q->remote_event, remote_event, remote_event_size);    }    else { q->remote_event = NULL; }    if (local_event_size > 0){        q->local_event = malloc(local_event_size);        memcpy(q->local_event, local_event, local_event_size);    }    else { q->local_event = NULL; }    mn_sched_queue *s = sched;    s->queue_len++;    qlist_add_tail(&q->ql, &s->reqs);    dprintf("%lu (mn):    adding %srequest from %lu to %lu, size %lu, at %lf/n",            lp->gid, req->is_pull ? "pull " : "", req->src_lp,            req->final_dest_lp, req->msg_size, tw_now(lp));}

/* * @return 1 if forwarded, 0 if discarded */int ForwardPktOnLink (tw_lp * lp, Link * lnk, Packet * p){    double t, now = tw_now (lp);  if (lnk->lastPktOutTime_ <= now) {      /* No packet is being sent out. Therefore forward it right now */    t = p->size_ * 8 / ((double) lnk->bw_); /* Transmission delay */    lnk->lastPktOutTime_ = now + t;    p->inPort_ = lnk->nbPort_; /* This will copied to the outgoing packet */    return ForwardPacket (lp, tw_getlp (lnk->neighbour_), p, t + lnk->latency_);  }        if (p->size_ + (lnk->lastPktOutTime_ - now) / 8 * lnk->bw_       > lnk->bufSize_ + SMALL_FLOAT) {    /* Buffer is full. Discard the packet     * The buffered packets include the one being sent except the portion     * that has been sent. */#ifdef DEBUG_PKT_DISCARD    printf ("%lf : pkt(%d:%lu) at %d discarded/n",            tw_now (lp), _addrToIdx[p->srcAddr_], p->od_.sqn_, lp->id);    fflush (stdout);#endif          return 0;  }    /* The buffer is not zero, some packet is being sent.   * Schedule to forward this packet at a future time */  p->inPort_ = lnk->nbPort_; /* This will copied to the outgoing packet */  lnk->lastPktOutTime_ += p->size_ * 8 / ((double) lnk->bw_);      return ForwardPacket (lp, tw_getlp (lnk->neighbour_), p,                         lnk->lastPktOutTime_ + lnk->latency_ - now);}

void TfmccSrcStart (Agent * agent){  TfmccSrcInfo * tsi = (TfmccSrcInfo *) agent->info_;  tw_lp * lp = tw_getlp (agent->nodeId_);  double now = tw_now (lp);    tsi->active_ = 1;    tsi->pktSize_ = TFMCC_SRC_DATA_PKT_SIZE;  tsi->fairsize_ = TFMCC_SRC_DATA_PKT_SIZE;  tsi->ndatapack_ = 0;  tsi->overhead_ = 0;  tsi->ssmult_ = 2.0;  tsi->bval_ = 1;  tsi->t_factor_ = 6;  tsi->seqno_ = 0;  tsi->oldrate_ = tsi->rate_ = TFMCC_SRC_INIT_RATE;  tsi->delta_ = 0;  tsi->rate_change_ = TFMCC_SRC_STATE_INIT_RATE;  tsi->last_change_ = 0;  tsi->round_begin_ = now; // not 0, otherwise "no-feedback reduction"                            // in first round  tsi->maxrate_ = 0;  tsi->ndatapack_ = 0;  tsi->round_id = 0;  tsi->expected_rate = 0 ;  tsi->inter_packet = 0;  tsi->max_rtt_ = TFMCC_SRC_INIT_RTT;  tsi->supp_rate_ = DBL_MAX;  tsi->rtt_recv_id = -1;  tsi->rtt_prio = 0;  tsi->rtt_recv_timestamp = 0;  tsi->rtt_recv_last_feedback = 0;  tsi->rtt_rate = 0;  // CLR  tsi->clr_id = -1;  tsi->clr_timestamp = 0;  tsi->clr_last_feedback = 0;#ifdef TRACE_TFMCC_THROUGHPUT  tsi->thruSampleTime_ = tw_now (tw_getlp (agent->nodeId_));  tsi->totalSentBytes_ = 0;#endif    // send the first packet  TfmccSrcSendPkt (agent);  // ... at initial rate  tsi->sendPktTimerEvnt_ = ScheduleTimer     (agent, TFMCC_SRC_TIMER_SEND_PKT, tsi->pktSize_ / tsi->rate_);  // ... and start timer so we can cut rate   // in half if we do not get feedback  tsi->noFdbkTimerEvnt_ = ScheduleTimer    (agent, TFMCC_SRC_TIMER_NO_FDBK,      TFMCC_SRC_DEC_NO_REPORT * tsi->pktSize_/ tsi->rate_);}

voidtmr_event_handler(tmr_state * s, tw_bf * bf, tmr_message * m, tw_lp * lp){	tw_lpid	 dest;	* (int *) bf = 0;	if(s->timer)	{		// if current event being processed is the timer		if(tw_event_data(s->timer) == m)		{			bf->c1 = 1;			m->old_timer = s->timer;			s->timer = NULL;			fprintf(f, "%lld: tmr fired at %lf/n", 				lp->gid, tw_now(lp));			return;		} else		{			bf->c2 = 1;			m->old_time = s->timer->recv_ts;			tw_timer_reset(lp, &s->timer, tw_now(lp) + 100.0);			if(s->timer == NULL)				fprintf(f, "%lld: reset tmr failed %lf/n", 					lp->gid, tw_now(lp) + 100.0);			else				fprintf(f, "%lld: reset tmr to %lf/n", 					lp->gid, tw_now(lp) + 100.0);		}	}	if(tw_rand_unif(lp->rng) <= percent_remote)	{		bf->c3 = 1;		dest = tw_rand_integer(lp->rng, 0, ttl_lps - 1);		dest += offset_lpid;		if(dest >= ttl_lps)			dest -= ttl_lps;	} else	{		bf->c3 = 0;		dest = lp->gid;	}	if(!lp->gid)		dest = 0;	tw_event_send(tw_event_new(dest, tw_rand_exponential(lp->rng, mean), lp));}

voidtcp_timeout(tcp_state * state, tw_bf * bf, tcp_message * in, tw_lp * lp){	if(in->seq_num < state->unack || state->unack >= state->len)		return;	bf->c1 = 1;#if TCP_DEBUG//if(lp->id == 9377)	printf("/n/t%lld: timeout %lf/n", lp->id, tw_now(lp));#endif	//in->dst = state->ssthresh;	state->ssthresh = (min(((int)state->cwnd + 1), state->recv_wnd) / 2) * state->mss;	//in->RC.cwnd = state->cwnd;	state->cwnd = 1;	//in->RC.lastsent = state->lastsent;	tcp_event_send(state, lp->id, 0.0, state->connection, TCP_MTU, state->unack, 0, lp);	in->ack = state->seq_num;	//state->seq_num = state->unack + state->mss;	state->stats->tout++;	state->stats->sent++;	//in->RC.dup_count = state->rto;	state->rto = state->rto * 2;	state->rto_seq++;	if(state->rto < 60.0)	{#if TCP_DEBUG//if(lp->id == 9377)	printf("/t%lld: timeout reset to %lf (was %lf) (unack %d)/n/n", lp->id, tw_now(lp) + state->rto, state->timer->recv_ts, state->unack);#endif		state->timer = tcp_timer_reset(state->timer, bf, in, 					tw_now(lp) + state->rto, state->unack, lp);	} else	{#if TCP_DEBUG//if(lp->id == 9377)	printf("/t%lld: timeout cancelled (was %lf) (unack %d)/n/n", lp->id, state->timer->recv_ts, state->unack);		state->timer = NULL;#endif		rn_event_send(rn_event_new(lp->gid, 0.0, lp, UPSTREAM, 0));	}	// state-save into event	//in->RC.rtt_seq = state->rtt_seq;	state->rtt_seq = state->unack;	//in->RC.rtt_time = state->rtt_time;	state->rtt_time = 0;}

voidpacket_process(nodes_state * s, tw_bf * bf, nodes_message * msg, tw_lp * lp){  int i;  tw_event *e;  nodes_message *m;  bf->c3 = 1;  // One packet leaves the queue  s->node_queue_length[msg->source_direction][msg->source_dim]--;  s->N_wait_to_be_processed--;    if(lp->gid==msg->dest_lp)    {      // one packet arrives and dies      bf->c3 = 0;	N_finished++;	int index = floor(N_COLLECT_POINTS*(tw_now(lp)/g_tw_ts_end));	N_finished_storage[index]++;	total_time += tw_now(lp) - msg->travel_start_time;	if (max_latency<tw_now(lp) - msg->travel_start_time)	  max_latency=tw_now(lp) - msg->travel_start_time;	total_hops += msg->my_N_hop;	total_queue_length += msg->my_N_queue;	queueing_times_sum += msg->queueing_times;	//total_queue_length += msg->accumulate_queue_length;    }  else    {      e = tw_event_new(lp->gid, MEAN_PROCESS, lp);      m = tw_event_data(e);      m->type = SEND;            // Carry on the message info      for( i = 0; i < N_dims; i++ )	m->dest[i] = msg->dest[i];      m->dest_lp = msg->dest_lp;      m->transmission_time = msg->transmission_time;            m->source_dim = msg->source_dim;      m->source_direction = msg->source_direction;            m->packet_ID = msg->packet_ID;	        m->travel_start_time = msg->travel_start_time;      m->my_N_hop = msg->my_N_hop;      m->my_N_queue = msg->my_N_queue;      m->queueing_times = msg->queueing_times;      tw_event_send(e);          }  }

tw_event	*forward(ip_state * state, tw_bf * bf, rn_message * msg, tw_lp * lp){	tw_memory	*b;	tw_event	*e;	tw_stime	 ts = 0.0;	rn_link		*l;	ip_message	*ip;	state->stats->s_nforward++;	if(NULL == (l = getroute(state, msg, lp)))		return NULL;	b = tw_memory_alloc(lp, g_ip_fd);	b->ts = tw_now(lp);	tw_memoryq_push(state->link_q, b);	ip = tw_memory_data(b);	ip->link = l;	ip->last_sent = l->last_sent;	if(tw_now(lp) > l->last_sent)		l->last_sent = tw_now(lp);	else		ts = l->last_sent - tw_now(lp);	l->last_sent += (msg->size / l->bandwidth);	ts += ((msg->size / l->bandwidth) + l->delay);	l->avg_delay += msg->size;	e = rn_event_new(l->addr, ts, lp, DOWNSTREAM, msg->size);#if 0	if((bf->c31 = (e == lp->pe->abort_event && e->recv_ts <= g_tw_ts_end)))	{		tw_error(TW_LOC, "Got abort event ...");		state->stats->s_nnet_failures++;		msg->ttl++;	}#endif#if VERIFY_IP	printf("/t/t%lld IP FWD: to %lld at ts %lf (sz %d bw %lf del %lf)/n", 		lp->gid, l->addr, e->recv_ts, msg->size, l->bandwidth, l->delay);#endif	return e;}

static void svr_finalize(    svr_state * ns,    tw_lp * lp){    printf("server %llu : size:%d requests:%d time:%lf rate:%lf/n",           (unsigned long long)lp->gid,           PAYLOAD_SZ,           NUM_REQS,           ns_to_s((tw_now(lp)-ns->start_ts)),           (double)(PAYLOAD_SZ*NUM_REQS)/(1024.0*1024.0)/ns_to_s((tw_now(lp)-ns->start_ts)));    return;}

void analysis_event(analysis_state *s, tw_bf *bf, analysis_msg *m, tw_lp *lp){    int i;    tw_lp *model_lp;    lp->pe->stats.s_alp_nevent_processed++; //don't undo in RC    if ((g_st_model_stats == VT_STATS || g_st_model_stats == ALL_STATS) && s->num_lps > 0)    {        model_sample_data *sample = s->model_samples_current;        // TODO handle this situation better        if (sample == s->model_samples_tail)            printf("WARNING: last available sample space for analysis lp!/n");        sample->timestamp = tw_now(lp);        m->timestamp = tw_now(lp);        // call the model sampling function for each LP on this KP        for (i = 0; i < s->num_lps; i++)        {            if (s->lp_list[i] == ULLONG_MAX)                break;            model_lp = tw_getlocal_lp(s->lp_list[i]);            if (model_lp->model_types == NULL || model_lp->model_types->sample_struct_sz == 0)                continue;            model_lp->model_types->sample_event_fn(model_lp->cur_state, bf, lp, sample->lp_data[i]);        }        s->model_samples_current = s->model_samples_current->next;    }    // sim engine sampling    if (g_tw_synchronization_protocol != SEQUENTIAL &&             (g_st_engine_stats == VT_STATS || g_st_engine_stats == ALL_STATS))    {#ifdef USE_DAMARIS        if (g_st_damaris_enabled)            st_damaris_expose_data(lp->pe, tw_now(lp), ANALYSIS_LP);        else            st_collect_engine_data(lp->pe, ANALYSIS_LP);#else        st_collect_engine_data(lp->pe, ANALYSIS_LP);#endif    }        //collect_sim_engine_data(lp->pe, lp, s, (tw_stime) tw_clock_read() / g_tw_clock_rate);        // create next sampling event    st_create_sample_event(lp);}

voidnum_file_start(num_state * state, tw_bf * bf, unsigned int size, tw_lp * lp){#if NUM_VERIFY	if(lp->id == DEBUG_LP)		printf("%ld T%d %s: FILE_START size: %.2lfB/n", 			lp->id, state->type, get_tod(tw_now(lp)), (double) size / 8.0);#endif	// send event down to TCP	state->file_size = size;	state->stats->s_nstart++;	state->start = tw_now(lp);	rn_event_send(rn_event_new(lp, 0.0, lp, DOWNSTREAM, size/8));}

void cont_create_process( CON_state* s, tw_bf* bf, MsgData* msg, tw_lp* lp ){  tw_event * e;  tw_stime ts;  MsgData * m;#ifdef TRACE  printf("create %d process at controller travel time is %lf/n",         msg->message_CN_source,         tw_now(lp) - msg->travel_start_time );#endif  ts = CONT_CONT_msg_prep_time;  e = tw_event_new( lp->gid, ts , lp );  m = tw_event_data(e);  m->event_type = CREATE_ACK;  m->travel_start_time = msg->travel_start_time;  m->io_offset = msg->io_offset;  m->io_payload_size = msg->io_payload_size;  m->collective_group_size = msg->collective_group_size;  m->collective_group_rank = msg->collective_group_rank;  m->collective_master_node_id = msg->collective_master_node_id;  m->io_type = msg->io_type;  m->message_ION_source = msg->message_ION_source;  m->message_FS_source = msg->message_FS_source;  m->message_CN_source = msg->message_CN_source;  m->io_tag = msg->io_tag;  tw_event_send(e);}

void TfmccSrcIncrRate (Agent * agent){  TfmccSrcInfo * tsi = (TfmccSrcInfo *) agent->info_;  tw_lp * lp = tw_getlp (agent->nodeId_);  double now = tw_now (lp), next;  double mult = (now - tsi->last_change_) / tsi->max_rtt_;  if (mult > 2) mult = 2;    tsi->rate_ = tsi->rate_ + (tsi->fairsize_ / tsi->max_rtt_) * mult;  if (tsi->rate_ > tsi->expected_rate) tsi->rate_ = tsi->expected_rate;  if (tsi->rate_ < tsi->fairsize_ / tsi->max_rtt_) {    tsi->rate_ = tsi->fairsize_ / tsi->max_rtt_;  }  tsi->last_change_ = now;  // check if rate increase should impact send timer  next = tsi->pktSize_ / tsi->rate_;  if (tsi->inter_packet >= 2 * next) {    assert(next > TFMCC_SMALLFLOAT);    ReschedTimer (agent, TFMCC_SRC_TIMER_SEND_PKT, next,       & (tsi->sendPktTimerEvnt_));  }}

/* * lsm_lp_init *   - initialize the lsm model *   - sets the disk to be idle now */static void lsm_lp_init (lsm_state_t *ns, tw_lp *lp){    memset(ns, 0, sizeof(*ns));    ns->next_idle = tw_now(lp);    // set the correct model    const char *anno = codes_mapping_get_annotation_by_lpid(lp->gid);    if (anno == NULL)        ns->model = &model_unanno;    else {        int id = configuration_get_annotation_index(anno, anno_map);        ns->model = &models_anno[id];    }    // initialize the scheduler if need be    ns->use_sched = ns->model->use_sched > 0;    if (ns->use_sched) {        ns->sched.num_prios = ns->model->use_sched;        ns->sched.active_count = 0;        rc_stack_create(&ns->sched.freelist);        ns->sched.queues =            malloc(ns->sched.num_prios * sizeof(*ns->sched.queues));        for (int i = 0; i < ns->sched.num_prios; i++)            INIT_QLIST_HEAD(&ns->sched.queues[i]);    }    return;}

/* write stats to buffer * currently does not overwrite in cases of overflow, just records the amount of overflow in bytes * for later reporting */void st_buffer_push(int type, char *data, int size){    int size1, size2;    if (!g_st_disable_out && st_buffer_free_space(g_st_buffer[type]) < size)    {        if (!buffer_overflow_warned)        {            printf("WARNING: Stats buffer overflow on rank %lu/n", g_tw_mynode);            buffer_overflow_warned = 1;            printf("tw_now() = %f/n", tw_now(g_tw_lp[0]));        }        missed_bytes += size;        size = 0; // if we can't push it all, don't push anything to buffer    }    if (size)    {        if ((size1 = g_st_buffer[type]->size - g_st_buffer[type]->write_pos) >= size)        {            // can use only one memcpy here            memcpy(st_buffer_write_ptr(g_st_buffer[type]), data, size);            g_st_buffer[type]->write_pos += size;        }        else // data to be stored wraps around end of physical array        {            size2 = size - size1;            memcpy(st_buffer_write_ptr(g_st_buffer[type]), data, size1);            memcpy(g_st_buffer[type]->buffer, data + size1, size2);            g_st_buffer[type]->write_pos = size2;        }    }    g_st_buffer[type]->count += size;    //printf("PE %ld wrote %d bytes to buffer; %d bytes of free space left/n", g_tw_mynode, size, st_buffer_free_space(g_st_buffer[type]));}

voidtcp_timer_cancel(tw_event * timer, tw_bf * bf, tcp_message * old, tw_lp * lp){#if VERIFY_TCP_TIMER	printf("%lld: TCP cancel at %lf: old %lf/n", 		lp->gid, tw_now(lp), timer->recv_ts);#endif	old->RC.timer_ts = timer->recv_ts;	//if(timer->recv_ts <= tw_now(lp) || timer->state.uint == 0)	if(timer == lp->pe->cur_event)		return;	if(!timer->memory)		tw_error(TW_LOC, "No membuf on timer!");	tw_memory_free(lp, timer->memory, g_tcp_fd);	timer->memory = NULL;	rn_timer_cancel(lp, &timer);#if VERIFY_TCP_TIMER	printf("/t%lld: CANCELLED RTO for %lf/n/n/n", 		lp->gid, old->RC.timer_ts);#endif}

/* * epi_agent_send: create new event, and pack on the agent membuf and * associated pathogen membufs */voidepi_agent_send(tw_lp * src, tw_memory * buf, tw_stime offset, tw_lpid dst){	tw_event	*e;	tw_memory	*b;	tw_memory	*next;	epi_agent	*a;	e = tw_event_new(dst, 0.0, src);	a = tw_memory_data(buf);	printf("%lld: Sending agent to %lld (%d, %lld) at %lf/n", 		src->id, dst, a->curr, a->loc[a->curr], tw_now(src));	// place agent membuf on event, and pack pathogen membufs on after that	// in case this event/agent is sent over the network.	for(next = b = a->pathogens; b; b = next)	{		next = b->next;		tw_event_memory_set(e, b, g_epi_pathogen_fd);	}	tw_event_memory_set(e, buf, g_epi_fd);	a->pathogens = NULL;	tw_event_send(e);}

void traffic_light_intersection_startup(intersection_state* SV, tw_lp* LP) {    tw_stime ts = 0;    tw_event* current_event;    message_data* new_message;    // Initialize the total number of cars arrived:    SV->total_cars_arrived = 0;    SV->total_cars_finished = 0;    // Initialize the number of cars arriving into the intersection:    SV->num_cars_south = 0;    SV->num_cars_west = 0;    SV->num_cars_north = 0;    SV->num_cars_east = 0;    SV->num_cars_south_left = 0;    SV->num_cars_west_left = 0;    SV->num_cars_north_left = 0;    SV->num_cars_east_left = 0;    SV->north_south_green_until = -1;    SV->north_south_left_green_until = -1;    SV->east_west_green_until = -1;    SV->east_west_left_green_until = -1;    // Initialize a random direction to be green    SV->traffic_direction = tw_rand_ulong(LP->rng, NORTH_SOUTH, EAST_WEST_LEFT);    // Schedule the first light change    current_event = tw_event_new(LP->gid, ts, LP);    new_message = (message_data*)tw_event_data(current_event);    new_message->event_type = LIGHT_CHANGE;    tw_event_send(current_event);    // Put cars on the road    int i;    for(i = 0; i < g_traffic_start_events; i++) {        // Arrival time        ts = tw_rand_exponential(LP->rng, INITIAL_ARRIVAL_MEAN);        current_event = tw_event_new(LP->gid, ts, LP);        new_message = (message_data*)tw_event_data(current_event);        new_message->event_type = CAR_ARRIVES;        assign_rand_dest:        new_message->car.x_to_go = tw_rand_integer(LP->rng, -MAX_TRAVEL_DISTANCE, MAX_TRAVEL_DISTANCE);        new_message->car.y_to_go = tw_rand_integer(LP->rng, -MAX_TRAVEL_DISTANCE, MAX_TRAVEL_DISTANCE);        new_message->car.x_to_go_original = new_message->car.x_to_go;        new_message->car.y_to_go_original = new_message->car.y_to_go;        if (new_message->car.y_to_go == 0)            new_message->car.has_turned = 1;        else            new_message->car.has_turned = 0;        new_message->car.start_time = tw_now(LP);        tw_event_send(current_event);    }}

void TfmccSrcSendPkt (Agent * agent){  TfmccSrcInfo * tsi = (TfmccSrcInfo *) agent->info_;  tw_lp * lp = tw_getlp (agent->nodeId_);  double now = tw_now (lp);  Packet p;  TfmccData * data = (TfmccData *) & (p.type.td_);    if (! tsi->active_) return;  p.type_ = TFMCC_DATA;    p.srcAddr_ = _nodeTable[agent->nodeId_].addr_;  p.srcAgent_ = agent->id_;  p.dstAddr_ = tsi->pktDstAddr_;  p.dstAgent_ = tsi->pktDstAgentId_;  p.size_ = tsi->pktSize_;  p.inPort_ = (uint16) -1;    data->seqno = tsi->seqno_++;  data->timestamp = now;  data->supp_rate = tsi->supp_rate_;  data->round_id = tsi->round_id;  data->max_rtt = tsi->max_rtt_;  if (tsi->clr_id == tsi->rtt_recv_id || tsi->rtt_recv_id == -1) {    data->rtt_recv_id = tsi->clr_id;    data->timestamp_echo =  tsi->clr_timestamp;    data->timestamp_offset = now - tsi->clr_last_feedback;    data->is_clr = 1;  } else {    data->rtt_recv_id = tsi->rtt_recv_id;    data->timestamp_echo = tsi->rtt_recv_timestamp;    data->timestamp_offset = now - tsi->rtt_recv_last_feedback;    data->is_clr = 0;  }  SendPacket (lp, &p);  tsi->ndatapack_++;  tsi->rtt_recv_id = -1;  tsi->rtt_prio = 0;#ifdef TRACE_TFMCC_THROUGHPUT  if (! tsi->trackStat_) return;  tsi->totalSentBytes_ += p.size_;  if (now - tsi->thruSampleTime_ < TFMCC_THROUGHPUT_SAMPLE_ITVL) {      return;  }     /* For rate tracing */  printf ("%lf : at %d.%ld , throughput rate = %lf Mbps, itvl = %lf/n",    now, lp->id, agent->id_,     tsi->totalSentBytes_ / (now - tsi->thruSampleTime_) / 125000,    now - tsi->thruSampleTime_);  tsi->thruSampleTime_ = now;  tsi->totalSentBytes_ = 0;#endif}

void TfmccSrcDecrRate (Agent * agent){  TfmccSrcInfo * tsi = (TfmccSrcInfo *) agent->info_;  tw_lp * lp = tw_getlp (agent->nodeId_);  double now = tw_now (lp);  tsi->oldrate_ = tsi->rate_ = tsi->expected_rate;  tsi->last_change_ = now;}

/* reverse handler code for a MPI process LP */void mpi_event_rc_handler( mpi_process * p,                           tw_bf * bf,                           nodes_message * msg,                           tw_lp * lp){ switch(msg->type)  {     case MPI_SEND:                {		  if(bf->c4)		    return;		  if(bf->c3)		       tw_rand_reverse_unif(lp->rng);		    int i; 		    for( i=0; i < num_packets; i++ )		     {			     tw_rand_reverse_unif(lp->rng);			     p->available_time = msg->saved_available_time;			     p->message_counter--;					     }		    tw_rand_reverse_unif(lp->rng);	     }     break;     case MPI_RECV:		  {			int index = floor( N_COLLECT_POINTS * ( tw_now( lp ) / g_tw_ts_end ) );			N_finished_msgs--;			N_finished_packets--;			N_finished_storage[index]--;			N_num_hops[index] -= msg->my_N_hop;			total_time -= tw_now( lp ) - msg->travel_start_time;			total_hops-=msg->my_N_hop;			if(bf->c3)			  max_latency= msg->saved_available_time;		  }      break;  }}

/* handle recving ack */static void handle_ack_event(    svr_state * ns,    tw_bf * b,    svr_msg * m,    tw_lp * lp){    svr_msg * m_local = malloc(sizeof(svr_msg));    svr_msg * m_remote = malloc(sizeof(svr_msg));//    m_local->svr_event_type = REQ;    m_local->svr_event_type = LOCAL;    m_local->src = lp->gid;    memcpy(m_remote, m_local, sizeof(svr_msg));    m_remote->svr_event_type = (do_pull) ? ACK : REQ;    //printf("handle_ack_event(), lp %llu./n", (unsigned long long)lp->gid);    /* safety check that this request got to the right server *///    printf("/n m->src %d lp->gid %d ", m->src, lp->gid);    int opt_offset = 0;       if(net_id == DRAGONFLY && (lp->gid % lps_per_rep == num_servers_per_rep - 1))      opt_offset = num_servers_per_rep + num_routers_per_rep; /* optional offset due to dragonfly mapping */    	    tw_lpid dest_id = (lp->gid + offset + opt_offset)%(num_servers*2 + num_routers);    /* in the "pull" case, src should actually be self */    if (do_pull){        assert(m->src == lp->gid);    }    else{        assert(m->src == dest_id);    }    if(ns->msg_sent_count < NUM_REQS)    {        /* send another request */        if (do_pull){            model_net_pull_event(net_id, "test", dest_id, PAYLOAD_SZ, 0.0,                    sizeof(svr_msg), (const void*)m_remote, lp);        }        else{            model_net_event(net_id, "test", dest_id, PAYLOAD_SZ, 0.0, sizeof(svr_msg), (const void*)m_remote, sizeof(svr_msg), (const void*)m_local, lp);        }        ns->msg_sent_count++;        m->incremented_flag = 1;    }    else    {        ns->end_ts = tw_now(lp);        m->incremented_flag = 0;    }    return;}

static void svr_finalize(    svr_state * ns,    tw_lp * lp){    ns->end_ts = tw_now(lp);    printf("server %llu recvd %d bytes in %f seconds, %f MiB/s sent_count %d recvd_count %d local_count %d /n", (unsigned long long)lp->gid, PAYLOAD_SZ*ns->msg_recvd_count, ns_to_s(ns->end_ts-ns->start_ts),        ((double)(PAYLOAD_SZ*ns->msg_sent_count)/(double)(1024*1024)/ns_to_s(ns->end_ts-ns->start_ts)), ns->msg_sent_count, ns->msg_recvd_count, ns->local_recvd_count);    return;}

static void st_create_sample_event(tw_lp *lp){    if (tw_now(lp) + g_st_vt_interval <= g_st_sampling_end)    {        tw_event *e = tw_event_new(lp->gid, g_st_vt_interval, lp);        analysis_msg *m = (analysis_msg*) tw_event_data(e);        m->src = lp->gid;        tw_event_send(e);    }}

void cont_read_ack( CON_state* s, tw_bf* bf, MsgData* msg, tw_lp* lp ){  tw_event * e;  tw_stime ts;  MsgData * m;  double transmission_time = msg->io_payload_size/DDN_FS_out_bw;#ifdef TRACE  printf("read %d ACKed at controller travel time is %lf/n",         msg->message_CN_source,         tw_now(lp) - msg->travel_start_time );#endif  s->fs_sender_next_available_time = max(s->fs_sender_next_available_time, tw_now(lp) );  ts = s->fs_sender_next_available_time - tw_now(lp);  s->fs_sender_next_available_time += transmission_time;  e = tw_event_new( msg->message_FS_source, ts + transmission_time , lp );  m = tw_event_data(e);  m->event_type = READ_ACK;  m->travel_start_time = msg->travel_start_time;  m->io_offset = msg->io_offset;  m->io_payload_size = msg->io_payload_size;  m->collective_group_size = msg->collective_group_size;  m->collective_group_rank = msg->collective_group_rank;  m->collective_master_node_id = msg->collective_master_node_id;  m->io_type = msg->io_type;  m->message_ION_source = msg->message_ION_source;  m->message_FS_source = msg->message_FS_source;  m->message_CN_source = msg->message_CN_source;  m->io_tag = msg->io_tag;  m->IsLastPacket = msg->IsLastPacket;  tw_event_send(e);}

	/*	 * Adding an agent means placing them into our PQ.  If agent	 * remove ts is < PQ->min, then must update LP remove timer.	 *	 * Check time for stage change.  Place in queue by minimum	 * stage change time or remove time.  Set timer for stage	 * change time (and event) or remove time (and event).	 */voidepi_agent_add(epi_state * state, tw_bf * bf, tw_memory * buf, tw_lp * lp){	epi_agent	*a;	a = tw_memory_data(buf);	if(5 == a->id && 0)		printf("%lld: added %d at %lf /n", lp->gid, a->id && 0, tw_now(lp));	// Test to make sure this location is this lp	if (lp->gid != a->loc[a->curr])		tw_error(TW_LOC, "epi_agent_add adding agent to lp %d but loc[%d] is %d",			 lp->gid, a->curr, a->loc[a->curr]);	// if agent is contagious, increment number of contagious agents here	epi_agent_contagious(state, buf, +1);	a->ts_remove = tw_now(lp) + a->dur[a->curr];	epi_agent_update_ts(state, bf, a, lp);	pq_enqueue(g_epi_pq[lp->id], buf);	if(pq_get_size(g_epi_pq[lp->id]) > state->max_queue_size)		state->max_queue_size = pq_get_size(g_epi_pq[lp->id]);	// If not at home, if agent is a worried well, if there are already enough symtomatic	// people here then starting tomorrow, this agent will stay home for the number	// of days remaining.  days_remaining is checked in agent_remove().		if(!a->days_remaining && a->nloc > 1 && a->loc[a->curr] != a->loc[0] && 		(a->behavior_flags & EPI_AGENT_WORRIED_WELL))	{		if ((float) state->ncontagious_tot / (float) state->max_queue_size > 				g_epi_ww_threshold)		{			a->days_remaining = g_epi_ww_duration;			a->behavior_flags = 0;			g_epi_hospital_ww[state->hospital][0]++;		}	}}

voidpm_connection_print(pm_state * state, pm_connection * c, tw_lp * lp){#if 0	if(!c)		return;	fprintf(F, "%4.4lf,%d,%ld,%2.4lf,%2.4lf,%ld/n",		tw_now(lp), c->id, lp->id, state->x_pos, state->y_pos, c->to);#endif}