Initial distributor implementation

appengine/cmd/dm/model/attempt.go

Index: appengine/cmd/dm/model/attempt.go
diff --git a/appengine/cmd/dm/model/attempt.go b/appengine/cmd/dm/model/attempt.go
index ec8ce5c6ce8935097ba8f721b8fd20d8cddf5c27..03c476e312d421181d2e9dc641cf398a7956935d 100644
--- a/appengine/cmd/dm/model/attempt.go
+++ b/appengine/cmd/dm/model/attempt.go
@@ -10,12 +10,29 @@ import (
 
 	"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
@@ -29,7 +46,21 @@ type Attempt struct {
 	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
@@ -45,22 +76,21 @@ type Attempt struct {
 	// (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
@@ -77,6 +107,9 @@ func MakeAttempt(c context.Context, aid *dm.Attempt_ID) *Attempt {
 // 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
 	}
@@ -93,15 +126,6 @@ func (a *Attempt) ModifyState(c context.Context, newState dm.Attempt_State) erro
 	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}
@@ -112,32 +136,19 @@ func (a *Attempt) ToProto(withData bool) *dm.Attempt {
 }
 
 // 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))
 	}