Added syscall hooks, 'finished' produce/consume code.

9 files changed, 504 insertions, 71 deletions

diff --git a/include/litmus/litmus.h b/include/litmus/litmus.h
index 683407328f5a..cfafef3a795d 100644
--- a/include/litmus/litmus.h
+++ b/include/litmus/litmus.h
@@ -126,26 +126,14 @@ void preempt_if_preemptable(struct task_struct* t, int on_cpu);
         (is_pgm(t) && \
          tsk_rt(t)->pgm_params.produce != NULL && \
          tsk_rt(t)->pgm_params.consume == NULL)
-#define is_pgm_sink(t) \
-        (is_pgm(t) && \
-         tsk_rt(t)->pgm_params.produce == NULL && \
-         tsk_rt(t)->pgm_params.consume != NULL)
 #define is_pgm_intermediate(t) \
         (is_pgm(t) && \
          tsk_rt(t)->pgm_params.produce != NULL && \
          tsk_rt(t)->pgm_params.consume != NULL)
-#ifdef CONFIG_PGM
-int is_pgm_satisfied(struct task_struct* t);
-void pgm_produce(struct task_struct* t);
-void pgm_consume(struct task_struct* t);
-int pgm_complete(struct task_struct* t);
-#else
-/* nothing */
-#define is_pgm_satisfied(t)
-#define pgm_produce(t)
-#define pgm_consume(t)
-#define pgm_complete(t)
-#endif
+#define is_pgm_sink(t) \
+        (is_pgm_intermediate(t) && \
+         tsk_rt(t)->pgm_params.produce == NULL && \
+         tsk_rt(t)->pgm_params.consume != NULL)
 
 #ifdef CONFIG_LITMUS_LOCKING
 void srp_ceiling_block(void);
diff --git a/include/litmus/pgm.h b/include/litmus/pgm.h
new file mode 100644
index 000000000000..688fc3b9bbe2
--- /dev/null
+++ b/include/litmus/pgm.h
@@ -0,0 +1,28 @@
+/*
+ * Routines for constructing and working with PGMs.
+ */
+
+#ifndef __UNC_PGM_H__
+#define __UNC_PGM_H__
+
+#include <linux/sched.h>
+
+#include <litmus/litmus.h>
+#include <litmus/rt_param.h>
+
+int is_pgm_satisfied(struct task_struct* t);
+
+void pgm_prepare_for_next_period(struct task_struct *t);
+
+int pgm_complete_job(struct task_struct* t);
+
+int do_pgm_add_edge(struct task_struct *a, int nr_gen,
+        struct task_struct *b, int nr_req);
+int do_pgm_signal_ready(struct task_struct *src);
+int do_pgm_set_nr_children(int nr_children);
+int do_pgm_wait_for_children(void);
+
+// TODO
+//int pgm_remove_edge(struct task_struct *a, struct task_struct *b);
+
+#endif
diff --git a/include/litmus/rt_param.h b/include/litmus/rt_param.h
index 613e9f98294e..a767e52882f4 100644
--- a/include/litmus/rt_param.h
+++ b/include/litmus/rt_param.h
@@ -77,6 +77,14 @@ struct control_page {
         /* to be extended */
 };
 
+struct pgm_edge {
+        pid_t a;
+        int a_nr_produced;
+
+        pid_t b;
+        int b_nr_consumed;
+};
+
 /* don't export internal data structures to user space (liblitmus) */
 #ifdef __KERNEL__
 
@@ -100,28 +108,29 @@ struct pgm_production
 struct pgm_produce
 {
         unsigned long nr_productions;           /* num productions/children */
-        struct pgm_production produce[0];       /* array of productions */
+        struct pgm_production produce[MAX_FAN]; /* array of productions. TODO: make dyn alloc */
 };
 
 struct pgm_consumption
 {
+        /* assumption required == number consumed */
         unsigned long required;         /* number tokens required before firing */
         long nr_tokens;                         /* constraint met when >= required */
 };
 
 struct pgm_consume
 {
-        spinlock_t      lock;
+        spinlock_t      lock;                   /* TODO: Use atomics instead of a lock. */
         unsigned long nr_consumptions;                          /* num of consumption constraints */
         unsigned long unsatisfied[NR_FAN_CHUNKS];       /* bit-array of unsatisfied constraints */
-        struct pgm_consumption consume[0];      /* array of consumption constraints */
-        //atomic_t nr_pending_release;
+        struct pgm_consumption consume[MAX_FAN];        /* array of consumption constraints. TODO: make dyn alloc */
 };
 
 struct rt_pgm {
         struct pgm_produce *produce;    /* output and current state */
         struct pgm_consume *consume;    /* input constraints and current state */ 
-        struct task_struct *src;                /* only not NULL if task is sink */
+        struct task_struct *root;               /* root node of the PGM */
+//      struct task_struct *src;                /* only not NULL if task is sink */
 };
 
 #endif /* end PGM */
