impl/memory: Replace gkvlite with "treapstore".

impl/memory/datastore_index.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 memory
 
 import (
         "bytes"
         "fmt"
         "sort"
 
         ds "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/datastore/serialize"
-        "github.com/luci/gkvlite"
 )
 
 type qIndexSlice []*ds.IndexDefinition
 
 func (s qIndexSlice) Len() int           { return len(s) }
 func (s qIndexSlice) Swap(i, j int)      { s[i], s[j] = s[j], s[i] }
 func (s qIndexSlice) Less(i, j int) bool { return s[i].Less(s[j]) }
 
 func defaultIndexes(kind string, pmap ds.PropertyMap) []*ds.IndexDefinition {
         ret := make(qIndexSlice, 0, 2*len(pmap)+1)
@@ skipping 151 matching lines @@
         }
         itrDef := iterDefinition{c: idxColl}
 
         if endsWith != nil {
                 full := serialize.ToBytes(*endsWith.Flip())
                 // chop off the null terminating byte
                 itrDef.prefix = full[:len(full)-1]
         }
 
         it := itrDef.mkIter()
-        defer it.stop()
-        for !it.stopped {
-                it.next(nil, func(i *gkvlite.Item) {
-                        if i == nil {
-                                return
-                        }
-                        qi, err := serialize.ReadIndexDefinition(bytes.NewBuffer(i.Key))
-                        memoryCorruption(err)
-                        if !cb(qi.Flip()) {
-                                it.stop()
-                        }
-                })
+        for ent := it.next(); ent != nil; ent = it.next() {
+                qi, err := serialize.ReadIndexDefinition(bytes.NewBuffer(ent.key))
+                memoryCorruption(err)
+                if !cb(qi.Flip()) {
+                        break
+                }
         }
 }
 
 func mergeIndexes(ns string, store, oldIdx, newIdx memStore) {
         prefixBuf := []byte("idx:" + ns + ":")
         origPrefixBufLen := len(prefixBuf)
 
         oldIdx = oldIdx.Snapshot()
         newIdx = newIdx.Snapshot()
 
-        gkvCollide(oldIdx.GetCollection("idx"), newIdx.GetCollection("idx"), func(k, ov, nv []byte) {
+        memStoreCollide(oldIdx.GetCollection("idx"), newIdx.GetCollection("idx"), func(k, ov, nv []byte) {
                 prefixBuf = append(prefixBuf[:origPrefixBufLen], k...)
                 ks := string(prefixBuf)
 
                 coll := store.GetOrCreateCollection(ks)
 
                 oldColl := oldIdx.GetCollection(ks)
                 newColl := newIdx.GetCollection(ks)
 
                 switch {
                 case ov == nil && nv != nil: // all additions
-                        newColl.VisitItemsAscend(nil, false, func(i *gkvlite.Item) bool {
-                                coll.Set(i.Key, []byte{})
+                        newColl.ForEachItem(func(k, _ []byte) bool {
+                                coll.Set(k, []byte{})
                                 return true
                         })
                 case ov != nil && nv == nil: // all deletions
-                        oldColl.VisitItemsAscend(nil, false, func(i *gkvlite.Item) bool {
-                                coll.Delete(i.Key)
+                        oldColl.ForEachItem(func(k, _ []byte) bool {
+                                coll.Delete(k)
                                 return true
                         })
                 case ov != nil && nv != nil: // merge
-                        gkvCollide(oldColl, newColl, func(k, ov, nv []byte) {
+                        memStoreCollide(oldColl, newColl, func(k, ov, nv []byte) {
                                 if nv == nil {
                                         coll.Delete(k)
                                 } else {
                                         coll.Set(k, []byte{})
                                 }
                         })
                 default:
-                        impossible(fmt.Errorf("both values from gkvCollide were nil?"))
+                        impossible(fmt.Errorf("both values from memStoreCollide were nil?"))
                 }
                 // TODO(riannucci): remove entries from idxColl and remove index collections
                 // when there are no index entries for that index any more.
         })
 }
 
 func addIndexes(store memStore, aid string, compIdx []*ds.IndexDefinition) {
         normalized := make([]*ds.IndexDefinition, len(compIdx))
         idxColl := store.GetOrCreateCollection("idx")
         for i, idx := range compIdx {
                 normalized[i] = idx.Normalize()
                 idxColl.Set(serialize.ToBytes(*normalized[i].PrepForIdxTable()), []byte{})
         }
 
         for _, ns := range namespaces(store) {
+                kctx := ds.MkKeyContext(aid, ns)
                 if allEnts := store.Snapshot().GetCollection("ents:" + ns); allEnts != nil {
-                        allEnts.VisitItemsAscend(nil, true, func(i *gkvlite.Item) bool {
-                                pm, err := rpm(i.Val)
+                        allEnts.ForEachItem(func(ik, iv []byte) bool {
+                                pm, err := rpm(iv)
                                 memoryCorruption(err)
 
-                                prop, err := serialize.ReadProperty(bytes.NewBuffer(i.Key), serialize.WithoutContext, ds.MkKeyContext(aid, ns))
+                                prop, err := serialize.ReadProperty(bytes.NewBuffer(ik), serialize.WithoutContext, kctx)
                                 memoryCorruption(err)
 
                                 k := prop.Value().(*ds.Key)
 
                                 sip := serialize.PropertyMapPartially(k, pm)
 
                                 mergeIndexes(ns, store,
                                         newMemStore(),
                                         indexEntries(k, sip, normalized))
                                 return true
@@ skipping 12 matching lines @@
         var compIdx []*ds.IndexDefinition
         walkCompIdxs(store.Snapshot(), nil, func(i *ds.IndexDefinition) bool {
                 compIdx = append(compIdx, i)
                 return true
         })
 
         mergeIndexes(key.Namespace(), store,
                 indexEntriesWithBuiltins(key, oldEnt, compIdx),
                 indexEntriesWithBuiltins(key, newEnt, compIdx))
 }