modelnet sampling interface and event loop

NOTE: no models implement it yet

codes/model-net-lp.h

@@ -97,6 +97,8 @@ enum model_net_base_event_type {
     MN_BASE_NEW_MSG,
     // schedule next packet
     MN_BASE_SCHED_NEXT,
+    // gather a sample from the underlying model
+    MN_BASE_SAMPLE,
     // message goes directly down to topology-specific event handler
     MN_BASE_PASS
 };

codes/model-net-method.h

@@ -54,7 +54,11 @@ struct model_net_method
     int (*mn_get_msg_sz)();
     void (*mn_report_stats)();
     void (*mn_collective_call)(char const * category, int message_size, int remote_event_size, const void* remote_event, tw_lp* sender);
-    void (*mn_collective_call_rc)(int message_size, tw_lp* sender);    
+    void (*mn_collective_call_rc)(int message_size, tw_lp* sender);
+    event_f mn_sample_fn;
+    revent_f mn_sample_rc_fn;
+    init_f mn_sample_init_fn;
+    final_f mn_sample_fini_fn;
 };
 
 extern struct model_net_method * method_array[];

codes/model-net.h

@@ -140,6 +140,14 @@ void model_net_register();
  * modelnet_order configuration parameter */
 int* model_net_configure(int *id_count);
 
+/* Sets up a sampling loop for model net events. The sampling data provided by
+ * each modelnet lp is model-defined. This is a PE-wide setting. Data is sent
+ * to LP-IO with the category modelnet-samples */
+void model_net_enable_sampling(tw_stime interval);
+
+/* Returns 1 if modelnet is performing sampling, 0 otherwise */
+int model_net_sampling_enabled(void);
+
 /* Initialize/configure the network(s) based on the CODES configuration.
  * returns an array of the network ids, indexed in the order given by the 
  * modelnet_order configuration parameter 

src/networks/model-net/dragonfly.c

@@ -2851,6 +2851,10 @@ struct model_net_method dragonfly_method =
     .mn_get_msg_sz = dragonfly_get_msg_sz,
     .mn_report_stats = dragonfly_report_stats,
     .mn_collective_call = dragonfly_collective,
-    .mn_collective_call_rc = dragonfly_collective_rc   
+    .mn_collective_call_rc = dragonfly_collective_rc,
+    .mn_sample_fn = NULL,
+    .mn_sample_rc_fn = NULL,
+    .mn_sample_init_fn = NULL,
+    .mn_sample_fini_fn = NULL
 };
 

src/networks/model-net/loggp.c

@@ -155,7 +155,11 @@ struct model_net_method loggp_method =
     .mn_get_msg_sz = loggp_get_msg_sz,
     .mn_report_stats = loggp_report_stats,
     .mn_collective_call = loggp_collective,
-    .mn_collective_call_rc = loggp_collective_rc
+    .mn_collective_call_rc = loggp_collective_rc,
+    .mn_sample_fn = NULL,
+    .mn_sample_rc_fn = NULL,
+    .mn_sample_init_fn = NULL,
+    .mn_sample_fini_fn = NULL
 };
 
 static void loggp_init(

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

@@ -35,6 +35,9 @@ static int                       num_params = 0;
 static const char              * annos[CONFIGURATION_MAX_ANNOS];
 static model_net_base_params     all_params[CONFIGURATION_MAX_ANNOS];
 
+static tw_stime mn_sample_interval = 0.0;
+static int mn_sample_enabled = 0;
+
 typedef struct model_net_base_state {
     int net_id;
     // whether scheduler loop is running
@@ -112,6 +115,26 @@ tw_lptype model_net_base_lp = {
 
 /**** BEGIN IMPLEMENTATIONS ****/
 