diff --git a/include/litmus/unistd_32.h b/include/litmus/unistd_32.h
index 94264c27d9ac..92e733c42041 100644
--- a/include/litmus/unistd_32.h
+++ b/include/litmus/unistd_32.h
@@ -17,5 +17,9 @@
 #define __NR_wait_for_ts_release __LSC(9)
 #define __NR_release_ts         __LSC(10)
 #define __NR_null_call          __LSC(11)
+#define __NR_add_pgm_edge       __LSC(12)
+#define __NR_pgm_signal_ready   __LSC(13)
+#define __NR_pgm_set_nr_children        __LSC(14)
+#define __NR_pgm_wait_for_children      __LSC(15)
 
-#define NR_litmus_syscalls 12
+#define NR_litmus_syscalls 16
diff --git a/include/litmus/unistd_64.h b/include/litmus/unistd_64.h
index d5ced0d2642c..84125395dfe6 100644
--- a/include/litmus/unistd_64.h
+++ b/include/litmus/unistd_64.h
@@ -27,7 +27,15 @@ __SYSCALL(__NR_wait_for_job_release, sys_wait_for_job_release)
 __SYSCALL(__NR_wait_for_ts_release, sys_wait_for_ts_release)
 #define __NR_release_ts                         __LSC(10)
 __SYSCALL(__NR_release_ts, sys_release_ts)
-#define __NR_null_call                          __LSC(11)
+#define __NR_null_call              __LSC(11)
 __SYSCALL(__NR_null_call, sys_null_call)
+#define __NR_add_pgm_edge                       __LSC(12)
+__SYSCALL(__NR_add_pgm_edge, sys_add_pgm_edge)
+#define __NR_pgm_signal_ready           __LSC(13)
+__SYSCALL(__NR_pgm_signal_ready, sys_pgm_signal_ready)
+#define __NR_pgm_set_nr_children        __LSC(14)
+__SYSCALL(__NR_pgm_set_nr_children, sys_pgm_set_nr_children)
+#define __NR_pgm_wait_for_children      __LSC(15)
+__SYSCALL(__NR_pgm_wait_for_children, sys_pgm_wait_for_children)
 
-#define NR_litmus_syscalls 12
+#define NR_litmus_syscalls 16
diff --git a/litmus/jobs.c b/litmus/jobs.c
index 36e314625d86..ab2b05df91a9 100644
--- a/litmus/jobs.c
+++ b/litmus/jobs.c
@@ -6,6 +6,10 @@
 #include <litmus/litmus.h>
 #include <litmus/jobs.h>
 
