1 files changed, 127 insertions, 98 deletions

diff --git a/litmus/sched_pfair.c b/litmus/sched_pfair.c
index 0a64273daa47..16f1065bbdca 100644
--- a/litmus/sched_pfair.c
+++ b/litmus/sched_pfair.c
@@ -1,7 +1,8 @@
 /*
  * kernel/sched_pfair.c
  *
- * Implementation of the (global) Pfair scheduling algorithm.
+ * Implementation of the PD^2 pfair scheduling algorithm. This
+ * implementation realizes "early releasing," i.e., it is work-conserving.
  *
  */
 
@@ -76,36 +77,29 @@ struct pfair_state {
         struct task_struct* local;     /* the local copy of linked          */
         struct task_struct* scheduled; /* what is actually scheduled        */
 
-        unsigned long missed_quanta;
         lt_t offset;                    /* stagger offset */
+        unsigned int missed_updates;
+        unsigned int missed_quanta;
 };
 
-/* Currently, we limit the maximum period of any task to 2000 quanta.
- * The reason is that it makes the implementation easier since we do not
- * need to reallocate the release wheel on task arrivals.
- * In the future
- */
-#define PFAIR_MAX_PERIOD 2000
-
 struct pfair_cluster {
         struct scheduling_cluster topology;
 
         /* The "global" time in this cluster. */
         quanta_t pfair_time; /* the "official" PFAIR clock */
-        quanta_t merge_time; /* Updated after the release queue has been
-                              * merged. Used by drop_all_references().
-                              */
 
         /* The ready queue for this cluster. */
         rt_domain_t pfair;
 
-        /* This is the release queue wheel for this cluster. It is indexed by
-         * pfair_time % PFAIR_MAX_PERIOD.  Each heap is ordered by PFAIR
-         * priority, so that it can be merged with the ready queue.
+        /* The set of jobs that should have their release enacted at the next
+         * quantum boundary.
          */
-        struct bheap release_queue[PFAIR_MAX_PERIOD];
+        struct bheap release_queue;
+        raw_spinlock_t release_lock;
 };
 
+#define RT_F_REQUEUE 0x2
+
 static inline struct pfair_cluster* cpu_cluster(struct pfair_state* state)
 {
         return container_of(state->topology.cluster, struct pfair_cluster, topology);
@@ -121,6 +115,11 @@ static inline struct pfair_state* from_cluster_list(struct list_head* pos)
         return list_entry(pos, struct pfair_state, topology.cluster_list);
 }
 
+static inline struct pfair_cluster* from_domain(rt_domain_t* rt)
+{
+        return container_of(rt, struct pfair_cluster, pfair);
+}
+
 static inline raw_spinlock_t* cluster_lock(struct pfair_cluster* cluster)
 {
         /* The ready_lock is used to serialize all scheduling events. */
@@ -161,21 +160,11 @@ static quanta_t cur_deadline(struct task_struct* t)
         return cur_subtask(t)->deadline +  tsk_pfair(t)->release;
 }
 
-
-static quanta_t cur_sub_release(struct task_struct* t)
-{
-        return cur_subtask(t)->release +  tsk_pfair(t)->release;
-}
-
 static quanta_t cur_release(struct task_struct* t)
 {
-#ifdef EARLY_RELEASE
-        /* only the release of the first subtask counts when we early
-         * release */
+        /* This is early releasing: only the release of the first subtask
+         * counts. */
         return tsk_pfair(t)->release;
-#else
-        return cur_sub_release(t);
-#endif
 }
 
 static quanta_t cur_overlap(struct task_struct* t)
@@ -235,11 +224,16 @@ int pfair_ready_order(struct bheap_node* a, struct bheap_node* b)
         return pfair_higher_prio(bheap2task(a), bheap2task(b));
 }
 
-/* return the proper release queue for time t */
-static struct bheap* relq(struct pfair_cluster* cluster, quanta_t t)
+static void pfair_release_jobs(rt_domain_t* rt, struct bheap* tasks)
 {
-        struct bheap* rq = cluster->release_queue + (t % PFAIR_MAX_PERIOD);
-        return rq;
+        struct pfair_cluster* cluster = from_domain(rt);
+        unsigned long flags;
+
+        raw_spin_lock_irqsave(&cluster->release_lock, flags);
+
+        bheap_union(pfair_ready_order, &cluster->release_queue, tasks);
+
+        raw_spin_unlock_irqrestore(&cluster->release_lock, flags);
 }
 
 static void prepare_release(struct task_struct* t, quanta_t at)
@@ -248,25 +242,12 @@ static void prepare_release(struct task_struct* t, quanta_t at)
         tsk_pfair(t)->cur        = 0;
 }
 