+void model_net_enable_sampling(tw_stime interval)
+{
+    mn_sample_interval = interval;
+    mn_sample_enabled = 1;
+}
+
+int model_net_sampling_enabled(void)
+{
+    return mn_sample_enabled;
+}
+
+// schedule sample event - want to be precise, so no noise here
+static void issue_sample_event(tw_lp *lp)
+{
+    tw_event *e = tw_event_new(lp->gid, mn_sample_interval, lp);
+    model_net_wrap_msg *m = tw_event_data(e);
+    msg_set_header(model_net_base_magic, MN_BASE_SAMPLE, lp->gid, &m->h);
+    tw_event_send(e);
+}
+
 void model_net_base_register(int *do_config_nets){
     // here, we initialize ALL lp types to use the base type
     for (int i = 0; i < MAX_NETS; i++){
@@ -319,6 +342,28 @@ void model_net_base_lp_init(
 
     // initialize the model-net method
     ns->sub_type->init(ns->sub_state, lp);
+
+    // check validity of sampling function
+    event_f  sample  = method_array[ns->net_id]->mn_sample_fn;
+    revent_f rsample = method_array[ns->net_id]->mn_sample_rc_fn;
+    if (model_net_sampling_enabled()) {
+        if (sample == NULL) {
+            tw_error(TW_LOC,
+                    "Sampling requested for a model that doesn't provide it\n");
+        }
+        else if (rsample == NULL &&
+                (g_tw_synchronization_protocol == OPTIMISTIC ||
+                 g_tw_synchronization_protocol == OPTIMISTIC_DEBUG)) {
+            tw_error(TW_LOC,
+                    "Sampling requested for a model that doesn't provide it\n");
+        }
+        else {
+            init_f sinit = method_array[ns->net_id]->mn_sample_init_fn;
+            if (sinit != NULL)
+                sinit(ns->sub_state, lp);
+            issue_sample_event(lp);
+        }
+    }
 }
 
 void model_net_base_event(
@@ -327,7 +372,8 @@ void model_net_base_event(
         model_net_wrap_msg * m,
         tw_lp * lp){
     assert(m->h.magic == model_net_base_magic);
-    
+
+    void * sub_msg;
     switch (m->h.event_type){
         case MN_BASE_NEW_MSG:
             handle_new_msg(ns, b, m, lp);
@@ -335,8 +381,15 @@ void model_net_base_event(
         case MN_BASE_SCHED_NEXT:
             handle_sched_next(ns, b, m, lp);
             break;
+        case MN_BASE_SAMPLE: ;
+            event_f sample = method_array[ns->net_id]->mn_sample_fn;
+            assert(model_net_sampling_enabled() && sample != NULL);
+            sub_msg = ((char*)m)+msg_offsets[ns->net_id];
+            sample(ns->sub_state, b, sub_msg, lp);
+            issue_sample_event(lp);
+            break;
         case MN_BASE_PASS: ;
-            void * sub_msg = ((char*)m)+msg_offsets[ns->net_id];
+            sub_msg = ((char*)m)+msg_offsets[ns->net_id];
             ns->sub_type->event(ns->sub_state, b, sub_msg, lp);
             break;
         /* ... */
@@ -352,7 +405,8 @@ void model_net_base_event_rc(
         model_net_wrap_msg * m,
         tw_lp * lp){
     assert(m->h.magic == model_net_base_magic);
-    
+
+    void * sub_msg;
     switch (m->h.event_type){
         case MN_BASE_NEW_MSG:
             handle_new_msg_rc(ns, b, m, lp);
@@ -360,8 +414,14 @@ void model_net_base_event_rc(
         case MN_BASE_SCHED_NEXT:
             handle_sched_next_rc(ns, b, m, lp);
             break;
+        case MN_BASE_SAMPLE: ;
+            revent_f sample_rc = method_array[ns->net_id]->mn_sample_rc_fn;
+            assert(model_net_sampling_enabled() && sample_rc != NULL);
+            sub_msg = ((char*)m)+msg_offsets[ns->net_id];
+            sample_rc(ns->sub_state, b, sub_msg, lp);
+            break;
         case MN_BASE_PASS: ;
-            void * sub_msg = ((char*)m)+msg_offsets[ns->net_id];
+            sub_msg = ((char*)m)+msg_offsets[ns->net_id];
             ns->sub_type->revent(ns->sub_state, b, sub_msg, lp);
             break;
         /* ... */
@@ -369,13 +429,14 @@ void model_net_base_event_rc(
             assert(!"model_net_base event type not known");
             break;
     }
-
-    *(int*)b = 0;
 }
 
 void model_net_base_finalize(
         model_net_base_state * ns,
         tw_lp * lp){
+    final_f sfini = method_array[ns->net_id]->mn_sample_fini_fn;
+    if (sfini != NULL)
+        sfini(ns->sub_state, lp);
     ns->sub_type->final(ns->sub_state, lp);
     free(ns->sub_state);
 }

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

@@ -120,8 +120,12 @@ struct model_net_method simplenet_method =
     .mn_get_lp_type = sn_get_lp_type,
     .mn_get_msg_sz = sn_get_msg_sz,
     .mn_report_stats = sn_report_stats,
-    .mn_collective_call = simple_net_collective,	
-    .mn_collective_call_rc = simple_net_collective_rc
+    .mn_collective_call = simple_net_collective,
+    .mn_collective_call_rc = simple_net_collective_rc,
+    .mn_sample_fn = NULL,
+    .mn_sample_rc_fn = NULL,
+    .mn_sample_init_fn = NULL,
+    .mn_sample_fini_fn = NULL
 };
 
 static void sn_init(

src/networks/model-net/simplep2p.c

@@ -151,7 +151,11 @@ struct model_net_method simplep2p_method =
     .mn_get_msg_sz = sp_get_msg_sz,
     .mn_report_stats = sp_report_stats,
     .mn_collective_call = simple_wan_collective,
-    .mn_collective_call_rc = simple_wan_collective_rc  
+    .mn_collective_call_rc = simple_wan_collective_rc,
+    .mn_sample_fn = NULL,
+    .mn_sample_rc_fn = NULL,
+    .mn_sample_init_fn = NULL,
+    .mn_sample_fini_fn = NULL
 };
 
 static void sp_init(

src/networks/model-net/torus.c

@@ -1981,7 +1981,11 @@ struct model_net_method torus_method =
    .mn_get_msg_sz = torus_get_msg_sz,
    .mn_report_stats = torus_report_stats,
    .mn_collective_call = torus_collective,
-   .mn_collective_call_rc = torus_collective_rc
+   .mn_collective_call_rc = torus_collective_rc,
+   .mn_sample_fn = NULL,
+   .mn_sample_rc_fn = NULL,
+   .mn_sample_init_fn = NULL,
+   .mn_sample_fini_fn = NULL
 };
 
 /* user-facing modelnet functions */