+#ifdef CONFIG_PGM
+#include <litmus/pgm.h>
+#endif
+
 void prepare_for_next_period(struct task_struct *t)
 {
         BUG_ON(!t);
@@ -33,8 +37,17 @@ void release_at(struct task_struct *t, lt_t start)
  */
 long complete_job(void)
 {
+        struct task_struct *t = current;
+
+#ifdef CONFIG_PGM
+        if(is_pgm(t)) {
+                /* produce tokens, etc. */
+                pgm_complete_job(t);
+        }
+#endif
+
         /* Mark that we do not excute anymore */
-        set_rt_flags(current, RT_F_SLEEP);
+        set_rt_flags(t, RT_F_SLEEP);
         /* call schedule, this will return when a new job arrives
          * it also takes care of preparing for the next release
          */
diff --git a/litmus/litmus.c b/litmus/litmus.c
index 301390148d02..f6d29a42332a 100644
--- a/litmus/litmus.c
+++ b/litmus/litmus.c
@@ -21,6 +21,10 @@
 #include <litmus/affinity.h>
 #endif
 
+#ifdef CONFIG_PGM
+#include <litmus/pgm.h>
+#endif
+
 /* Number of RT tasks that exist in the system */
 atomic_t rt_task_count          = ATOMIC_INIT(0);
 static DEFINE_RAW_SPINLOCK(task_transition_lock);
@@ -286,6 +290,122 @@ asmlinkage long sys_null_call(cycles_t __user *ts)
         return ret;
 }
 
+
+#ifdef CONIG_PGM
+asmlinkage long sys_add_pgm_edge(void* __user *user_edge)
+{
+        int retval = -EINVAL;
+        struct pgm_edge edge;
+        struct task_struct *a, *b;
+
+        if(!access_ok(VERIFY_READ, user_edge, sizeof(struct pgm_edge)))
+                goto out;
+
+        if(__copy_from_user(&edge, user_edge, sizeof(struct pgm_edge)))
+                goto out;
+
+        if(edge.a < 0 || edge.b < 0)
+                goto out;
+
+        read_lock_irq(&tasklist_lock);
+        a = find_task_by_vpid(edge.a);
+        b = find_task_by_vpid(edge.b);
+        read_unlock_irq(&tasklist_lock);        
+                                
+        if(!a || !b)
+                goto out;
+
+        if(!is_realtime(a) || !is_realtime(b))
+                goto out;
+
+        retval = do_add_pgm_edge(a, edge.a_nr_produced,
+                b, edge.b.nr_consumed);
+
+out:
+        return retval;
+}
+
+asmlinkage long sys_pgm_signal_ready(pid_t r)
+{
+        int retval = -EINVAL;
+
+        struct task_struct *root;
+
+        if(p < 0)
+                goto out;
+
+        read_lock_irq(&tasklist_lock);
+        root = find_task_by_vpid(r);
+        read_unlock_irq(&tasklist_lock);
+
+        if(!root)
+                goto out;
+
+        if(!is_realtime(root))
+                goto out;
+
+        if(!is_pgm_src(root))
+                goto out;
+
+        retval = do_pgm_signal_ready(root);
+
+out:
+        return retval;
+}
+
+asmlinkage long sys_pgm_set_nr_children(int nr_children)
+{
+        int retval = -EINVAL;
+
+        if(nr_children <= 0)
+                goto out;
+
+        retval = do_pgm_set_nr_children(nr_children);
+
+out:
+        return retval;
+}
+
+asmlinkage long sys_pgm_wait_for_children(void)
+{
+        int retval = -EINVAL;
+
+        if(!is_realtime(current) || !is_pgm_src(current))
+                goto out;
+
+        retval = do_pgm_wait_for_children();
+
+out:
+        return retval;
+}
+
+
+#else
+asmlinkage long sys_add_pgm_edge(struct pgm_edge __user *edge)
+{
+        return -EINVAL;
+}
+
+asmlinkage long sys_pgm_signal_ready(pid_t r)
+{
+        return -EINVAL;
+}
+
+asmlinkage long sys_pgm_set_nr_children(int nr_children)
+{
+        return -EINVAL;
+}
+
+asmlinkage long sys_pgm_wait_for_children(void)
+{
+        return -EINVAL;
+}
+#endif
+
+
+
+
+
 /* p is a real-time task. Re-init its state as a best-effort task. */
 static void reinit_litmus_state(struct task_struct* p, int restore)
 {
diff --git a/litmus/pgm.c b/litmus/pgm.c
index 6710644eae8e..661fb2065dea 100644
--- a/litmus/pgm.c
+++ b/litmus/pgm.c
@@ -1,48 +1,60 @@
-#include <linux/sched.h>
-#include <linux/bitops.h>
+
+#include <linux/slab.h>
 #include <linux/spinlock.h>
 
-#include <litmus/litmus.h>
+#include <litmus/pgm.h>
 #include <litmus/sched_plugin.h>
-#include <litmus/rt_param.h>
+#include <litmus/jobs.h>
+
+#include <litmus/trace.h>
+
 
+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 inline int __is_pgm_satisfied(struct task_struct* t)
+
+static inline int __is_pgm_satisfied(struct task_struct *t)
 {
         unsigned long unsatisfied =
                 find_first_bit(tsk_rt(t)->pgm_params.consume->unsatisfied,
                                            tsk_rt(t)->pgm_params.consume->nr_consumptions);
-        
-        return (unsatisfied == tsk_rt(t)->pgm_params.consume->nr_consumptions);
+
+        int is_sat = (unsatisfied ==
+                tsk_rt(t)->pgm_params.consume->nr_consumptions);
+
+        TRACE_TASK(t, "is_satisfied = %d (unsatisifed = %x)\n",
+                                        is_sat,
+                                        unsatisfied);
+
+        return (is_sat);
 }
 
-int is_pgm_satisfied(struct task_struct* t)
+
+static void pgm_produce(struct task_struct *t)
 {
-        int is_satisfied;
-        unsigned long flags;
-        
         BUG_ON(!t);
         BUG_ON(!is_realtime(t));
-        BUG_ON(!is_pgm(t) || is_pgm_src(t));
-        
-        spin_lock_irqsave(&tsk_rt(t)->pgm_params.consume->lock, flags);
-        
-        is_satisfied = __is_pgm_satisfied(t);
-        
-        spin_unlock_irqrestore(&tsk_rt(t)->pgm_params.consume->lock, flags);
-        
-        return (is_satisfied);
+        BUG_ON(!is_pgm(t));
+
+        if(!is_pgm_sink(t)) {
+                pgm_produce_intermediate(t);
+        }
+        else {
+                pgm_produce_root(t);
+        }
 }
 
-void pgm_produce(struct task_struct* t)
+
+static void pgm_produce_intermediate(struct task_struct *t)
 {
         unsigned long i;
         struct pgm_produce *prod;
         
-        BUG_ON(!t);
-        BUG_ON(!is_realtime(t));
-        BUG_ON(!is_pgm(t) || is_pgm_sink(t));
-        
+        BUG_ON(is_pgm_sink(t));
+
+        TRACE_TASK(t, "pgm_produce_intermediate()\n");
+
         prod = tsk_rt(t)->pgm_params.produce;
         
         for(i = 0; i < prod->nr_productions; ++i) {
@@ -51,7 +63,7 @@ void pgm_produce(struct task_struct* t)
                 struct pgm_consumption *child_input;
                 int unsatisfied;
                 unsigned long flags;
-
+                int release_child = 0;
                 
                 BUG_ON(!is_pgm(child) || is_pgm_src(child));
                 
@@ -60,25 +72,69 @@ void pgm_produce(struct task_struct* t)
                 spin_lock_irqsave(&tsk_rt(child)->pgm_params.consume->lock, flags);
                 
                 child_input->nr_tokens += to_produce->nr_tokens;
+
+                TRACE_TASK(t, "producing %d tokens to %s/%d (tot now = %d, wants = %d)\n",
+                                                to_produce->nr_tokens,
+                                                child->comm, child->pid,
+                                                child_input->nr_tokens,
+                                                child_input->required);
                 
                 if(child_input->nr_tokens >= child_input->required) {
                         unsatisfied = __test_and_clear_bit(to_produce->production_idx,
                                 tsk_rt(child)->pgm_params.consume->unsatisfied);
                         
                         if(unsatisfied) {
-                                /* we satisfied the child's requirement */
+                                /* we satisfied the child's requirement for this input */
+                                TRACE_TASK(t, "completed a req of %s/%d\n",
+                                        child->comm, child->pid);
+
                                 if(__is_pgm_satisfied(child)) {
-                                        /* all requirements met: release the child */
-                                        
+                                        /* all requirements have been met: release the child */
+
+                                        TRACE_TASK(t, "completed all reqs of %s\%d!\n",
+                                                child->comm, child->pid);
+
+                                        release_child = 1;
                                 }
                         }
+                        else {
+                                TRACE_TASK(t, "%s/%d already had constraint satisifed???\n",
+                                        child->comm, child->pid);
+                        }
                 }
-                
+        
+                /* drop the child's lock before we try to wake it up, otherwise we may
+                 * deadlock with scheduler's locks
+                 */
                 spin_unlock_irqrestore(&tsk_rt(child)->pgm_params.consume->lock, flags);
+
+                if(release_child)
+                {
+                        pgm_consume(child);
+
+                        TRACE_TASK(t, "waking up %s/%d\n", child->comm, child->pid);
+                        litmus->task_wake_up(child);
+                }
         }
 }
 
-void pgm_consume(struct task_struct* t)
+
+static void pgm_produce_root(struct task_struct *t)
+{
+        struct task_struct *root;
+        BUG_ON(!is_pgm_sink(t));
+
+        root = tsk_rt(t)->pgm_params.root;
+
+        TRACE_TASK(t, "waking up PGM root %s/%d\n",
+                root->comm, root->pid);
+
+        /* wake up the new task, the task will sleep if it is too soon. */
+        litmus->task_wake_up(root);
+}
+
+
+static void pgm_consume(struct task_struct *t)
 {
         struct pgm_consume* cons;
         unsigned long i;
@@ -88,6 +144,9 @@ void pgm_consume(struct task_struct* t)
         BUG_ON(!is_realtime(t));
         BUG_ON(!is_pgm(t) || is_pgm_src(t));
         
+        TRACE("%s/%d is consuming tokens! Munch munch... yum!\n",
+                t->comm, t->pid);
+
         cons = tsk_rt(t)->pgm_params.consume;
         
         spin_lock_irqsave(&cons->lock, flags);
@@ -99,33 +158,174 @@ void pgm_consume(struct task_struct* t)
         spin_unlock_irqrestore(&cons->lock, flags);
 }
 
-int pgm_complete(struct task_struct* t)
+
+/**********************************************/
+
+
+int is_pgm_satisfied(struct task_struct *t)
 {
-        struct pgm_consume* cons;
-        unsigned long i;
+        int is_satisfied;
         unsigned long flags;
-        int keep_going = 0;
         
         BUG_ON(!t);
         BUG_ON(!is_realtime(t));
         BUG_ON(!is_pgm(t) || is_pgm_src(t));
         
-        cons = tsk_rt(t)->pgm_params.consume;
+        spin_lock_irqsave(&tsk_rt(t)->pgm_params.consume->lock, flags);
         
-        spin_lock_irqsave(&cons->lock, flags);
+        is_satisfied = __is_pgm_satisfied(t);
         
-        for(i = 0; i < cons->nr_consumptions; ++i) {
-                if(cons->consume[i].nr_tokens <= cons->consume[i].required) {
-                        __set_bit(i, cons->unsatisfied);
-                }
-        }
+        spin_unlock_irqrestore(&tsk_rt(t)->pgm_params.consume->lock, flags);
+        
+        return (is_satisfied);
+}
+
+
+void pgm_prepare_for_next_period(struct task_struct *t)
+{
+        TRACE_TASK(t, "pgm_prepage_for_next_period().\n");
+
+        prepare_for_next_period(t);
+}
+
+
+int pgm_complete_job(struct task_struct *t)
+{
+        struct pgm_consume *cons;
+        unsigned long i;
+        unsigned long flags;
+        int still_satisfied = 0;
         
-        keep_going = __is_pgm_satisfied(t);
+        BUG_ON(!t);
+        BUG_ON(!is_realtime(t));
+        BUG_ON(!is_pgm(t));
+
+        TRACE_TASK(t, "pgm_complete_job().  [src/inter/sink] = [%d/%d/%d]\n",
+                is_pgm_src(t), is_pgm_intermediate(t), is_pgm_sink(t));
+
+        if(!is_pgm_src(t)) {
+
+                TRACE_TASK(t, "re-evaluating constraints\n");
+
+                cons = tsk_rt(t)->pgm_params.consume;
+
+                if(cons->nr_consumptions != 0) {
+                        /* flag constraints as unmet */
+                        spin_lock_irqsave(&cons->lock, flags);
         
-        spin_unlock_irqrestore(&cons->lock, flags);
+                        for(i = 0; i < cons->nr_consumptions; ++i) {
+                                if(cons->consume[i].nr_tokens <= cons->consume[i].required) {
+                                        __set_bit(i, cons->unsatisfied);
+                                }
+                        }
+        
+                        still_satisfied = __is_pgm_satisfied(t);
         
-        return(keep_going);
+                        spin_unlock_irqrestore(&cons->lock, flags);
+                }
+
+                if(still_satisfied) {
+                        TRACE_TASK(t, "all constraints are still satisfied!\n");
+                }
+        }
+
+        /* produce new tokens, possibly waking up children,
+         * or setup for the next PGM root.
+         */
+        pgm_produce(t);
+
+        return(still_satisfied);
 }
 
 
+int do_pgm_add_edge(struct task_struct *a, int nr_gen,
+        struct task_struct *b, int nr_req)
+{
+        int retval = 0;
+        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);
+                memset(tsk_rt(a)->pgm_params.produce, 0, sizeof(pgm_produce));
+        }
+
+        if(tsk_rt(b)->pgm_params.consume == NULL) {
+                tsk_rt(b)->pgm_params.consume =
+                        kmalloc(sizeof(pgm_consume), GFP_KERNEL);
+                memset(tsk_rt(b)->pgm_params.consume, 0, sizeof(pgm_consume));
+
+                spin_lock_init(&(tsk_rt(b)->pgm_params.consume->lock));
+        }
+
+        prod = tsk_rt(a)->pgm_params.produce;
+        cons = tsk_rt(b)->pgm_params.consume;
+
+        if(prod->nr_productions + 1 >= MAX_FAN ||
+           cons->nr_consumptions + 1 >= MAX_FAN) {
+                retval = -EINVAL;
+                goto out;
+        }
+
+        // set up parent
+        prod->produce[prod->nr_productions].child = b;
+        prod->produce[prod->nr_productions].production_idx = cons->nr_consumptions;
+        prod->produce[prod->nr_productions].nr_tokens = nr_gen;
+        ++(prod->nr_productions);
+
+
+        // set up child
+        cons->consume[cons->nr_consumptions].required = nr_req;
+        cons->consume[cons->nr_consumptions].nr_tokens = 0;
+        __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;
+
+out:
+        return retval;
+}
+
+int do_pgm_signal_ready(struct task_struct *root)
+{
+        int retval = -EINVAL;
+
+        tsk_rt(current)->pgm_params.root = root;
+
+        /* decrement the parent's semaphore and then suspend */
+
+        return retval;
+}
+
+int do_pgm_set_nr_children(int nr_children)
+{
+        int retval = -EINVAL;
+
+        /* TODO: set up a counting semaphore */
+
+        return retval;
+}
+
+int do_pgm_wait_for_children(void)
+{
+        int retval = -EINVAL;
+        struct task_struct *root = current;
+
+        /* TODO: sleep on a counting semaphore */
+
+        return retval;
+}
 
