model-net 'pull' (e.g., RDMA) support

- pull itself not yet tested

codes/dragonfly.h

@@ -65,7 +65,7 @@ static tw_lpid dragonfly_find_local_device(tw_lp *sender);
 /* dragonfly packet event method called by modelnet, this method triggers the packet
  * generate event of dragonfly and attached remote and local events to the last packet
  * of the message */
-static tw_stime dragonfly_packet_event(char* category, tw_lpid final_dest_lp, uint64_t packet_size, tw_stime offset, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lp *sender, int is_last_pckt);
+static tw_stime dragonfly_packet_event(char* category, tw_lpid final_dest_lp, uint64_t packet_size, int is_pull, uint64_t pull_size, tw_stime offset, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lp *sender, int is_last_pckt);
 
 /* returns dragonfly message size */
 static int dragonfly_get_msg_sz(void);
@@ -187,6 +187,8 @@ struct terminal_message
    short vc_index;
    int input_chan;
    int output_chan;
+    int is_pull;
+    uint64_t pull_size;
    
    tw_stime saved_available_time;
    tw_stime saved_credit_time;

codes/model-net-method.h

@@ -18,6 +18,8 @@ struct model_net_method
         char* category, 
         tw_lpid final_dest_lp, 
         uint64_t packet_size, 
+        int is_pull,
+        uint64_t pull_size, /* only used when is_pull==1 */
         tw_stime offset,
         int remote_event_size,  /* 0 means don't deliver remote event */
         const void* remote_event,

codes/model-net.h

@@ -121,6 +121,9 @@ int model_net_set_params();
 
 // setup the modelnet parameters
 int model_net_setup(char* net_name, uint64_t packet_size, const void* net_params);
+
+/* utility function to get the modelnet ID post-setup */
+int model_net_get_id(char *net_name);
 /* allocate and transmit a new event that will pass through model_net to 
  * arrive at its destination:
  *
@@ -154,6 +157,7 @@ void model_net_event(
     char* category, 
     tw_lpid final_dest_lp, 
     uint64_t message_size, 
+    tw_stime offset,
     int remote_event_size,
     const void* remote_event,
     int self_event_size,
@@ -181,6 +185,32 @@ void model_net_event_rc(
     tw_lp *sender,
     uint64_t message_size);
 
+
+/* Issue a 'pull' from the memory of the destination LP, without
+ * requiring the destination LP to do event processing. This is meant as a
+ * simulation-based abstraction of RDMA. A control packet will be sent to the
+ * destination LP, the payload will be sent back to the requesting LP, and the
+ * requesting LP will be issued it's given completion event.
+ *
+ * Parameters are largely the same as model_net_event, with the following
+ * exceptions:
+ * - final_dest_lp is the lp to pull data from
+ * - self_event_size, self_event are applied at the requester upon receipt of 
+ *   the payload from the dest
+ */
+void model_net_pull_event(
+        int net_id,
+        char *category,
+        tw_lpid final_dest_lp,
+        uint64_t message_size,
+        tw_stime offset,
+        int self_event_size,
+        const void *self_event,
+        tw_lp *sender);
+void model_net_pull_event_rc(
+        int net_id,
+        tw_lp *sender);
+
 /* returns pointer to LP information for simplenet module */
 const tw_lptype* model_net_get_lp_type(int net_id);
 

codes/torus.h

@@ -63,6 +63,8 @@ static tw_stime torus_packet_event(
 		       char* category,
 		       tw_lpid final_dest_lp,
 		       uint64_t packet_size,
+                       int is_pull,
+                       uint64_t pull_size,
                        tw_stime offset,
 		       int remote_event_size,
 		       const void* remote_event,
@@ -170,6 +172,9 @@ struct nodes_message
   /* chunk id of the flit (distinguishes flits) */
   short chunk_id;
 
+  int is_pull;
+  uint64_t pull_size;
+
   /* for codes local and remote events, only carried by the last packet of the message */
   int local_event_size_bytes;
   int remote_event_size_bytes;

src/models/networks/model-net/dragonfly.c

@@ -56,7 +56,7 @@ static void dragonfly_report_stats()
    return;
 }
 /* dragonfly packet event , generates a dragonfly packet on the compute node */
