Bump deps

```
* use go 1.17 semantics in go.mod
* pin tonistiigi/fsutil dep to match docker/docker's dep
go get -u ./...
go mod vendor
```
This commit is contained in:
Jason Hall
2022-01-21 16:58:24 +00:00
committed by GitHub
parent a59970f4ac
commit c19ffc26a2
3595 changed files with 429699 additions and 120494 deletions
-3
View File
@@ -1,3 +0,0 @@
language: go
go:
- tip
+23
View File
@@ -0,0 +1,23 @@
# UNRELEASED
# 1.3.0 (September 17th, 2020)
FEATURES
* Add reverse tree traversal [[GH-30](https://github.com/hashicorp/go-immutable-radix/pull/30)]
# 1.2.0 (March 18th, 2020)
FEATURES
* Adds a `Clone` method to `Txn` allowing transactions to be split either into two independently mutable trees. [[GH-26](https://github.com/hashicorp/go-immutable-radix/pull/26)]
# 1.1.0 (May 22nd, 2019)
FEATURES
* Add `SeekLowerBound` to allow for range scans. [[GH-24](https://github.com/hashicorp/go-immutable-radix/pull/24)]
# 1.0.0 (August 30th, 2018)
* go mod adopted
+26 -1
View File
@@ -1,4 +1,4 @@
go-immutable-radix [![Build Status](https://travis-ci.org/hashicorp/go-immutable-radix.png)](https://travis-ci.org/hashicorp/go-immutable-radix)
go-immutable-radix [![CircleCI](https://circleci.com/gh/hashicorp/go-immutable-radix/tree/master.svg?style=svg)](https://circleci.com/gh/hashicorp/go-immutable-radix/tree/master)
=========
Provides the `iradix` package that implements an immutable [radix tree](http://en.wikipedia.org/wiki/Radix_tree).
@@ -39,3 +39,28 @@ if string(m) != "foo" {
}
```
Here is an example of performing a range scan of the keys.
```go
// Create a tree
r := iradix.New()
r, _, _ = r.Insert([]byte("001"), 1)
r, _, _ = r.Insert([]byte("002"), 2)
r, _, _ = r.Insert([]byte("005"), 5)
r, _, _ = r.Insert([]byte("010"), 10)
r, _, _ = r.Insert([]byte("100"), 10)
// Range scan over the keys that sort lexicographically between [003, 050)
it := r.Root().Iterator()
it.SeekLowerBound([]byte("003"))
for key, _, ok := it.Next(); ok; key, _, ok = it.Next() {
if key >= "050" {
break
}
fmt.Println(key)
}
// Output:
// 005
// 010
```
-6
View File
@@ -1,6 +0,0 @@
module github.com/hashicorp/go-immutable-radix
require (
github.com/hashicorp/go-uuid v1.0.0
github.com/hashicorp/golang-lru v0.5.0
)
-4
View File
@@ -1,4 +0,0 @@
github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCSnM=
github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+14
View File
@@ -86,6 +86,20 @@ func (t *Tree) Txn() *Txn {
return txn
}
// Clone makes an independent copy of the transaction. The new transaction
// does not track any nodes and has TrackMutate turned off. The cloned transaction will contain any uncommitted writes in the original transaction but further mutations to either will be independent and result in different radix trees on Commit. A cloned transaction may be passed to another goroutine and mutated there independently however each transaction may only be mutated in a single thread.
func (t *Txn) Clone() *Txn {
// reset the writable node cache to avoid leaking future writes into the clone
t.writable = nil
txn := &Txn{
root: t.root,
snap: t.snap,
size: t.size,
}
return txn
}
// TrackMutate can be used to toggle if mutations are tracked. If this is enabled
// then notifications will be issued for affected internal nodes and leaves when
// the transaction is committed.
+117 -3
View File
@@ -1,6 +1,8 @@
package iradix
import "bytes"
import (
"bytes"
)
// Iterator is used to iterate over a set of nodes
// in pre-order
@@ -18,7 +20,7 @@ func (i *Iterator) SeekPrefixWatch(prefix []byte) (watch <-chan struct{}) {
watch = n.mutateCh
search := prefix
for {
// Check for key exhaution
// Check for key exhaustion
if len(search) == 0 {
i.node = n
return
@@ -53,12 +55,124 @@ func (i *Iterator) SeekPrefix(prefix []byte) {
i.SeekPrefixWatch(prefix)
}
func (i *Iterator) recurseMin(n *Node) *Node {
// Traverse to the minimum child
if n.leaf != nil {
return n
}
nEdges := len(n.edges)
if nEdges > 1 {
// Add all the other edges to the stack (the min node will be added as
// we recurse)
i.stack = append(i.stack, n.edges[1:])
}
if nEdges > 0 {
return i.recurseMin(n.edges[0].node)
}
// Shouldn't be possible
return nil
}
// SeekLowerBound is used to seek the iterator to the smallest key that is
// greater or equal to the given key. There is no watch variant as it's hard to
// predict based on the radix structure which node(s) changes might affect the
// result.
func (i *Iterator) SeekLowerBound(key []byte) {
// Wipe the stack. Unlike Prefix iteration, we need to build the stack as we
// go because we need only a subset of edges of many nodes in the path to the
// leaf with the lower bound. Note that the iterator will still recurse into
// children that we don't traverse on the way to the reverse lower bound as it
// walks the stack.
i.stack = []edges{}
// i.node starts off in the common case as pointing to the root node of the
// tree. By the time we return we have either found a lower bound and setup
// the stack to traverse all larger keys, or we have not and the stack and
// node should both be nil to prevent the iterator from assuming it is just
// iterating the whole tree from the root node. Either way this needs to end
// up as nil so just set it here.
n := i.node
i.node = nil
search := key
found := func(n *Node) {
i.stack = append(i.stack, edges{edge{node: n}})
}
findMin := func(n *Node) {
n = i.recurseMin(n)
if n != nil {
found(n)
return
}
}
for {
// Compare current prefix with the search key's same-length prefix.
var prefixCmp int
if len(n.prefix) < len(search) {
prefixCmp = bytes.Compare(n.prefix, search[0:len(n.prefix)])
} else {
prefixCmp = bytes.Compare(n.prefix, search)
}
if prefixCmp > 0 {
// Prefix is larger, that means the lower bound is greater than the search
// and from now on we need to follow the minimum path to the smallest
// leaf under this subtree.
findMin(n)
return
}
if prefixCmp < 0 {
// Prefix is smaller than search prefix, that means there is no lower
// bound
i.node = nil
return
}
// Prefix is equal, we are still heading for an exact match. If this is a
// leaf and an exact match we're done.
if n.leaf != nil && bytes.Equal(n.leaf.key, key) {
found(n)
return
}
// Consume the search prefix if the current node has one. Note that this is
// safe because if n.prefix is longer than the search slice prefixCmp would
// have been > 0 above and the method would have already returned.
search = search[len(n.prefix):]
if len(search) == 0 {
// We've exhausted the search key, but the current node is not an exact
// match or not a leaf. That means that the leaf value if it exists, and
// all child nodes must be strictly greater, the smallest key in this
// subtree must be the lower bound.
findMin(n)
return
}
// Otherwise, take the lower bound next edge.
idx, lbNode := n.getLowerBoundEdge(search[0])
if lbNode == nil {
return
}
// Create stack edges for the all strictly higher edges in this node.
if idx+1 < len(n.edges) {
i.stack = append(i.stack, n.edges[idx+1:])
}
// Recurse
n = lbNode
}
}
// Next returns the next node in order
func (i *Iterator) Next() ([]byte, interface{}, bool) {
// Initialize our stack if needed
if i.stack == nil && i.node != nil {
i.stack = []edges{
edges{
{
edge{node: i.node},
},
}
+42
View File
@@ -79,6 +79,18 @@ func (n *Node) getEdge(label byte) (int, *Node) {
return -1, nil
}
func (n *Node) getLowerBoundEdge(label byte) (int, *Node) {
num := len(n.edges)
idx := sort.Search(num, func(i int) bool {
return n.edges[i].label >= label
})
// we want lower bound behavior so return even if it's not an exact match
if idx < num {
return idx, n.edges[idx].node
}
return -1, nil
}
func (n *Node) delEdge(label byte) {
num := len(n.edges)
idx := sort.Search(num, func(i int) bool {
@@ -199,6 +211,12 @@ func (n *Node) Iterator() *Iterator {
return &Iterator{node: n}
}
// ReverseIterator is used to return an iterator at
// the given node to walk the tree backwards
func (n *Node) ReverseIterator() *ReverseIterator {
return NewReverseIterator(n)
}
// rawIterator is used to return a raw iterator at the given node to walk the
// tree.
func (n *Node) rawIterator() *rawIterator {
@@ -212,6 +230,11 @@ func (n *Node) Walk(fn WalkFn) {
recursiveWalk(n, fn)
}
// WalkBackwards is used to walk the tree in reverse order
func (n *Node) WalkBackwards(fn WalkFn) {
reverseRecursiveWalk(n, fn)
}
// WalkPrefix is used to walk the tree under a prefix
func (n *Node) WalkPrefix(prefix []byte, fn WalkFn) {
search := prefix
@@ -290,3 +313,22 @@ func recursiveWalk(n *Node, fn WalkFn) bool {
}
return false
}
// reverseRecursiveWalk is used to do a reverse pre-order
// walk of a node recursively. Returns true if the walk
// should be aborted
func reverseRecursiveWalk(n *Node, fn WalkFn) bool {
// Visit the leaf values if any
if n.leaf != nil && fn(n.leaf.key, n.leaf.val) {
return true
}
// Recurse on the children in reverse order
for i := len(n.edges) - 1; i >= 0; i-- {
e := n.edges[i]
if reverseRecursiveWalk(e.node, fn) {
return true
}
}
return false
}
+1 -1
View File
@@ -41,7 +41,7 @@ func (i *rawIterator) Next() {
// Initialize our stack if needed.
if i.stack == nil && i.node != nil {
i.stack = []rawStackEntry{
rawStackEntry{
{
edges: edges{
edge{node: i.node},
},
+239
View File
@@ -0,0 +1,239 @@
package iradix
import (
"bytes"
)
// ReverseIterator is used to iterate over a set of nodes
// in reverse in-order
type ReverseIterator struct {
i *Iterator
// expandedParents stores the set of parent nodes whose relevant children have
// already been pushed into the stack. This can happen during seek or during
// iteration.
//
// Unlike forward iteration we need to recurse into children before we can
// output the value stored in an internal leaf since all children are greater.
// We use this to track whether we have already ensured all the children are
// in the stack.
expandedParents map[*Node]struct{}
}
// NewReverseIterator returns a new ReverseIterator at a node
func NewReverseIterator(n *Node) *ReverseIterator {
return &ReverseIterator{
i: &Iterator{node: n},
}
}
// SeekPrefixWatch is used to seek the iterator to a given prefix
// and returns the watch channel of the finest granularity
func (ri *ReverseIterator) SeekPrefixWatch(prefix []byte) (watch <-chan struct{}) {
return ri.i.SeekPrefixWatch(prefix)
}
// SeekPrefix is used to seek the iterator to a given prefix
func (ri *ReverseIterator) SeekPrefix(prefix []byte) {
ri.i.SeekPrefixWatch(prefix)
}
// SeekReverseLowerBound is used to seek the iterator to the largest key that is
// lower or equal to the given key. There is no watch variant as it's hard to
// predict based on the radix structure which node(s) changes might affect the
// result.
func (ri *ReverseIterator) SeekReverseLowerBound(key []byte) {
// Wipe the stack. Unlike Prefix iteration, we need to build the stack as we
// go because we need only a subset of edges of many nodes in the path to the
// leaf with the lower bound. Note that the iterator will still recurse into
// children that we don't traverse on the way to the reverse lower bound as it
// walks the stack.
ri.i.stack = []edges{}
// ri.i.node starts off in the common case as pointing to the root node of the
// tree. By the time we return we have either found a lower bound and setup
// the stack to traverse all larger keys, or we have not and the stack and
// node should both be nil to prevent the iterator from assuming it is just
// iterating the whole tree from the root node. Either way this needs to end
// up as nil so just set it here.
n := ri.i.node
ri.i.node = nil
search := key
if ri.expandedParents == nil {
ri.expandedParents = make(map[*Node]struct{})
}
found := func(n *Node) {
ri.i.stack = append(ri.i.stack, edges{edge{node: n}})
// We need to mark this node as expanded in advance too otherwise the
// iterator will attempt to walk all of its children even though they are
// greater than the lower bound we have found. We've expanded it in the
// sense that all of its children that we want to walk are already in the
// stack (i.e. none of them).
ri.expandedParents[n] = struct{}{}
}
for {
// Compare current prefix with the search key's same-length prefix.
var prefixCmp int
if len(n.prefix) < len(search) {
prefixCmp = bytes.Compare(n.prefix, search[0:len(n.prefix)])
} else {
prefixCmp = bytes.Compare(n.prefix, search)
}
if prefixCmp < 0 {
// Prefix is smaller than search prefix, that means there is no exact
// match for the search key. But we are looking in reverse, so the reverse
// lower bound will be the largest leaf under this subtree, since it is
// the value that would come right before the current search key if it
// were in the tree. So we need to follow the maximum path in this subtree
// to find it. Note that this is exactly what the iterator will already do
// if it finds a node in the stack that has _not_ been marked as expanded
// so in this one case we don't call `found` and instead let the iterator
// do the expansion and recursion through all the children.
ri.i.stack = append(ri.i.stack, edges{edge{node: n}})
return
}
if prefixCmp > 0 {
// Prefix is larger than search prefix, or there is no prefix but we've
// also exhausted the search key. Either way, that means there is no
// reverse lower bound since nothing comes before our current search
// prefix.
return
}
// If this is a leaf, something needs to happen! Note that if it's a leaf
// and prefixCmp was zero (which it must be to get here) then the leaf value
// is either an exact match for the search, or it's lower. It can't be
// greater.
if n.isLeaf() {
// Firstly, if it's an exact match, we're done!
if bytes.Equal(n.leaf.key, key) {
found(n)
return
}
// It's not so this node's leaf value must be lower and could still be a
// valid contender for reverse lower bound.
// If it has no children then we are also done.
if len(n.edges) == 0 {
// This leaf is the lower bound.
found(n)
return
}
// Finally, this leaf is internal (has children) so we'll keep searching,
// but we need to add it to the iterator's stack since it has a leaf value
// that needs to be iterated over. It needs to be added to the stack
// before its children below as it comes first.
ri.i.stack = append(ri.i.stack, edges{edge{node: n}})
// We also need to mark it as expanded since we'll be adding any of its
// relevant children below and so don't want the iterator to re-add them
// on its way back up the stack.
ri.expandedParents[n] = struct{}{}
}
// Consume the search prefix. Note that this is safe because if n.prefix is
// longer than the search slice prefixCmp would have been > 0 above and the
// method would have already returned.
search = search[len(n.prefix):]
if len(search) == 0 {
// We've exhausted the search key but we are not at a leaf. That means all
// children are greater than the search key so a reverse lower bound
// doesn't exist in this subtree. Note that there might still be one in
// the whole radix tree by following a different path somewhere further
// up. If that's the case then the iterator's stack will contain all the
// smaller nodes already and Previous will walk through them correctly.
return
}
// Otherwise, take the lower bound next edge.
idx, lbNode := n.getLowerBoundEdge(search[0])
// From here, we need to update the stack with all values lower than
// the lower bound edge. Since getLowerBoundEdge() returns -1 when the
// search prefix is larger than all edges, we need to place idx at the
// last edge index so they can all be place in the stack, since they
// come before our search prefix.
if idx == -1 {
idx = len(n.edges)
}
// Create stack edges for the all strictly lower edges in this node.
if len(n.edges[:idx]) > 0 {
ri.i.stack = append(ri.i.stack, n.edges[:idx])
}
// Exit if there's no lower bound edge. The stack will have the previous
// nodes already.
if lbNode == nil {
return
}
// Recurse
n = lbNode
}
}
// Previous returns the previous node in reverse order
func (ri *ReverseIterator) Previous() ([]byte, interface{}, bool) {
// Initialize our stack if needed
if ri.i.stack == nil && ri.i.node != nil {
ri.i.stack = []edges{
{
edge{node: ri.i.node},
},
}
}
if ri.expandedParents == nil {
ri.expandedParents = make(map[*Node]struct{})
}
for len(ri.i.stack) > 0 {
// Inspect the last element of the stack
n := len(ri.i.stack)
last := ri.i.stack[n-1]
m := len(last)
elem := last[m-1].node
_, alreadyExpanded := ri.expandedParents[elem]
// If this is an internal node and we've not seen it already, we need to
// leave it in the stack so we can return its possible leaf value _after_
// we've recursed through all its children.
if len(elem.edges) > 0 && !alreadyExpanded {
// record that we've seen this node!
ri.expandedParents[elem] = struct{}{}
// push child edges onto stack and skip the rest of the loop to recurse
// into the largest one.
ri.i.stack = append(ri.i.stack, elem.edges)
continue
}
// Remove the node from the stack
if m > 1 {
ri.i.stack[n-1] = last[:m-1]
} else {
ri.i.stack = ri.i.stack[:n-1]
}
// We don't need this state any more as it's no longer in the stack so we
// won't visit it again
if alreadyExpanded {
delete(ri.expandedParents, elem)
}
// If this is a leaf, return it
if elem.leaf != nil {
return elem.leaf.key, elem.leaf.val, true
}
// it's not a leaf so keep walking the stack to find the previous leaf
}
return nil, nil, false
}
+2
View File
@@ -22,3 +22,5 @@ _testmain.go
*.exe
*.test
*.prof
.idea
-7
View File
@@ -1,7 +0,0 @@
language: go
go:
- "1.10"
script:
- go test
+74 -26
View File
@@ -1,4 +1,4 @@
# go-memdb
# go-memdb [![CircleCI](https://circleci.com/gh/hashicorp/go-memdb/tree/master.svg?style=svg)](https://circleci.com/gh/hashicorp/go-memdb/tree/master)
Provides the `memdb` package that implements a simple in-memory database
built on immutable radix trees. The database provides Atomicity, Consistency
@@ -21,7 +21,7 @@ The database provides the following:
a single field index, or more advanced compound field indexes. Certain types like
UUID can be efficiently compressed from strings into byte indexes for reduced
storage requirements.
* Watches - Callers can populate a watch set as part of a query, which can be used to
detect when a modification has been made to the database which affects the query
results. This lets callers easily watch for changes in the database in a very general
@@ -32,50 +32,62 @@ For the underlying immutable radix trees, see [go-immutable-radix](https://githu
Documentation
=============
The full documentation is available on [Godoc](http://godoc.org/github.com/hashicorp/go-memdb).
The full documentation is available on [Godoc](https://pkg.go.dev/github.com/hashicorp/go-memdb).
Example
=======
Below is a simple example of usage
Below is a [simple example](https://play.golang.org/p/gCGE9FA4og1) of usage
```go
// Create a sample struct
type Person struct {
Email string
Name string
Age int
Email string
Name string
Age int
}
// Create the DB schema
schema := &memdb.DBSchema{
Tables: map[string]*memdb.TableSchema{
"person": &memdb.TableSchema{
Name: "person",
Indexes: map[string]*memdb.IndexSchema{
"id": &memdb.IndexSchema{
Name: "id",
Unique: true,
Indexer: &memdb.StringFieldIndex{Field: "Email"},
},
},
},
},
Tables: map[string]*memdb.TableSchema{
"person": &memdb.TableSchema{
Name: "person",
Indexes: map[string]*memdb.IndexSchema{
"id": &memdb.IndexSchema{
Name: "id",
Unique: true,
Indexer: &memdb.StringFieldIndex{Field: "Email"},
},
"age": &memdb.IndexSchema{
Name: "age",
Unique: false,
Indexer: &memdb.IntFieldIndex{Field: "Age"},
},
},
},
},
}
// Create a new data base
db, err := memdb.NewMemDB(schema)
if err != nil {
panic(err)
panic(err)
}
// Create a write transaction
txn := db.Txn(true)
// Insert a new person
p := &Person{"joe@aol.com", "Joe", 30}
if err := txn.Insert("person", p); err != nil {
panic(err)
// Insert some people
people := []*Person{
&Person{"joe@aol.com", "Joe", 30},
&Person{"lucy@aol.com", "Lucy", 35},
&Person{"tariq@aol.com", "Tariq", 21},
&Person{"dorothy@aol.com", "Dorothy", 53},
}
for _, p := range people {
if err := txn.Insert("person", p); err != nil {
panic(err)
}
}
// Commit the transaction
@@ -88,11 +100,47 @@ defer txn.Abort()
// Lookup by email
raw, err := txn.First("person", "id", "joe@aol.com")
if err != nil {
panic(err)
panic(err)
}
// Say hi!
fmt.Printf("Hello %s!", raw.(*Person).Name)
fmt.Printf("Hello %s!\n", raw.(*Person).Name)
// List all the people
it, err := txn.Get("person", "id")
if err != nil {
panic(err)
}
fmt.Println("All the people:")
for obj := it.Next(); obj != nil; obj = it.Next() {
p := obj.(*Person)
fmt.Printf(" %s\n", p.Name)
}
// Range scan over people with ages between 25 and 35 inclusive
it, err = txn.LowerBound("person", "age", 25)
if err != nil {
panic(err)
}
fmt.Println("People aged 25 - 35:")
for obj := it.Next(); obj != nil; obj = it.Next() {
p := obj.(*Person)
if p.Age > 35 {
break
}
fmt.Printf(" %s is aged %d\n", p.Name, p.Age)
}
// Output:
// Hello Joe!
// All the people:
// Dorothy
// Joe
// Lucy
// Tariq
// People aged 25 - 35:
// Joe is aged 30
// Lucy is aged 35
```
+34
View File
@@ -0,0 +1,34 @@
package memdb
// Changes describes a set of mutations to memDB tables performed during a
// transaction.
type Changes []Change
// Change describes a mutation to an object in a table.
type Change struct {
Table string
Before interface{}
After interface{}
// primaryKey stores the raw key value from the primary index so that we can
// de-duplicate multiple updates of the same object in the same transaction
// but we don't expose this implementation detail to the consumer.
primaryKey []byte
}
// Created returns true if the mutation describes a new object being inserted.
func (m *Change) Created() bool {
return m.Before == nil && m.After != nil
}
// Updated returns true if the mutation describes an existing object being
// updated.
func (m *Change) Updated() bool {
return m.Before != nil && m.After != nil
}
// Deleted returns true if the mutation describes an existing object being
// deleted.
func (m *Change) Deleted() bool {
return m.Before != nil && m.After == nil
}
+8 -3
View File
@@ -13,17 +13,22 @@ type FilterIterator struct {
iter ResultIterator
}
func NewFilterIterator(wrap ResultIterator, filter FilterFunc) *FilterIterator {
// NewFilterIterator wraps a ResultIterator. The filter function is applied
// to each value returned by a call to iter.Next.
//
// See the documentation for ResultIterator to understand the behaviour of the
// returned FilterIterator.
func NewFilterIterator(iter ResultIterator, filter FilterFunc) *FilterIterator {
return &FilterIterator{
filter: filter,
iter: wrap,
iter: iter,
}
}
// WatchCh returns the watch channel of the wrapped iterator.
func (f *FilterIterator) WatchCh() <-chan struct{} { return f.iter.WatchCh() }
// Next returns the next non-filtered result from the wrapped iterator
// Next returns the next non-filtered result from the wrapped iterator.
func (f *FilterIterator) Next() interface{} {
for {
if value := f.iter.Next(); value == nil || !f.filter(value) {
+330 -12
View File
@@ -3,7 +3,9 @@ package memdb
import (
"encoding/binary"
"encoding/hex"
"errors"
"fmt"
"math/bits"
"reflect"
"strings"
)
@@ -63,9 +65,16 @@ func (s *StringFieldIndex) FromObject(obj interface{}) (bool, []byte, error) {
v = reflect.Indirect(v) // Dereference the pointer if any
fv := v.FieldByName(s.Field)
if !fv.IsValid() {
isPtr := fv.Kind() == reflect.Ptr
fv = reflect.Indirect(fv)
if !isPtr && !fv.IsValid() {
return false, nil,
fmt.Errorf("field '%s' for %#v is invalid", s.Field, obj)
fmt.Errorf("field '%s' for %#v is invalid %v ", s.Field, obj, isPtr)
}
if isPtr && !fv.IsValid() {
val := ""
return false, []byte(val), nil
}
val := fv.String()
@@ -188,6 +197,16 @@ func (s *StringSliceFieldIndex) PrefixFromArgs(args ...interface{}) ([]byte, err
// StringMapFieldIndex is used to extract a field of type map[string]string
// from an object using reflection and builds an index on that field.
//
// Note that although FromArgs in theory supports using either one or
// two arguments, there is a bug: FromObject only creates an index
// using key/value, and does not also create an index using key. This
// means a lookup using one argument will never actually work.
//
// It is currently left as-is to prevent backwards compatibility
// issues.
//
// TODO: Fix this in the next major bump.
type StringMapFieldIndex struct {
Field string
Lowercase bool
@@ -233,6 +252,8 @@ func (s *StringMapFieldIndex) FromObject(obj interface{}) (bool, [][]byte, error
return true, vals, nil
}
// WARNING: Because of a bug in FromObject, this function will never return
// a value when using the single-argument version.
func (s *StringMapFieldIndex) FromArgs(args ...interface{}) ([]byte, error) {
if len(args) > 2 || len(args) == 0 {
return nil, fmt.Errorf("must provide one or two arguments")
@@ -262,6 +283,79 @@ func (s *StringMapFieldIndex) FromArgs(args ...interface{}) ([]byte, error) {
return []byte(key), nil
}
// IntFieldIndex is used to extract an int field from an object using
// reflection and builds an index on that field.
type IntFieldIndex struct {
Field string
}
func (i *IntFieldIndex) FromObject(obj interface{}) (bool, []byte, error) {
v := reflect.ValueOf(obj)
v = reflect.Indirect(v) // Dereference the pointer if any
fv := v.FieldByName(i.Field)
if !fv.IsValid() {
return false, nil,
fmt.Errorf("field '%s' for %#v is invalid", i.Field, obj)
}
// Check the type
k := fv.Kind()
size, ok := IsIntType(k)
if !ok {
return false, nil, fmt.Errorf("field %q is of type %v; want an int", i.Field, k)
}
// Get the value and encode it
val := fv.Int()
buf := make([]byte, size)
binary.PutVarint(buf, val)
return true, buf, nil
}
func (i *IntFieldIndex) FromArgs(args ...interface{}) ([]byte, error) {
if len(args) != 1 {
return nil, fmt.Errorf("must provide only a single argument")
}
v := reflect.ValueOf(args[0])
if !v.IsValid() {
return nil, fmt.Errorf("%#v is invalid", args[0])
}
k := v.Kind()
size, ok := IsIntType(k)
if !ok {
return nil, fmt.Errorf("arg is of type %v; want a int", k)
}
val := v.Int()
buf := make([]byte, size)
binary.PutVarint(buf, val)
return buf, nil
}
// IsIntType returns whether the passed type is a type of int and the number
// of bytes needed to encode the type.
func IsIntType(k reflect.Kind) (size int, okay bool) {
switch k {
case reflect.Int:
return binary.MaxVarintLen64, true
case reflect.Int8:
return 2, true
case reflect.Int16:
return binary.MaxVarintLen16, true
case reflect.Int32:
return binary.MaxVarintLen32, true
case reflect.Int64:
return binary.MaxVarintLen64, true
default:
return 0, false
}
}
// UintFieldIndex is used to extract a uint field from an object using
// reflection and builds an index on that field.
type UintFieldIndex struct {
@@ -287,8 +381,7 @@ func (u *UintFieldIndex) FromObject(obj interface{}) (bool, []byte, error) {
// Get the value and encode it
val := fv.Uint()
buf := make([]byte, size)
binary.PutUvarint(buf, val)
buf := encodeUInt(val, size)
return true, buf, nil
}
@@ -310,31 +403,82 @@ func (u *UintFieldIndex) FromArgs(args ...interface{}) ([]byte, error) {
}
val := v.Uint()
buf := make([]byte, size)
binary.PutUvarint(buf, val)
buf := encodeUInt(val, size)
return buf, nil
}
func encodeUInt(val uint64, size int) []byte {
buf := make([]byte, size)
switch size {
case 1:
buf[0] = uint8(val)
case 2:
binary.BigEndian.PutUint16(buf, uint16(val))
case 4:
binary.BigEndian.PutUint32(buf, uint32(val))
case 8:
binary.BigEndian.PutUint64(buf, val)
}
return buf
}
// IsUintType returns whether the passed type is a type of uint and the number
// of bytes needed to encode the type.
func IsUintType(k reflect.Kind) (size int, okay bool) {
switch k {
case reflect.Uint:
return binary.MaxVarintLen64, true
return bits.UintSize / 8, true
case reflect.Uint8:
return 2, true
return 1, true
case reflect.Uint16:
return binary.MaxVarintLen16, true
return 2, true
case reflect.Uint32:
return binary.MaxVarintLen32, true
return 4, true
case reflect.Uint64:
return binary.MaxVarintLen64, true
return 8, true
default:
return 0, false
}
}
// BoolFieldIndex is used to extract an boolean field from an object using
// reflection and builds an index on that field.
type BoolFieldIndex struct {
Field string
}
func (i *BoolFieldIndex) FromObject(obj interface{}) (bool, []byte, error) {
v := reflect.ValueOf(obj)
v = reflect.Indirect(v) // Dereference the pointer if any
fv := v.FieldByName(i.Field)
if !fv.IsValid() {
return false, nil,
fmt.Errorf("field '%s' for %#v is invalid", i.Field, obj)
}
// Check the type
k := fv.Kind()
if k != reflect.Bool {
return false, nil, fmt.Errorf("field %q is of type %v; want a bool", i.Field, k)
}
// Get the value and encode it
buf := make([]byte, 1)
if fv.Bool() {
buf[0] = 1
}
return true, buf, nil
}
func (i *BoolFieldIndex) FromArgs(args ...interface{}) ([]byte, error) {
return fromBoolArgs(args)
}
// UUIDFieldIndex is used to extract a field from an object
// using reflection and builds an index on that field by treating
// it as a UUID. This is an optimization to using a StringFieldIndex
@@ -540,7 +684,7 @@ func (c *CompoundIndex) FromObject(raw interface{}) (bool, []byte, error) {
func (c *CompoundIndex) FromArgs(args ...interface{}) ([]byte, error) {
if len(args) != len(c.Indexes) {
return nil, fmt.Errorf("less arguments than index fields")
return nil, fmt.Errorf("non-equivalent argument count and index fields")
}
var out []byte
for i, arg := range args {
@@ -579,3 +723,177 @@ func (c *CompoundIndex) PrefixFromArgs(args ...interface{}) ([]byte, error) {
}
return out, nil
}
// CompoundMultiIndex is used to build an index using multiple
// sub-indexes.
//
// Unlike CompoundIndex, CompoundMultiIndex can have both
// SingleIndexer and MultiIndexer sub-indexers. However, each
// MultiIndexer adds considerable overhead/complexity in terms of
// the number of indexes created under-the-hood. It is not suggested
// to use more than one or two, if possible.
//
// Another change from CompoundIndexer is that if AllowMissing is
// set, not only is it valid to have empty index fields, but it will
// still create index values up to the first empty index. This means
// that if you have a value with an empty field, rather than using a
// prefix for lookup, you can simply pass in less arguments. As an
// example, if {Foo, Bar} is indexed but Bar is missing for a value
// and AllowMissing is set, an index will still be created for {Foo}
// and it is valid to do a lookup passing in only Foo as an argument.
// Note that the ordering isn't guaranteed -- it's last-insert wins,
// but this is true if you have two objects that have the same
// indexes not using AllowMissing anyways.
//
// Because StringMapFieldIndexers can take a varying number of args,
// it is currently a requirement that whenever it is used, two
// arguments must _always_ be provided for it. In theory we only
// need one, except a bug in that indexer means the single-argument
// version will never work. You can leave the second argument nil,
// but it will never produce a value. We support this for whenever
// that bug is fixed, likely in a next major version bump.
//
// Prefix-based indexing is not currently supported.
type CompoundMultiIndex struct {
Indexes []Indexer
// AllowMissing results in an index based on only the indexers
// that return data. If true, you may end up with 2/3 columns
// indexed which might be useful for an index scan. Otherwise,
// CompoundMultiIndex requires all indexers to be satisfied.
AllowMissing bool
}
func (c *CompoundMultiIndex) FromObject(raw interface{}) (bool, [][]byte, error) {
// At each entry, builder is storing the results from the next index
builder := make([][][]byte, 0, len(c.Indexes))
// Start with something higher to avoid resizing if possible
out := make([][]byte, 0, len(c.Indexes)^3)
forloop:
// This loop goes through each indexer and adds the value(s) provided to the next
// entry in the slice. We can then later walk it like a tree to construct the indices.
for i, idxRaw := range c.Indexes {
switch idx := idxRaw.(type) {
case SingleIndexer:
ok, val, err := idx.FromObject(raw)
if err != nil {
return false, nil, fmt.Errorf("single sub-index %d error: %v", i, err)
}
if !ok {
if c.AllowMissing {
break forloop
} else {
return false, nil, nil
}
}
builder = append(builder, [][]byte{val})
case MultiIndexer:
ok, vals, err := idx.FromObject(raw)
if err != nil {
return false, nil, fmt.Errorf("multi sub-index %d error: %v", i, err)
}
if !ok {
if c.AllowMissing {
break forloop
} else {
return false, nil, nil
}
}
// Add each of the new values to each of the old values
builder = append(builder, vals)
default:
return false, nil, fmt.Errorf("sub-index %d does not satisfy either SingleIndexer or MultiIndexer", i)
}
}
// We are walking through the builder slice essentially in a depth-first fashion,
// building the prefix and leaves as we go. If AllowMissing is false, we only insert
// these full paths to leaves. Otherwise, we also insert each prefix along the way.
// This allows for lookup in FromArgs when AllowMissing is true that does not contain
// the full set of arguments. e.g. for {Foo, Bar} where an object has only the Foo
// field specified as "abc", it is valid to call FromArgs with just "abc".
var walkVals func([]byte, int)
walkVals = func(currPrefix []byte, depth int) {
if depth == len(builder)-1 {
// These are the "leaves", so append directly
for _, v := range builder[depth] {
out = append(out, append(currPrefix, v...))
}
return
}
for _, v := range builder[depth] {
nextPrefix := append(currPrefix, v...)
if c.AllowMissing {
out = append(out, nextPrefix)
}
walkVals(nextPrefix, depth+1)
}
}
walkVals(nil, 0)
return true, out, nil
}
func (c *CompoundMultiIndex) FromArgs(args ...interface{}) ([]byte, error) {
var stringMapCount int
var argCount int
for _, index := range c.Indexes {
if argCount >= len(args) {
break
}
if _, ok := index.(*StringMapFieldIndex); ok {
// We require pairs for StringMapFieldIndex, but only got one
if argCount+1 >= len(args) {
return nil, errors.New("invalid number of arguments")
}
stringMapCount++
argCount += 2
} else {
argCount++
}
}
argCount = 0
switch c.AllowMissing {
case true:
if len(args) > len(c.Indexes)+stringMapCount {
return nil, errors.New("too many arguments")
}
default:
if len(args) != len(c.Indexes)+stringMapCount {
return nil, errors.New("number of arguments does not equal number of indexers")
}
}
var out []byte
var val []byte
var err error
for i, idx := range c.Indexes {
if argCount >= len(args) {
// We're done; should only hit this if AllowMissing
break
}
if _, ok := idx.(*StringMapFieldIndex); ok {
if args[argCount+1] == nil {
val, err = idx.FromArgs(args[argCount])
} else {
val, err = idx.FromArgs(args[argCount : argCount+2]...)
}
argCount += 2
} else {
val, err = idx.FromArgs(args[argCount])
argCount++
}
if err != nil {
return nil, fmt.Errorf("sub-index %d error: %v", i, err)
}
out = append(out, val...)
}
return out, nil
}
+17 -6
View File
@@ -10,11 +10,19 @@ import (
"github.com/hashicorp/go-immutable-radix"
)
// MemDB is an in-memory database.
// MemDB is an in-memory database providing Atomicity, Consistency, and
// Isolation from ACID. MemDB doesn't provide Durability since it is an
// in-memory database.
//
// MemDB provides a table abstraction to store objects (rows) with multiple
// indexes based on inserted values. The database makes use of immutable radix
// trees to provide transactions and MVCC.
//
// Objects inserted into MemDB are not copied. It is **extremely important**
// that objects are not modified in-place after they are inserted since they
// are stored directly in MemDB. It remains unsafe to modify inserted objects
// even after they've been deleted from MemDB since there may still be older
// snapshots of the DB being read from other goroutines.
type MemDB struct {
schema *DBSchema
root unsafe.Pointer // *iradix.Tree underneath
@@ -24,7 +32,7 @@ type MemDB struct {
writer sync.Mutex
}
// NewMemDB creates a new MemDB with the given schema
// NewMemDB creates a new MemDB with the given schema.
func NewMemDB(schema *DBSchema) (*MemDB, error) {
// Validate the schema
if err := schema.Validate(); err != nil {
@@ -50,7 +58,7 @@ func (db *MemDB) getRoot() *iradix.Tree {
return root
}
// Txn is used to start a new transaction, in either read or write mode.
// Txn is used to start a new transaction in either read or write mode.
// There can only be a single concurrent writer, but any number of readers.
func (db *MemDB) Txn(write bool) *Txn {
if write {
@@ -64,9 +72,12 @@ func (db *MemDB) Txn(write bool) *Txn {
return txn
}
// Snapshot is used to capture a point-in-time snapshot
// of the database that will not be affected by any write
// operations to the existing DB.
// Snapshot is used to capture a point-in-time snapshot of the database that
// will not be affected by any write operations to the existing DB.
//
// If MemDB is storing reference-based values (pointers, maps, slices, etc.),
// the Snapshot will not deep copy those values. Therefore, it is still unsafe
// to modify any inserted values in either DB.
func (db *MemDB) Snapshot() *MemDB {
clone := &MemDB{
schema: db.schema,
+358 -21
View File
@@ -7,7 +7,7 @@ import (
"sync/atomic"
"unsafe"
"github.com/hashicorp/go-immutable-radix"
iradix "github.com/hashicorp/go-immutable-radix"
)
const (
@@ -33,19 +33,35 @@ type Txn struct {
rootTxn *iradix.Txn
after []func()
// changes is used to track the changes performed during the transaction. If
// it is nil at transaction start then changes are not tracked.
changes Changes
modified map[tableIndex]*iradix.Txn
}
// readableIndex returns a transaction usable for reading the given
// index in a table. If a write transaction is in progress, we may need
// to use an existing modified txn.
// TrackChanges enables change tracking for the transaction. If called at any
// point before commit, subsequent mutations will be recorded and can be
// retrieved using ChangeSet. Once this has been called on a transaction it
// can't be unset. As with other Txn methods it's not safe to call this from a
// different goroutine than the one making mutations or committing the
// transaction.
func (txn *Txn) TrackChanges() {
if txn.changes == nil {
txn.changes = make(Changes, 0, 1)
}
}
// readableIndex returns a transaction usable for reading the given index in a
// table. If the transaction is a write transaction with modifications, a clone of the
// modified index will be returned.
func (txn *Txn) readableIndex(table, index string) *iradix.Txn {
// Look for existing transaction
if txn.write && txn.modified != nil {
key := tableIndex{table, index}
exist, ok := txn.modified[key]
if ok {
return exist
return exist.Clone()
}
}
@@ -101,6 +117,7 @@ func (txn *Txn) Abort() {
// Clear the txn
txn.rootTxn = nil
txn.modified = nil
txn.changes = nil
// Release the writer lock since this is invalid
txn.db.writer.Unlock()
@@ -153,7 +170,11 @@ func (txn *Txn) Commit() {
}
}
// Insert is used to add or update an object into the given table
// Insert is used to add or update an object into the given table.
//
// When updating an object, the obj provided should be a copy rather
// than a value updated in-place. Modifying values in-place that are already
// inserted into MemDB is not supported behavior.
func (txn *Txn) Insert(table string, obj interface{}) error {
if !txn.write {
return fmt.Errorf("cannot insert in read-only transaction")
@@ -265,11 +286,19 @@ func (txn *Txn) Insert(table string, obj interface{}) error {
indexTxn.Insert(val, obj)
}
}
if txn.changes != nil {
txn.changes = append(txn.changes, Change{
Table: table,
Before: existing, // might be nil on a create
After: obj,
primaryKey: idVal,
})
}
return nil
}
// Delete is used to delete a single object from the given table
// This object must already exist in the table
// Delete is used to delete a single object from the given table.
// This object must already exist in the table.
func (txn *Txn) Delete(table string, obj interface{}) error {
if !txn.write {
return fmt.Errorf("cannot delete in read-only transaction")
@@ -332,6 +361,14 @@ func (txn *Txn) Delete(table string, obj interface{}) error {
}
}
}
if txn.changes != nil {
txn.changes = append(txn.changes, Change{
Table: table,
Before: existing,
After: nil, // Now nil indicates deletion
primaryKey: idVal,
})
}
return nil
}
@@ -376,6 +413,19 @@ func (txn *Txn) DeletePrefix(table string, prefix_index string, prefix string) (
if !ok {
return false, fmt.Errorf("object missing primary index")
}
if txn.changes != nil {
// Record the deletion
idTxn := txn.writableIndex(table, id)
existing, ok := idTxn.Get(idVal)
if ok {
txn.changes = append(txn.changes, Change{
Table: table,
Before: existing,
After: nil, // Now nil indicates deletion
primaryKey: idVal,
})
}
}
// Remove the object from all the indexes except the given prefix index
for name, indexSchema := range tableSchema.Indexes {
if name == deletePrefixIndex {
@@ -413,6 +463,7 @@ func (txn *Txn) DeletePrefix(table string, prefix_index string, prefix string) (
}
}
}
}
if foundAny {
indexTxn := txn.writableIndex(table, deletePrefixIndex)
@@ -489,6 +540,34 @@ func (txn *Txn) FirstWatch(table, index string, args ...interface{}) (<-chan str
return watch, value, nil
}
// LastWatch is used to return the last matching object for
// the given constraints on the index along with the watch channel
func (txn *Txn) LastWatch(table, index string, args ...interface{}) (<-chan struct{}, interface{}, error) {
// Get the index value
indexSchema, val, err := txn.getIndexValue(table, index, args...)
if err != nil {
return nil, nil, err
}
// Get the index itself
indexTxn := txn.readableIndex(table, indexSchema.Name)
// Do an exact lookup
if indexSchema.Unique && val != nil && indexSchema.Name == index {
watch, obj, ok := indexTxn.GetWatch(val)
if !ok {
return watch, nil, nil
}
return watch, obj, nil
}
// Handle non-unique index by using an iterator and getting the last value
iter := indexTxn.Root().ReverseIterator()
watch := iter.SeekPrefixWatch(val)
_, value, _ := iter.Previous()
return watch, value, nil
}
// First is used to return the first matching object for
// the given constraints on the index
func (txn *Txn) First(table, index string, args ...interface{}) (interface{}, error) {
@@ -496,6 +575,13 @@ func (txn *Txn) First(table, index string, args ...interface{}) (interface{}, er
return val, err
}
// Last is used to return the last matching object for
// the given constraints on the index
func (txn *Txn) Last(table, index string, args ...interface{}) (interface{}, error) {
_, val, err := txn.LastWatch(table, index, args...)
return val, err
}
// LongestPrefix is used to fetch the longest prefix match for the given
// constraints on the index. Note that this will not work with the memdb
// StringFieldIndex because it adds null terminators which prevent the
@@ -581,29 +667,48 @@ func (txn *Txn) getIndexValue(table, index string, args ...interface{}) (*IndexS
return indexSchema, val, err
}
// ResultIterator is used to iterate over a list of results
// from a Get query on a table.
// ResultIterator is used to iterate over a list of results from a query on a table.
//
// When a ResultIterator is created from a write transaction, the results from
// Next will reflect a snapshot of the table at the time the ResultIterator is
// created.
// This means that calling Insert or Delete on a transaction while iterating is
// allowed, but the changes made by Insert or Delete will not be observed in the
// results returned from subsequent calls to Next. For example if an item is deleted
// from the index used by the iterator it will still be returned by Next. If an
// item is inserted into the index used by the iterator, it will not be returned
// by Next. However, an iterator created after a call to Insert or Delete will
// reflect the modifications.
//
// When a ResultIterator is created from a write transaction, and there are already
// modifications to the index used by the iterator, the modification cache of the
// index will be invalidated. This may result in some additional allocations if
// the same node in the index is modified again.
type ResultIterator interface {
WatchCh() <-chan struct{}
// Next returns the next result from the iterator. If there are no more results
// nil is returned.
Next() interface{}
}
// Get is used to construct a ResultIterator over all the
// rows that match the given constraints of an index.
// Get is used to construct a ResultIterator over all the rows that match the
// given constraints of an index. The index values must match exactly (this
// is not a range-based or prefix-based lookup) by default.
//
// Prefix lookups: if the named index implements PrefixIndexer, you may perform
// prefix-based lookups by appending "_prefix" to the index name. In this
// scenario, the index values given in args are treated as prefix lookups. For
// example, a StringFieldIndex will match any string with the given value
// as a prefix: "mem" matches "memdb".
//
// See the documentation for ResultIterator to understand the behaviour of the
// returned ResultIterator.
func (txn *Txn) Get(table, index string, args ...interface{}) (ResultIterator, error) {
// Get the index value to scan
indexSchema, val, err := txn.getIndexValue(table, index, args...)
indexIter, val, err := txn.getIndexIterator(table, index, args...)
if err != nil {
return nil, err
}
// Get the index itself
indexTxn := txn.readableIndex(table, indexSchema.Name)
indexRoot := indexTxn.Root()
// Get an interator over the index
indexIter := indexRoot.Iterator()
// Seek the iterator to the appropriate sub-set
watchCh := indexIter.SeekPrefixWatch(val)
@@ -615,6 +720,198 @@ func (txn *Txn) Get(table, index string, args ...interface{}) (ResultIterator, e
return iter, nil
}
// GetReverse is used to construct a Reverse ResultIterator over all the
// rows that match the given constraints of an index.
// The returned ResultIterator's Next() will return the next Previous value.
//
// See the documentation on Get for details on arguments.
// See the documentation for ResultIterator to understand the behaviour of the
// returned ResultIterator.
func (txn *Txn) GetReverse(table, index string, args ...interface{}) (ResultIterator, error) {
indexIter, val, err := txn.getIndexIteratorReverse(table, index, args...)
if err != nil {
return nil, err
}
// Seek the iterator to the appropriate sub-set
watchCh := indexIter.SeekPrefixWatch(val)
// Create an iterator
iter := &radixReverseIterator{
iter: indexIter,
watchCh: watchCh,
}
return iter, nil
}
// LowerBound is used to construct a ResultIterator over all the the range of
// rows that have an index value greater than or equal to the provide args.
// Calling this then iterating until the rows are larger than required allows
// range scans within an index. It is not possible to watch the resulting
// iterator since the radix tree doesn't efficiently allow watching on lower
// bound changes. The WatchCh returned will be nill and so will block forever.
//
// See the documentation for ResultIterator to understand the behaviour of the
// returned ResultIterator.
func (txn *Txn) LowerBound(table, index string, args ...interface{}) (ResultIterator, error) {
indexIter, val, err := txn.getIndexIterator(table, index, args...)
if err != nil {
return nil, err
}
// Seek the iterator to the appropriate sub-set
indexIter.SeekLowerBound(val)
// Create an iterator
iter := &radixIterator{
iter: indexIter,
}
return iter, nil
}
// ReverseLowerBound is used to construct a Reverse ResultIterator over all the
// the range of rows that have an index value less than or equal to the
// provide args. Calling this then iterating until the rows are lower than
// required allows range scans within an index. It is not possible to watch the
// resulting iterator since the radix tree doesn't efficiently allow watching
// on lower bound changes. The WatchCh returned will be nill and so will block
// forever.
//
// See the documentation for ResultIterator to understand the behaviour of the
// returned ResultIterator.
func (txn *Txn) ReverseLowerBound(table, index string, args ...interface{}) (ResultIterator, error) {
indexIter, val, err := txn.getIndexIteratorReverse(table, index, args...)
if err != nil {
return nil, err
}
// Seek the iterator to the appropriate sub-set
indexIter.SeekReverseLowerBound(val)
// Create an iterator
iter := &radixReverseIterator{
iter: indexIter,
}
return iter, nil
}
// objectID is a tuple of table name and the raw internal id byte slice
// converted to a string. It's only converted to a string to make it comparable
// so this struct can be used as a map index.
type objectID struct {
Table string
IndexVal string
}
// mutInfo stores metadata about mutations to allow collapsing multiple
// mutations to the same object into one.
type mutInfo struct {
firstBefore interface{}
lastIdx int
}
// Changes returns the set of object changes that have been made in the
// transaction so far. If change tracking is not enabled it wil always return
// nil. It can be called before or after Commit. If it is before Commit it will
// return all changes made so far which may not be the same as the final
// Changes. After abort it will always return nil. As with other Txn methods
// it's not safe to call this from a different goroutine than the one making
// mutations or committing the transaction. Mutations will appear in the order
// they were performed in the transaction but multiple operations to the same
// object will be collapsed so only the effective overall change to that object
// is present. If transaction operations are dependent (e.g. copy object X to Y
// then delete X) this might mean the set of mutations is incomplete to verify
// history, but it is complete in that the net effect is preserved (Y got a new
// value, X got removed).
func (txn *Txn) Changes() Changes {
if txn.changes == nil {
return nil
}
// De-duplicate mutations by key so all take effect at the point of the last
// write but we keep the mutations in order.
dups := make(map[objectID]mutInfo)
for i, m := range txn.changes {
oid := objectID{
Table: m.Table,
IndexVal: string(m.primaryKey),
}
// Store the latest mutation index for each key value
mi, ok := dups[oid]
if !ok {
// First entry for key, store the before value
mi.firstBefore = m.Before
}
mi.lastIdx = i
dups[oid] = mi
}
if len(dups) == len(txn.changes) {
// No duplicates found, fast path return it as is
return txn.changes
}
// Need to remove the duplicates
cs := make(Changes, 0, len(dups))
for i, m := range txn.changes {
oid := objectID{
Table: m.Table,
IndexVal: string(m.primaryKey),
}
mi := dups[oid]
if mi.lastIdx == i {
// This was the latest value for this key copy it with the before value in
// case it's different. Note that m is not a pointer so we are not
// modifying the txn.changeSet here - it's already a copy.
m.Before = mi.firstBefore
// Edge case - if the object was inserted and then eventually deleted in
// the same transaction, then the net affect on that key is a no-op. Don't
// emit a mutation with nil for before and after as it's meaningless and
// might violate expectations and cause a panic in code that assumes at
// least one must be set.
if m.Before == nil && m.After == nil {
continue
}
cs = append(cs, m)
}
}
// Store the de-duped version in case this is called again
txn.changes = cs
return cs
}
func (txn *Txn) getIndexIterator(table, index string, args ...interface{}) (*iradix.Iterator, []byte, error) {
// Get the index value to scan
indexSchema, val, err := txn.getIndexValue(table, index, args...)
if err != nil {
return nil, nil, err
}
// Get the index itself
indexTxn := txn.readableIndex(table, indexSchema.Name)
indexRoot := indexTxn.Root()
// Get an iterator over the index
indexIter := indexRoot.Iterator()
return indexIter, val, nil
}
func (txn *Txn) getIndexIteratorReverse(table, index string, args ...interface{}) (*iradix.ReverseIterator, []byte, error) {
// Get the index value to scan
indexSchema, val, err := txn.getIndexValue(table, index, args...)
if err != nil {
return nil, nil, err
}
// Get the index itself
indexTxn := txn.readableIndex(table, indexSchema.Name)
indexRoot := indexTxn.Root()
// Get an interator over the index
indexIter := indexRoot.ReverseIterator()
return indexIter, val, nil
}
// Defer is used to push a new arbitrary function onto a stack which
// gets called when a transaction is committed and finished. Deferred
// functions are called in LIFO order, and only invoked at the end of
@@ -642,3 +939,43 @@ func (r *radixIterator) Next() interface{} {
}
return value
}
type radixReverseIterator struct {
iter *iradix.ReverseIterator
watchCh <-chan struct{}
}
func (r *radixReverseIterator) Next() interface{} {
_, value, ok := r.iter.Previous()
if !ok {
return nil
}
return value
}
func (r *radixReverseIterator) WatchCh() <-chan struct{} {
return r.watchCh
}
// Snapshot creates a snapshot of the current state of the transaction.
// Returns a new read-only transaction or nil if the transaction is already
// aborted or committed.
func (txn *Txn) Snapshot() *Txn {
if txn.rootTxn == nil {
return nil
}
snapshot := &Txn{
db: txn.db,
rootTxn: txn.rootTxn.Clone(),
}
// Commit sub-transactions into the snapshot
for key, subTxn := range txn.modified {
path := indexPath(key.Table, key.Index)
final := subTxn.CommitOnly()
snapshot.rootTxn.Insert(path, final)
}
return snapshot
}
+15
View File
@@ -127,3 +127,18 @@ func (w WatchSet) watchMany(ctx context.Context) error {
return ctx.Err()
}
}
// WatchCh returns a channel that is used to wait for either the watch set to trigger
// or for the context to be cancelled. WatchCh creates a new goroutine each call, so
// callers may need to cache the returned channel to avoid creating extra goroutines.
func (w WatchSet) WatchCh(ctx context.Context) <-chan error {
// Create the outgoing channel
triggerCh := make(chan error, 1)
// Create a goroutine to collect the error from WatchCtx
go func() {
triggerCh <- w.WatchCtx(ctx)
}()
return triggerCh
}
+1 -1
View File
@@ -2,7 +2,7 @@ package memdb
//go:generate sh -c "go run watch-gen/main.go >watch_few.go"
import(
import (
"context"
)
-23
View File
@@ -1,23 +0,0 @@
# Compiled Object files, Static and Dynamic libs (Shared Objects)
*.o
*.a
*.so
# Folders
_obj
_test
# Architecture specific extensions/prefixes
*.[568vq]
[568vq].out
*.cgo1.go
*.cgo2.c
_cgo_defun.c
_cgo_gotypes.go
_cgo_export.*
_testmain.go
*.exe
*.test
-223
View File
@@ -1,223 +0,0 @@
package lru
import (
"fmt"
"sync"
"github.com/hashicorp/golang-lru/simplelru"
)
const (
// Default2QRecentRatio is the ratio of the 2Q cache dedicated
// to recently added entries that have only been accessed once.
Default2QRecentRatio = 0.25
// Default2QGhostEntries is the default ratio of ghost
// entries kept to track entries recently evicted
Default2QGhostEntries = 0.50
)
// TwoQueueCache is a thread-safe fixed size 2Q cache.
// 2Q is an enhancement over the standard LRU cache
// in that it tracks both frequently and recently used
// entries separately. This avoids a burst in access to new
// entries from evicting frequently used entries. It adds some
// additional tracking overhead to the standard LRU cache, and is
// computationally about 2x the cost, and adds some metadata over
// head. The ARCCache is similar, but does not require setting any
// parameters.
type TwoQueueCache struct {
size int
recentSize int
recent simplelru.LRUCache
frequent simplelru.LRUCache
recentEvict simplelru.LRUCache
lock sync.RWMutex
}
// New2Q creates a new TwoQueueCache using the default
// values for the parameters.
func New2Q(size int) (*TwoQueueCache, error) {
return New2QParams(size, Default2QRecentRatio, Default2QGhostEntries)
}
// New2QParams creates a new TwoQueueCache using the provided
// parameter values.
func New2QParams(size int, recentRatio float64, ghostRatio float64) (*TwoQueueCache, error) {
if size <= 0 {
return nil, fmt.Errorf("invalid size")
}
if recentRatio < 0.0 || recentRatio > 1.0 {
return nil, fmt.Errorf("invalid recent ratio")
}
if ghostRatio < 0.0 || ghostRatio > 1.0 {
return nil, fmt.Errorf("invalid ghost ratio")
}
// Determine the sub-sizes
recentSize := int(float64(size) * recentRatio)
evictSize := int(float64(size) * ghostRatio)
// Allocate the LRUs
recent, err := simplelru.NewLRU(size, nil)
if err != nil {
return nil, err
}
frequent, err := simplelru.NewLRU(size, nil)
if err != nil {
return nil, err
}
recentEvict, err := simplelru.NewLRU(evictSize, nil)
if err != nil {
return nil, err
}
// Initialize the cache
c := &TwoQueueCache{
size: size,
recentSize: recentSize,
recent: recent,
frequent: frequent,
recentEvict: recentEvict,
}
return c, nil
}
// Get looks up a key's value from the cache.
func (c *TwoQueueCache) Get(key interface{}) (value interface{}, ok bool) {
c.lock.Lock()
defer c.lock.Unlock()
// Check if this is a frequent value
if val, ok := c.frequent.Get(key); ok {
return val, ok
}
// If the value is contained in recent, then we
// promote it to frequent
if val, ok := c.recent.Peek(key); ok {
c.recent.Remove(key)
c.frequent.Add(key, val)
return val, ok
}
// No hit
return nil, false
}
// Add adds a value to the cache.
func (c *TwoQueueCache) Add(key, value interface{}) {
c.lock.Lock()
defer c.lock.Unlock()
// Check if the value is frequently used already,
// and just update the value
if c.frequent.Contains(key) {
c.frequent.Add(key, value)
return
}
// Check if the value is recently used, and promote
// the value into the frequent list
if c.recent.Contains(key) {
c.recent.Remove(key)
c.frequent.Add(key, value)
return
}
// If the value was recently evicted, add it to the
// frequently used list
if c.recentEvict.Contains(key) {
c.ensureSpace(true)
c.recentEvict.Remove(key)
c.frequent.Add(key, value)
return
}
// Add to the recently seen list
c.ensureSpace(false)
c.recent.Add(key, value)
return
}
// ensureSpace is used to ensure we have space in the cache
func (c *TwoQueueCache) ensureSpace(recentEvict bool) {
// If we have space, nothing to do
recentLen := c.recent.Len()
freqLen := c.frequent.Len()
if recentLen+freqLen < c.size {
return
}
// If the recent buffer is larger than
// the target, evict from there
if recentLen > 0 && (recentLen > c.recentSize || (recentLen == c.recentSize && !recentEvict)) {
k, _, _ := c.recent.RemoveOldest()
c.recentEvict.Add(k, nil)
return
}
// Remove from the frequent list otherwise
c.frequent.RemoveOldest()
}
// Len returns the number of items in the cache.
func (c *TwoQueueCache) Len() int {
c.lock.RLock()
defer c.lock.RUnlock()
return c.recent.Len() + c.frequent.Len()
}
// Keys returns a slice of the keys in the cache.
// The frequently used keys are first in the returned slice.
func (c *TwoQueueCache) Keys() []interface{} {
c.lock.RLock()
defer c.lock.RUnlock()
k1 := c.frequent.Keys()
k2 := c.recent.Keys()
return append(k1, k2...)
}
// Remove removes the provided key from the cache.
func (c *TwoQueueCache) Remove(key interface{}) {
c.lock.Lock()
defer c.lock.Unlock()
if c.frequent.Remove(key) {
return
}
if c.recent.Remove(key) {
return
}
if c.recentEvict.Remove(key) {
return
}
}
// Purge is used to completely clear the cache.
func (c *TwoQueueCache) Purge() {
c.lock.Lock()
defer c.lock.Unlock()
c.recent.Purge()
c.frequent.Purge()
c.recentEvict.Purge()
}
// Contains is used to check if the cache contains a key
// without updating recency or frequency.
func (c *TwoQueueCache) Contains(key interface{}) bool {
c.lock.RLock()
defer c.lock.RUnlock()
return c.frequent.Contains(key) || c.recent.Contains(key)
}
// Peek is used to inspect the cache value of a key
// without updating recency or frequency.
func (c *TwoQueueCache) Peek(key interface{}) (value interface{}, ok bool) {
c.lock.RLock()
defer c.lock.RUnlock()
if val, ok := c.frequent.Peek(key); ok {
return val, ok
}
return c.recent.Peek(key)
}
-25
View File
@@ -1,25 +0,0 @@
golang-lru
==========
This provides the `lru` package which implements a fixed-size
thread safe LRU cache. It is based on the cache in Groupcache.
Documentation
=============
Full docs are available on [Godoc](http://godoc.org/github.com/hashicorp/golang-lru)
Example
=======
Using the LRU is very simple:
```go
l, _ := New(128)
for i := 0; i < 256; i++ {
l.Add(i, nil)
}
if l.Len() != 128 {
panic(fmt.Sprintf("bad len: %v", l.Len()))
}
```
-257
View File
@@ -1,257 +0,0 @@
package lru
import (
"sync"
"github.com/hashicorp/golang-lru/simplelru"
)
// ARCCache is a thread-safe fixed size Adaptive Replacement Cache (ARC).
// ARC is an enhancement over the standard LRU cache in that tracks both
// frequency and recency of use. This avoids a burst in access to new
// entries from evicting the frequently used older entries. It adds some
// additional tracking overhead to a standard LRU cache, computationally
// it is roughly 2x the cost, and the extra memory overhead is linear
// with the size of the cache. ARC has been patented by IBM, but is
// similar to the TwoQueueCache (2Q) which requires setting parameters.
type ARCCache struct {
size int // Size is the total capacity of the cache
p int // P is the dynamic preference towards T1 or T2
t1 simplelru.LRUCache // T1 is the LRU for recently accessed items
b1 simplelru.LRUCache // B1 is the LRU for evictions from t1
t2 simplelru.LRUCache // T2 is the LRU for frequently accessed items
b2 simplelru.LRUCache // B2 is the LRU for evictions from t2
lock sync.RWMutex
}
// NewARC creates an ARC of the given size
func NewARC(size int) (*ARCCache, error) {
// Create the sub LRUs
b1, err := simplelru.NewLRU(size, nil)
if err != nil {
return nil, err
}
b2, err := simplelru.NewLRU(size, nil)
if err != nil {
return nil, err
}
t1, err := simplelru.NewLRU(size, nil)
if err != nil {
return nil, err
}
t2, err := simplelru.NewLRU(size, nil)
if err != nil {
return nil, err
}
// Initialize the ARC
c := &ARCCache{
size: size,
p: 0,
t1: t1,
b1: b1,
t2: t2,
b2: b2,
}
return c, nil
}
// Get looks up a key's value from the cache.
func (c *ARCCache) Get(key interface{}) (value interface{}, ok bool) {
c.lock.Lock()
defer c.lock.Unlock()
// If the value is contained in T1 (recent), then
// promote it to T2 (frequent)
if val, ok := c.t1.Peek(key); ok {
c.t1.Remove(key)
c.t2.Add(key, val)
return val, ok
}
// Check if the value is contained in T2 (frequent)
if val, ok := c.t2.Get(key); ok {
return val, ok
}
// No hit
return nil, false
}
// Add adds a value to the cache.
func (c *ARCCache) Add(key, value interface{}) {
c.lock.Lock()
defer c.lock.Unlock()
// Check if the value is contained in T1 (recent), and potentially
// promote it to frequent T2
if c.t1.Contains(key) {
c.t1.Remove(key)
c.t2.Add(key, value)
return
}
// Check if the value is already in T2 (frequent) and update it
if c.t2.Contains(key) {
c.t2.Add(key, value)
return
}
// Check if this value was recently evicted as part of the
// recently used list
if c.b1.Contains(key) {
// T1 set is too small, increase P appropriately
delta := 1
b1Len := c.b1.Len()
b2Len := c.b2.Len()
if b2Len > b1Len {
delta = b2Len / b1Len
}
if c.p+delta >= c.size {
c.p = c.size
} else {
c.p += delta
}
// Potentially need to make room in the cache
if c.t1.Len()+c.t2.Len() >= c.size {
c.replace(false)
}
// Remove from B1
c.b1.Remove(key)
// Add the key to the frequently used list
c.t2.Add(key, value)
return
}
// Check if this value was recently evicted as part of the
// frequently used list
if c.b2.Contains(key) {
// T2 set is too small, decrease P appropriately
delta := 1
b1Len := c.b1.Len()
b2Len := c.b2.Len()
if b1Len > b2Len {
delta = b1Len / b2Len
}
if delta >= c.p {
c.p = 0
} else {
c.p -= delta
}
// Potentially need to make room in the cache
if c.t1.Len()+c.t2.Len() >= c.size {
c.replace(true)
}
// Remove from B2
c.b2.Remove(key)
// Add the key to the frequently used list
c.t2.Add(key, value)
return
}
// Potentially need to make room in the cache
if c.t1.Len()+c.t2.Len() >= c.size {
c.replace(false)
}
// Keep the size of the ghost buffers trim
if c.b1.Len() > c.size-c.p {
c.b1.RemoveOldest()
}
if c.b2.Len() > c.p {
c.b2.RemoveOldest()
}
// Add to the recently seen list
c.t1.Add(key, value)
return
}
// replace is used to adaptively evict from either T1 or T2
// based on the current learned value of P
func (c *ARCCache) replace(b2ContainsKey bool) {
t1Len := c.t1.Len()
if t1Len > 0 && (t1Len > c.p || (t1Len == c.p && b2ContainsKey)) {
k, _, ok := c.t1.RemoveOldest()
if ok {
c.b1.Add(k, nil)
}
} else {
k, _, ok := c.t2.RemoveOldest()
if ok {
c.b2.Add(k, nil)
}
}
}
// Len returns the number of cached entries
func (c *ARCCache) Len() int {
c.lock.RLock()
defer c.lock.RUnlock()
return c.t1.Len() + c.t2.Len()
}
// Keys returns all the cached keys
func (c *ARCCache) Keys() []interface{} {
c.lock.RLock()
defer c.lock.RUnlock()
k1 := c.t1.Keys()
k2 := c.t2.Keys()
return append(k1, k2...)
}
// Remove is used to purge a key from the cache
func (c *ARCCache) Remove(key interface{}) {
c.lock.Lock()
defer c.lock.Unlock()
if c.t1.Remove(key) {
return
}
if c.t2.Remove(key) {
return
}
if c.b1.Remove(key) {
return
}
if c.b2.Remove(key) {
return
}
}
// Purge is used to clear the cache
func (c *ARCCache) Purge() {
c.lock.Lock()
defer c.lock.Unlock()
c.t1.Purge()
c.t2.Purge()
c.b1.Purge()
c.b2.Purge()
}
// Contains is used to check if the cache contains a key
// without updating recency or frequency.
func (c *ARCCache) Contains(key interface{}) bool {
c.lock.RLock()
defer c.lock.RUnlock()
return c.t1.Contains(key) || c.t2.Contains(key)
}
// Peek is used to inspect the cache value of a key
// without updating recency or frequency.
func (c *ARCCache) Peek(key interface{}) (value interface{}, ok bool) {
c.lock.RLock()
defer c.lock.RUnlock()
if val, ok := c.t1.Peek(key); ok {
return val, ok
}
return c.t2.Peek(key)
}
-21
View File
@@ -1,21 +0,0 @@
// Package lru provides three different LRU caches of varying sophistication.
//
// Cache is a simple LRU cache. It is based on the
// LRU implementation in groupcache:
// https://github.com/golang/groupcache/tree/master/lru
//
// TwoQueueCache tracks frequently used and recently used entries separately.
// This avoids a burst of accesses from taking out frequently used entries,
// at the cost of about 2x computational overhead and some extra bookkeeping.
//
// ARCCache is an adaptive replacement cache. It tracks recent evictions as
// well as recent usage in both the frequent and recent caches. Its
// computational overhead is comparable to TwoQueueCache, but the memory
// overhead is linear with the size of the cache.
//
// ARC has been patented by IBM, so do not use it if that is problematic for
// your program.
//
// All caches in this package take locks while operating, and are therefore
// thread-safe for consumers.
package lru
-3
View File
@@ -1,3 +0,0 @@
module github.com/hashicorp/golang-lru
go 1.12
-134
View File
@@ -1,134 +0,0 @@
package lru
import (
"sync"
"github.com/hashicorp/golang-lru/simplelru"
)
// Cache is a thread-safe fixed size LRU cache.
type Cache struct {
lru simplelru.LRUCache
lock sync.RWMutex
}
// New creates an LRU of the given size.
func New(size int) (*Cache, error) {
return NewWithEvict(size, nil)
}
// NewWithEvict constructs a fixed size cache with the given eviction
// callback.
func NewWithEvict(size int, onEvicted func(key interface{}, value interface{})) (*Cache, error) {
lru, err := simplelru.NewLRU(size, simplelru.EvictCallback(onEvicted))
if err != nil {
return nil, err
}
c := &Cache{
lru: lru,
}
return c, nil
}
// Purge is used to completely clear the cache.
func (c *Cache) Purge() {
c.lock.Lock()
c.lru.Purge()
c.lock.Unlock()
}
// Add adds a value to the cache. Returns true if an eviction occurred.
func (c *Cache) Add(key, value interface{}) (evicted bool) {
c.lock.Lock()
evicted = c.lru.Add(key, value)
c.lock.Unlock()
return evicted
}
// Get looks up a key's value from the cache.
func (c *Cache) Get(key interface{}) (value interface{}, ok bool) {
c.lock.Lock()
value, ok = c.lru.Get(key)
c.lock.Unlock()
return value, ok
}
// Contains checks if a key is in the cache, without updating the
// recent-ness or deleting it for being stale.
func (c *Cache) Contains(key interface{}) bool {
c.lock.RLock()
containKey := c.lru.Contains(key)
c.lock.RUnlock()
return containKey
}
// Peek returns the key value (or undefined if not found) without updating
// the "recently used"-ness of the key.
func (c *Cache) Peek(key interface{}) (value interface{}, ok bool) {
c.lock.RLock()
value, ok = c.lru.Peek(key)
c.lock.RUnlock()
return value, ok
}
// ContainsOrAdd checks if a key is in the cache without updating the
// recent-ness or deleting it for being stale, and if not, adds the value.
// Returns whether found and whether an eviction occurred.
func (c *Cache) ContainsOrAdd(key, value interface{}) (ok, evicted bool) {
c.lock.Lock()
defer c.lock.Unlock()
if c.lru.Contains(key) {
return true, false
}
evicted = c.lru.Add(key, value)
return false, evicted
}
// Remove removes the provided key from the cache.
func (c *Cache) Remove(key interface{}) (present bool) {
c.lock.Lock()
present = c.lru.Remove(key)
c.lock.Unlock()
return
}
// Resize changes the cache size.
func (c *Cache) Resize(size int) (evicted int) {
c.lock.Lock()
evicted = c.lru.Resize(size)
c.lock.Unlock()
return evicted
}
// RemoveOldest removes the oldest item from the cache.
func (c *Cache) RemoveOldest() (key interface{}, value interface{}, ok bool) {
c.lock.Lock()
key, value, ok = c.lru.RemoveOldest()
c.lock.Unlock()
return
}
// GetOldest returns the oldest entry
func (c *Cache) GetOldest() (key interface{}, value interface{}, ok bool) {
c.lock.Lock()
key, value, ok = c.lru.GetOldest()
c.lock.Unlock()
return
}
// Keys returns a slice of the keys in the cache, from oldest to newest.
func (c *Cache) Keys() []interface{} {
c.lock.RLock()
keys := c.lru.Keys()
c.lock.RUnlock()
return keys
}
// Len returns the number of items in the cache.
func (c *Cache) Len() int {
c.lock.RLock()
length := c.lru.Len()
c.lock.RUnlock()
return length
}