Add Task DB implementation using a local BoltDB.

build_scheduler/go/db/task.go

 package db
 
 import (
         "bytes"
         "encoding/gob"
         "fmt"
         "time"
 
         "go.skia.org/infra/go/util"
 
@@ skipping 38 matching lines @@
 // that can not be executed on Swarming, but can be added to the DB and
 // displayed as if it were a real TaskSpec.
 type Task struct {
         // Commits are the commits which were tested in this Task. The list may
         // change due to backfilling/bisecting.
         Commits []string
 
         // Created is the creation timestamp.
         Created time.Time
 
-        // Id is a generated unique identifier for this Task instance. Must be URL-safe.
+        // Id is a generated unique identifier for this Task instance. Must be
+        // URL-safe.
         Id string
 
         // IsolatedOutput is the isolated hash of any outputs produced by this
         // Task. Filled in when the task is completed.
         IsolatedOutput string
 
         // Name is a human-friendly descriptive name for this Task. All Tasks
         // generated from the same TaskSpec have the same name.
         Name string
 
@@ skipping 72 matching lines @@
                 return fmt.Errorf("Unable to parse task creation time for task %s. %v %v", t.Id, err, s)
         }
 
         switch s.TaskResult.State {
         case SWARMING_STATE_BOT_DIED, SWARMING_STATE_CANCELED, SWARMING_STATE_EXPIRED, SWARMING_STATE_TIMED_OUT:
                 t.Status = TASK_STATUS_MISHAP
         case SWARMING_STATE_PENDING, SWARMING_STATE_RUNNING:
                 t.Status = TASK_STATUS_PENDING
         case SWARMING_STATE_COMPLETED:
                 if s.TaskResult.Failure {
-                        // TODO(benjaminwagner): Choose FAILURE or MISHAP depending on ExitCode?
+                        // TODO(benjaminwagner): Choose FAILURE or MISHAP depending on
+                        // ExitCode?
                         t.Status = TASK_STATUS_FAILURE
                 } else {
                         t.Status = TASK_STATUS_SUCCESS
                 }
         default:
                 return fmt.Errorf("Unknown Swarming State %v in %v", s.TaskResult.State, s)
         }
 
         if s.TaskResult.OutputsRef == nil {
                 t.IsolatedOutput = ""
@@ skipping 19 matching lines @@
         if err := gob.NewEncoder(&buf).Encode(t); err != nil {
                 glog.Fatal(err)
         }
         var rv Task
         if err := gob.NewDecoder(&buf).Decode(&rv); err != nil {
                 glog.Fatal(err)
         }
         return &rv
 }
 
+// TaskSlice implements sort.Interface. To sort tasks []*Task, use
+// sort.Sort(TaskSlice(tasks)).
 type TaskSlice []*Task
 
 func (s TaskSlice) Len() int { return len(s) }
 
 func (s TaskSlice) Less(i, j int) bool {
         return s[i].Created.Before(s[j].Created)
 }
 
 func (s TaskSlice) Swap(i, j int) {
         s[i], s[j] = s[j], s[i]
 }
+
+// TaskEncoder encodes Tasks into bytes via GOB encoding. Not safe for
+// concurrent use.
+// TODO(benjaminwagner): Encode in parallel.
+type TaskEncoder struct {
+        err    error
+        tasks  []*Task
+        result [][]byte
+}
+
+// Process encodes the Task into a byte slice that will be returned from Next()
+// (in arbitrary order). Returns false if Next is certain to return an error.
+// Caller must ensure t does not change until after the first call to Next().
+// May not be called after calling Next().
+func (e *TaskEncoder) Process(t *Task) bool {
+        if e.err != nil {
+                return false
+        }
+        var buf bytes.Buffer
+        if err := gob.NewEncoder(&buf).Encode(t); err != nil {
+                e.err = err
+                e.tasks = nil
+                e.result = nil
+                return false
+        }
+        e.tasks = append(e.tasks, t)
+        e.result = append(e.result, buf.Bytes())
+        return true
+}
+
+// Next returns one of the Tasks provided to Process (in arbitrary order) and
+// its serialized bytes. If any tasks remain, returns the task, the serialized
+// bytes, nil. If all tasks have been returned, returns nil, nil, nil. If an
+// error is encountered, returns nil, nil, error.
+func (e *TaskEncoder) Next() (*Task, []byte, error) {
+        if e.err != nil {
+                return nil, nil, e.err
+        }
+        if len(e.tasks) == 0 {
+                return nil, nil, nil
+        }
+        t := e.tasks[0]
+        e.tasks = e.tasks[1:]
+        serialized := e.result[0]
+        e.result = e.result[1:]
+        return t, serialized, nil
+}
+
+// TaskDecoder decodes bytes into Tasks via GOB decoding. Not safe for
+// concurrent use.
+// TODO(benjaminwagner): Decode in parallel.
+type TaskDecoder struct {
+        err    error
+        result []*Task
+}
+
+// Process decodes the byte slice into a Task and includes it in Result() (in
+// arbitrary order). Returns false if Result is certain to return an error.
+// Caller must ensure b does not change until after Result() returns.
+func (d *TaskDecoder) Process(b []byte) bool {
+        if d.err != nil {
+                return false
+        }
+        var t Task
+        if err := gob.NewDecoder(bytes.NewReader(b)).Decode(&t); err != nil {
+                d.err = err
+                d.result = nil
+                return false
+        }
+        d.result = append(d.result, &t)
+        return true
+}
+
+// Result returns all decoded Tasks provided to Process (in arbitrary order), or
+// any error encountered.
+func (d *TaskDecoder) Result() ([]*Task, error) {
+        // Allow TaskDecoder to be used without initialization.
+        if d.err == nil && d.result == nil {
+                return []*Task{}, nil
+        }
+        return d.result, d.err
+}