model-net scheduler implementation and integration

- NOTE: the current scheduling policy of FCFS is hard-coded, this will change
  in future revisions
- NOTE: random number generation has changed due to the differences in
  scheduling and packet ordering. Hence, results may be slightly different from
  runs prior to this commit

codes/model-net-lp.h

@@ -14,6 +14,8 @@
 #define MODEL_NET_LP_H
 
 #include <ross.h>
+#include "model-net.h"
+#include "model-net-sched.h"
 #include "net/dragonfly.h"
 #include "net/loggp.h"
 #include "net/simplenet-upd.h"
@@ -57,23 +59,13 @@ enum model_net_base_event_type {
     MN_BASE_PASS
 };
 
-typedef struct model_net_request {
-    tw_lpid  final_dest_lp;
-    uint64_t msg_size;
-    int      net_id;
-    int      is_pull;
-    int      remote_event_size;
-    int      self_event_size;
-    char     category[CATEGORY_NAME_MAX];
-} model_net_request;
-
 typedef struct model_net_base_msg {
     // no need for event type - in wrap message
-    tw_lpid src;
     union {
         model_net_request req;
         struct {} sched; // needs nothing at the moment
     } u;
+    model_net_sched_rc rc; // rc for scheduling events
 } model_net_base_msg;
 
 typedef struct model_net_wrap_msg {

codes/model-net-sched.h

+/*
+ * Copyright (C) 2014 University of Chicago.
+ * See COPYRIGHT notice in top-level directory.
+ *
+ */
+
+/* A scheduler interface for use by model-net. */
+
+#ifndef MODEL_NET_SCHED_H
+#define MODEL_NET_SCHED_H
+
+#include <ross.h>
+#include "model-net.h"
+#include "model-net-method.h"
+
+/// scheduler decls
+
+typedef struct model_net_sched_s model_net_sched;
+typedef struct model_net_sched_rc_s model_net_sched_rc;
+
+/// interface to be implemented by schedulers
+/// see corresponding general functions
+typedef struct model_net_sched_interface {
+    void (*init)(struct model_net_method *method, void ** sched);
+    void (*destroy)(void * sched);
+    void (*add)(
+            model_net_request *req, 
+            int remote_event_size,
+            void * remote_event,
+            int local_event_size,
+            void * local_event,
+            void *sched, 
+            model_net_sched_rc *rc, 
+            tw_lp *lp);
+    void (*add_rc)(void *sched, model_net_sched_rc *rc, tw_lp *lp);
+    int  (*next)(
+            tw_stime *poffset,
+            void *sched, 
+            model_net_sched_rc *rc, 
+            tw_lp *lp);
+    void (*next_rc)(void * sched, model_net_sched_rc *rc, tw_lp *lp);
+} model_net_sched_interface;
+
+enum sched_type {
+    MN_SCHED_FCFS, // issue all packets at once (orig. model-net behavior)
+    MN_SCHED_RR    // round-robin packet scheduling
+}; 
+
+/// overall scheduler struct - type puns the actual data structure
+
+struct model_net_sched_s {
+    enum sched_type type;
+    void * dat;
+    model_net_sched_interface *impl;
+};
+
+/// reverse computation structure - this is expected to be held by upper-level
+/// model-net LPs and passed to the scheduler functions to allow proper rc
+/// NOTE: since modelnet LPs will be stashing this in their event structs, 
+/// need to provide full definition in header
+struct model_net_sched_rc_s {
+    // NOTE: sched implementations may need different types, but for now they
+    // are equivalent 
+    int rtn; // return code from a sched_next 
+};
+
+void model_net_sched_init(
+        enum sched_type type, 
+        struct model_net_method *method,
+        model_net_sched *sched);
+
+/// schedules the next chunk, storing any info needed for rc in sched_rc
+/// packet issue time is placed in poffset to be able to separate packet calls
+/// between multiple scheduler events
+/// returns: 
+/// * 0 on success,
+/// * 1 on success and the corresponding request is finished. In this case,
+///   out_req is set to the underlying request
+/// * -1 when there is nothing to be scheduled
+int model_net_sched_next(
+        tw_stime *poffset,
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp);
+
+void model_net_sched_next_rc(
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp);
+
+/// enter a new request into the scheduler, storing any info needed for rc in
+/// sched_rc
+void model_net_sched_add(
+        model_net_request *req,
+        int remote_event_size,
+        void * remote_event,
+        int local_event_size,
+        void * local_event,
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp);
+
+void model_net_sched_add_rc(
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp);
+
+#endif /* end of include guard: MODEL_NET_SCHED_H */
+
+/*
+ * Local variables:
+ *  c-indent-level: 4
+ *  c-basic-offset: 4
+ * End:
+ *
+ * vim: ft=c ts=8 sts=4 sw=4 expandtab
+ */

codes/model-net.h

@@ -66,6 +66,20 @@ enum NETWORKS
 extern char * model_net_lp_config_names[];
 extern char * model_net_method_names[];
 
+// request structure that gets passed around (by the model-net implementation,
+// not the user)
+typedef struct model_net_request {
+    tw_lpid  final_dest_lp;
+    tw_lpid  src_lp;
+    uint64_t msg_size;
+    uint64_t packet_size;
+    int      net_id;
+    int      is_pull;
+    int      remote_event_size;
+    int      self_event_size;
+    char     category[CATEGORY_NAME_MAX];
+} model_net_request;
+
 /* data structure for tracking network statistics */
 struct mn_stats
 {

src/models/Makefile.subdir

@@ -4,6 +4,7 @@ nobase_include_HEADERS = \
  codes/model-net.h	\
  codes/model-net-method.h \
  codes/model-net-lp.h \
+ codes/model-net-sched.h \
  codes/net/dragonfly.h \
  codes/net/loggp.h \
  codes/net/simplenet-upd.h \
@@ -18,4 +19,5 @@ src_libcodes_net_a_SOURCES = \
  src/models/networks/model-net/dragonfly.c \
  src/models/networks/model-net/loggp.c \
  src/models/networks/model-net/simplewan.c \
- src/models/networks/model-net/model-net-lp.c
+ src/models/networks/model-net/model-net-lp.c \
+ src/models/networks/model-net/model-net-sched.c

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

@@ -9,6 +9,7 @@
 #include "codes/model-net.h"
 #include "codes/model-net-method.h"
 #include "codes/model-net-lp.h"
+#include "codes/model-net-sched.h"
 #include "codes/codes_mapping.h"
 #include "codes/jenkins-hash.h"
 
@@ -24,6 +25,10 @@ static int msg_offsets[MAX_NETS];
 
 typedef struct model_net_base_state {
     int net_id;
+    // whether scheduler loop is running
+    int in_sched_loop;
+    // model-net scheduler
+    model_net_sched *sched;
     // lp type and state of underlying model net method - cache here so we
     // don't have to constantly look up
     const tw_lptype *sub_type;
@@ -126,6 +131,10 @@ void model_net_base_lp_init(
         }
     }
 
+    // TODO: parameterize scheduler type
+    ns->sched = malloc(sizeof(model_net_sched));
+    model_net_sched_init(MN_SCHED_FCFS, method_array[ns->net_id], ns->sched);
+
     ns->sub_type = model_net_get_lp_type(ns->net_id);
     // NOTE: some models actually expect LP state to be 0 initialized...
     // *cough anything that uses mn_stats_array cough*
@@ -147,8 +156,6 @@ void model_net_base_event(
             handle_new_msg(ns, b, m, lp);
             break;
         case MN_BASE_SCHED_NEXT:
-            // TODO: write the scheduler
-            assert(0);
             handle_sched_next(ns, b, m, lp);
             break;
         case MN_BASE_PASS: ;
@@ -194,91 +201,91 @@ void model_net_base_finalize(
     free(ns->sub_state);
 }
 
-/* event type handlers */
+/// bitfields used:
+/// c0 - we initiated a sched_next event
 void handle_new_msg(
         model_net_base_state * ns,
         tw_bf *b,
         model_net_wrap_msg * m,
         tw_lp * lp){
-    // TODO: write the scheduler, this currently issues all packets at once
-    // (i.e. the orig modelnet implementation)
+    // simply pass down to the scheduler
     model_net_request *r = &m->msg.m_base.u.req;
-
-    uint64_t packet_size = model_net_get_packet_size(r->net_id);
-    uint64_t send_msg_size = r->is_pull ? PULL_MSG_SIZE : r->msg_size;
-    uint64_t num_packets = send_msg_size/packet_size; /* Number of packets to be issued by the API */
-    uint64_t i;
-    int last = 0;
-
-    if(send_msg_size % packet_size)
-        num_packets++; /* Handle the left out data if message size is not exactly divisible by packet size */
-
-    /* compute the (possible) remote/self event locations */
-    void * e_msg = (m+1);
-    void *remote_event = NULL, *self_event = NULL;
+    void * m_data = m+1;
+    void *remote = NULL, *local = NULL;
     if (r->remote_event_size > 0){
-        remote_event = e_msg;
-        e_msg = (char*)e_msg + r->remote_event_size;
+        remote = m_data;
+        m_data = (char*)m_data + r->remote_event_size;
     }
     if (r->self_event_size > 0){
-        self_event = e_msg;
+        local = m_data;
     }
-
-    /* issue N packets using method API */
-    /* somehow mark the final packet as the one responsible for delivering
-     * the self event and remote event 
-     *
-     * local event is delivered to caller of this function, remote event is
-     * passed along through network hops and delivered to final_dest_lp
-     */
-
-    tw_stime poffset = 0.0;
-    for( i = 0; i < num_packets; i++ )
-    {
-        /*Mark the last packet to the net method API*/
-        if(i == num_packets - 1)
-        {
-            last = 1;
-            /* also calculate the last packet's size */
-            packet_size = send_msg_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*/
-        poffset += method_array[r->net_id]->model_net_method_packet_event(r->category,
-                r->final_dest_lp, packet_size, r->is_pull, r->msg_size, poffset,
-                r->remote_event_size, remote_event, r->self_event_size, self_event, m->msg.m_base.src, lp, last);
+    
+    model_net_sched_add(r, r->remote_event_size, remote, r->self_event_size,
+            local, ns->sched, &m->msg.m_base.rc, lp);
+    
+    if (ns->in_sched_loop == 0){
+        b->c0 = 1;
+        tw_event *e = codes_event_new(lp->gid, codes_local_latency(lp), lp);
+        model_net_wrap_msg *m = tw_event_data(e);
+        m->event_type = MN_BASE_SCHED_NEXT;
+        m->magic = model_net_base_magic;
+        // m_base not used in sched event
+        tw_event_send(e);
+        ns->in_sched_loop = 1;
     }
-    return;
 }
-void handle_sched_next(
-        model_net_base_state * ns,
+
+void handle_new_msg_rc(
+        model_net_base_state *ns,
         tw_bf *b,
-        model_net_wrap_msg * m,
-        tw_lp * lp){
-    // currently unused
+        model_net_wrap_msg *m,
+        tw_lp *lp){
+    model_net_sched_add_rc(ns->sched, &m->msg.m_base.rc, lp);
+    if (b->c0){
+        codes_local_latency_reverse(lp);
+        ns->in_sched_loop = 0;
+    }
 }
-void handle_new_msg_rc(
+
+/// bitfields used
+/// c0 - scheduler loop is finished
+void handle_sched_next(
         model_net_base_state * ns,
         tw_bf *b,
         model_net_wrap_msg * m,
         tw_lp * lp){
-    model_net_request *r = &m->msg.m_base.u.req;
-
-    uint64_t packet_size = model_net_get_packet_size(r->net_id);
-    uint64_t send_msg_size = r->is_pull ? PULL_MSG_SIZE : r->msg_size;
-    uint64_t num_packets = send_msg_size/packet_size; /* Number of packets to be issued by the API */
-
-    if(send_msg_size % packet_size)
-        num_packets++; /* Handle the left out data if message size is not exactly divisible by packet size */
-
-    for (uint64_t i = 0; i < num_packets; i++) {
-        method_array[r->net_id]->model_net_method_packet_event_rc(lp);
+    tw_stime poffset;
+    int ret = model_net_sched_next(&poffset, ns->sched, &m->msg.m_base.rc, lp);
+    // we only need to know whether scheduling is finished or not - if not,
+    // go to the 'next iteration' of the loop
+    if (ret == -1){
+        b->c0 = 1;
+        ns->in_sched_loop = 0;
+    }
+    else {
+        tw_event *e = codes_event_new(lp->gid, 
+                poffset+codes_local_latency(lp), lp);
+        model_net_wrap_msg *m = tw_event_data(e);
+        m->event_type = MN_BASE_SCHED_NEXT;
+        m->magic = model_net_base_magic;
+        // no need to set m_base here
+        tw_event_send(e);
     }
 }
+
 void handle_sched_next_rc(
         model_net_base_state * ns,
         tw_bf *b,
         model_net_wrap_msg * m,
         tw_lp * lp){
+    model_net_sched_next_rc(ns->sched, &m->msg.m_base.rc, lp);
+
+    if (b->c0){
+        ns->in_sched_loop = 1;
+    }
+    else{
+        codes_local_latency_reverse(lp);
+    }
 }
 
 /**** END IMPLEMENTATIONS ****/

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

+/*
+ * Copyright (C) 2014 University of Chicago.
+ * See COPYRIGHT notice in top-level directory.
+ *
+ */
+
+/* A scheduler interface for use by model-net. */
+
+#include <assert.h>
+#include <ross.h>
+
+#include "codes/model-net-sched.h"
+#include "codes/model-net-lp.h"
+#include "codes/quicklist.h"
+
+/// scheduler-specific data structures (TODO: split specific schedulers into
+/// their own files if we move beyond just these two)
+/// NOTE: for now, scheduler data structures are the same - this may change in
+/// later versions
+
+typedef struct mn_sched {
+    // method containing packet event to call
+    struct model_net_method *method;
+    struct qlist_head reqs; // of type mn_sched_qitem
+    // this is an unfortunate result - we have to basically not free anything
+    // in order to keep around the remote and local events
+    // we desperately need GVT hooks to run our own garbage collection
+    struct qlist_head free_reqs;
+} mn_sched;
+
+// at the moment, rr and fcfs only differ in how req queue is modified, queue
+// items themselves are equivalent
+
+typedef struct mn_sched_qitem {
+    model_net_request req;
+    // remaining bytes to send
+    uint64_t rem;
+    // pointers to event structures 
+    // sizes are given in the request struct
+    void * remote_event;
+    void * local_event;
+    struct qlist_head ql;
+} mn_sched_qitem;
+
+/// scheduler-specific function decls and function table
+
+// void used to avoid ptr-to-ptr conv warnings
+static void fcfs_init (struct model_net_method *method, void ** sched);
+static void fcfs_destroy (void *sched);
+static void fcfs_add (
+        model_net_request *req, 
+        int remote_event_size,
+        void * remote_event,
+        int local_event_size,
+        void * local_event,
+        void *sched, 
+        model_net_sched_rc *rc, 
+        tw_lp *lp);
+static void fcfs_add_rc(void *sched, model_net_sched_rc *rc, tw_lp *lp);
+static int  fcfs_next(tw_stime *poffset, void *sched, model_net_sched_rc *rc, tw_lp *lp);
+static void fcfs_next_rc(void *sched, model_net_sched_rc *rc, tw_lp *lp);
+
+static model_net_sched_interface fcfs_tab = 
+{ &fcfs_init, &fcfs_destroy, &fcfs_add, &fcfs_add_rc, &fcfs_next, &fcfs_next_rc};
+
+static void rr_init (struct model_net_method *method, void ** sched);
+static void rr_destroy (void *sched);
+static void rr_add (
+        model_net_request *req,
+        int remote_event_size,
+        void * remote_event,
+        int local_event_size,
+        void * local_event,
+        void *sched,
+        model_net_sched_rc *rc,
+        tw_lp *lp);
+static void rr_add_rc(void *sched, model_net_sched_rc *rc, tw_lp *lp);
+static int  rr_next(tw_stime *poffset, void *sched, model_net_sched_rc *rc, tw_lp *lp);
+static void rr_next_rc (void *sched, model_net_sched_rc *rc, tw_lp *lp);
+
+static model_net_sched_interface rr_tab = 
+{ &rr_init, &rr_destroy, &rr_add, &rr_add_rc, &rr_next, &rr_next_rc};
+
+/// general scheduler functions
+
+void model_net_sched_init(
+        enum sched_type type, 
+        struct model_net_method *method,
+        model_net_sched *sched){
+    sched->type = type;
+    switch (type){
+        case MN_SCHED_FCFS:
+            sched->impl = &fcfs_tab;
+            break;
+        case MN_SCHED_RR:
+            sched->impl = &rr_tab;
+            break;
+        default:
+            fprintf(stderr, "unknown scheduler type");
+            abort();
+    }
+    sched->impl->init(method, &sched->dat);
+}
+
+int model_net_sched_next(
+        tw_stime *poffset,
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp){
+    return sched->impl->next(poffset, sched->dat, sched_rc, lp);
+}
+
+void model_net_sched_next_rc(
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp) {
+    sched->impl->next_rc(sched->dat, sched_rc, lp);
+}
+
+void model_net_sched_add(
+        model_net_request *req,
+        int remote_event_size,
+        void * remote_event,
+        int local_event_size,
+        void * local_event,
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp){
+    sched->impl->add(req, remote_event_size, remote_event, local_event_size,
+            local_event, sched->dat, sched_rc, lp);
+}
+
+void model_net_sched_add_rc(
+        model_net_sched *sched,
+        model_net_sched_rc *sched_rc,
+        tw_lp *lp){
+    sched->impl->add_rc(sched->dat, sched_rc, lp);
+}
+
+/// specific scheduler implementations
+
+void fcfs_init(struct model_net_method *method, void ** sched){
+    *sched = malloc(sizeof(mn_sched));
+    mn_sched *ss = *sched;
+    ss->method = method;
+    INIT_QLIST_HEAD(&ss->reqs);
+    INIT_QLIST_HEAD(&ss->free_reqs);
+}
+
+void fcfs_destroy(void *sched){
+    free(sched);
+}
+
+void fcfs_add (
+        model_net_request *req, 
+        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->req = *req;
+    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);
+    }
+    if (local_event_size > 0){
+        q->local_event = malloc(local_event_size);
+        memcpy(q->local_event, local_event, local_event_size);
+    }
+    mn_sched *s = sched;
+    qlist_add_tail(&q->ql, &s->reqs);
+}
+
+void fcfs_add_rc(void *sched, model_net_sched_rc *rc, tw_lp *lp){
+    mn_sched *s = sched;
+    struct qlist_head *ent = qlist_pop_back(&s->reqs);
+    assert(ent != NULL);
+    mn_sched_qitem *q = qlist_entry(ent, mn_sched_qitem, ql);
+    if (q->remote_event) free(q->remote_event);
+    if (q->local_event)  free(q->local_event);
+    free(q);
+}
+
+int fcfs_next(tw_stime *poffset, void *sched, model_net_sched_rc *rc, tw_lp *lp){
+    mn_sched *s = sched;
+    struct qlist_head *ent = s->reqs.next;
+    if (ent == &s->reqs){
+        rc->rtn = -1;
+        return -1;
+    }
+    mn_sched_qitem *q = qlist_entry(ent, mn_sched_qitem, ql);
+    
+    // issue the next packet
+    int is_last_packet;
+    uint64_t psize;
+    if (q->req.packet_size >= q->rem) {
+        psize = q->rem;
+        is_last_packet = 1;
+    }
+    else{
+        psize = q->req.packet_size;
+        is_last_packet = 0;
+    }
+
+    *poffset = s->method->model_net_method_packet_event(q->req.category,
+            q->req.final_dest_lp, psize, q->req.is_pull, q->req.msg_size, 0.0,
+            q->req.remote_event_size, q->remote_event, q->req.self_event_size,
+            q->local_event, q->req.src_lp, lp, is_last_packet);
+
+    // if last packet - remove from list, put into free list
+    if (is_last_packet){
+        qlist_pop(&s->reqs);
+        qlist_add_tail(&q->ql, &s->free_reqs);
+        rc->rtn = 1;
+    }
+    else{
+        q->rem -= psize;
+        rc->rtn = 0;
+    }
+    return rc->rtn;
+}
+
+void fcfs_next_rc(void *sched, model_net_sched_rc *rc, tw_lp *lp){
+    mn_sched *s = sched;
+    if (rc->rtn == -1){
+        // no op
+    }
+    else{