Fair Scheduling on a Priority Queue

I'm in St. Luis for a math conference! The time zone is different, but it's still after midnight. This merges the past two projects, implementing the fair scheduler on the priority queue (which should be a performance gain, in theory).

Here is the (untested!) code.

(* Priority Queue *)

(* Tasks are of type unit -> unit *)
type priority_queue = {mutable tasklist: ((unit -> unit) * float) option array; mutable next: int}

let new_priority_queue = {tasklist = Array.make 2 None; next = 1}

let rec sift_up place pq =
  if (place > 1) then match (pq.tasklist.(place), pq.tasklist.(place/2)) with
    (None, p) -> ()
  | (Some (t1,f1), Some (t2,f2)) when f1 <= f2 -> ()
  | (Some (t1,f1), Some (t2,f2)) when f1 > f2 ->
      pq.tasklist.(place)   <- Some (t2, f2);
      pq.tasklist.(place/2) <- Some (t1, f1);
      sift_up (place/2) pq;
  | (q, r) -> raise (Invalid_argument "sift_up");;

let double_size pq =
  pq.tasklist <- Array.append pq.tasklist (Array.make (Array.length pq.tasklist) None);;

let push_item item priority pq =
  (if not (pq.next < Array.length pq.tasklist) then double_size pq);
  pq.tasklist.(pq.next) <- Some (item, priority);
  sift_up pq.next pq;
  pq.next <- (pq.next + 1);;

let rec sift_down place pq =
  if place*2 < Array.length pq.tasklist then
    if place*2+1 < pq.next then
      let (largervalue,largerplace) =
        (match (pq.tasklist.(place*2), pq.tasklist.(place*2+1)) with
          (Some (t1,f1), Some (t2,f2)) when f1 < f2 -> (f2,place*2+1)
        | (Some (t1,f1), Some (t2,f2))              -> (f1,place*2)
        | _ -> raise (Invalid_argument "sift_down") )
      in
      match pq.tasklist.(place) with
        Some (t,f) when f < largervalue ->
          (pq.tasklist.(place) <- pq.tasklist.(largerplace);
           pq.tasklist.(largerplace) <- Some (t,f);
           sift_down largerplace pq )
      | Some (t,f) -> ()
      | _ -> raise (Invalid_argument "sift_down")
    else if place*2 < pq.next then
      match pq.tasklist.(place) with
        Some (t,f) when f < (match pq.tasklist.(place*2) with Some (t,f) -> f | _ -> 0.0) ->
          (pq.tasklist.(place) <- pq.tasklist.(place*2);
           pq.tasklist.(place*2) <- Some (t,f) )
      | Some (t,f) -> ()
      | _ -> raise (Invalid_argument "sift_down");;


let pop_item pq =
  let result = pq.tasklist.(1) in
  pq.tasklist.(1) <- pq.tasklist.(pq.next-1);
  pq.tasklist.(pq.next-1) <- None;
  if pq.next > 1 then pq.next <- pq.next-1;
  sift_down 1 pq;
  result;;

let loop_items pq =
  while not (pq.tasklist.(1) = None) do
    match pop_item pq with
      Some (item,f) ->
    item();
    |  None -> ()
  done;;

(* Fair Scheduler, implemented as a persistent codelet *)

type supertask_data = {parentq: priority_queue;
            childq: priority_queue;
            ratio: float;
            mutable priority: float}

let rec supertask data unit =
  let starttime =
    Unix.gettimeofday() in
  (match (pop_item data.childq) with
    Some (item, f) ->
      item();
  | None -> ());
  let tasktime =
    (Unix.gettimeofday() -. starttime) /. data.ratio  in
  data.priority <- (data.priority -. tasktime);
  push_item
    (supertask data)
    data.priority
    data.parentq;;