Add transaction buffer filter.

impl/memory/datastore_data.go

 // 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 memory
 
 import (
         "bytes"
         "fmt"
         "sync"
         "sync/atomic"
 
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/datastore/serialize"
         "github.com/luci/luci-go/common/errors"
         "golang.org/x/net/context"
 )
 
 //////////////////////////////// dataStoreData /////////////////////////////////
 
 type dataStoreData struct {
         rwlock sync.RWMutex
         // See README.md for head schema.
         head *memStore
+        // if snap is nil, that means that this is always-consistent, and
+        // getQuerySnaps will return (head, head)
         snap *memStore
         // For testing, see SetTransactionRetryCount.
         txnFakeRetry int
-        // true means that head always == snap
-        consistent bool

[iannucci, 2015/09/29 03:21:38]

This change means that we avoid taking a snapshot on every mutation when we're
in full-consistency mode. This saves a couple .1s on the tests.

         // true means that queries with insufficent indexes will pause to add them
         // and then continue instead of failing.
         autoIndex bool
         // true means that all of the __...__ keys which are normally automatically
         // maintained will be omitted. This also means that Put with an incomplete
         // key will become an error.
         disableSpecialEntities bool
 }
 
 var (
@@ skipping 20 matching lines @@
 func (d *dataStoreData) setTxnRetry(count int) {
         d.Lock()
         defer d.Unlock()
         d.txnFakeRetry = count
 }
 
 func (d *dataStoreData) setConsistent(always bool) {
         d.Lock()
         defer d.Unlock()
 
-        d.consistent = always
-        if d.consistent {
+        if always {
+                d.snap = nil
+        } else {
                 d.snap = d.head.Snapshot()
         }
 }
 
 func (d *dataStoreData) addIndexes(ns string, idxs []*ds.IndexDefinition) {
         d.Lock()
         defer d.Unlock()
         addIndexes(d.head, ns, idxs)
-        if d.consistent {
-                d.snap = d.head.Snapshot()
-        }
 }
 
 func (d *dataStoreData) setAutoIndex(enable bool) {
         d.Lock()
         defer d.Unlock()
         d.autoIndex = enable
 }
 
 func (d *dataStoreData) maybeAutoIndex(err error) bool {
         mi, ok := err.(*ErrMissingIndex)
@@ skipping 21 matching lines @@
 
 func (d *dataStoreData) getDisableSpecialEntities() bool {
         d.rwlock.RLock()
         defer d.rwlock.RUnlock()
         return d.disableSpecialEntities
 }
 
 func (d *dataStoreData) getQuerySnaps(consistent bool) (idx, head *memStore) {
         d.rwlock.RLock()
         defer d.rwlock.RUnlock()
-        if d.consistent {
-                // snap is already a consistent snapshot of head
-                return d.snap, d.snap
+        if d.snap == nil {
+                // we're 'always consistent'
+                return d.head, d.head
         }
 
         head = d.head.Snapshot()
         if consistent {
                 idx = head
         } else {
                 idx = d.snap
         }
         return
 }
 
 func (d *dataStoreData) takeSnapshot() *memStore {
         d.rwlock.RLock()
         defer d.rwlock.RUnlock()
-        if d.consistent {
-                return d.snap
-        }
         return d.head.Snapshot()
 }
 
 func (d *dataStoreData) setSnapshot(snap *memStore) {
         d.rwlock.Lock()
         defer d.rwlock.Unlock()
-        if d.consistent {
+        if d.snap == nil {
+                // we're 'always consistent'
                 return
         }
         d.snap = snap
 }
 
 func (d *dataStoreData) catchupIndexes() {
         d.rwlock.Lock()
         defer d.rwlock.Unlock()
-        if d.consistent {
+        if d.snap == nil {
+                // we're 'always consistent'
                 return
         }
         d.snap = d.head.Snapshot()
 }
 
 /////////////////////////// indexes(dataStoreData) ////////////////////////////
 
 func groupMetaKey(key *ds.Key) []byte {
         return keyBytes(ds.NewKey("", "", "__entity_group__", "", 1, key.Root()))
 }
@@ skipping 101 matching lines @@
 
                         old := ents.Get(keyBytes(ret))
                         oldPM := ds.PropertyMap(nil)
                         if old != nil {
                                 if oldPM, err = rpmWoCtx(old, ns); err != nil {
                                         return
                                 }
                         }
                         updateIndexes(d.head, ret, oldPM, pmap)
                         ents.Set(keyBytes(ret), dataBytes)
-                        if d.consistent {
-                                d.snap = d.head.Snapshot()
-                        }
                         return
                 }()
                 if cb != nil {
                         cb(k, err)
                 }
         }
 }
 
 func getMultiInner(keys []*ds.Key, cb ds.GetMultiCB, getColl func() (*memCollection, error)) error {
         ents, err := getColl()
@@ skipping 41 matching lines @@
                                 if !d.disableSpecialEntities {
                                         incrementLocked(ents, groupMetaKey(k), 1)
                                 }
                                 if old := ents.Get(kb); old != nil {
                                         oldPM, err := rpmWoCtx(old, ns)
                                         if err != nil {
                                                 return err
                                         }
                                         updateIndexes(d.head, k, oldPM, nil)
                                         ents.Delete(kb)
-                                        if d.consistent {
-                                                d.snap = d.head.Snapshot()
-                                        }
                                 }
                                 return nil
                         }()
                         if cb != nil {
                                 cb(err)
                         }
                 }
         } else if cb != nil {
                 for range keys {
                         cb(nil)
@@ skipping 200 matching lines @@
 
 func rpmWoCtx(data []byte, ns string) (ds.PropertyMap, error) {
         return serialize.ReadPropertyMap(bytes.NewBuffer(data),
                 serialize.WithoutContext, globalAppID, ns)
 }
 
 func rpm(data []byte) (ds.PropertyMap, error) {
         return serialize.ReadPropertyMap(bytes.NewBuffer(data),
                 serialize.WithContext, "", "")
 }