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

voidphold_init(phold_state * s, tw_lp * lp){	int              i;	if( stagger )	  {	    for (i = 0; i < g_phold_start_events; i++)	      {		tw_event_send(			      tw_event_new(lp->gid,					   tw_rand_exponential(lp->rng, mean) + lookahead + (tw_stime)(lp->gid % (unsigned int)g_tw_ts_end),					   lp));	      }	  }	else	  {	    for (i = 0; i < g_phold_start_events; i++)	      {		tw_event_send(			      tw_event_new(lp->gid,					   tw_rand_exponential(lp->rng, mean) + lookahead,					   lp));	      }	  }}

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 Airport_EventHandler(Airport_State *SV, tw_bf *CV, Msg_Data *M, 		      tw_lp *lp) {  tw_stime ts;  tw_event *CurEvent;  Msg_Data *NewM;  *(int *)CV = (int)0;  switch(M->event_type) {      case ARRIVAL:     // Schedule a landing in the future    SV->InTheAir++;        if((CV->c1 = (SV->RunwayFree == 1))){      SV->RunwayFree = 0;         ts = tw_rand_exponential(lp->id, R);      CurEvent = tw_event_new(lp, ts, lp);      NewM = (Msg_Data *)tw_event_data(CurEvent);      NewM->event_type = LAND;      tw_event_send(CurEvent);    }    break;        case LAND:     SV->InTheAir--;    SV->OnTheGround++;    SV->NumLanded++;        ts = tw_rand_exponential(lp->id, G);    CurEvent = tw_event_new(lp, ts, lp);    NewM = (Msg_Data *)tw_event_data(CurEvent);    NewM->event_type = DEPARTURE;    tw_event_send(CurEvent);        if ((CV->c1 = (SV->InTheAir > 0))){      ts = tw_rand_exponential(lp->id, R);      CurEvent = tw_event_new(lp, ts, lp);      NewM = (Msg_Data *)tw_event_data(CurEvent);      NewM->event_type = LAND;      tw_event_send(CurEvent);    }    else      SV->RunwayFree = 1;    break;      case DEPARTURE:     SV->OnTheGround--;        ts = tw_rand_exponential(lp->id, A);    CurEvent = tw_event_new(tw_getlp((lp->id + 1) % 3) , ts, lp);    NewM = (Msg_Data *) tw_event_data(CurEvent);    NewM->event_type = ARRIVAL;    tw_event_send(CurEvent);    break;  }}

static void svr_init(    svr_state * ns,    tw_lp * lp){    tw_event *e;    svr_msg *m;    tw_stime kickoff_time;        memset(ns, 0, sizeof(*ns));    /* each server sends a dummy event to itself that will kick off the real     * simulation     */    //printf("/n Initializing servers %d ", (int)lp->gid);    /* skew each kickoff event slightly to help avoid event ties later on */    kickoff_time = g_tw_lookahead + tw_rand_unif(lp->rng);     e = codes_event_new(lp->gid, kickoff_time, lp);    m = tw_event_data(e);    m->svr_event_type = KICKOFF;    tw_event_send(e);    return;}

void event_handler(ping_pong_state * s, tw_bf * bf, ping_pong_message * in_m, tw_lp * lp){	tw_stime ts;	tw_event *e;	ping_pong_message *m;	int dest=0;	ts=1;//	if(g_tw_mynode == 0)	if(lp->gid == 0)	{		dest = 1;	}//	else if (g_tw_mynode == 1)	else if (lp->gid == 1)	{		dest = 0;	}	else	{		cout<<"invalid core number"<<endl;		assert(false);	}	e = tw_event_new(dest, ts, lp);	m = (ping_pong_message *)tw_event_data(e);	tw_event_send(e);}

/** * Initializer for OLSR */void olsr_init(node_state *s, tw_lp *lp){    hello *h;    tw_event *e;    olsr_msg_data *msg;    tw_stime ts;        //s->num_tuples = 0;    s->num_neigh  = 0;    s->num_two_hop = 0;    s->local_address = lp->gid;    s->lng = tw_rand_unif(lp->rng) * GRID_MAX;    s->lat = tw_rand_unif(lp->rng) * GRID_MAX;        ts = tw_rand_unif(lp->rng) * STAGGER_MAX;        e = tw_event_new(lp->gid, ts, lp);    msg = tw_event_data(e);    msg->type = HELLO_TX;    msg->originator = s->local_address;    msg->lng = s->lng;    msg->lat = s->lat;    h = &msg->mt.h;    h->num_neighbors = 0;    //h->neighbor_addrs[0] = s->local_address;    tw_event_send(e);}

/* * 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);}

static void s_event(s_state *ns, tw_bf *bf, s_msg *m, tw_lp *lp){    assert(m->h.magic == s_magic);    switch(m->h.event_type){        case S_KICKOFF: ;            msg_header h;            msg_set_header(s_magic, S_ALLOC_ACK, lp->gid, &h);            resource_lp_get(bsize, 0, lp, CODES_MCTX_DEFAULT, 0, &h, &ns->cb);            break;        case S_ALLOC_ACK:            if (m->c.ret == 0){                ns->mem += bsize;                m->mem_max_prev = ns->mem_max;                ns->mem_max = maxu64(ns->mem, ns->mem_max);                msg_header h;                msg_set_header(s_magic, S_ALLOC_ACK, lp->gid, &h);                resource_lp_get(bsize, 0, lp, CODES_MCTX_DEFAULT, 0, &h,                        &ns->cb);                break;            }            /* else fall into the free stmt */         case S_FREE:            resource_lp_free(bsize, lp, CODES_MCTX_DEFAULT);            ns->mem -= bsize;            if (ns->mem > 0){                tw_event *e =                     codes_event_new(lp->gid, codes_local_latency(lp), lp);                s_msg *m = tw_event_data(e);                msg_set_header(s_magic, S_FREE, lp->gid, &m->h);                tw_event_send(e);            }            break;    }}

/*Sends a 8-byte credit back to the torus node LP that sent the message */static void credit_send( nodes_state * s,                         tw_bf * bf,                         tw_lp * lp,                         nodes_message * msg){#if DEBUG    //printf("/n (%lf) sending credit tmp_dir %d tmp_dim %d %lf ", tw_now(lp), msg->source_direction, msg->source_dim, s->params->credit_delay );#endif    bf->c1 = 0;    tw_event * buf_e;    nodes_message *m;    tw_stime ts;    int src_dir = msg->source_direction;    int src_dim = msg->source_dim;    msg->saved_available_time = s->next_credit_available_time[(2 * src_dim) + src_dir][0];    s->next_credit_available_time[(2 * src_dim) + src_dir][0] = maxd(s->next_credit_available_time[(2 * src_dim) + src_dir][0], tw_now(lp));    ts =  s->params->credit_delay +          tw_rand_exponential(lp->rng, s->params->credit_delay/1000);    s->next_credit_available_time[(2 * src_dim) + src_dir][0] += ts;    //buf_e = tw_event_new( msg->sender_lp, s->next_credit_available_time[(2 * src_dim) + src_dir][0] - tw_now(lp), lp);    //m = tw_event_data(buf_e);    buf_e = model_net_method_event_new(msg->sender_node,                                       s->next_credit_available_time[(2*src_dim) + src_dir][0] - tw_now(lp),                                       lp, TORUS, (void**)&m, NULL);    m->source_direction = msg->source_direction;    m->source_dim = msg->source_dim;    m->type = CREDIT;    tw_event_send( buf_e );}

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;    }}

/* collective operation for the torus network */void torus_collective(char* category, int message_size, int remote_event_size, const void* remote_event, tw_lp* sender){    tw_event * e_new;    tw_stime xfer_to_nic_time;    nodes_message * msg;    tw_lpid local_nic_id;    char* tmp_ptr;    // TODO: be annotation-aware    codes_mapping_get_lp_info(sender->gid, grp_name, &mapping_grp_id, NULL,                              &mapping_type_id, NULL, &mapping_rep_id, &mapping_offset);    codes_mapping_get_lp_id(grp_name, LP_CONFIG_NM, NULL, 1,                            mapping_rep_id, mapping_offset, &local_nic_id);    xfer_to_nic_time = g_tw_lookahead + codes_local_latency(sender);    e_new = model_net_method_event_new(local_nic_id, xfer_to_nic_time,                                       sender, TORUS, (void**)&msg, (void**)&tmp_ptr);    msg->remote_event_size_bytes = message_size;    strcpy(msg->category, category);    msg->sender_svr=sender->gid;    msg->type = T_COLLECTIVE_INIT;    tmp_ptr = (char*)msg;    tmp_ptr += torus_get_msg_sz();    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;    }    tw_event_send(e_new);    return;}

void wifi_station_arrival(wifi_access_point_state * s, tw_bf * bf, wifi_message * m, tw_lp * lp){  unsigned int rng_calls=0;  tw_event *e=NULL;  wifi_message *m_new=NULL;  // packets coming from access point have much more power and so better snr  s->stations[m->station].total_packets++;  s->stations[m->station].station_snr = tw_rand_normal_sd(lp->rng,4.0,8.0, &rng_calls);  // New Function - to add prop loss, but not ready yet.  //s->stations[m->station].station_snr = calcRxPower (txPowerDbm, distance, minDistance, lambda, systemLoss);  s->stations[m->station].station_success_rate =    WiFi_80211b_DsssDqpskCck11_SuccessRate(s->stations[m->station].station_snr, num_of_bits);  if( tw_rand_normal_sd(lp->rng,0.5,0.1, &rng_calls) <      s->stations[m->station].station_success_rate)    {      bf->c1 = 1;      s->stations[m->station].failed_packets++;    }  // schedule event back to AP w/ exponential service time  e = tw_event_new(lp->gid, tw_rand_exponential(lp->rng, 10.0), lp);  m_new = (wifi_message *) tw_event_data(e);  m_new->type = WIFI_PACKET_ARRIVAL_AT_ACCESS_POINT;  m_new->station = m->station;  tw_event_send(e);}

void testsvr_lp_init(        testsvr_state * ns,        tw_lp * lp){    /* for test, just use dummy way (assume 1 svr / 1 modelnet) */    ns->idx = lp->gid / 2;        /* expect exactly three servers */    assert(ns->idx <= 2);    memset(ns->req_stat, 0x0, NUM_REQS*sizeof(int));    /* create kickoff event only if we're a request server */    if (ns->idx == 0 || ns->idx == 2){        tw_event *e = codes_event_new(lp->gid, codes_local_latency(lp), lp);        testsvr_msg *m_local = tw_event_data(e);        m_local->magic = testsvr_magic;        m_local->event_type = KICKOFF;        /* dummy values for kickoff */        m_local->idx_src = INT_MAX;        m_local->lp_src  = INT_MAX;        m_local->req_num = INT_MAX;        tw_event_send(e);    }#if TEST_DEBUG    char name[32];    sprintf(name, "testsvr.%d.%lu", ns->idx, lp->gid);    ns->fdebug = fopen(name, "w");    setvbuf(ns->fdebug, NULL, _IONBF, 0);    assert(ns->fdebug != NULL);    ns->event_ctr = 0;#endif}

voidtorus_init(nodes_state * s, tw_lp * lp){  tw_event *e;  tw_stime ts;  nodes_message *m;  /*   * Set up the initial state for each LP (node)   * according to the number of dimensions   */  torus_setup(s, lp);  /*   * Start a GENERATE event on each LP   */  //ts = tw_rand_exponential(lp->rng, MEAN_INTERVAL);  ts = tw_rand_exponential(lp->rng, 10);  int i;  for(i=0;i<N_nodes;i++)    {      e = tw_event_new(lp->gid, ts, lp);      m = tw_event_data(e);      m->type = GENERATE;            m->next_stop = i;            m->dest_lp = lp->gid;      tw_event_send(e);    }}

static void svr_init(    svr_state * ns,    tw_lp * lp){    ns->server_idx = lp->gid / 2;    if (ns->server_idx < NUM_SERVERS-1){        for (int i = 0; i < NUM_PRIOS; i++){            ns->random_order[i] = -1;        }        for (int i = 0; i < NUM_PRIOS; i++){            for (;;){                int idx = tw_rand_integer(lp->rng, 0, NUM_PRIOS-1);                // not sure whether rand_integer is inclusive or not...                assert(idx < NUM_PRIOS);                if (ns->random_order[idx] == -1){                    ns->random_order[idx] = i;                    break;                }            }        }        tw_event *e = codes_event_new(lp->gid, codes_local_latency(lp), lp);        svr_msg * m = tw_event_data(e);        msg_set_header(666, KICKOFF, lp->gid, &m->h);        tw_event_send(e);    }    else {        memset(ns->num_recv, 0, NUM_SERVERS*sizeof(*ns->num_recv));    }}

/* handle initial event (initialize job submission) */void handle_kick_off_event(    awe_server_state * ns,    tw_bf * b,    awe_msg * m,    tw_lp * lp){    printf("%lf;awe_server;%lu]Start serving/n", now_sec(lp), lp->gid);    GHashTableIter iter;    gpointer key, value;    g_hash_table_iter_init(&iter, job_map);        while (g_hash_table_iter_next(&iter, &key, &value)) {        Job* job = (Job*)value;        tw_event *e;        awe_msg *msg;        tw_stime submit_time;        submit_time =  s_to_ns(etime_to_stime(job->stats.created)) + ns_tw_lookahead;        if (fraction < 1.0) {        	submit_time = submit_time * fraction;        }        e = codes_event_new(lp->gid, submit_time, lp);        msg = tw_event_data(e);        msg->event_type = JOB_SUBMIT;        strcpy(msg->object_id, job->id);        tw_event_send(e);    }    return;}

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 lpf_awe_server_init(    awe_server_state * ns,    tw_lp * lp){    tw_event *e;    awe_msg *m;    tw_stime kickoff_time;    memset(ns, 0, sizeof(*ns));    work_queue = g_queue_new();    client_req_queue = g_queue_new();        /* skew each kickoff event slightly to help avoid event ties later on */    kickoff_time = 0;    /* first create the event (time arg is an offset, not absolute time) */    e = codes_event_new(lp->gid, kickoff_time, lp);    /* after event is created, grab the allocated message and set msg-specific     * data */     m = tw_event_data(e);    m->event_type = KICK_OFF;    m->src = lp->gid;    /* event is ready to be processed, send it off */    tw_event_send(e);    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;}

void packet_buffer_process( nodes_state * s, tw_bf * bf, nodes_message * msg, tw_lp * lp ){   msg->wait_loc = -1;   s->buffer[ msg->source_direction + ( msg->source_dim * 2 ) ][  0 ]-=1;    tw_event * e_h;  nodes_message * m;  tw_stime ts;  bf->c3 = 0;  int loc=s->wait_count, j=0;  for(j = 0; j < loc; j++)   {    if( s->waiting_list[j].dim == msg->source_dim && s->waiting_list[j].dir == msg->source_direction)     {        bf->c3=1;	ts = tw_rand_exponential(lp->rng, MEAN_INTERVAL/100);        e_h = tw_event_new( lp->gid, ts, lp );        m = tw_event_data( e_h );     	memcpy(m, s->waiting_list[j].packet, sizeof(nodes_message));        //        For reverse computation, also copy data to the msg    	memcpy(msg, s->waiting_list[j].packet, sizeof(nodes_message));	msg->wait_loc = j;	msg->type = CREDIT;	m->type = SEND;        tw_event_send(e_h);        waiting_packet_free(s, j);        break;    }  }}

	/*	 * Create and send new boundary particle	 */tw_event	*rm_particle_send(tw_lpid gid, tw_stime offset, tw_lp * src_lp){	tw_event	*e;	//tw_memory	*b;	rm_message	*m;	//rm_particle	*p;	//printf("%ld: sending particle to cell %ld /n", src_lp->id, gid);	if(gid == 0)		tw_error(TW_LOC, "here");	e = tw_event_new(gid, offset, src_lp);	m = tw_event_data(e);	m->type = RM_PARTICLE;#if DWB	b = tw_memory_alloc(src_lp, g_rm_fd);	tw_event_memory_set(e, b, g_rm_fd);	p = tw_memory_data(b);	p->user_lp = src_lp;#endif	tw_event_send(e);	return e;}

/*Sends a 8-byte credit back to the torus node LP that sent the message */void credit_send( nodes_state * s, 	    tw_bf * bf, 	    tw_lp * lp, 	    nodes_message * msg){    //if(lp->gid == TRACK_LP)    //	printf("/n (%lf) sending credit tmp_dir %d tmp_dim %d %lf ", tw_now(lp), msg->source_direction, msg->source_dim, credit_delay );        tw_event * buf_e;    nodes_message *m;    tw_stime ts;        int src_dir = msg->source_direction;    int src_dim = msg->source_dim;    msg->saved_available_time = s->next_credit_available_time[(2 * src_dim) + src_dir][0];    s->next_credit_available_time[(2 * src_dim) + src_dir][0] = max(s->next_credit_available_time[(2 * src_dim) + src_dir][0], tw_now(lp));    ts =  credit_delay + tw_rand_exponential(lp->rng, credit_delay/1000);    s->next_credit_available_time[(2 * src_dim) + src_dir][0] += ts;    buf_e = tw_event_new( msg->sender_lp, s->next_credit_available_time[(2 * src_dim) + src_dir][0] - tw_now(lp), lp);    m = tw_event_data(buf_e);    m->source_direction = msg->source_direction;    m->source_dim = msg->source_dim;    m->type = CREDIT;    tw_event_send( buf_e );}

voidmem_init(mem_state * s, tw_lp * lp){    tw_event	*e;    tw_memory	*b;    int              i;    for (i = 0; i < g_mem_start_events; i++)    {        e = tw_event_new(lp->gid, tw_rand_exponential(lp->rng, mean), lp);        for(i = 0; i < nbufs; i++)        {            b = mem_alloc(lp);            mem_fill(b);            tw_event_memory_set(e, b, my_fd);            s->stats.s_mem_alloc++;        }        tw_event_send(e);        s->stats.s_sent++;    }}

/** * Initialize the LPs with BB capacity and cluster flag then kickoff * @Params ns node state * 		   m message * 		   lp LP */void node_lp_init(node_state * ns, tw_lp * lp) {	burst_buffer_capacity = ((long) (burst_buffer_max_capacity)) * 1000000000;	//burst_buffer_capacity = ((long) (burst_buffer_max_capacity))*10;	//printf("Burst Buffer Capacity:%li/n",burst_buffer_capacity);	printf("In node_lp_init/n");	ns->num_processed = 0;	// nodes are addressed in their logical id space (0...num_client_nodes-1 and	// 0...num_svr_nodes-1, respectively). LPs are computed upon use with	// model-net, other events	ns->id_clust = codes_mapping_get_lp_relative_id(lp->gid, 1, 0);	int id_all = codes_mapping_get_lp_relative_id(lp->gid, 0, 0);	// track which cluster we're in	ns->is_in_client = (id_all < num_client_nodes);	ns->is_in_server = (id_all < (num_svr_nodes + num_client_nodes)			&& (id_all >= num_client_nodes));	ns->is_in_bb = (id_all			< (num_svr_nodes + num_client_nodes + num_burst_buffer_nodes)			&& (id_all >= num_svr_nodes + num_client_nodes));	printf("is_in_client=%d/nis_in_svr=%d/nis_in_bb=%d/n", ns->is_in_client,			ns->is_in_server, ns->is_in_bb);	printf("id_all= %d/nnum_client_nodes= %d/n", id_all, num_client_nodes);	// send a self kickoff event	tw_event *e = codes_event_new(lp->gid, codes_local_latency(lp), lp);	node_msg *m = tw_event_data(e);	msg_set_header(node_magic, NODE_KICKOFF, lp->gid, &m->h);	tw_event_send(e);}

void handle_work_checkout_event(    awe_server_state * ns,    tw_bf * b,    awe_msg * m,    tw_lp * lp){    tw_event *e;    awe_msg *msg;    e = codes_event_new(m->src, ns_tw_lookahead, lp);    msg = tw_event_data(e);    msg->event_type = WORK_CHECKOUT;    memset(msg->object_id, 0, sizeof(msg->object_id));        tw_lpid client_id = m->src;//    char group_name[MAX_LENGTH_GROUP];    //char lp_type_name[MAX_LENGTH_GROUP];    //int  lp_type_id, grp_id, grp_rep_id, offset; //   codes_mapping_get_lp_info(client_id, group_name, &grp_id, &lp_type_id,   //         lp_type_name, &grp_rep_id, &offset);    int group_id = 0;    group_id = get_group_id(client_id);    /*if queue is empty, msg->object_id is "", otherwise msg->object-id is the dequeued workid*/    int got_work = 0;    char workid[MAX_LENGTH_ID];    if (!g_queue_is_empty(work_queue)) {        if (group_id == 1 && sched_policy>0) {  //client from remote site            char* work = NULL;            if (sched_policy==1) {                work = get_first_work_by_stage(5); //checkout task 5 (blat) only for remote site            } else if (sched_policy==2) {            	work = get_first_work_by_greedy(WorkOrder);            }            if (work) {            	strcpy(workid, work);            	got_work = 1;            }        } else {        	strcpy(workid, g_queue_pop_head(work_queue));        	got_work = 1;        }    }    if (got_work) { //eligible work found, send back to the requesting client        fprintf(event_log, "%lf;awe_server;%lu;WC;work=%s client=%lu/n", now_sec(lp), lp->gid, workid, m->src);        assert (strlen(workid) > 10);        strcpy(msg->object_id, workid);        tw_event_send(e);    } else {  //no eligible work found, put client request to the waiting queue        tw_lpid *clientid = NULL;        clientid = malloc(sizeof(tw_lpid));        *clientid = m->src;        g_queue_push_tail(client_req_queue, clientid);    }    return;}

voidmem_event_handler(mem_state * s, tw_bf * bf, mem_message * m, tw_lp * lp){    tw_lpid		 dest;    tw_event	*e;    tw_memory	*b;    int		 i;    s->stats.s_recv++;    // read membufs off inbound event, check it and free it    for(i = 0; i < nbufs; i++)    {        b = tw_event_memory_get(lp);        if(!b)            tw_error(TW_LOC, "Missing memory buffers: %d of %d",                     i+1, nbufs);        mem_verify(b);        tw_memory_free(lp, b, my_fd);        s->stats.s_mem_free++;        s->stats.s_mem_get++;    }    if(tw_rand_unif(lp->rng) <= percent_remote)    {        bf->c1 = 1;        dest = tw_rand_integer(lp->rng, 0, ttl_lps - 1);        dest += offset_lpid;        if(dest >= ttl_lps)            dest -= ttl_lps;    } else    {        bf->c1 = 0;        dest = lp->gid;    }    e = tw_event_new(dest, tw_rand_exponential(lp->rng, mean), lp);    // allocate membufs and attach them to the event    for(i = 0; i < nbufs; i++)    {        b = mem_alloc(lp);        if(!b)            tw_error(TW_LOC, "no membuf allocated!");        mem_fill(b);        tw_event_memory_set(e, b, my_fd);        s->stats.s_mem_alloc++;    }    tw_event_send(e);}

// Helpersvoid raid_controller_gen_send( tw_lpid dest, Event event_type, tw_stime event_time, tw_lp* lp ){    // Generate and send message to dest    tw_event* event = tw_event_new( dest, event_time, lp );    MsgData* message = (MsgData*)tw_event_data( event );    message->event_type = event_type;    tw_event_send( event );}

void lsm_io_event(        const char * lp_io_category,        uint64_t io_object,        int64_t  io_offset,        uint64_t io_size_bytes,        int      io_type,        tw_stime delay,        tw_lp *sender,        struct codes_mctx const * map_ctx,        int return_tag,        msg_header const * return_header,        struct codes_cb_info const * cb){    assert(strlen(lp_io_category) < CATEGORY_NAME_MAX-1);    assert(strlen(lp_io_category) > 0);    SANITY_CHECK_CB(cb, lsm_return_t);    tw_lpid lsm_id = codes_mctx_to_lpid(map_ctx, LSM_NAME, sender->gid);    tw_stime delta = delay + codes_local_latency(sender);    if (lsm_in_sequence) {        tw_stime tmp = lsm_msg_offset;        lsm_msg_offset += delta;        delta += tmp;    }    tw_event *e = tw_event_new(lsm_id, delta, sender);    lsm_message_t *m = tw_event_data(e);    m->magic = lsm_magic;    m->event = (lsm_event_t) io_type;    m->data.object = io_object;    m->data.offset = io_offset;    m->data.size   = io_size_bytes;    strcpy(m->data.category, lp_io_category);    // get the priority count for checking    int num_prios = lsm_get_num_priorities(map_ctx, sender->gid);    // prio checks and sets    if (num_prios <= 0) // disabled scheduler - ignore        m->data.prio = 0;    else if (temp_prio < 0) // unprovided priority - defer to max possible        m->data.prio = num_prios-1;    else if (temp_prio < num_prios) // valid priority        m->data.prio = temp_prio;    else        tw_error(TW_LOC,                "LP %lu, LSM LP %lu: Bad priority (%d supplied, %d lanes)/n",                sender->gid, lsm_id, temp_prio, num_prios);    // reset temp_prio    temp_prio = -1;    m->cb.info = *cb;    m->cb.h = *return_header;    m->cb.tag = return_tag;    tw_event_send(e);}

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));}

void plan_work_enqueue_event(char* work_id, tw_lp *lp) {    tw_event *e;    awe_msg *msg;    e = codes_event_new(lp->gid, ns_tw_lookahead, lp);    msg = tw_event_data(e);    msg->event_type = WORK_ENQUEUE;    strcpy(msg->object_id, work_id);    tw_event_send(e);}