Initial distributor implementation

appengine/cmd/dm/model/attempt.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package model
 
 import (
         "fmt"
         "time"
 
         "golang.org/x/net/context"
 
-        "github.com/luci/luci-go/common/api/dm/service/v1"
-        "github.com/luci/luci-go/common/bit_field"
+        dm "github.com/luci/luci-go/common/api/dm/service/v1"
+        bf "github.com/luci/luci-go/common/bit_field"
         "github.com/luci/luci-go/common/clock"
         google_pb "github.com/luci/luci-go/common/proto/google"
 )
 
+// AttemptRetryState indicates the current state of the Attempt's retry
+// counters.
+type AttemptRetryState struct {
+        Failed   uint32
+        Expired  uint32
+        TimedOut uint32
+        Crashed  uint32
+}
+
+// Reset resets all of the AttemptRetryState counters.
+func (a *AttemptRetryState) Reset() {
+        a.Failed = 0

[dnj (Google), 2016/06/09 18:00:56]

*a = AttemptRetryState{} !

[iannucci, 2016/06/15 00:46:01]

mmmm tasty.

+        a.Expired = 0
+        a.TimedOut = 0
+        a.Crashed = 0
+}
+
 // Attempt is the datastore model for a DM Attempt. It has no parent key, but
 // it may have the following children entities:
 //   * FwdDep
 //   * AttemptResult
 //
 // Additionally, every Attempt has an associated BackDepGroup whose ID equals
 // the ID of this Attempt.
 type Attempt struct {
         ID dm.Attempt_ID `gae:"$id"`
 
         Created  time.Time
         Modified time.Time
 
-        State dm.Attempt_State
+        State      dm.Attempt_State
+        RetryState AttemptRetryState
+
+        // Only valid when State == ABNORMAL_FINISHED
+        AbnormalFinish dm.AbnormalFinish

[dnj (Google), 2016/06/09 18:00:56]

Why not make these pointers?

[iannucci, 2016/06/15 00:46:01]

because luci/gae doesn't allow embed-by-pointer in this context, since these
implement PLS on (*struct). The issue is that if it's nil then Load wouldn't
have a way to populate it.

+
+        // Only valid when State == FINISHED
+        ResultExpiration time.Time `gae:",noindex"`
+        ResultSize       uint32    `gae:",noindex"`
+
+        // PersistentState is the last successful execution's returned
+        // PersistentState. It is set whenever an execution for this Attempt finishes
+        // sucessfully. This is denormalized with the Execution's
+        // ResultPersistentState field.
+        PersistentState string `gae:",noindex"`
 
         // TODO(iannucci): Use CurExecution as a 'deps block version'
         // then we can have an 'ANY' directive which executes the attempt as soon
         // as any of the dependencies are ready. If it adds more deps in ANY mode,
         // the bitmaps get /extended/, and new deps bit indices are added to the
         // existing max.
         // If it adds more deps in ALL mode, it just converts from ANY -> ALL mode
         // and follows the current behavior.
 
         // CurExecution is the maximum Execution ID for this Attempt so far. Execution
         // IDs are contiguous from [1, CurExecution]. If the State is not currently
         // Executing, then CurExecution represents the execution that JUST finished
         // (or 0 if no Executions have been made yet).
         CurExecution uint32
 
-        // AddingDepsBitmap is valid only while Attempt is in 'AddingDeps'.
-        // A field value of 0 means the backdep hasn't been added yet.
-        AddingDepsBitmap bf.BitField `gae:",noindex" json:"-"`
-
-        // WaitingDepBitmap is valid only while Attempt is in a Status of 'AddingDeps'
-        // or 'Blocked'.
-        // A field value of 0 means that the dep is currently waiting.
-        WaitingDepBitmap bf.BitField `gae:",noindex" json:"-"`
-
-        // Only valid while Attempt is Finished
-        ResultExpiration time.Time
-        ResultSize       uint32
+        // DepMap is valid only while Attempt is in a State of EXECUTING or WAITING.
+        //
+        // The size of this field is inspected to deteremine what the next state after
+        // EXECUTING is. If the size == 0, it means the Attempt should move to the
+        // FINISHED state. Otherwise it means that the Attempt should move to the
+        // WAITING state.
+        //
+        // A bit field value of 0 means that the dep is currently waiting, and a bit
+        // value of 1 means that the coresponding dep is satisfined. The Attempt can
+        // be unblocked from WAITING back to SCHEDULING when all bits are set to 1.
+        DepMap bf.BitField `gae:",noindex" json:"-"`

[iannucci, 2016/06/08 02:54:24]

this is part of the state machine simplification; there's no real need to
distinguish between 'adding deps' and 'blocked'. Tumble adds backdeps in the
background, but the state of THIS attempt is just 'waiting'. It can't proceed
from waiting until (all backdeps added && dependencies finished), which is
enforced through the mutations.

 
         // A lazily-updated boolean to reflect that this Attempt is expired for
         // queries.
-        Expired bool
+        ResultExpired bool
 }
 
 // MakeAttempt is a convenience function to create a new Attempt model in
 // the NeedsExecution state.
 func MakeAttempt(c context.Context, aid *dm.Attempt_ID) *Attempt {
         now := clock.Now(c).UTC()
         return &Attempt{
                 ID:       *aid,
                 Created:  now,
                 Modified: now,
         }
 }
 
 // ModifyState changes the current state of this Attempt and updates its
 // Modified timestamp.
 func (a *Attempt) ModifyState(c context.Context, newState dm.Attempt_State) error {
+        if a.State == newState {
+                return nil
+        }
         if err := a.State.Evolve(newState); err != nil {
                 return err
         }
         now := clock.Now(c).UTC()
         if now.After(a.Modified) {
                 a.Modified = now
         } else {
                 // Microsecond is the smallest granularity that datastore can store
                 // timestamps, so use that to disambiguate: the goal here is that any
                 // modification always increments the modified time, and never decrements
                 // it.
                 a.Modified = a.Modified.Add(time.Microsecond)
         }
         return nil
 }
 
