A transparent cache for datastore, backed by memcache.

filter/dscache/ds.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 dscache
+
+import (
+        "time"
+
+        ds "github.com/luci/gae/service/datastore"
+        "github.com/luci/gae/service/memcache"
+        "golang.org/x/net/context"
+)
+
+type dsCache struct {
+        ds.RawInterface
+
+        *supportContext
+}
+
+var _ ds.RawInterface = (*dsCache)(nil)
+
+func (d *dsCache) DeleteMulti(keys []ds.Key, cb ds.DeleteMultiCB) error {
+        return d.mutation(keys, func() error {
+                return d.RawInterface.DeleteMulti(keys, cb)
+        })
+}
+
+func (d *dsCache) PutMulti(keys []ds.Key, vals []ds.PropertyMap, cb ds.PutMultiCB) error {
+        return d.mutation(keys, func() error {
+                return d.RawInterface.PutMulti(keys, vals, cb)
+        })
+}
+
+func (d *dsCache) GetMulti(keys []ds.Key, metas ds.MultiMetaGetter, cb ds.GetMultiCB) error {
+        lockItems, nonce := d.mkRandLockItems(keys, metas)
+        if len(lockItems) == 0 {
+                return d.RawInterface.GetMulti(keys, metas, cb)
+        }
+
+        if err := d.mc.AddMulti(lockItems); err != nil {

[Vadim Sh., 2015/08/06 01:23:33]

what happens if only one item is failed to lock? (Is AddMulti atomic btw)?

[iannucci, 2015/08/06 02:37:33]

it's not atomic, but it doesn't matter. Either:
  * it's being put, so we get from datastore and don't update the cache
  * someone already filled the value in and so Add is a noop (and then we just
decode from memcache)
  * someone else is locking, so we get from datastore and don't update the cache
  * memcache had a brain fart, so we... that thing ^
  * we did add it, but it still doesn't matter, since we need to re-get it in
order to compare and swap at the end of the function.

+                d.log.Errorf("dscache: mc.AddMulti: %s", err)

[dnj, 2015/08/05 18:32:17]

(Debugf)

[iannucci, 2015/08/06 01:54:01]

Did warn

+        }
+        if err := d.mc.GetMulti(lockItems); err != nil {
+                d.log.Errorf("dscache: mc.GetMulti: %s", err)

[dnj, 2015/08/05 18:32:17]

(Debugf)

[iannucci, 2015/08/06 01:54:01]

Did warn

+        }
+
+        p := makePlan(d.aid, d.ns, d.log, &facts{keys, metas, lockItems, nonce})
+
+        if !p.empty() {

[dnj, 2015/08/05 18:32:17]

Invert: if p.empty() { return nil }

[iannucci, 2015/08/06 01:54:01]

can't, this modifies p.decoded for the loop below it.

[dnj, 2015/08/07 16:30:06]

Acknowledged.

+                toCas := []memcache.Item{}

[dnj, 2015/08/05 18:32:17]

Might as well optimistically allocate capacity.

[iannucci, 2015/08/06 01:54:01]

Eh... estimating the amount to allocate is hard enough and append does quadratic
allocation which should basically be fine.

[dnj, 2015/08/07 16:30:06]

Acknowledged.

+                j := 0
+                err := d.RawInterface.GetMulti(p.toGet, p.toGetMeta, func(pm ds.PropertyMap, err error) {
+                        i := p.idxMap[j]
+                        toSave := p.toSave[j]

[Vadim Sh., 2015/08/06 01:23:33]

ah, I guess appending nils is by design. Looked werid though. Consider using a
list of structs instead of two "parallel" arrays.

[Vadim Sh., 2015/08/06 01:23:33]

nit: move j++ to this line. I forgot what "j" was when saw it at the bottom.

+
+                        data := []byte(nil)
+
+                        // 0 == do not save
+                        // 1 == save
+                        // 2 == lock until next put
+                        saveState := 1
+                        if err == nil {
+                                p.decoded[i] = pm
+                                if toSave != nil {
+                                        data = encodeItemValue(pm)
+                                        if len(data) > internalValueSizeLimit {
+                                                saveState = 2

[Vadim Sh., 2015/08/06 01:23:33]

heh... smart :)

[iannucci, 2015/08/06 02:37:33]

managed to reduce this to just a boolean, but yeah, locking this one until the
next put is just an all around good idea.

+                                                d.log.Warningf("dscache: encoded entity too big (%d/%d)!",
+                                                        len(data), internalValueSizeLimit)
+                                        }
+                                }
+                        }
+
+                        if err != nil {

[dnj, 2015/08/05 18:32:17]

} else {

[iannucci, 2015/08/06 01:54:01]

Done.

+                                p.lme.Assign(i, err)
+                                if err != ds.ErrNoSuchEntity {
+                                        saveState = 0
+                                }
+                        }
+
+                        if saveState > 0 && toSave != nil {
+                                if saveState == 1 { // save
+                                        expSecs := metas.GetMetaDefault(i, CacheExpirationMeta, CacheTimeSeconds).(int64)
+                                        toSave.SetFlags(uint32(ItemHasData))
+                                        toSave.SetExpiration(time.Duration(expSecs) * time.Second)
+                                        toSave.SetValue(data)
+                                } else {
+                                        // Set a lock with an infinite timeout. No one else should try to
+                                        // serialize this item to memcache until something Put/Delete's it.
+                                        toSave.SetFlags(uint32(ItemHasLock))
+                                        toSave.SetExpiration(0)
+                                        toSave.SetValue(nonce)
+                                }
+                                toCas = append(toCas, toSave)
+                        }
+
+                        j++
+                })
+                if err != nil {
+                        return err
+                }
+                if len(toCas) > 0 {
+                        if err := d.mc.CompareAndSwapMulti(toCas); err != nil {
+                                d.log.Errorf("dscache: CompareAndSwapMulti: %s", err)
+                        }
+                }
+        }
+
+        for i, dec := range p.decoded {
+                cb(dec, p.lme.GetOne(i))
+        }
+
+        return nil
+}
+
+func (d *dsCache) RunInTransaction(f func(context.Context) error, opts *ds.TransactionOptions) error {
+        txnState := dsTxnState{}
+        err := d.RawInterface.RunInTransaction(func(ctx context.Context) error {
+                txnState.Reset()
+                err := f(context.WithValue(ctx, dsTxnCacheKey, &txnState))
+                if err == nil {
+                        txnState.Apply(d.supportContext)
+                }
+                return err
+        }, opts)
+        if err == nil {
+                txnState.Release(d.supportContext)
+        }
+        return err
+}