src/player/registry.go

   1 // registry.go
   2 //
   3 // Job Registry.
   4 //
   5 // The Registry provides a 'threadsafe' interface to various global
   6 // information stores.
   7 //
   8 // The registry dispatch thread is forbidden from performing any work
   9 // that is likely to block.  Result channels must be buffered with
  10 // enough space for the full set of results.
  11
  12 package main
  13
  14 const (
  15         requestAddTask                  = iota
  16         requestGetTask
  17
  18         requestQueueSize                = 10
  19 )
  20
  21 type registryRequest struct {
  22         operation               int
  23         id                      uint64
  24         task                    *TaskRequest
  25         responseChannel         chan *registryResponse
  26 }
  27
  28 type registryResponse struct {
  29         success                 bool
  30         task                    *TaskRequest
  31 }
  32
  33 var chanRequest = make(chan *registryRequest, requestQueueSize)
  34
  35 // bake a minimal request structure together.
  36 func newRequest(wants_response bool) (r *registryRequest) {
  37         r = new(registryRequest)
  38         if wants_response {
  39                 r.responseChannel = make(chan *registryResponse, 1)
  40         }
  41
  42         return r
  43 }
  44
  45 // Add a Task to the registry.  Return true if successful, returns
  46 // false if the task is lacking critical information (such as a Job Id)
  47 // and can't be registered.
  48 func TaskAdd(task *TaskRequest) bool {
  49         rr := newRequest(true)
  50         rr.operation = requestAddTask
  51         rr.task = task
  52
  53         chanRequest <- rr
  54         resp := <- rr.responseChannel
  55         return resp.success
  56 }
  57
  58 // Get a Task from the registry.  Returns the task if successful,
  59 // returns nil if the task couldn't be found.
  60 func TaskGet(id uint64) *TaskRequest {
  61         rr := newRequest(true)
  62         rr.operation = requestGetTask
  63         rr.id = id
  64
  65         chanRequest <- rr
  66         resp := <- rr.responseChannel
  67         return resp.task
  68 }
  69
  70 func manageRegistry() {
  71         taskRegister := make(map[uint64]*TaskRequest)
  72
  73         for {
  74                 req := <- chanRequest
  75                 resp := new (registryResponse)
  76                 switch (req.operation) {
  77                 case requestAddTask:
  78                         if nil != req.task {
  79                                 // and register the job
  80                                 taskRegister[req.task.Id] = req.task
  81                                 resp.success = true
  82                         } else {
  83                                 resp.success = false
  84                         }
  85                 case requestGetTask:
  86                         task, exists := taskRegister[req.id]
  87                         resp.success = exists
  88                         if exists {
  89                                 resp.task = task
  90                         }
  91                 }
  92                 if req.responseChannel != nil {
  93                         req.responseChannel <- resp
  94                 }
  95         }
  96 }
  97
  98 func init() {
  99         go manageRegistry()
 100 }