Create unbuffered Tumble entry point for LogDog.

tumble/tumble.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 tumble
 
 import (
         "fmt"
 
-        "github.com/luci/gae/filter/txnBuf"
-        ds "github.com/luci/gae/service/datastore"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/data/stringset"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logging"
+
+        "github.com/luci/gae/filter/txnBuf"
+        ds "github.com/luci/gae/service/datastore"
+
         "golang.org/x/net/context"
 )
 
 // RunMutation immediately runs the Mutation `m` in a transaction. This method
 // should be used to start a tumble chain when you have transactional checks
 // to do (e.g. `m` implements the first transactional link in the chain).
 //
 // Usually this is called from your application's handlers to begin a tumble
 // state machine as a result of some API interaction.
 func RunMutation(c context.Context, m Mutation) error {
         cfg := getConfig(c)
-        shardSet, _, _, err := enterTransactionInternal(txnBuf.FilterRDS(c), cfg, m, 0)
+        shardSet, _, _, err := enterTransactionMutation(txnBuf.FilterRDS(c), cfg, m, 0)
         if err != nil {
                 return err
         }
         fireTasks(c, cfg, shardSet)
         return nil
 }
 
-func enterTransactionInternal(c context.Context, cfg *Config, m Mutation, round uint64) (map[taskShard]struct{}, []Mutation, []*ds.Key, error) {
-        fromRoot := m.Root(c)
+// RunUnbuffered opens a lightweight unbuffered transaction on "root"
+// runs "fn" inside of it. Any mutations returned by "fn" will be registered
+// at the end of the transaction if "fn" doesn't return an error.
+//
+// This is useful as an initial starting point without incurring any of the
+// overhead of spinning up a new buffered transaction.
+//
+// During "fn"'s execution, standard Tumble operations such as PutNamedMutation
+// and CancelNamedMutation may be performed.
+func RunUnbuffered(c context.Context, root *ds.Key, fn func(context.Context) ([]Mutation, error)) error {
+        cfg := getConfig(c)
+        shardSet, _, _, err := enterTransactionInternal(c, cfg, root, fn, 0)
+        if err != nil {
+                return err
+        }
+        fireTasks(c, cfg, shardSet)
+        return nil
+}
 
-        if fromRoot == nil {
-                return nil, nil, nil, fmt.Errorf("tumble: Passing nil as fromRoot is illegal")
+func enterTransactionMutation(c context.Context, cfg *Config, m Mutation, round uint64) (
+        map[taskShard]struct{}, []Mutation, []*ds.Key, error) {
+
+        return enterTransactionInternal(c, cfg, m.Root(c), m.RollForward, round)
+}
+
+func enterTransactionInternal(c context.Context, cfg *Config, root *ds.Key, fn func(context.Context) ([]Mutation, error), round uint64) (
+        map[taskShard]struct{}, []Mutation, []*ds.Key, error) {
+
+        if root == nil {
+                return nil, nil, nil, fmt.Errorf("tumble: Passing nil as root is illegal")
         }
 
         shardSet := map[taskShard]struct{}(nil)
         retMuts := []Mutation(nil)
         retMutKeys := []*ds.Key(nil)
 
         err := ds.RunInTransaction(c, func(c context.Context) error {
-                // do a Get on the fromRoot to ensure that this transaction is associated
+                // do a Get on the root to ensure that this transaction is associated
                 // with that entity group.
-                _, _ = ds.Exists(c, fromRoot)
+                _, _ = ds.Exists(c, root)
 
-                muts, err := m.RollForward(c)
+                muts, err := fn(c)
                 if err != nil {
                         return err
                 }
 
                 retMuts = muts
-                shardSet, retMutKeys, err = putMutations(c, cfg, fromRoot, muts, round)
+                shardSet, retMutKeys, err = putMutations(c, cfg, root, muts, round)
 
                 return err
         }, nil)
         if err != nil {
                 return nil, nil, nil, err
         }
 
         return shardSet, retMuts, retMutKeys, nil
 }
 
@@ skipping 44 matching lines @@
 // CancelNamedMutations does a best-effort cancellation of the named mutations.
 func CancelNamedMutations(c context.Context, parent *ds.Key, names ...string) error {
         toDel := make([]*ds.Key, 0, len(names))
         nameSet := stringset.NewFromSlice(names...)
         nameSet.Iter(func(name string) bool {
                 toDel = append(toDel, ds.NewKey(c, "tumble.Mutation", "n:"+name, 0, parent))
                 return true
         })
         return errors.Filter(ds.Delete(c, toDel), ds.ErrNoSuchEntity)
 }