dynamic tasks

DynamicTaskOp.scala

  class Task[A] {
    type self = Task
    
    // Should be a staged block
    val block: Exp[A]

    // All dependencies must be Tasks that have been previously submitted, including
    // Tasks submitted before this Task in the current generation and the previous
    // generation
    val dependencies : Array[self]
  }

  /**
   * A dynamic task. Retrieve tasks by calling tasks. When done is true and there are no more tasks, the task is finished.
   */
  case class DeliteDynamicTask[A] {
    tasks: Rep[List[Task[V]]]
    done: Exp[Boolean]
    result: Exp[A]
  } extends DeliteOp[A]

GeneratedTaskOp.scala

class Task[A] {
    type self = Task
    
    // A series of kernels, that return A
    val block: A
    
    // All dependencies must be Tasks that have been previously submitted, including
    // Tasks submitted before this Task in the current generation and the previous
    // generation
    val dependencies : Array[self]

    def status : READY / INFLIGHT / DONE
  }

 /**
   * A dynamic task. Retrieve tasks by calling tasks. When done is true and there are no more tasks, the task is finished.
   */
  class DeliteDynamicTask[A] {
    tasks: List[Task[V]]
    done: Boolean
    result: A
  } extends DeliteOp[A]

// What the scheduler should do. It should keep asking the dynamic task for tasks until it says it is done
  {
    val scheduler : DeliteRuntimeScheduler
    
    val dynamic : DeliteDynamicTask

    /* This is a polling based approach. */
    while(dynamic.hasTasksInFlight || !dynamic.done) {
      // Throw an exception or something if task definitions are invalid...
      scheduler.addTasks(dynamic.tasks)
      
      // Run tasks that have no unsatisfied dependencies
      // Tasks MAY ask for multiple threads
      scheduler.submitReadyTasks()
    }

    /* This is an interrupt based approach */
    while(dynamic.hasTasksInFlight || !dynamic.done) {
      try {
        dynamic.wait
      }
      catch(InterruptedException e) {
        // Throw an exception or something if task definitions are invalid...
        scheduler.addTasks(dynamic.tasks)
        
        // Run tasks that have no unsatisfied dependencies
        scheduler.submitReadyTasks()
      }
    }
    
    dynamic.result
  }