Merge "syncbase/store: fix a race condition"
diff --git a/lib/vdl/codegen/java/file_array.go b/lib/vdl/codegen/java/file_array.go
index 633e8a1..44f710d 100644
--- a/lib/vdl/codegen/java/file_array.go
+++ b/lib/vdl/codegen/java/file_array.go
@@ -56,6 +56,17 @@
super(VDL_TYPE, convert(arr));
}
+ /**
+ * Converts the array into its primitive (array) type.
+ */
+ public {{ .ElemPrimitiveType }}[] toPrimitiveArray() {
+ {{ .ElemPrimitiveType }}[] ret = new {{ .ElemPrimitiveType }}[size()];
+ for (int i = 0; i < size(); ++i) {
+ ret[i] = get(i);
+ }
+ return ret;
+ }
+
private static {{ .ElemType }}[] convert({{ .ElemPrimitiveType }}[] arr) {
{{ .ElemType }}[] ret = new {{ .ElemType }}[arr.length];
for (int i = 0; i < arr.length; ++i) {
diff --git a/services/syncbase/server/app.go b/services/syncbase/server/app.go
index a2b1c4f..37b9152 100644
--- a/services/syncbase/server/app.go
+++ b/services/syncbase/server/app.go
@@ -89,11 +89,11 @@
}
// Check perms.
sn := a.s.st.NewSnapshot()
+ defer sn.Abort()
if err := util.GetWithAuth(ctx, call, sn, a.stKey(), &appData{}); err != nil {
- sn.Abort()
return err
}
- return util.GlobChildren(ctx, call, matcher, sn, sn.Abort, util.JoinKeyParts(util.DbInfoPrefix, a.name))
+ return util.GlobChildren(ctx, call, matcher, sn, util.JoinKeyParts(util.DbInfoPrefix, a.name))
}
////////////////////////////////////////
diff --git a/services/syncbase/server/dispatcher.go b/services/syncbase/server/dispatcher.go
index a5b9bba..6cf2359 100644
--- a/services/syncbase/server/dispatcher.go
+++ b/services/syncbase/server/dispatcher.go
@@ -33,8 +33,6 @@
var auth security.Authorizer = security.AllowEveryone()
func (disp *dispatcher) Lookup(ctx *context.T, suffix string) (interface{}, security.Authorizer, error) {
- // TODO(sadovsky): Can we drop this TrimPrefix?
- suffix = strings.TrimPrefix(suffix, "/")
if len(suffix) == 0 {
return wire.ServiceServer(disp.s), auth, nil
}
diff --git a/services/syncbase/server/nosql/database.go b/services/syncbase/server/nosql/database.go
index 40da19e..c8afa10 100644
--- a/services/syncbase/server/nosql/database.go
+++ b/services/syncbase/server/nosql/database.go
@@ -17,6 +17,7 @@
wire "v.io/v23/services/syncbase/nosql"
"v.io/v23/syncbase/nosql/query"
"v.io/v23/syncbase/nosql/query/exec"
+ pubutil "v.io/v23/syncbase/util"
"v.io/v23/vdl"
"v.io/v23/verror"
"v.io/v23/vom"
@@ -320,22 +321,53 @@
if !d.exists {
return verror.New(verror.ErrNoExist, ctx, d.name)
}
- impl := func(sntx store.SnapshotOrTransaction, closeSntx func() error) error {
+ impl := func(sntx store.SnapshotOrTransaction) error {
// Check perms.
- sn := d.st.NewSnapshot()
- if err := util.GetWithAuth(ctx, call, sn, d.stKey(), &databaseData{}); err != nil {
- sn.Abort()
+ if err := util.GetWithAuth(ctx, call, sntx, d.stKey(), &databaseData{}); err != nil {
return err
}
- return util.GlobChildren(ctx, call, matcher, sn, closeSntx, util.TablePrefix)
+ return util.GlobChildren(ctx, call, matcher, sntx, util.TablePrefix)
}
if d.batchId != nil {
- return impl(d.batchReader(), func() error {
- return nil
- })
+ return impl(d.batchReader())
} else {
sn := d.st.NewSnapshot()
- return impl(sn, sn.Abort)
+ defer sn.Abort()
+ return impl(sn)
+ }
+}
+
+// See comment in v.io/v23/services/syncbase/nosql/service.vdl for why we can't
+// implement ListTables using Glob.
+func (d *databaseReq) ListTables(ctx *context.T, call rpc.ServerCall) ([]string, error) {
+ if !d.exists {
+ return nil, verror.New(verror.ErrNoExist, ctx, d.name)
+ }
+ impl := func(sntx store.SnapshotOrTransaction) ([]string, error) {
+ // Check perms.
+ if err := util.GetWithAuth(ctx, call, sntx, d.stKey(), &databaseData{}); err != nil {
+ return nil, err
+ }
+ it := sntx.Scan(util.ScanPrefixArgs(util.TablePrefix, ""))
+ key := []byte{}
+ res := []string{}
+ for it.Advance() {
+ key = it.Key(key)
+ parts := util.SplitKeyParts(string(key))
+ // For explanation of Escape(), see comment in server/nosql/dispatcher.go.
+ res = append(res, pubutil.Escape(parts[len(parts)-1]))
+ }
+ if err := it.Err(); err != nil {
+ return nil, err
+ }
+ return res, nil
+ }
+ if d.batchId != nil {
+ return impl(d.batchReader())
+ } else {
+ sntx := d.st.NewSnapshot()
+ defer sntx.Abort()
+ return impl(sntx)
}
}
diff --git a/services/syncbase/server/nosql/dispatcher.go b/services/syncbase/server/nosql/dispatcher.go
index 87ff495..022a86f 100644
--- a/services/syncbase/server/nosql/dispatcher.go
+++ b/services/syncbase/server/nosql/dispatcher.go
@@ -50,8 +50,6 @@
// restricted to be valid identifiers, such that neither "<appName>:<dbName>"
// nor "<tableName>:<rowKey>" can be ambiguous.
func (disp *dispatcher) Lookup(ctx *context.T, suffix string) (interface{}, security.Authorizer, error) {
- // TODO(sadovsky): Can we drop this TrimPrefix?
- suffix = strings.TrimPrefix(suffix, "/")
parts := strings.SplitN(suffix, "/", 3) // db, table, row
// Note, the slice returned by strings.SplitN is guaranteed to contain at
diff --git a/services/syncbase/server/nosql/table.go b/services/syncbase/server/nosql/table.go
index 7da77fa..d5d4c10 100644
--- a/services/syncbase/server/nosql/table.go
+++ b/services/syncbase/server/nosql/table.go
@@ -117,9 +117,9 @@
if t.d.batchId != nil {
return impl(t.d.batchReader())
} else {
- sntx := t.d.st.NewSnapshot()
- defer sntx.Abort()
- return impl(sntx)
+ sn := t.d.st.NewSnapshot()
+ defer sn.Abort()
+ return impl(sn)
}
}
@@ -365,21 +365,19 @@
}
func (t *tableReq) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error {
- impl := func(sntx store.SnapshotOrTransaction, closeSntx func() error) error {
+ impl := func(sntx store.SnapshotOrTransaction) error {
// Check perms.
if err := t.checkAccess(ctx, call, sntx, ""); err != nil {
- closeSntx()
return err
}
- return util.GlobChildren(ctx, call, matcher, sntx, closeSntx, util.JoinKeyParts(util.RowPrefix, t.name))
+ return util.GlobChildren(ctx, call, matcher, sntx, util.JoinKeyParts(util.RowPrefix, t.name))
}
if t.d.batchId != nil {
- return impl(t.d.batchReader(), func() error {
- return nil
- })
+ return impl(t.d.batchReader())
} else {
sn := t.d.st.NewSnapshot()
- return impl(sn, sn.Abort)
+ defer sn.Abort()
+ return impl(sn)
}
}
diff --git a/services/syncbase/server/service.go b/services/syncbase/server/service.go
index 2ddf195..1677a39 100644
--- a/services/syncbase/server/service.go
+++ b/services/syncbase/server/service.go
@@ -161,11 +161,11 @@
func (s *service) GlobChildren__(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element) error {
// Check perms.
sn := s.st.NewSnapshot()
+ defer sn.Abort()
if err := util.GetWithAuth(ctx, call, sn, s.stKey(), &serviceData{}); err != nil {
- sn.Abort()
return err
}
- return util.GlobChildren(ctx, call, matcher, sn, sn.Abort, util.AppPrefix)
+ return util.GlobChildren(ctx, call, matcher, sn, util.AppPrefix)
}
////////////////////////////////////////
diff --git a/services/syncbase/server/util/glob_children.go b/services/syncbase/server/util/glob_children.go
index e874163..13007e4 100644
--- a/services/syncbase/server/util/glob_children.go
+++ b/services/syncbase/server/util/glob_children.go
@@ -10,27 +10,31 @@
"v.io/v23/naming"
"v.io/v23/rpc"
pubutil "v.io/v23/syncbase/util"
+ "v.io/v23/verror"
"v.io/x/ref/services/syncbase/store"
)
// Note, Syncbase handles Glob requests by implementing GlobChildren__ at each
// level (service, app, database, table).
-// GlobChildren performs a glob. It calls closeSntx to close sntx.
-func GlobChildren(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element, sntx store.SnapshotOrTransaction, closeSntx func() error, stKeyPrefix string) error {
+// GlobChildren implements glob over the Syncbase namespace.
+func GlobChildren(ctx *context.T, call rpc.GlobChildrenServerCall, matcher *glob.Element, sntx store.SnapshotOrTransaction, stKeyPrefix string) error {
prefix, _ := matcher.FixedPrefix()
- it := sntx.Scan(ScanPrefixArgs(stKeyPrefix, prefix))
- defer closeSntx()
+ unescPrefix, ok := pubutil.Unescape(prefix)
+ if !ok {
+ return verror.New(verror.ErrBadArg, ctx, prefix)
+ }
+ it := sntx.Scan(ScanPrefixArgs(stKeyPrefix, unescPrefix))
key := []byte{}
for it.Advance() {
key = it.Key(key)
parts := SplitKeyParts(string(key))
- name := parts[len(parts)-1]
+ // For explanation of Escape(), see comment in server/nosql/dispatcher.go.
+ name := pubutil.Escape(parts[len(parts)-1])
if matcher.Match(name) {
// TODO(rogulenko): Check for resolve access. (For table glob, this means
// checking prefix perms.)
- // For explanation of Escape(), see comment in server/nosql/dispatcher.go.
- if err := call.SendStream().Send(naming.GlobChildrenReplyName{Value: pubutil.Escape(name)}); err != nil {
+ if err := call.SendStream().Send(naming.GlobChildrenReplyName{Value: name}); err != nil {
return err
}
}
diff --git a/services/syncbase/testutil/layer.go b/services/syncbase/testutil/layer.go
index 755aae9..7c4e6b0 100644
--- a/services/syncbase/testutil/layer.go
+++ b/services/syncbase/testutil/layer.go
@@ -7,6 +7,7 @@
import (
"fmt"
"reflect"
+ "sort"
"testing"
"v.io/v23/context"
@@ -103,16 +104,13 @@
for _, name := range notOkNames {
if err := parent.Child(name).Create(ctx, nil); err == nil {
t.Fatalf("Create(%q) should have failed: %v", name, err)
+ } else if name != "" && verror.ErrorID(err) != wire.ErrInvalidName.ID {
// TODO(sadovsky): Currently for name "" we cannot check for
// ErrInvalidName, since parent.Child("") dispatches on the parent
// component. Create will still fail, but for a different reason. If/when
// we add client-side name validation, we'll be able to check for
// ErrInvalidName specifically.
- // TODO(sadovsky): Will dropping the "TrimPrefix" calls in our server-side
- // dispatchers fix this?
- if name != "" && verror.ErrorID(err) != wire.ErrInvalidName.ID {
- t.Fatalf("Create(%q) should have failed with ErrInvalidName: %v", name, err)
- }
+ t.Fatalf("Create(%q) should have failed with ErrInvalidName: %v", name, err)
}
}
}
@@ -199,43 +197,36 @@
assertExists(t, ctx, self, "self", false)
}
-func TestListChildren(t *testing.T, ctx *context.T, i interface{}) {
+func TestListChildren(t *testing.T, ctx *context.T, i interface{}, names []string) {
self := makeLayer(i)
var got, want []string
var err error
- got, err = self.ListChildren(ctx)
- want = []string{}
- if err != nil {
- t.Fatalf("self.ListChildren() failed: %v", err)
- }
- if !reflect.DeepEqual([]string(got), want) {
- t.Fatalf("Lists do not match: got %v, want %v", got, want)
- }
+ sortedNames := make([]string, len(names))
+ copy(sortedNames, names)
+ sort.Strings(sortedNames)
+ names = sortedNames
- if err := self.Child("y").Create(ctx, nil); err != nil {
- t.Fatalf("y.Create() failed: %v", err)
- }
- got, err = self.ListChildren(ctx)
- want = []string{"y"}
- if err != nil {
- t.Fatalf("self.ListChildren() failed: %v", err)
- }
- if !reflect.DeepEqual(got, want) {
- t.Fatalf("Lists do not match: got %v, want %v", got, want)
- }
-
- if err := self.Child("x").Create(ctx, nil); err != nil {
- t.Fatalf("x.Create() failed: %v", err)
- }
- got, err = self.ListChildren(ctx)
- want = []string{"x", "y"}
- if err != nil {
- t.Fatalf("self.ListChildren() failed: %v", err)
- }
- if !reflect.DeepEqual(got, want) {
- t.Fatalf("Lists do not match: got %v, want %v", got, want)
+ for i := 0; i <= len(names); i++ {
+ got, err = self.ListChildren(ctx)
+ want = names[:i]
+ if err != nil {
+ t.Fatalf("self.ListChildren() failed: %v", err)
+ }
+ if got == nil {
+ got = []string{}
+ }
+ if !reflect.DeepEqual(got, want) {
+ t.Fatalf("Lists do not match: got %v, want %v", got, want)
+ }
+ if i == len(names) {
+ break
+ }
+ name := names[i]
+ if err := self.Child(name).Create(ctx, nil); err != nil {
+ t.Fatalf("Create(%q) failed: %v", name, err)
+ }
}
}
@@ -340,6 +331,8 @@
type layer interface {
util.AccessController
+ Name() string
+ FullName() string
Create(ctx *context.T, perms access.Permissions) error
Destroy(ctx *context.T) error
Exists(ctx *context.T) (bool, error)
@@ -351,6 +344,9 @@
syncbase.Service
}
+func (s *service) Name() string {
+ panic(notAvailable)
+}
func (s *service) Create(ctx *context.T, perms access.Permissions) error {
panic(notAvailable)
}
@@ -411,6 +407,9 @@
nosql.Row
}
+func (r *row) Name() string {
+ return r.Key()
+}
func (r *row) Create(ctx *context.T, perms access.Permissions) error {
if perms != nil {
panic(fmt.Sprintf("bad perms: %v", perms))
diff --git a/services/xproxyd/proxy_test.go b/services/xproxyd/proxy_test.go
index 2414920..4a08676 100644
--- a/services/xproxyd/proxy_test.go
+++ b/services/xproxyd/proxy_test.go
@@ -7,6 +7,7 @@
import (
"bufio"
"fmt"
+ "math/rand"
"strings"
"sync"
"testing"
@@ -163,21 +164,19 @@
p3ep := startProxy(t, pctx, address{"v23", p2ep.String()}, address{"kill", "127.0.0.1:0"})
- ch := make(chan struct{})
- var allEps []naming.Endpoint
- idx := 0
+ done := make(chan struct{})
update := func(eps []naming.Endpoint) {
// TODO(suharshs): Fix this test once we have the proxy send update messages to the
- // server when it reconnects to a proxy.
- if len(eps) == 3 {
- allEps = eps
- }
+ // server when it reconnects to a proxy. This test only really tests the first connection
+ // currently because the connections are cached. So we need to kill connections and
+ // wait for them to reestablish but we need proxies to update communicate their new endpoints
+ // to each other and to the server. For now we at least check a random endpoint so the
+ // test will at least fail over many runs if something is wrong.
if len(eps) > 0 {
- if err := testEndToEndConnection(t, dctx, actx, dm, am, allEps[idx]); err != nil {
+ if err := testEndToEndConnection(t, dctx, actx, dm, am, eps[rand.Int()%3]); err != nil {
t.Error(err)
}
- idx++
- ch <- struct{}{}
+ close(done)
}
}
@@ -185,13 +184,7 @@
t.Fatal(err)
}
- <-ch
- // Test the other two endpoints.
- for i := 0; i < 2; i++ {
- // Kill the connections to test reconnection.
- kp.KillConnections()
- <-ch
- }
+ <-done
}
func testEndToEndConnection(t *testing.T, dctx, actx *context.T, dm, am flow.Manager, aep naming.Endpoint) error {
