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
+
+        // the 'appid' of this datastore
+        aid string
+
         // 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
         // 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 (
         _ = memContextObj((*dataStoreData)(nil))
         _ = sync.Locker((*dataStoreData)(nil))
 )
 
-func newDataStoreData() *dataStoreData {
+func newDataStoreData(aid string) *dataStoreData {
         head := newMemStore()
         return &dataStoreData{
+                aid:  aid,
                 head: head,
                 snap: head.Snapshot(), // empty but better than a nil pointer.
         }
 }
 
 func (d *dataStoreData) Lock() {
         d.rwlock.Lock()
 }
 
 func (d *dataStoreData) Unlock() {
         d.rwlock.Unlock()
 }
 
 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()
-        }
+        addIndexes(d.head, d.aid, ns, idxs)
 }
 
 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 95 matching lines @@
                         if err != nil {
                                 return
                         }
                         if !d.disableSpecialEntities {
                                 incrementLocked(ents, groupMetaKey(ret), 1)
                         }
 
                         old := ents.Get(keyBytes(ret))
                         oldPM := ds.PropertyMap(nil)
                         if old != nil {
-                                if oldPM, err = rpmWoCtx(old, ns); err != nil {
+                                if oldPM, err = rpm(old); 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()
         if err != nil {
                 return err
         }
         if ents == nil {
                 for range keys {
                         cb(nil, ds.ErrNoSuchEntity)
                 }
                 return nil
         }
 
         for _, k := range keys {
                 pdata := ents.Get(keyBytes(k))
                 if pdata == nil {
                         cb(nil, ds.ErrNoSuchEntity)
                         continue
                 }
-                cb(rpmWoCtx(pdata, k.Namespace()))
+                cb(rpm(pdata))
         }
         return nil
 }
 
 func (d *dataStoreData) getMulti(keys []*ds.Key, cb ds.GetMultiCB) error {
         return getMultiInner(keys, cb, func() (*memCollection, error) {
                 s := d.takeSnapshot()
 
                 return s.GetCollection("ents:" + keys[0].Namespace()), nil
         })
 }
 
 func (d *dataStoreData) delMulti(keys []*ds.Key, cb ds.DeleteMultiCB) {
         ns := keys[0].Namespace()
         ents := d.mutableEnts(ns)
 
         if ents != nil {
                 for _, k := range keys {
                         err := func() error {
                                 kb := keyBytes(k)
 
                                 d.Lock()
                                 defer d.Unlock()
 
                                 if !d.disableSpecialEntities {
                                         incrementLocked(ents, groupMetaKey(k), 1)
                                 }
                                 if old := ents.Get(kb); old != nil {
-                                        oldPM, err := rpmWoCtx(old, ns)
+                                        oldPM, err := rpm(old)
                                         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 191 matching lines @@
                         cb(err)
                 }
         }
         return nil
 }
 
 func keyBytes(key *ds.Key) []byte {
         return serialize.ToBytes(ds.MkProperty(key))
 }
 
-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, "", "")
 }