-static tw_stime dragonfly_packet_event(char* category, tw_lpid final_dest_lp, uint64_t packet_size, tw_stime offset, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lp *sender, int is_last_pckt)
+static tw_stime dragonfly_packet_event(char* category, tw_lpid final_dest_lp, uint64_t packet_size, int is_pull, uint64_t pull_size, tw_stime offset, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lp *sender, int is_last_pckt)
 {
     tw_event * e_new;
     tw_stime xfer_to_nic_time;
@@ -83,6 +83,8 @@ static tw_stime dragonfly_packet_event(char* category, tw_lpid final_dest_lp, ui
     msg->remote_event_size_bytes = 0;
     msg->local_event_size_bytes = 0;
     msg->type = T_GENERATE;
+    msg->is_pull = is_pull;
+    msg->pull_size = pull_size;
 
     if(is_last_pckt) /* Its the last packet so pass in remote and local event information*/
       {
@@ -363,13 +365,21 @@ if( msg->packet_ID == TRACK && msg->chunk_id == num_chunks-1)
 	// Trigger an event on receiving server
 	if(msg->remote_event_size_bytes)
 	{
-		ts = (1/cn_bandwidth) * msg->remote_event_size_bytes;
-		e = tw_event_new(msg->final_dest_gid, ts, lp);
-		m = tw_event_data(e);
-		char* tmp_ptr = (char*)msg;
-		tmp_ptr += dragonfly_get_msg_sz();                                                                                                            
-		memcpy(m, tmp_ptr, msg->remote_event_size_bytes);
-		tw_event_send(e); 
+            char* tmp_ptr = (char*)msg;
+            tmp_ptr += dragonfly_get_msg_sz();
+            ts = (1/cn_bandwidth) * msg->remote_event_size_bytes;
+            if (msg->is_pull){
+                int net_id = model_net_get_id("dragonfly");
+                model_net_event(net_id, msg->category, msg->sender_lp,
+                        msg->pull_size, ts, msg->remote_event_size_bytes,
+                        tmp_ptr, 0, NULL, lp);
+            }
+            else{
+                e = tw_event_new(msg->final_dest_gid, ts, lp);
+                m = tw_event_data(e);
+                memcpy(m, tmp_ptr, msg->remote_event_size_bytes);
+                tw_event_send(e); 
+            }
 	}
   }
 
@@ -896,6 +906,12 @@ void router_rc_event_handler(router_state * s, tw_bf * bf, terminal_message * ms
 			tw_rand_reverse_unif(lp->rng);
 			int output_port = msg->saved_vc/num_vcs;
 			s->next_credit_available_time[output_port] = 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("dragonfly");
+                            model_net_event_rc(net_id, lp, msg->pull_size);
+                        }
 		    }
 	    break;
 

src/models/networks/model-net/loggp.c

@@ -48,6 +48,8 @@ struct loggp_message
     int event_size_bytes;     /* size of simulator event message that will be tunnelled to destination */
     int local_event_size_bytes;     /* size of simulator event message that delivered locally upon local completion */
     char category[CATEGORY_NAME_MAX]; /* category for communication */
+    int is_pull;
+    uint64_t pull_size;
 
     /* for reverse computation */
     tw_stime net_send_next_idle_saved;
@@ -109,6 +111,8 @@ static tw_stime loggp_packet_event(
      char* category, 
      tw_lpid final_dest_lp, 
      uint64_t packet_size, 
+     int is_pull,
+     uint64_t pull_size, /* only used when is_pull==1 */
      tw_stime offset,
      int remote_event_size, 
      const void* remote_event, 
@@ -299,6 +303,11 @@ static void handle_msg_ready_rev_event(
     stat->recv_bytes -= m->net_msg_size_bytes;
     stat->recv_time -= m->recv_time_saved;
 
+    if (m->event_size_bytes && m->is_pull){
+        int net_id = model_net_get_id("loggp");
+        model_net_event_rc(net_id, lp, m->pull_size);
+    }
+
     return;
 }
 
@@ -350,12 +359,21 @@ static void handle_msg_ready_event(
       /* schedule event to final destination for when the recv is complete */
 //      printf("\n Remote message to LP %d ", m->final_dest_gid); 
 
-      e_new = tw_event_new(m->final_dest_gid, recv_queue_time, lp);
-      m_new = tw_event_data(e_new);
-      char* tmp_ptr = (char*)m;
-      tmp_ptr += loggp_get_msg_sz();
-      memcpy(m_new, tmp_ptr, m->event_size_bytes);
-      tw_event_send(e_new);
+        char* tmp_ptr = (char*)m;
+        tmp_ptr += loggp_get_msg_sz();
+        if (m->is_pull){
+            /* call the model-net event */
+            int net_id = model_net_get_id("loggp");
+            model_net_event(net_id, m->category, m->src_gid, m->pull_size,
+                    recv_queue_time, m->event_size_bytes, tmp_ptr, 0, NULL,
+                    lp);
+        }
+        else{
+            e_new = tw_event_new(m->final_dest_gid, recv_queue_time, lp);
+            m_new = tw_event_data(e_new);
+            memcpy(m_new, tmp_ptr, m->event_size_bytes);
+            tw_event_send(e_new);
+        }
     }
 
     return;
@@ -485,6 +503,8 @@ static tw_stime loggp_packet_event(
 		char* category,
 		tw_lpid final_dest_lp,
 		uint64_t packet_size,
+                int is_pull,
+                uint64_t pull_size, /* only used when is_pull==1 */
                 tw_stime offset,
 		int remote_event_size,
 		const void* remote_event,
@@ -515,6 +535,8 @@ static tw_stime loggp_packet_event(
      msg->event_size_bytes = 0;
      msg->local_event_size_bytes = 0;
      msg->event_type = MSG_START;
+     msg->is_pull = is_pull;
+     msg->pull_size = pull_size;
 
      tmp_ptr = (char*)msg;
      tmp_ptr += loggp_get_msg_sz();

src/models/networks/model-net/model-net.c

@@ -10,6 +10,8 @@
 #include "codes/model-net.h"
 #include "codes/model-net-method.h"
 
+#define PULL_MSG_SIZE 128
+
 #define STR_SIZE 16
 #define PROC_TIME 10.0
 
@@ -48,6 +50,16 @@ int model_net_setup(char* name,
      return -1; // indicating error
 }
 
+int model_net_get_id(char *name){
+    int i;
+    for(i=0; method_array[i] != NULL; i++) {
+        if(strcmp(method_array[i]->method_name, name) == 0) {
+            return i;
+        }
+    }
+    return -1;
+}
+
 void model_net_write_stats(tw_lpid lpid, struct mn_stats* stat)
 {
     int ret;
@@ -130,17 +142,19 @@ struct mn_stats* model_net_find_stats(const char* category, mn_stats mn_stats_ar
     return(&mn_stats_array[i]);
 }
 
-void model_net_event(
-    int net_id,
-    char* category, 
-    tw_lpid final_dest_lp, 
-    uint64_t message_size, 
-    int remote_event_size,
-    const void* remote_event,
-    int self_event_size,
-    const void* self_event,
-    tw_lp *sender)
-{
+static void model_net_event_impl(
+        int net_id,
+        char* category, 
+        tw_lpid final_dest_lp, 
+        uint64_t message_size, 
+        int is_pull,
+        uint64_t pull_msg_size,
+        tw_stime offset,
+        int remote_event_size,
+        const void* remote_event,
+        int self_event_size,
+        const void* self_event,
+        tw_lp *sender) {
     /* determine packet size for underlying method */
      uint64_t packet_size = model_net_get_packet_size(net_id);
      uint64_t num_packets = message_size/packet_size; /* Number of packets to be issued by the API */
@@ -178,7 +192,7 @@ void model_net_event(
      * passed along through network hops and delivered to final_dest_lp
      */
 
-     tw_stime offset = (in_sequence) ? mn_msg_offset : 0.0;
+     tw_stime poffset = (in_sequence) ? mn_msg_offset : 0.0;
      for( i = 0; i < num_packets; i++ )
        {
 	  /*Mark the last packet to the net method API*/
@@ -189,14 +203,49 @@ void model_net_event(
               packet_size = message_size - ((num_packets-1)*packet_size);
             }
 	  /* Number of packets and packet ID is passed to the underlying network to mark the final packet for local event completion*/
-	  offset += method_array[net_id]->model_net_method_packet_event(category,
-                  final_dest_lp, packet_size, offset, remote_event_size, remote_event,
-                  self_event_size, self_event, sender, last);
+	  poffset += method_array[net_id]->model_net_method_packet_event(category,
+                  final_dest_lp, packet_size, is_pull, pull_msg_size, poffset+offset,
+                  remote_event_size, remote_event, self_event_size, self_event, sender, last);
        }
-    if (in_sequence) mn_msg_offset = offset;
+    if (in_sequence) mn_msg_offset = poffset;
     return;
 }
 
+
+void model_net_event(
+    int net_id,
+    char* category, 
+    tw_lpid final_dest_lp, 
+    uint64_t message_size, 
+    tw_stime offset,
+    int remote_event_size,
+    const void* remote_event,
+    int self_event_size,
+    const void* self_event,
+    tw_lp *sender)
+{
+    model_net_event_impl(net_id, category, final_dest_lp, message_size, 0, 0,
+            offset, remote_event_size, remote_event, self_event_size,
+            self_event, sender); 
+}
+
+void model_net_pull_event(
+        int net_id,
+        char *category,
+        tw_lpid final_dest_lp,
+        uint64_t message_size,
+        tw_stime offset,
+        int self_event_size,
+        const void *self_event,
+        tw_lp *sender){
+    /* NOTE: for a pull, we are filling the *remote* event - it will be remote
+     * from the destination's POV */
+    model_net_event_impl(net_id, category, final_dest_lp, PULL_MSG_SIZE, 1,
+            message_size, offset, self_event_size, self_event, 0, NULL,
+            sender); 
+}
+
+
 int model_net_set_params()
 {
   char mn_name[MAX_NAME_LENGTH];
@@ -393,7 +442,7 @@ int model_net_set_params()
        printf("\n Invalid network argument %s ", mn_name);
   return net_id;
 }
-void model_net_event_rc(
+static void model_net_event_impl_rc(
     int net_id,
     tw_lp *sender,
     uint64_t message_size)
@@ -418,6 +467,19 @@ void model_net_event_rc(
     return;
 } 
 
+void model_net_event_rc(
+        int net_id,
+        tw_lp *sender,
+        uint64_t message_size){
+    model_net_event_impl_rc(net_id,sender,message_size);
+}
+
+void model_net_pull_event_rc(
+        int net_id,
+        tw_lp *sender) {
+    model_net_event_impl_rc(net_id, sender, PULL_MSG_SIZE);
+}
+
 /* returns the message size, can be either simplenet, dragonfly or torus message size*/
 static int model_net_get_msg_sz(int net_id)
 {

src/models/networks/model-net/simplenet-upd.c

@@ -48,6 +48,8 @@ struct sn_message
     int event_size_bytes;     /* size of simulator event message that will be tunnelled to destination */
     int local_event_size_bytes;     /* size of simulator event message that delivered locally upon local completion */
     char category[CATEGORY_NAME_MAX]; /* category for communication */
+    int is_pull; /* this message represents a pull request from the destination LP to the source */
+    uint64_t pull_size; /* data size to pull from dest LP */
 
     /* for reverse computation */
     tw_stime net_send_next_idle_saved;
@@ -102,6 +104,8 @@ static tw_stime simplenet_packet_event(
      char* category, 
      tw_lpid final_dest_lp, 
      uint64_t packet_size, 
+     int is_pull,
+     uint64_t pull_size, /* only used when is_pull==1 */
      tw_stime offset,
      int remote_event_size, 
      const void* remote_event, 
@@ -317,6 +321,11 @@ static void handle_msg_ready_rev_event(
     stat->recv_bytes -= m->net_msg_size_bytes;
     stat->recv_time = m->recv_time_saved;
 
+    if (m->event_size_bytes && m->is_pull){
+        int net_id = model_net_get_id("simplenet");
+        model_net_event_rc(net_id, lp, m->pull_size);
+    }
+
     return;
 }
 
@@ -357,15 +366,23 @@ static void handle_msg_ready_event(
     /* copy only the part of the message used by higher level */
     if(m->event_size_bytes)
     {
+        char *tmp_ptr = (char*)m;
+        tmp_ptr += sn_get_msg_sz();
       /* schedule event to final destination for when the recv is complete */
 //      printf("\n Remote message to LP %d ", m->final_dest_gid); 
-
-      e_new = tw_event_new(m->final_dest_gid, recv_queue_time, lp);
-      m_new = tw_event_data(e_new);
-      char* tmp_ptr = (char*)m;
-      tmp_ptr += sn_get_msg_sz();
-      memcpy(m_new, tmp_ptr, m->event_size_bytes);
-      tw_event_send(e_new);
+        if (m->is_pull){
+            /* call the model-net event */
+            int net_id = model_net_get_id("simplenet");
+            model_net_event(net_id, m->category, m->src_gid, m->pull_size,
+                    recv_queue_time, m->event_size_bytes, tmp_ptr, 0, NULL,
+                    lp);
+        }
+        else{
+            e_new = tw_event_new(m->final_dest_gid, recv_queue_time, lp);
+            m_new = tw_event_data(e_new);
+            memcpy(m_new, tmp_ptr, m->event_size_bytes);
+            tw_event_send(e_new);
+        }
     }
 
     return;
@@ -481,6 +498,8 @@ static tw_stime simplenet_packet_event(
 		char* category,
 		tw_lpid final_dest_lp,
 		uint64_t packet_size,
+                int is_pull,
+                uint64_t pull_size, /* only used when is_pull == 1 */
                 tw_stime offset,
 		int remote_event_size,
 		const void* remote_event,
@@ -511,6 +530,8 @@ static tw_stime simplenet_packet_event(
      msg->event_size_bytes = 0;
      msg->local_event_size_bytes = 0;
      msg->event_type = MSG_START;
+     msg->is_pull = is_pull;
+     msg->pull_size = pull_size;
 
      tmp_ptr = (char*)msg;
      tmp_ptr += sn_get_msg_sz();

src/models/networks/model-net/simplewan.c

@@ -74,6 +74,9 @@ struct sw_message
     int local_event_size_bytes;     /* size of simulator event message that delivered locally upon local completion */
     char category[CATEGORY_NAME_MAX]; /* category for communication */
     
+    int is_pull;
+    uint64_t pull_size; 
+
     /* for reverse computation */
     tw_stime send_next_idle_saved;
     tw_stime recv_next_idle_saved;
@@ -141,16 +144,18 @@ static void sw_setup(const void* net_params);
 
 /* Issues a simplewan packet event call */
 static tw_stime simplewan_packet_event(
-     char* category, 
-     tw_lpid final_dest_lp, 
-     uint64_t packet_size, 
-     tw_stime offset,
-     int remote_event_size, 
-     const void* remote_event, 
-     int self_event_size,
-     const void* self_event,
-     tw_lp *sender,
-     int is_last_pckt);
+        char* category,
+        tw_lpid final_dest_lp,
+        uint64_t packet_size,
+        int is_pull,
+        uint64_t pull_size, /* only used when is_pull == 1 */
+        tw_stime offset,
+        int remote_event_size,
+        const void* remote_event,
+        int self_event_size,
+        const void* self_event,
+        tw_lp *sender,
+        int is_last_pckt);
 static void simplewan_packet_event_rc(tw_lp *sender);
 
 static void simplewan_packet_event_rc(tw_lp *sender);
@@ -505,6 +510,11 @@ static void handle_msg_ready_rev_event(
     idles->recv_next_idle_all = m->recv_next_idle_all_saved;
     idles->recv_prev_idle_all = m->recv_prev_idle_all_saved;
 
+    if (m->event_size_bytes && m->is_pull){
+        int net_id = model_net_get_id("simplewan");
+        model_net_event_rc(net_id, lp, m->pull_size);
+    }
+
     return;
 }
 
@@ -591,13 +601,20 @@ static void handle_msg_ready_event(
     /* copy only the part of the message used by higher level */
     if(m->event_size_bytes)
     {
-      /* schedule event to final destination for when the recv is complete */
-      e_new = tw_event_new(m->final_dest_gid, recv_queue_time, lp);
-      m_new = tw_event_data(e_new);
-      char* tmp_ptr = (char*)m;
-      tmp_ptr += sw_get_msg_sz();
-      memcpy(m_new, tmp_ptr, m->event_size_bytes);
-      tw_event_send(e_new);
+        char* tmp_ptr = (char*)m;
+        tmp_ptr += sw_get_msg_sz();
+        if (m->is_pull){
+            int net_id = model_net_get_id("simplewan");
+            model_net_event(net_id, m->category, m->src_gid, m->pull_size,
+                    recv_queue_time, m->event_size_bytes, tmp_ptr, 0, NULL, lp);
+        }
+        else{
+            /* schedule event to final destination for when the recv is complete */
+            e_new = tw_event_new(m->final_dest_gid, recv_queue_time, lp);
+            m_new = tw_event_data(e_new);
+            memcpy(m_new, tmp_ptr, m->event_size_bytes);
+            tw_event_send(e_new);
+        }
     }
 
     return;
@@ -760,16 +777,18 @@ static void handle_msg_start_event(
 /*This method will serve as an intermediate layer between simplewan and modelnet. 
  * It takes the packets from modelnet layer and calls underlying simplewan methods*/
 static tw_stime simplewan_packet_event(
-		char* category,
-		tw_lpid final_dest_lp,
-		uint64_t packet_size,
-                tw_stime offset,
-		int remote_event_size,
-		const void* remote_event,
-		int self_event_size,
-		const void* self_event,
-		tw_lp *sender,
-		int is_last_pckt)
+        char* category,
+        tw_lpid final_dest_lp,
+        uint64_t packet_size,
+        int is_pull,
+        uint64_t pull_size, /* only used when is_pull == 1 */
+        tw_stime offset,
+        int remote_event_size,
+        const void* remote_event,
+        int self_event_size,
+        const void* self_event,
+        tw_lp *sender,
+        int is_last_pckt)
 {
      tw_event * e_new;
      tw_stime xfer_to_nic_time;
@@ -801,6 +820,8 @@ static tw_stime simplewan_packet_event(
      msg->event_size_bytes = 0;
      msg->local_event_size_bytes = 0;
      msg->event_type = MSG_START;
+     msg->is_pull = is_pull;
+     msg->pull_size = pull_size;
 
      tmp_ptr = (char*)msg;
      tmp_ptr += sw_get_msg_sz();

src/models/networks/model-net/torus.c

@@ -38,7 +38,7 @@ static void torus_packet_event_rc(tw_lp *sender)
 }
 
 /* torus packet event , generates a torus packet on the compute node */
-static tw_stime torus_packet_event(char* category, tw_lpid final_dest_lp, uint64_t packet_size, tw_stime offset, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lp *sender, int is_last_pckt)
+static tw_stime torus_packet_event(char* category, tw_lpid final_dest_lp, uint64_t packet_size, int is_pull, uint64_t pull_size, tw_stime offset, int remote_event_size, const void* remote_event, int self_event_size, const void* self_event, tw_lp *sender, int is_last_pckt)
 {
     tw_event * e_new;
     tw_stime xfer_to_nic_time;
@@ -66,6 +66,8 @@ static tw_stime torus_packet_event(char* category, tw_lpid final_dest_lp, uint64
     msg->remote_event_size_bytes = 0;
     msg->local_event_size_bytes = 0;
     msg->type = GENERATE;
+    msg->is_pull = is_pull;
+    msg->pull_size = pull_size;
     
     num_chunks = msg->packet_size/chunk_size;
 
@@ -475,12 +477,20 @@ static void packet_arrive( nodes_state * s,
 	    if(msg->remote_event_size_bytes)
 	    {
 	       ts = (1/link_bandwidth) * msg->remote_event_size_bytes;
-	       e = tw_event_new(msg->final_dest_gid, ts, lp);
-	       m = tw_event_data(e);
 	       char* tmp_ptr = (char*)msg;
 	       tmp_ptr += torus_get_msg_sz();
-	       memcpy(m, tmp_ptr, msg->remote_event_size_bytes);
-	       tw_event_send(e);
+               if (msg->is_pull){
+                   int net_id = model_net_get_id("torus");
+                   model_net_event(net_id, msg->category, msg->sender_lp,
+                           msg->pull_size, ts, msg->remote_event_size_bytes,
+                           tmp_ptr, 0, NULL, lp);
+               }
+               else{
+                   e = tw_event_new(msg->final_dest_gid, ts, lp);
+                   m = tw_event_data(e);
+                   memcpy(m, tmp_ptr, msg->remote_event_size_bytes);
+                   tw_event_send(e);
+               }
 	    }
        }
     }
@@ -572,6 +582,12 @@ static void node_rc_handler(nodes_state * s, tw_bf * bf, nodes_message * msg, tw