barak/cayley
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Fix data race in gremlin timeout handling

Browse source 
Fixes issue #95.
This commit is contained in:
kortschak 2014-08-02 22:01:02 +09:30
parent cb177aa390
commit fe9ca5ffcc
2 changed files with 39 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

28
query/gremlin/finals.go
View file

@ -148,8 +148,10 @@ func runIteratorToArray(it graph.Iterator, ses *Session, limit int) []map[string
count := 0
it, _ = it.Optimize()
for {
if ses.doHalt {
select {
case <-ses.kill:
return nil
default:
}
_, ok := graph.Next(it)
if !ok {
@ -163,8 +165,10 @@ func runIteratorToArray(it graph.Iterator, ses *Session, limit int) []map[string
break
}
for it.NextResult() == true {
if ses.doHalt {
select {
case <-ses.kill:
return nil
default:
}
tags := make(map[string]graph.Value)
it.TagResults(tags)
@ -184,8 +188,10 @@ func runIteratorToArrayNoTags(it graph.Iterator, ses *Session, limit int) []stri
count := 0
it, _ = it.Optimize()
for {
if ses.doHalt {
select {
case <-ses.kill:
return nil
default:
}
val, ok := graph.Next(it)
if !ok {
@ -205,8 +211,10 @@ func runIteratorWithCallback(it graph.Iterator, ses *Session, callback otto.Valu
count := 0
it, _ = it.Optimize()
for {
if ses.doHalt {
select {
case <-ses.kill:
return
default:
}
_, ok := graph.Next(it)
if !ok {
@ -221,8 +229,10 @@ func runIteratorWithCallback(it graph.Iterator, ses *Session, callback otto.Valu
break
}
for it.NextResult() == true {
if ses.doHalt {
select {
case <-ses.kill:
return
default:
}
tags := make(map[string]graph.Value)
it.TagResults(tags)
@ -246,8 +256,10 @@ func runIteratorOnSession(it graph.Iterator, ses *Session) {
glog.V(2).Infoln(it.DebugString(0))
for {
// TODO(barakmich): Better halting.
if ses.doHalt {
select {
case <-ses.kill:
return
default:
}
_, ok := graph.Next(it)
if !ok {
@ -260,8 +272,10 @@ func runIteratorOnSession(it graph.Iterator, ses *Session) {
break
}
for it.NextResult() == true {
if ses.doHalt {
select {
case <-ses.kill:
return
default:
}
tags := make(map[string]graph.Value)
it.TagResults(tags)

24
query/gremlin/session.go
View file

@ -18,6 +18,7 @@ import (
"errors"
"fmt"
"sort"
"sync"
"time"
"github.com/robertkrimen/otto"
@ -30,6 +31,7 @@ type Session struct {
ts graph.TripleStore
currentChannel chan interface{}
env *otto.Otto
envLock sync.Mutex
debug bool
limit int
count int
@ -38,7 +40,7 @@ type Session struct {
queryShape map[string]interface{}
err error
script *otto.Script
doHalt bool
kill chan struct{}
timeoutSec time.Duration
emptyEnv *otto.Otto
}
@ -95,8 +97,10 @@ func (s *Session) SendResult(result *GremlinResult) bool {
if s.limit >= 0 && s.limit == s.count {
return false
}
if s.doHalt {
select {
case <-s.kill:
return false
default:
}
if s.currentChannel != nil {
s.currentChannel <- result
@ -113,7 +117,7 @@ func (s *Session) SendResult(result *GremlinResult) bool {
var halt = errors.New("Query Timeout")
func (s *Session) runUnsafe(input interface{}) (otto.Value, error) {
s.doHalt = false
s.kill = make(chan struct{})
defer func() {
if caught := recover(); caught != nil {
if caught == halt {
@ -129,7 +133,9 @@ func (s *Session) runUnsafe(input interface{}) (otto.Value, error) {
if s.timeoutSec != -1 {
go func() {
time.Sleep(s.timeoutSec * time.Second) // Stop after two seconds
s.doHalt = true
close(s.kill)
s.envLock.Lock()
defer s.envLock.Unlock()
if s.env != nil {
s.env.Interrupt <- func() {
panic(halt)
@ -139,6 +145,8 @@ func (s *Session) runUnsafe(input interface{}) (otto.Value, error) {
}()
}
s.envLock.Lock()
defer s.envLock.Unlock()
return s.env.Run(input) // Here be dragons (risky code)
}
@ -166,7 +174,9 @@ func (s *Session) ExecInput(input string, out chan interface{}, limit int) {
}
s.currentChannel = nil
s.script = nil
s.envLock.Lock()
s.env = s.emptyEnv
s.envLock.Unlock()
return
}
@ -256,10 +266,12 @@ func (ses *Session) GetJson() ([]interface{}, error) {
if ses.err != nil {
return nil, ses.err
}
if ses.doHalt {
select {
case <-ses.kill:
return nil, halt
default:
return ses.dataOutput, nil
}
return ses.dataOutput, nil
}
func (ses *Session) ClearJson() {