src/conductor/dispatch.go

   1 /* dispatch.go
   2 */
   3
   4 package main
   5
   6 import (
   7         "container/list"
   8         o "orchestra"
   9 )
  10
  11 func NewRequest() (req *JobRequest) {
  12         req = NewJobRequest()
  13
  14         return req
  15 }
  16
  17 const messageBuffer = 10
  18
  19 var newJob              = make(chan *JobRequest, messageBuffer)
  20 var rqTask              = make(chan *TaskRequest, messageBuffer)
  21 var playerIdle          = make(chan *ClientInfo, messageBuffer)
  22 var playerDead          = make(chan *ClientInfo, messageBuffer)
  23 var statusRequest       = make(chan(chan *QueueInformation))
  24
  25 func PlayerWaitingForJob(player *ClientInfo) {
  26         playerIdle <- player
  27 }
  28
  29 func PlayerDied(player *ClientInfo) {
  30         playerDead <- player
  31 }
  32
  33 func DispatchTask(task *TaskRequest) {
  34         rqTask <- task
  35 }
  36
  37 type QueueInformation struct {
  38         idlePlayers     []string
  39         waitingTasks    int
  40 }
  41
  42 func DispatchStatus() (waitingTasks int, waitingPlayers []string) {
  43         r := make(chan *QueueInformation)
  44
  45         statusRequest <- r
  46         s := <- r
  47
  48         return s.waitingTasks, s.idlePlayers
  49 }
  50
  51 func InitDispatch() {
  52         go masterDispatch(); // go!
  53 }
  54
  55 func QueueJob(job *JobRequest) {
  56         /* first, allocate the Job it's ID */
  57         job.Id = NextRequestId()
  58         /* first up, split the job up into it's tasks. */
  59         job.Tasks = job.MakeTasks()
  60         /* add it to the registry */
  61         JobAdd(job)
  62         /* an enqueue all of the tasks */
  63         for i := range job.Tasks {
  64                 DispatchTask(job.Tasks[i])
  65         }
  66 }
  67
  68 func masterDispatch() {
  69         pq := list.New()
  70         tq := list.New()
  71
  72         for {
  73                 select {
  74                 case player := <-playerIdle:
  75                         o.Debug("Dispatch: Player")
  76                         /* first, scan to see if we have anything for this player */
  77                         i := tq.Front()
  78                         for {
  79                                 if (nil == i) {
  80                                         /* Out of items! */
  81                                         /* Append this player to the waiting players queue */
  82                                         pq.PushBack(player)
  83                                         break;
  84                                 }
  85                                 t,_ := i.Value.(*TaskRequest)
  86                                 if t.IsTarget(player.Player) {
  87                                         /* Found a valid job. Send it to the player, and remove it from our pending
  88                                          * list */
  89                                         tq.Remove(i)
  90                                         player.TaskQ <- t
  91                                         break;
  92                                 }
  93                                 i = i.Next()
  94                         }
  95                 case player := <-playerDead:
  96                         o.Debug("Dispatch: Dead Player")
  97                         for i := pq.Front(); i != nil; i = i.Next() {
  98                                 p, _ := i.Value.(*ClientInfo)
  99                                 if player.Player == p.Player {
 100                                         pq.Remove(i)
 101                                         break;
 102                                 }
 103                         }
 104                 case task := <-rqTask:
 105                         o.Debug("Dispatch: Task")
 106                         /* first, scan to see if we have valid pending player for this task */
 107                         i := pq.Front()
 108                         for {
 109                                 if (nil == i) {
 110                                         /* Out of players! */
 111                                         /* Append this task to the waiting tasks queue */
 112                                         tq.PushBack(task)
 113                                         break;
 114                                 }
 115                                 p,_ := i.Value.(*ClientInfo)
 116                                 if task.IsTarget(p.Player) {
 117                                         /* Found it. */
 118                                         pq.Remove(i)
 119                                         p.TaskQ <- task
 120                                         break;
 121                                 }
 122                                 i = i.Next();
 123                         }
 124                 case respChan := <-statusRequest:
 125                         o.Debug("Status!")
 126                         response := new(QueueInformation)
 127                         response.waitingTasks = tq.Len()
 128                         pqLen := pq.Len()
 129                         response.idlePlayers = make([]string, pqLen)
 130
 131                         idx := 0
 132                         for i := pq.Front(); i != nil; i = i.Next() {
 133                                 player,_ := i.Value.(*ClientInfo)
 134                                 response.idlePlayers[idx] = player.Player
 135                                 idx++
 136                         }
 137                         respChan <- response
 138                 }
 139         }
 140 }