blob: 5106b029ed53f82cdcee4b35afe239ff44900427 [file] [log] [blame]
// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package namespace
import (
"io"
"strings"
"sync"
"v.io/v23"
"v.io/v23/context"
"v.io/v23/glob"
"v.io/v23/naming"
"v.io/v23/options"
"v.io/v23/rpc"
"v.io/v23/verror"
"v.io/x/ref/lib/apilog"
)
type tracks struct {
m sync.Mutex
places map[string]struct{}
}
func (tr *tracks) beenThereDoneThat(servers []naming.MountedServer, pstr string) bool {
tr.m.Lock()
defer tr.m.Unlock()
found := false
for _, s := range servers {
x := naming.JoinAddressName(s.Server, "") + "!" + pstr
if _, ok := tr.places[x]; ok {
found = true
}
tr.places[x] = struct{}{}
}
return found
}
// task is a sub-glob that has to be performed against a mount table. Tasks are
// done in parrallel to speed up the glob.
type task struct {
pattern *glob.Glob // pattern to match
er *naming.GlobError // error for that particular point in the name space
me *naming.MountEntry // server to match at
error error // any error performing this task
depth int // number of mount tables traversed recursively
}
// globAtServer performs a Glob on the servers at a mount point. It cycles through the set of
// servers until it finds one that replies.
func (ns *namespace) globAtServer(ctx *context.T, t *task, replies chan *task, tr *tracks, opts []rpc.CallOpt) {
defer func() {
if t.error == nil {
replies <- nil
} else {
replies <- t
}
}()
client := v23.GetClient(ctx)
pstr := t.pattern.String()
ctx.VI(2).Infof("globAtServer(%v, %v)", *t.me, pstr)
// If there are no servers to call, this isn't a mount point. No sense
// trying to call servers that aren't there.
if len(t.me.Servers) == 0 {
t.error = nil
return
}
// If we've been there before with the same request, give up.
if tr.beenThereDoneThat(t.me.Servers, pstr) {
t.error = nil
return
}
// t.me.Name has already been matched at this point to so don't pass it to the Call. Kind of sleazy to do this
// but it avoids making yet another copy of the MountEntry.
on := t.me.Name
t.me.Name = ""
call, err := client.StartCall(withTimeout(ctx), "", rpc.GlobMethod, []interface{}{pstr}, append(opts, options.Preresolved{t.me})...)
t.me.Name = on
if err != nil {
t.error = err
return
}
// At this point we're commited to the server that answered the call
// first. Cycle through all replies from that server.
for {
// If the mount table returns an error, we're done. Send the task to the channel
// including the error. This terminates the task.
var gr naming.GlobReply
err := call.Recv(&gr)
if err == io.EOF {
break
}
if err != nil {
t.error = err
return
}
var x *task
switch v := gr.(type) {
case naming.GlobReplyEntry:
// Convert to the ever so slightly different name.MountTable version of a MountEntry
// and add it to the list.
x = &task{
me: &naming.MountEntry{
Name: naming.Join(t.me.Name, v.Value.Name),
Servers: v.Value.Servers,
ServesMountTable: v.Value.ServesMountTable,
IsLeaf: v.Value.IsLeaf,
},
depth: t.depth + 1,
}
case naming.GlobReplyError:
// Pass on the error.
x = &task{
er: &v.Value,
depth: t.depth + 1,
}
}
// x.depth is the number of servers we've walked through since we've gone
// recursive (i.e. with pattern length of 0). Limit the depth of globs.
// TODO(p): return an error?
if t.pattern.Len() == 0 {
if x.depth > ns.maxRecursiveGlobDepth {
continue
}
}
replies <- x
}
t.error = call.Finish()
return
}
// depth returns the directory depth of a given name. It is used to pick off the unsatisfied part of the pattern.
func depth(name string) int {
name = strings.Trim(naming.Clean(name), "/")
if name == "" {
return 0
}
return strings.Count(name, "/") + 1
}
// globLoop fires off a go routine for each server and reads backs replies.
func (ns *namespace) globLoop(ctx *context.T, e *naming.MountEntry, prefix string, pattern *glob.Glob, reply chan naming.GlobReply, tr *tracks, opts []rpc.CallOpt) {
defer close(reply)
// Provide enough buffers to avoid too much switching between the readers and the writers.
// This size is just a guess.
replies := make(chan *task, 100)
defer close(replies)
// Push the first task into the channel to start the ball rolling. This task has the
// root of the search and the full pattern. It will be the first task fired off in the for
// loop that follows.
replies <- &task{me: e, pattern: pattern}
replies <- nil
inFlight := 1
// Perform a parallel search of the name graph. Each task will send what it learns
// on the replies channel. If the reply is a mount point and the pattern is not completely
// fulfilled, a new task will be fired off to handle it.
for inFlight != 0 {
t := <-replies
// A nil reply represents a successfully terminated task.
// If no tasks are running, return.
if t == nil {
inFlight--
continue
}
// We want to output this entry if there was a real error other than
// "not a mount table".
//
// An error reply is also a terminated task.
// If no tasks are running, return.
if t.error != nil {
if !notAnMT(t.error) {
reply <- &naming.GlobReplyError{Value: naming.GlobError{Name: naming.Join(prefix, t.me.Name), Error: t.error}}
}
inFlight--
continue
}
// If this is just an error from the mount table, pass it on.
if t.er != nil {
x := *t.er
x.Name = naming.Join(prefix, x.Name)
reply <- &naming.GlobReplyError{Value: x}
continue
}
// Get the pattern elements below the current path.
suffix := pattern
for i := depth(t.me.Name) - 1; i >= 0; i-- {
suffix = suffix.Tail()
}
// If we've satisfied the request and this isn't the root,
// reply to the caller.
if suffix.Len() == 0 && t.depth != 0 {
x := *t.me
x.Name = naming.Join(prefix, x.Name)
reply <- &naming.GlobReplyEntry{Value: x}
}
// If the pattern is finished (so we're only querying about the root on the
// remote server) and the server is not another MT, then we needn't send the
// query on since we know the server will not supply a new address for the
// current name.
if suffix.Empty() {
if !t.me.ServesMountTable {
continue
}
}
// If this is restricted recursive and not a mount table, don't descend into it.
if suffix.Restricted() && suffix.Len() == 0 && !t.me.ServesMountTable {
continue
}
// Perform a glob at the next server.
inFlight++
t.pattern = suffix
go ns.globAtServer(ctx, t, replies, tr, opts)
}
}
// Glob implements naming.MountTable.Glob.
func (ns *namespace) Glob(ctx *context.T, pattern string, opts ...naming.NamespaceOpt) (<-chan naming.GlobReply, error) {
defer apilog.LogCallf(ctx, "pattern=%.10s...,opts...=%v", pattern, opts)(ctx, "") // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
// Root the pattern. If we have no servers to query, give up.
e, patternWasRooted := ns.rootMountEntry(pattern)
if len(e.Servers) == 0 {
return nil, verror.New(naming.ErrNoMountTable, ctx)
}
// If the name doesn't parse, give up.
g, err := glob.Parse(e.Name)
if err != nil {
return nil, err
}
tr := &tracks{places: make(map[string]struct{})}
// If pattern was already rooted, make sure we tack that root
// onto all returned names. Otherwise, just return the relative
// name.
var prefix string
if patternWasRooted {
prefix = e.Servers[0].Server
}
e.Name = ""
reply := make(chan naming.GlobReply, 100)
go ns.globLoop(ctx, e, prefix, g, reply, tr, getCallOpts(opts))
return reply, nil
}