3 files changed, 78 insertions, 28 deletions

diff --git a/include/litmus/rt_param.h b/include/litmus/rt_param.h
index a767e52882f4..5411f1251e57 100644
--- a/include/litmus/rt_param.h
+++ b/include/litmus/rt_param.h
@@ -129,8 +129,11 @@ struct pgm_consume
 struct rt_pgm {
         struct pgm_produce *produce;    /* output and current state */
         struct pgm_consume *consume;    /* input constraints and current state */ 
-        struct task_struct *root;               /* root node of the PGM */
-//      struct task_struct *src;                /* only not NULL if task is sink */
+        struct task_struct *src;                /* root node of the PGM */
+
+        /* only valid for source nodes */
+        struct completion       children_ready;
+        atomic_t                        nr_children_pending;
 };
 
 #endif /* end PGM */
diff --git a/litmus/pgm.c b/litmus/pgm.c
index 661fb2065dea..fe67d8f95d17 100644
--- a/litmus/pgm.c
+++ b/litmus/pgm.c
@@ -1,6 +1,8 @@
 
 #include <linux/slab.h>
 #include <linux/spinlock.h>
+#include <linux/completion.h>
+#include <linux/atomic.h>
 
 #include <litmus/pgm.h>
 #include <litmus/sched_plugin.h>
@@ -11,7 +13,7 @@
 
 static void pgm_consume(struct task_struct *t);
 static void pgm_produce_intermediate(struct task_struct *t);
-static void pgm_produce_root(struct task_struct *t);
+static void pgm_produce_src(struct task_struct *t);
 
 
 static inline int __is_pgm_satisfied(struct task_struct *t)
@@ -41,10 +43,18 @@ static void pgm_produce(struct task_struct *t)
                 pgm_produce_intermediate(t);
         }
         else {
-                pgm_produce_root(t);
+                pgm_produce_src(t);
         }
 }
 
+static void pgm_prepare_to_wake(struct task_struct *t)
+{
+        // 't' is not a source node.
+
+        set_task_state(t, TASK_RUNNING);
+
+        /* TODO: Set our deadline! */
+}
 
 static void pgm_produce_intermediate(struct task_struct *t)
 {
@@ -111,6 +121,7 @@ static void pgm_produce_intermediate(struct task_struct *t)
                 if(release_child)
                 {
                         pgm_consume(child);
+                        pgm_prepare_to_wake(child);
 
                         TRACE_TASK(t, "waking up %s/%d\n", child->comm, child->pid);
                         litmus->task_wake_up(child);
@@ -119,18 +130,18 @@ static void pgm_produce_intermediate(struct task_struct *t)
 }
 
 
-static void pgm_produce_root(struct task_struct *t)
+static void pgm_produce_src(struct task_struct *t)
 {
-        struct task_struct *root;
+        struct task_struct *src;
         BUG_ON(!is_pgm_sink(t));
 
-        root = tsk_rt(t)->pgm_params.root;
+        src = tsk_rt(t)->pgm_params.src;
 
-        TRACE_TASK(t, "waking up PGM root %s/%d\n",
-                root->comm, root->pid);
+        TRACE_TASK(t, "waking up PGM src %s/%d\n",
+                src->comm, src->pid);
 
         /* wake up the new task, the task will sleep if it is too soon. */
-        litmus->task_wake_up(root);
+        litmus->task_wake_up(src);
 }
 
 
@@ -183,9 +194,27 @@ int is_pgm_satisfied(struct task_struct *t)
 
 void pgm_prepare_for_next_period(struct task_struct *t)
 {
+        BUG_ON(!t);
+
         TRACE_TASK(t, "pgm_prepage_for_next_period().\n");
 
-        prepare_for_next_period(t);
+        if(is_pgm_src(t)) {
+                prepare_for_next_period(t);
+        }
+        else {
+                // We defer setting up release and deadline times
+                // up to wake-time.  This makes handling the deadline
+                // of the first job properly.
+
+                tsk_rt(t)->job_params.exec_time = 0;
+                ++(tsk_rt(t)->job_params.job_no);
+                t->rt.time_slice = 1; // don't confuse Linux (??)
+        }
+
+        // DANGER! We're setting an UNINTERRUPTIBLE state
+        // without going on a wait queue!  Better hope the
+        // PGM sources do their work...
+        set_task_state(t, TASK_UNINTERRUPTIBLE);
 }
 
 
@@ -230,7 +259,7 @@ int pgm_complete_job(struct task_struct *t)
         }
 
         /* produce new tokens, possibly waking up children,
-         * or setup for the next PGM root.
+         * or setup for the next PGM src.
          */
         pgm_produce(t);
 
