Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

memstore add rwmutex, fix multi-thread issue #954

Open
wants to merge 8 commits into
base: master
Choose a base branch
from
Open

memstore add rwmutex, fix multi-thread issue #954

Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
9a69d8b
add rwmutex, fix thread safe issue
Jun 2, 2021
49cdd06
format code
Jun 2, 2021
c3bbf12
add test case for multi-thread rw
Jun 3, 2021
ce9cd4c
format code
Jun 3, 2021
28ea5d7
fix lint bug
Jun 3, 2021
8c05dca
fix map rw bug
Jun 21, 2021
fb132a0
fix bug, add rw
Jun 21, 2021
70bff3c
Merge branch 'master' of https://github.com/cayleygraph/cayley
Jun 21, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
68 changes: 66 additions & 2 deletions graph/memstore/quadstore.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"fmt"
"strconv"
"strings"
"sync"

"github.com/cayleygraph/cayley/graph"
"github.com/cayleygraph/cayley/graph/iterator"
Expand Down Expand Up @@ -60,7 +61,7 @@ type QuadDirectionIndex struct {
}

func NewQuadDirectionIndex() QuadDirectionIndex {
return QuadDirectionIndex{[...]map[int64]*Tree{
return QuadDirectionIndex{index: [...]map[int64]*Tree{
quad.Subject - 1: make(map[int64]*Tree),
quad.Predicate - 1: make(map[int64]*Tree),
quad.Object - 1: make(map[int64]*Tree),
Expand All @@ -72,6 +73,7 @@ func (qdi QuadDirectionIndex) Tree(d quad.Direction, id int64) *Tree {
if d < quad.Subject || d > quad.Label {
panic("illegal direction")
}

tree, ok := qdi.index[d-1][id]
if !ok {
tree = TreeNew(cmp)
Expand Down Expand Up @@ -143,6 +145,11 @@ type QuadStore struct {
index QuadDirectionIndex
horizon int64 // used only to assign ids to tx
// vip_index map[string]map[int64]map[string]map[int64]*b.Tree

vrw sync.RWMutex
RyouZhang marked this conversation as resolved.
Show resolved Hide resolved
qrw sync.RWMutex
prw sync.RWMutex
irw sync.RWMutex
}

// New creates a new in-memory quad store and loads provided quads.
Expand Down Expand Up @@ -178,7 +185,10 @@ func (qs *QuadStore) addPrimitive(p *Primitive) int64 {
}

func (qs *QuadStore) appendPrimitive(p *Primitive) {
qs.prw.Lock()
qs.prim[p.ID] = p
qs.prw.Unlock()

if !qs.reading {
qs.all = append(qs.all, p)
RyouZhang marked this conversation as resolved.
Show resolved Hide resolved
} else {
Expand All @@ -198,24 +208,36 @@ func (qs *QuadStore) resolveVal(v quad.Value, add bool) (int64, bool) {
n = n[len(internalBNodePrefix):]
id, err := strconv.ParseInt(string(n), 10, 64)
if err == nil && id != 0 {
qs.prw.RLock()
if p, ok := qs.prim[id]; ok || !add {
qs.prw.RUnlock()
if add {
p.refs++
}
return id, ok
}
qs.prw.RUnlock()
qs.appendPrimitive(&Primitive{ID: id, refs: 1})
return id, true
}
}
vs := v.String()
qs.vrw.RLock()
if id, exists := qs.vals[vs]; exists || !add {
RyouZhang marked this conversation as resolved.
Show resolved Hide resolved
qs.vrw.RUnlock()
if exists && add {
qs.prw.Lock()
qs.prim[id].refs++
qs.prw.Unlock()
}
return id, exists
}
qs.vrw.RUnlock()

id := qs.addPrimitive(&Primitive{Value: v})

qs.vrw.Lock()
defer qs.vrw.Unlock()
qs.vals[vs] = id
return id, true
}
Expand All @@ -237,7 +259,9 @@ func (qs *QuadStore) resolveQuad(q quad.Quad, add bool) (internalQuad, bool) {
}

func (qs *QuadStore) lookupVal(id int64) quad.Value {
qs.prw.RLock()
pv := qs.prim[id]
qs.prw.RUnlock()
if pv == nil || pv.Value == nil {
return quad.BNode(internalBNodePrefix + strconv.FormatInt(id, 10))
}
Expand Down Expand Up @@ -270,6 +294,9 @@ func (qs *QuadStore) AddValue(v quad.Value) (int64, bool) {
}

func (qs *QuadStore) indexesForQuad(q internalQuad) []*Tree {
qs.irw.Lock()
defer qs.irw.Unlock()

trees := make([]*Tree, 0, 4)
for dir := quad.Subject; dir <= quad.Label; dir++ {
v := q.Dir(dir)
Expand All @@ -285,13 +312,20 @@ func (qs *QuadStore) indexesForQuad(q internalQuad) []*Tree {
// False is returned as a second parameter if quad exists already.
func (qs *QuadStore) AddQuad(q quad.Quad) (int64, bool) {
p, _ := qs.resolveQuad(q, false)
qs.qrw.RLock()
if id := qs.quads[p]; id != 0 {
qs.qrw.RUnlock()
return id, false
}
qs.qrw.RUnlock()
p, _ = qs.resolveQuad(q, true)
pr := &Primitive{Quad: p}
id := qs.addPrimitive(pr)

qs.qrw.Lock()
qs.quads[p] = id
qs.qrw.Unlock()

for _, t := range qs.indexesForQuad(p) {
t.Set(id, pr)
}
Expand Down Expand Up @@ -345,32 +379,44 @@ func (qs *QuadStore) deleteQuadNodes(q internalQuad) {
if id == 0 {
continue
}
qs.prw.RLock()
if p := qs.prim[id]; p != nil {
qs.prw.RUnlock()
p.refs--
if p.refs < 0 {
panic("remove of deleted node")
} else if p.refs == 0 {
qs.Delete(id)
}
} else {
qs.prw.RUnlock()
}
}
}
func (qs *QuadStore) Delete(id int64) bool {
qs.prw.RLock()
p := qs.prim[id]
qs.prw.RUnlock()
if p == nil {
return false
}
// remove from value index
if p.Value != nil {
qs.vrw.Lock()
delete(qs.vals, p.Value.String())
qs.vrw.Unlock()
}
// remove from quad indexes
for _, t := range qs.indexesForQuad(p.Quad) {
t.Delete(id)
}
qs.qrw.Lock()
delete(qs.quads, p.Quad)
// remove primitive
qs.qrw.Unlock()
// remove Primitive
qs.prw.Lock()
delete(qs.prim, id)
qs.prw.Unlock()
di := -1
for i, p2 := range qs.all {
if p == p2 {
Expand Down Expand Up @@ -398,6 +444,8 @@ func (qs *QuadStore) findQuad(q quad.Quad) (int64, internalQuad, bool) {
if !ok {
return 0, p, false
}
qs.qrw.Lock()
defer qs.qrw.Unlock()
id := qs.quads[p]
return id, p, id != 0
}
Expand Down Expand Up @@ -456,7 +504,9 @@ func asID(v graph.Ref) (int64, bool) {
func (qs *QuadStore) quad(v graph.Ref) (q internalQuad, ok bool) {
switch v := v.(type) {
case bnode:
qs.prw.RLock()
p := qs.prim[int64(v)]
qs.prw.RUnlock()
if p == nil {
return
}
Expand All @@ -482,7 +532,9 @@ func (qs *QuadStore) QuadIterator(d quad.Direction, value graph.Ref) iterator.Sh
if !ok {
return iterator.NewNull()
}
qs.irw.RLock()
index, ok := qs.index.Get(d, id)
qs.irw.RUnlock()
if ok && index.Len() != 0 {
return qs.newIterator(index, d, id)
}
Expand All @@ -494,14 +546,20 @@ func (qs *QuadStore) QuadIteratorSize(ctx context.Context, d quad.Direction, v g
if !ok {
return refs.Size{Value: 0, Exact: true}, nil
}
qs.irw.RLock()
index, ok := qs.index.Get(d, id)
qs.irw.RUnlock()
if !ok {
return refs.Size{Value: 0, Exact: true}, nil
}
return refs.Size{Value: int64(index.Len()), Exact: true}, nil
}

func (qs *QuadStore) Stats(ctx context.Context, exact bool) (graph.Stats, error) {
qs.vrw.RLock()
defer qs.vrw.RUnlock()
qs.qrw.RLock()
defer qs.qrw.RUnlock()
return graph.Stats{
Nodes: refs.Size{
Value: int64(len(qs.vals)),
Expand All @@ -518,7 +576,10 @@ func (qs *QuadStore) ValueOf(name quad.Value) graph.Ref {
if name == nil {
return nil
}

qs.vrw.Lock()
id := qs.vals[name.String()]
qs.vrw.Unlock()
if id == 0 {
return nil
}
Expand All @@ -535,9 +596,12 @@ func (qs *QuadStore) NameOf(v graph.Ref) quad.Value {
if !ok {
return nil
}
qs.prw.RLock()
if _, ok = qs.prim[n]; !ok {
qs.prw.RUnlock()
return nil
}
qs.prw.RUnlock()
return qs.lookupVal(n)
}

Expand Down