diff --git a/litmus/sched_gsn_edf.c b/litmus/sched_gsn_edf.c
index 6ed504f4750e..c1145fc1dc8c 100644
--- a/litmus/sched_gsn_edf.c
+++ b/litmus/sched_gsn_edf.c
@@ -30,6 +30,10 @@
 
 #include <linux/module.h>
 
+#ifdef CONFIG_PGM
+#include <litmus/pgm.h>
+#endif
+
 /* Overview of GSN-EDF operations.
  *
  * For a detailed explanation of GSN-EDF have a look at the FMLP paper. This
@@ -250,12 +254,43 @@ static noinline void requeue(struct task_struct* task)
         /* sanity check before insertion */
         BUG_ON(is_queued(task));
 
+#ifndef CONFIG_PGM
         if (is_released(task, litmus_clock()))
                 __add_ready(&gsnedf, task);
         else {
                 /* it has got to wait */
                 add_release(&gsnedf, task);
         }
+#else
+        /* !!! DON'T ALLOW pgm_src NODES TO REQUEUE AFTER COMPLETION !!!
+         * !!! They are enqueued by their pgm_sink nodes             !!!
+         */
+        if (!is_pgm(task) || is_pgm_src(task)) {
+                if (is_released(task, litmus_clock())) {
+                        __add_ready(&gsnedf, task);
+                }
+                else {
+                        /* it has got to wait */
+                        add_release(&gsnedf, task);
+                }
+        }
+        else {
+                if (is_pgm_intermediate(task)) {
+                        if (is_pgm_satisfied(task)) {
+                                __add_ready(&gsnedf, task);
+                        }
+                        else {
+                                /* put in no queue.  parent will wake it up since
+                                   we are doing early releasing. */
+                                WARN_ON(1);
+                        }
+                }
+                else {
+                        /* should be unreachable */
+                        WARN_ON(1);
+                }
+        }
+#endif
 }
 
 #ifdef CONFIG_SCHED_CPU_AFFINITY
@@ -341,16 +376,44 @@ static noinline void job_completion(struct task_struct *t, int forced)
 
         /* set flags */
         set_rt_flags(t, RT_F_SLEEP);
+
+#ifndef CONFIG_PGM
         /* prepare for next period */
         prepare_for_next_period(t);
-        if (is_released(t, litmus_clock()))
+
+        if(is_released(t, litmus_clock())) {
                 sched_trace_task_release(t);
+        }
+
         /* unlink */
         unlink(t);
         /* requeue
          * But don't requeue a blocking task. */
         if (is_running(t))
                 gsnedf_job_arrival(t);
+
+#else
+        if(!is_pgm(t)) {
+                /* prepare for next period */
+                prepare_for_next_period(t);
+
+                if(is_released(t, litmus_clock())) {
+                        sched_trace_task_release(t);
+                }
+        }
+        else {
+                /* prepare for next period */
+                pgm_prepare_for_next_period(t);
+        }
+
+        /* unlink */
+        unlink(t);
+        /* requeue
+         * But don't requeue a blocking task, or pgm tasks */
+        if (!is_pgm(t) && is_running(t)) { 
+                gsnedf_job_arrival(t);
+        }
+#endif
 }
 
 /* gsnedf_tick - this function is called for every local timer