blob: 0f61885d6e67ec99e3a89a73d0a376cb88543693 [file] [log] [blame]
// Copyright 2015 The LUCI Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package mutate
import (
"context"
"go.chromium.org/luci/common/logging"
"go.chromium.org/luci/dm/api/service/v1"
"go.chromium.org/luci/dm/appengine/model"
ds "go.chromium.org/luci/gae/service/datastore"
"go.chromium.org/luci/tumble"
)
// AckFwdDep records the fact that a dependency was completed.
type AckFwdDep struct {
Dep *model.FwdEdge
}
// Root implements tumble.Mutation.
func (f *AckFwdDep) Root(c context.Context) *ds.Key {
return model.AttemptKeyFromID(c, f.Dep.From)
}
// RollForward implements tumble.Mutation.
func (f *AckFwdDep) RollForward(c context.Context) (muts []tumble.Mutation, err error) {
atmpt, fdep := f.Dep.Fwd(c)
err = ds.Get(c, atmpt, fdep)
if err != nil {
return
}
if (atmpt.State != dm.Attempt_EXECUTING && atmpt.State != dm.Attempt_WAITING) || atmpt.CurExecution != fdep.ForExecution {
logging.Errorf(c, "EARLY EXIT: %s: %s v %s", atmpt.State, atmpt.CurExecution, fdep.ForExecution)
return
}
idx := uint32(fdep.BitIndex)
if !atmpt.DepMap.IsSet(idx) {
atmpt.DepMap.Set(idx)
if atmpt.DepMap.All(true) && atmpt.State == dm.Attempt_WAITING {
if err = atmpt.ModifyState(c, dm.Attempt_SCHEDULING); err != nil {
return
}
atmpt.DepMap.Reset()
muts = append(muts, &ScheduleExecution{For: f.Dep.From})
}
err = ds.Put(c, atmpt)
}
return
}
func init() {
tumble.Register((*AckFwdDep)(nil))
}