Initial distributor implementation

appengine/cmd/dm/mutate/finish_execution.go

Index: appengine/cmd/dm/mutate/finish_execution.go
diff --git a/appengine/cmd/dm/mutate/finish_execution.go b/appengine/cmd/dm/mutate/finish_execution.go
new file mode 100644
index 0000000000000000000000000000000000000000..ecea9d59962d3a4f50719cf2b801f78c55854edb
--- /dev/null
+++ b/appengine/cmd/dm/mutate/finish_execution.go
@@ -0,0 +1,154 @@
+// Copyright 2015 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+package mutate
+
+import (
+	"fmt"
+	"golang.org/x/net/context"
+
+	"github.com/luci/gae/filter/txnBuf"
+	"github.com/luci/gae/service/datastore"
+
+	"github.com/luci/luci-go/appengine/tumble"
+	dm "github.com/luci/luci-go/common/api/dm/service/v1"
+
+	"github.com/luci/luci-go/appengine/cmd/dm/distributor"
+	"github.com/luci/luci-go/appengine/cmd/dm/model"
+)
+
+// FinishExecution records the final state of the Execution, and advances the
+// Attempt state machine.
+type FinishExecution struct {

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

this is the major state machine change. Previously:
  * An execution was started at the time any dependency was taken on an attempt
which did not exist, or an attempt went from blocked->unblocked
  * An execution would finish 'sometime after' the FinishAttempt rpc or
EnsureGraphData rpc, but the completion of the execution wasn't really
well-tracked.

With the addition of the 'state' api, an Attempt is "EXECUTING" any time it has
an unfinished execution. This means that even after FinishAttempt happens, we
still wait for the execution to finish (e.g. for swarming to tell DM that the
job is done). As part of the distributor telling DM that the job is done, it
also updates the 'persistent state' for the attempt (for swarming, this is the
isolated_outdir hash for the task). If the task fails between the FinishAttempt
rpc and the distributor telling DM about it, we treat the execution as failed,
and it's eligible for a retry.

The reduction in complexity comes from the fact that an Attempt is now
completely blocked on its execution completing; it can't fire another execution
or otherwise change state until the current execution completes (or times out).

+	EID    *dm.Execution_ID
+	Result *distributor.TaskResult
+}
+
+// Root implements tumble.Mutation
+func (f *FinishExecution) Root(c context.Context) *datastore.Key {
+	return model.ExecutionKeyFromID(c, f.EID)
+}
+
+// shouldRetry loads the quest for this attempt, to determine if the attempt can
+// be retried. As a side-effect, it increments the RetryState counter for the
+// indicated failure type.
+//
+// If stat is not a retryable AbnormalFinish_Status, this will panic.
+func shouldRetry(c context.Context, a *model.Attempt, stat dm.AbnormalFinish_Status) (retry bool, err error) {
+	if !stat.CouldRetry() {
+		return
+	}
+	q := model.QuestFromID(a.ID.Quest)
+	dsNoTxn := txnBuf.GetNoTxn(c)
+	if err = dsNoTxn.Get(q); err != nil {
+		return
+	}
+	var cur, max uint32
+	switch stat {
+	case dm.AbnormalFinish_FAILED:
+		cur, max = a.RetryState.Failed, q.Desc.Meta.Retry.Failed
+		a.RetryState.Failed++
+	case dm.AbnormalFinish_CRASHED:
+		cur, max = a.RetryState.Crashed, q.Desc.Meta.Retry.Crashed
+		a.RetryState.Crashed++
+	case dm.AbnormalFinish_EXPIRED:
+		cur, max = a.RetryState.Expired, q.Desc.Meta.Retry.Expired
+		a.RetryState.Expired++
+	case dm.AbnormalFinish_TIMED_OUT:
+		cur, max = a.RetryState.TimedOut, q.Desc.Meta.Retry.TimedOut
+		a.RetryState.TimedOut++
+	default:
+		panic(fmt.Errorf("do not know how to retry %q", stat))
+	}
+	retry = cur < max
+	return
+}
+
+// RollForward implements tumble.Mutation
+func (f *FinishExecution) RollForward(c context.Context) (muts []tumble.Mutation, err error) {
+	a := model.AttemptFromID(f.EID.AttemptID())
+	e := model.ExecutionFromID(c, f.EID)
+
+	ds := datastore.Get(c)
+	if err = ds.GetMulti([]interface{}{a, e}); err != nil {
+		return
+	}
+
+	if a.State == dm.Attempt_EXECUTING && a.CurExecution == f.EID.Id && !e.State.Terminal() {

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

For indentation purposes, maybe invert this and return early?

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

Done.

+		if f.Result.AbnormalFinish == nil && e.State != dm.Execution_STOPPING {
+			f.Result.AbnormalFinish = &dm.AbnormalFinish{
+				Status: dm.AbnormalFinish_FAILED,
+				Reason: fmt.Sprintf("distributor finished execution while it was in the %s state.", e.State),
+			}
+		}
+
+		if ab := f.Result.AbnormalFinish; ab != nil {
+			if err = e.ModifyState(c, dm.Execution_ABNORMAL_FINISHED); err != nil {
+				return
+			}
+			e.AbnormalFinish = *ab
+
+			var retry bool
+			if retry, err = shouldRetry(c, a, ab.Status); err != nil {
+				return
+			} else if retry {
+				if err = a.ModifyState(c, dm.Attempt_SCHEDULING); err != nil {
+					return
+				}
+				a.DepMap.Reset()
+				muts = append(muts, &ScheduleExecution{&a.ID})
+			} else {
+				// ran out of retries, or non-retriable error type
+				if err = a.ModifyState(c, dm.Attempt_ABNORMAL_FINISHED); err != nil {
+					return
+				}
+				a.AbnormalFinish = *ab
+			}
+		} else {
+			if err = e.ModifyState(c, dm.Execution_FINISHED); err != nil {
+				return
+			}
+			e.ResultPersistentState = string(f.Result.PersistentState)
+
+			a.PersistentState = string(f.Result.PersistentState)
+			a.RetryState.Reset()
+
+			if a.DepMap.Size() > 0 {
+				if err = a.ModifyState(c, dm.Attempt_WAITING); err != nil {
+					return
+				}
+			} else {
+				if err = a.ModifyState(c, dm.Attempt_FINISHED); err != nil {
+					return
+				}
+				muts = append(muts, &RecordCompletion{f.EID.AttemptID()})
+			}
+		}
+
+		// best-effort reset execution timeout
+		_ = ResetExecutionTimeout(c, e)
+
+		err = ds.PutMulti([]interface{}{a, e})
+	}
+	return
+}
+
+// FinishExecutionFn is the implementation of distributor.FinishExecutionFn.
+// It's defined here to avoid a circular dependency.
+func FinishExecutionFn(c context.Context, eid *dm.Execution_ID, rslt *distributor.TaskResult) ([]tumble.Mutation, error) {
+	return []tumble.Mutation{&FinishExecution{EID: eid, Result: rslt}}, nil
+}
+
+// NewFinishExecutionAbnormal is a shorthand to make a FinishExecution mutation
+// with some abnomal result.
+func NewFinishExecutionAbnormal(eid *dm.Execution_ID, status dm.AbnormalFinish_Status, reason string) *FinishExecution {
+	return &FinishExecution{
+		eid, &distributor.TaskResult{
+			AbnormalFinish: &dm.AbnormalFinish{
+				Status: status, Reason: reason}}}
+}
+
+func init() {
+	tumble.Register((*FinishExecution)(nil))
+}