CPAN-Blame/golang/analysis-scheduler.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "flag"
   6         "fmt"
   7         "github.com/garyburd/redigo/redis"
   8         "io/ioutil"
   9         "log"
  10         "net"
  11         "net/http"
  12         "regexp"
  13 )
  14
  15 // {"Id":"File-Locate-Iterator-21","CalcNext":1,"CalcTimestamp":0,"CalcNextETA":0,"DebugRating":500}
  16 type StateType struct {
  17         Id                string
  18         CalcNext          uint8 // not_scheduled(0) | scheduled(1) | in_progress(2)
  19         CalcTimestamp     uint64
  20         CalcNextETA       uint64
  21         Pass              uint32
  22         Fail              uint32
  23         DebugRating       uint32
  24         DebugRunningStart uint32
  25         DebugRunningSetId uint32
  26 }
  27
  28 // {"distv":"File-Locate-Iterator-21","start":1397696507}
  29 type JobqueueJobsXDescr struct {
  30         Distv string
  31         Start uint32
  32 }
  33
  34 // {"Id":"nonsense-0.01","VoteAccepted":true,"VoteRefusedBecause":nil,...
  35 type VoteType struct {
  36         Id                string
  37         VoteAccepted      bool
  38         VoteRefusedReason string
  39 }
  40
  41 // {"Id":"nonsense-0.01","LcPass":123,"LcFail":13,...
  42 type LastCalcType struct {
  43         Id     string
  44         LcPass uint32
  45         LcFail uint32
  46 }
  47
  48 func voteHandler(w http.ResponseWriter, r *http.Request, title string) {
  49         c, err := redis.Dial("tcp", ":6379")
  50         if err != nil {
  51                 panic(err)
  52         }
  53         defer c.Close()
  54
  55         s := new(StateType)
  56         s.Id = title
  57         set_passfail(c, s)
  58         set_calcnext(c, s)
  59
  60         l := new(LastCalcType)
  61         l.Id = s.Id
  62         set_lcpassfail(c, l)
  63
  64         v := new(VoteType)
  65         v.Id = s.Id
  66
  67         mustenqueue := false
  68         if s.CalcNext == 1 { // already scheduled unsure what the
  69                 // right thing to do is here; thinking about it hurts
  70                 // and costs time, let's discuss this later
  71                 v.VoteAccepted = true // but silently doing nothing
  72         } else if s.CalcNext == 2 {
  73                 v.VoteAccepted = false
  74                 v.VoteRefusedReason = "Calculation already running, voting closed"
  75         } else if l.LcPass == 0 && l.LcFail == 0 {
  76                 mustenqueue = true
  77         } else if l.LcPass == s.Pass && l.LcFail == s.Fail {
  78                 v.VoteAccepted = false
  79                 v.VoteRefusedReason = "No new results available that would justify a new calc"
  80         } else {
  81                 mustenqueue = true
  82         }
  83         if mustenqueue {
  84                 _, err := redis.Int(c.Do("ZADD", "analysis:jobqueue:q", 10000, title))
  85                 if err != nil {
  86                         v.VoteAccepted = false
  87                         v.VoteRefusedReason = "-ERR"
  88                 } else {
  89                         v.VoteAccepted = true
  90                 }
  91         }
  92
  93         b, err := json.Marshal(v)
  94         fmt.Fprintf(w, "%s\n", b)
  95 }
  96
  97 func stateHandler(w http.ResponseWriter, r *http.Request, title string) {
  98         c, err := redis.Dial("tcp", ":6379")
  99         if err != nil {
 100                 panic(err)
 101         }
 102         defer c.Close()
 103
 104         s := new(StateType)
 105         s.Id = title
 106
 107         set_passfail(c, s)
 108         set_calcnext(c, s)
 109         set_calctimestamp(c, s)
 110         set_eta(c, s)
 111
 112         b, err := json.Marshal(s)
 113         fmt.Fprintf(w, "%s\n", b)
 114 }
 115
 116 func set_calctimestamp(c redis.Conn, s *StateType) {
 117         calctimestamp := -1 // initialize to illegal value to detect nil.
 118         calctimestamp, err := redis.Int(c.Do("ZSCORE", "analysis:distv:calctimestamp", s.Id))
 119         if err != nil {
 120                 calctimestamp = -1
 121         }
 122         if calctimestamp > -1 {
 123                 s.CalcTimestamp = uint64(calctimestamp)
 124         }
 125 }
 126
 127 func set_eta(c redis.Conn, s *StateType) {
 128         eta := -1 // initialize to illegal value to detect nil.
 129         eta, err := redis.Int(c.Do("ZSCORE", "analysis:distv:eta", s.Id))
 130         if err != nil {
 131                 eta = -1
 132         }
 133         if eta > -1 {
 134                 s.CalcNextETA = uint64(eta)
 135         }
 136 }
 137
 138 func set_passfail(c redis.Conn, s *StateType) {
 139         fail := -1
 140         fail, err := redis.Int(c.Do("ZSCORE", "analysis:distv:fail", s.Id))
 141         if err != nil {
 142                 fail = -1
 143         }
 144         if fail > -1 {
 145                 s.Fail = uint32(fail)
 146         }
 147         pass := -1
 148         pass, err = redis.Int(c.Do("ZSCORE", "analysis:distv:pass", s.Id))
 149         if err != nil {
 150                 pass = -1
 151         }
 152         if pass > -1 {
 153                 s.Pass = uint32(pass)
 154         }
 155 }
 156
 157 func set_lcpassfail(c redis.Conn, l *LastCalcType) {
 158         fail := -1
 159         fail, err := redis.Int(c.Do("ZSCORE", "analysis:distv:lcfail", l.Id))
 160         if err != nil {
 161                 fail = 0
 162         }
 163         if fail > -1 {
 164                 l.LcFail = uint32(fail)
 165         }
 166         pass := -1
 167         pass, err = redis.Int(c.Do("ZSCORE", "analysis:distv:lcpass", l.Id))
 168         if err != nil {
 169                 pass = 0
 170         }
 171         if pass > -1 {
 172                 l.LcPass = uint32(pass)
 173         }
 174 }
 175
 176 func set_calcnext(c redis.Conn, s *StateType) {
 177         rating := -1 // initialize to illegal value to detect nil.
 178         rating, err := redis.Int(c.Do("ZSCORE", "analysis:jobqueue:q", s.Id))
 179         if err != nil {
 180                 rating = -1
 181         }
 182         if rating > -1 {
 183                 s.DebugRating = uint32(rating)
 184                 s.CalcNext = 1
 185         } else {
 186                 var tismember bool
 187                 var tstart uint32
 188                 var tid uint32
 189                 v, err := redis.Values(c.Do("SMEMBERS", "analysis:jobqueue:runningjobs"))
 190                 if err != nil {
 191                         panic(err)
 192                 }
 193                 for len(v) > 0 {
 194                         // $h{id} = $id;
 195                         var id int
 196                         v, err = redis.Scan(v, &id)
 197                         if err != nil {
 198                                 panic(err)
 199                         }
 200                         // my($json) = $redis->get("analysis:jobqueue:jobs:$id\:descr");
 201                         var jsonstr string
 202                         key := fmt.Sprintf("analysis:jobqueue:jobs:%d:descr", id)
 203                         jsonstr, err = redis.String(c.Do("GET", key))
 204                         if err != nil {
 205                                 fmt.Println("missing key: ", key)
 206                                 panic(err)
 207                         }
 208                         // my($j) = $jsonxs->decode($json);
 209                         b := []byte(jsonstr)
 210                         var j JobqueueJobsXDescr
 211                         err = json.Unmarshal(b, &j)
 212                         if err != nil {
 213                                 panic(err)
 214                         }
 215                         // $h{distv} = $j->{distv};
 216                         thisdistv := j.Distv
 217                         if thisdistv == s.Id {
 218                                 tismember = true
 219                                 tstart = j.Start
 220                                 tid = uint32(id)
 221                                 // last ???
 222                         }
 223                 }
 224                 if tismember {
 225                         s.CalcNext = 2
 226                         s.DebugRunningStart = tstart
 227                         s.DebugRunningSetId = tid
 228                 }
 229         }
 230 }
 231
 232 var validPath = regexp.MustCompile("^/scheduler/(state|vote)/([a-zA-Z0-9_.-]+)$")
 233
 234 func makeHandler(fn func(http.ResponseWriter, *http.Request, string)) http.HandlerFunc {
 235         return func(w http.ResponseWriter, r *http.Request) {
 236                 w.Header().Set("Content-Type", "application/json; charset=utf-8")
 237                 m := validPath.FindStringSubmatch(r.URL.Path)
 238                 if m == nil {
 239                         http.NotFound(w, r)
 240                         return
 241                 }
 242                 fn(w, r, m[2])
 243         }
 244 }
 245
 246 var (
 247         addr = flag.Bool("addr", false, "find open address and print to final-port.txt")
 248 )
 249
 250 func main() {
 251         flag.Parse()
 252         http.HandleFunc("/scheduler/state/", makeHandler(stateHandler))
 253         http.HandleFunc("/scheduler/vote/", makeHandler(voteHandler))
 254
 255         if *addr {
 256                 l, err := net.Listen("tcp", "127.0.0.1:0")
 257                 if err != nil {
 258                         log.Fatal(err)
 259                 }
 260                 err = ioutil.WriteFile("final-port.txt", []byte(l.Addr().String()), 0644)
 261                 if err != nil {
 262                         log.Fatal(err)
 263                 }
 264                 s := &http.Server{}
 265                 s.Serve(l)
 266                 return
 267         }
 268
 269         http.ListenAndServe(":3001", nil)
 270 }