-// MustModifyState is the same as ModifyState, except that it panics if the
-// state transition is invalid.
-func (a *Attempt) MustModifyState(c context.Context, newState dm.Attempt_State) {
-        err := a.ModifyState(c, newState)
-        if err != nil {
-                panic(err)
-        }
-}
-
 // ToProto returns a dm proto version of this Attempt.
 func (a *Attempt) ToProto(withData bool) *dm.Attempt {
         ret := dm.Attempt{Id: &a.ID}
         if withData {
                 ret.Data = a.DataProto()
         }
         return &ret
 }
 
 // DataProto returns an Attempt.Data message for this Attempt.
-func (a *Attempt) DataProto() *dm.Attempt_Data {
-        ret := (*dm.Attempt_Data)(nil)
+func (a *Attempt) DataProto() (ret *dm.Attempt_Data) {
         switch a.State {
-        case dm.Attempt_NEEDS_EXECUTION:
-                ret = dm.NewAttemptNeedsExecution(a.Modified).Data
-
+        case dm.Attempt_SCHEDULING:

[iannucci, 2016/06/08 02:54:24]

these changes are due to the state machine simplification as well.

+                ret = dm.NewAttemptScheduling().Data
         case dm.Attempt_EXECUTING:
                 ret = dm.NewAttemptExecuting(a.CurExecution).Data
-
-        case dm.Attempt_ADDING_DEPS:
-                addset := a.AddingDepsBitmap
-                waitset := a.WaitingDepBitmap
-                setlen := addset.Size()
-
-                ret = dm.NewAttemptAddingDeps(
-                        setlen-addset.CountSet(), setlen-waitset.CountSet()).Data
-
-        case dm.Attempt_BLOCKED:
-                waitset := a.WaitingDepBitmap
-                setlen := waitset.Size()
-
-                ret = dm.NewAttemptBlocked(setlen - waitset.CountSet()).Data
-
+        case dm.Attempt_WAITING:
+                ret = dm.NewAttemptWaiting(a.DepMap.Size() - a.DepMap.CountSet()).Data
         case dm.Attempt_FINISHED:
-                ret = dm.NewAttemptFinished(a.ResultExpiration, a.ResultSize, "").Data
-
+                ret = dm.NewAttemptFinished(a.ResultExpiration, a.ResultSize, "",
+                        string(a.PersistentState)).Data
+        case dm.Attempt_ABNORMAL_FINISHED:
+                ret = dm.NewAttemptAbnormalFinish(&a.AbnormalFinish).Data
         default:
                 panic(fmt.Errorf("unknown Attempt_State: %s", a.State))
         }
         ret.Created = google_pb.NewTimestamp(a.Created)
         ret.Modified = google_pb.NewTimestamp(a.Modified)
         ret.NumExecutions = a.CurExecution
         return ret
 }