@@ -245,13 +274,6 @@ int do_pgm_add_edge(struct task_struct *a, int nr_gen,
         struct pgm_produce *prod;
         struct pgm_consume *cons;
 
-        // validation of parameters takes place upstream.
-
-        if(!is_pgm(a)) {
-                // 'a' is the root. set itself as such.
-                tsk_rt(a)->pgm_params.root = a;
-        }
-
         if(tsk_rt(a)->pgm_params.produce == NULL) {
                 tsk_rt(a)->pgm_params.produce =
                         kmalloc(sizeof(pgm_produce), GFP_KERNEL);
@@ -288,9 +310,6 @@ int do_pgm_add_edge(struct task_struct *a, int nr_gen,
         __set_bit(cons->nr_consumptions, cons->unsatisfied);
         ++(cons->nr_consumptions);
 
-        // copy over root node from a to b
-        tsk_rt(b)->pgm_params.root = tsk_rt(a)->pgm_params.root;
-
         // mark as pgm nodes
         tsk_rt(a)->is_pgm_node = 1;
         tsk_rt(b)->is_pgm_node = 1;
@@ -299,13 +318,25 @@ out:
         return retval;
 }
 
-int do_pgm_signal_ready(struct task_struct *root)
+int do_pgm_signal_ready(struct task_struct *src)
 {
         int retval = -EINVAL;
+        int num_remaining;
+        struct task_struct *t = current;
 
-        tsk_rt(current)->pgm_params.root = root;
+        tsk_rt(t)->pgm_params.src = src;
+
+        num_remaining = 
+                atomic_dec_return(&tsk_rt(src)->pgm_params.nr_children_pending);
+
+        if(num_remaining == 0) {
+                complete(&tsk_rt(src)->pgm_params.children_ready);
+        }
 
-        /* decrement the parent's semaphore and then suspend */
+        // pray that our parent wakes us up.
+        set_task_state(t, TASK_UNINTERRUPTIBLE);
+        schedule();
+        retval = 0;
 
         return retval;
 }
@@ -313,19 +344,29 @@ int do_pgm_signal_ready(struct task_struct *root)
 int do_pgm_set_nr_children(int nr_children)
 {
         int retval = -EINVAL;
+        struct task_struct *src = current;
 
-        /* TODO: set up a counting semaphore */
+        if(!is_pgm_src(src))
+                goto out;
+
+        init_completion(&tsk_rt(src)->pgm_params.children_ready);       
+        atomic_set(&tsk_rt(src)->pgm_params.nr_children_pending, nr_children);
 
+out:
         return retval;
 }
 
 int do_pgm_wait_for_children(void)
 {
         int retval = -EINVAL;
-        struct task_struct *root = current;
+        struct task_struct *src = current;
 
-        /* TODO: sleep on a counting semaphore */
+        if(!is_pgm_src(src))
+                goto out;
+
+        wait_for_completion(&tsk_rt(src)->pgm_params.children_ready);
 
+out:
         return retval;
 }
 
diff --git a/litmus/sched_gsn_edf.c b/litmus/sched_gsn_edf.c
index c1145fc1dc8c..ca877938f729 100644
--- a/litmus/sched_gsn_edf.c
+++ b/litmus/sched_gsn_edf.c
@@ -640,6 +640,8 @@ static void gsnedf_task_wake_up(struct task_struct *task)
         TRACE_TASK(task, "wake_up at %llu\n", litmus_clock());
 
         raw_spin_lock_irqsave(&gsnedf_lock, flags);
+
+#if 0
         /* We need to take suspensions because of semaphores into
          * account! If a job resumes after being suspended due to acquiring
          * a semaphore, it should never be treated as a new job release.
@@ -661,6 +663,10 @@ static void gsnedf_task_wake_up(struct task_struct *task)
                         }
                 }
         }
+#else   // periodic task model
+        set_rt_flags(task, RT_F_RUNNING);
+#endif
+
         gsnedf_job_arrival(task);
         raw_spin_unlock_irqrestore(&gsnedf_lock, flags);
 }