-static void __pfair_add_release(struct task_struct* t, struct bheap* queue)
-{
-        bheap_insert(pfair_ready_order, queue,
-                    tsk_rt(t)->heap_node);
-}
-
-static void pfair_add_release(struct pfair_cluster* cluster,
-                              struct task_struct* t)
-{
-        BUG_ON(bheap_node_in_heap(tsk_rt(t)->heap_node));
-        __pfair_add_release(t, relq(cluster, cur_release(t)));
-}
-
 /* pull released tasks from the release queue */
-static void poll_releases(struct pfair_cluster* cluster,
-                          quanta_t time)
+static void poll_releases(struct pfair_cluster* cluster)
 {
-        __merge_ready(&cluster->pfair, relq(cluster, time));
-        cluster->merge_time = time;
+        raw_spin_lock(&cluster->release_lock);
+        __merge_ready(&cluster->pfair, &cluster->release_queue);
+        raw_spin_unlock(&cluster->release_lock);
 }
 
 static void check_preempt(struct task_struct* t)
@@ -292,16 +273,12 @@ static void drop_all_references(struct task_struct *t)
 {
         int cpu;
         struct pfair_state* s;
-        struct bheap* q;
         struct pfair_cluster* cluster;
         if (bheap_node_in_heap(tsk_rt(t)->heap_node)) {
-                /* figure out what queue the node is in */
+                /* It must be in the ready queue; drop references isn't called
+                 * when the job is in a release queue. */
                 cluster = tsk_pfair(t)->cluster;
-                if (time_before_eq(cur_release(t), cluster->merge_time))
-                        q = &cluster->pfair.ready_queue;
-                else
-                        q = relq(cluster, cur_release(t));
-                bheap_delete(pfair_ready_order, q,
+                bheap_delete(pfair_ready_order, &cluster->pfair.ready_queue,
                             tsk_rt(t)->heap_node);
         }
         for (cpu = 0; cpu < num_online_cpus(); cpu++) {
@@ -313,6 +290,17 @@ static void drop_all_references(struct task_struct *t)
                 if (s->scheduled  == t)
                         s->scheduled = NULL;
         }
+        /* make sure we don't have a stale linked_on field */
+        tsk_rt(t)->linked_on = NO_CPU;
+}
+
+static void pfair_prepare_next_period(struct task_struct* t)
+{
+        struct pfair_param* p = tsk_pfair(t);
+
+        prepare_for_next_period(t);
+        get_rt_flags(t) = RT_F_RUNNING;
+        p->release += p->period;
 }
 
 /* returns 1 if the task needs to go the release queue */
@@ -322,30 +310,26 @@ static int advance_subtask(quanta_t time, struct task_struct* t, int cpu)
         int to_relq;
         p->cur = (p->cur + 1) % p->quanta;
         if (!p->cur) {
-                sched_trace_task_completion(t, 1);
                 if (tsk_rt(t)->present) {
-                        /* we start a new job */
-                        prepare_for_next_period(t);
-                        sched_trace_task_release(t);
-                        get_rt_flags(t) = RT_F_RUNNING;
-                        p->release += p->period;
+                        /* The job overran; we start a new budget allocation. */
+                        pfair_prepare_next_period(t);
                 } else {
                         /* remove task from system until it wakes */
                         drop_all_references(t);
+                        tsk_rt(t)->flags = RT_F_REQUEUE;
                         TRACE_TASK(t, "on %d advanced to subtask %lu (not present)\n",
                                    cpu, p->cur);
                         return 0;
                 }
         }
         to_relq = time_after(cur_release(t), time);
-        TRACE_TASK(t, "on %d advanced to subtask %lu -> to_relq=%d\n",
-                   cpu, p->cur, to_relq);
+        TRACE_TASK(t, "on %d advanced to subtask %lu -> to_relq=%d (cur_release:%lu time:%lu)\n",
+                   cpu, p->cur, to_relq, cur_release(t), time);
         return to_relq;
 }
 
 static void advance_subtasks(struct pfair_cluster *cluster, quanta_t time)
 {
-        int missed;
         struct task_struct* l;
         struct pfair_param* p;
         struct list_head* pos;
@@ -354,14 +338,17 @@ static void advance_subtasks(struct pfair_cluster *cluster, quanta_t time)
         list_for_each(pos, &cluster->topology.cpus) {
                 cpu = from_cluster_list(pos);
                 l = cpu->linked;
-                missed = cpu->linked != cpu->local;
+                cpu->missed_updates += cpu->linked != cpu->local;
                 if (l) {
                         p = tsk_pfair(l);
                         p->last_quantum = time;
                         p->last_cpu     =  cpu_id(cpu);
                         if (advance_subtask(time, l, cpu_id(cpu))) {
-                                cpu->linked = NULL;
-                                pfair_add_release(cluster, l);
+                                //cpu->linked = NULL;
+                                PTRACE_TASK(l, "should go to release queue. "
+                                            "scheduled_on=%d present=%d\n",
+                                            tsk_rt(l)->scheduled_on,
+                                            tsk_rt(l)->present);
                         }
                 }
         }
@@ -445,6 +432,11 @@ static void schedule_subtasks(struct pfair_cluster *cluster, quanta_t time)
         list_for_each(pos, &cluster->topology.cpus) {
                 cpu_state = from_cluster_list(pos);
                 retry = 1;
+#ifdef CONFIG_RELEASE_MASTER
+                /* skip release master */
+                if (cluster->pfair.release_master == cpu_id(cpu_state))
+                        continue;
+#endif
                 while (retry) {
                         if (pfair_higher_prio(__peek_ready(&cluster->pfair),
                                               cpu_state->linked))
@@ -471,13 +463,13 @@ static void schedule_next_quantum(struct pfair_cluster *cluster, quanta_t time)
         sched_trace_quantum_boundary();
 
         advance_subtasks(cluster, time);
-        poll_releases(cluster, time);
+        poll_releases(cluster);
         schedule_subtasks(cluster, time);
 
         list_for_each(pos, &cluster->topology.cpus) {
                 cpu = from_cluster_list(pos);
                 if (cpu->linked)
-                        PTRACE_TASK(pstate[cpu]->linked,
+                        PTRACE_TASK(cpu->linked,
                                     " linked on %d.\n", cpu_id(cpu));
                 else
                         PTRACE("(null) linked on %d.\n", cpu_id(cpu));
@@ -612,12 +604,42 @@ static int safe_to_schedule(struct task_struct* t, int cpu)
 static struct task_struct* pfair_schedule(struct task_struct * prev)
 {
         struct pfair_state* state = &__get_cpu_var(pfair_state);
-        int blocks;
+        struct pfair_cluster* cluster = cpu_cluster(state);
+        int blocks, completion, out_of_time;
         struct task_struct* next = NULL;
 
+#ifdef CONFIG_RELEASE_MASTER
+        /* Bail out early if we are the release master.
+         * The release master never schedules any real-time tasks.
+         */
+        if (unlikely(cluster->pfair.release_master == cpu_id(state))) {
+                sched_state_task_picked();
+                return NULL;
+        }
+#endif
+
         raw_spin_lock(cpu_lock(state));
 
-        blocks  = is_realtime(prev) && !is_running(prev);
+        blocks      = is_realtime(prev) && !is_running(prev);
+        completion  = is_realtime(prev) && get_rt_flags(prev) == RT_F_SLEEP;
+        out_of_time = is_realtime(prev) && time_after(cur_release(prev),
+                                                      state->local_tick);
+
+        if (is_realtime(prev))
+            PTRACE_TASK(prev, "blocks:%d completion:%d out_of_time:%d\n",
+                        blocks, completion, out_of_time);
+
+        if (completion) {
+                sched_trace_task_completion(prev, 0);
+                pfair_prepare_next_period(prev);
+                prepare_release(prev, cur_release(prev));
+        }
+
+        if (!blocks && (completion || out_of_time)) {
+                drop_all_references(prev);
+                sched_trace_task_release(prev);
+                add_release(&cluster->pfair, prev);
+        }
 
         if (state->local && safe_to_schedule(state->local, cpu_id(state)))
                 next = state->local;
@@ -649,13 +671,19 @@ static void pfair_task_new(struct task_struct * t, int on_rq, int running)
         cluster = tsk_pfair(t)->cluster;
 
         raw_spin_lock_irqsave(cluster_lock(cluster), flags);
-        if (running)
-                t->rt_param.scheduled_on = task_cpu(t);
-        else
-                t->rt_param.scheduled_on = NO_CPU;
 
         prepare_release(t, cluster->pfair_time + 1);
-        pfair_add_release(cluster, t);
+
+        t->rt_param.scheduled_on = NO_CPU;
+
+        if (running) {
+#ifdef CONFIG_RELEASE_MASTER
+                if (task_cpu(t) != cluster->pfair.release_master)
+#endif
+                        t->rt_param.scheduled_on = task_cpu(t);
+                __add_ready(&cluster->pfair, t);
+        }
+
         check_preempt(t);
 
         raw_spin_unlock_irqrestore(cluster_lock(cluster), flags);
@@ -665,6 +693,7 @@ static void pfair_task_wake_up(struct task_struct *t)
 {
         unsigned long flags;
         lt_t now;
+        int requeue = 0;
         struct pfair_cluster* cluster;
 
         cluster = tsk_pfair(t)->cluster;
@@ -679,13 +708,20 @@ static void pfair_task_wake_up(struct task_struct *t)
          * (as if it never blocked at all). Otherwise, we have a
          * new sporadic job release.
          */
+        requeue = tsk_rt(t)->flags == RT_F_REQUEUE;
         now = litmus_clock();
         if (lt_before(get_deadline(t), now)) {
+                TRACE_TASK(t, "sporadic release!\n");
                 release_at(t, now);
                 prepare_release(t, time2quanta(now, CEIL));
                 sched_trace_task_release(t);
-                /* FIXME: race with pfair_time advancing */
-                pfair_add_release(cluster, t);
+        }
+
+        /* only add to ready queue if the task isn't still linked somewhere */
+        if (requeue) {
+                TRACE_TASK(t, "requeueing required\n");
+                tsk_rt(t)->flags = RT_F_RUNNING;
+                __add_ready(&cluster->pfair, t);
         }
 
         check_preempt(t);
@@ -744,15 +780,11 @@ static void pfair_release_at(struct task_struct* task, lt_t start)
         release_at(task, start);
         release = time2quanta(start, CEIL);
 
-        /* FIXME: support arbitrary offsets. */
-        if (release - cluster->pfair_time >= PFAIR_MAX_PERIOD)
-                release = cluster->pfair_time + PFAIR_MAX_PERIOD;
-
         TRACE_TASK(task, "sys release at %lu\n", release);
 
         drop_all_references(task);
         prepare_release(task, release);
-        pfair_add_release(cluster, task);
+        add_release(&cluster->pfair, task);
 
         raw_spin_unlock_irqrestore(cluster_lock(cluster), flags);
 }
@@ -834,13 +866,6 @@ static long pfair_admit_task(struct task_struct* t)
                        "The period of %s/%d is not a multiple of %llu.\n",
                        t->comm, t->pid, (unsigned long long) quantum_length);
 
-        if (period >= PFAIR_MAX_PERIOD) {
-                printk(KERN_WARNING
-                       "PFAIR: Rejecting task %s/%d; its period is too long.\n",
-                       t->comm, t->pid);
-                return -EINVAL;
-        }
-
         if (quanta == period) {
                 /* special case: task has weight 1.0 */
                 printk(KERN_INFO
@@ -880,12 +905,9 @@ static long pfair_admit_task(struct task_struct* t)
 
 static void pfair_init_cluster(struct pfair_cluster* cluster)
 {
-        int i;
-
-        /* initialize release queue */
-        for (i = 0; i < PFAIR_MAX_PERIOD; i++)
-                bheap_init(&cluster->release_queue[i]);
-        rt_domain_init(&cluster->pfair, pfair_ready_order, NULL, NULL);
+        rt_domain_init(&cluster->pfair, pfair_ready_order, NULL, pfair_release_jobs);
+        bheap_init(&cluster->release_queue);
+        raw_spin_lock_init(&cluster->release_lock);
         INIT_LIST_HEAD(&cluster->topology.cpus);
 }
 
@@ -899,8 +921,11 @@ static void cleanup_clusters(void)
         num_pfair_clusters = 0;
 
         /* avoid stale pointers */
-        for (i = 0; i < NR_CPUS; i++)
+        for (i = 0; i < num_online_cpus(); i++) {
                 pstate[i]->topology.cluster = NULL;
+                printk("P%d missed %u updates and %u quanta.\n", cpu_id(pstate[i]),
+                       pstate[i]->missed_updates, pstate[i]->missed_quanta);
+        }
 }
 
 static long pfair_activate_plugin(void)
@@ -936,6 +961,9 @@ static long pfair_activate_plugin(void)
                 pfair_init_cluster(cluster);
                 cluster->pfair_time = now;
                 clust[i] = &cluster->topology;
+#ifdef CONFIG_RELEASE_MASTER
+                cluster->pfair.release_master = atomic_read(&release_master_cpu);
+#endif
         }
 
         for (i = 0; i < num_online_cpus(); i++)  {
@@ -943,6 +971,7 @@ static long pfair_activate_plugin(void)
                 state->cur_tick   = now;
                 state->local_tick = now;
                 state->missed_quanta = 0;
+                state->missed_updates = 0;
                 state->offset     = cpu_stagger_offset(i);
                 printk(KERN_ERR "cpus[%d] set; %d\n", i, num_online_cpus());
                 cpus[i] = &state->topology;