ref: Change the shutdown mechanism for the new runtime.
Now we properly lame duck servers, and avoid error messages on clean
shutdowns..
MultiPart: 2/2
Change-Id: Ifb5e3080ff28a4496819c3ebdf2a8b0f7c1a02db
diff --git a/runtime/factories/fake/rpc.go b/runtime/factories/fake/rpc.go
index 22f0bf0..f020b7b 100644
--- a/runtime/factories/fake/rpc.go
+++ b/runtime/factories/fake/rpc.go
@@ -12,13 +12,22 @@
"v.io/x/ref/lib/apilog"
)
-// SetClient can be used to inject a mock client implementation into the context.
-func SetClient(ctx *context.T, client rpc.Client) *context.T {
+// SetClientFactory can be used to inject a mock Client implementation
+// into the context. When v23.WithNewClient is called passed function
+// will be invoked.
+func SetClientFactory(ctx *context.T, factory func(ctx *context.T, opts ...rpc.ClientOpt) rpc.Client) *context.T {
+ client := factory(ctx)
+ ctx = context.WithValue(ctx, clientFactoryKey, factory)
return context.WithValue(ctx, clientKey, client)
}
func (r *Runtime) WithNewClient(ctx *context.T, opts ...rpc.ClientOpt) (*context.T, rpc.Client, error) {
defer apilog.LogCallf(ctx, "opts...=%v", opts)(ctx, "") // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
- panic("unimplemented")
+ factory, ok := ctx.Value(clientFactoryKey).(func(ctx *context.T, opts ...rpc.ClientOpt) rpc.Client)
+ if !ok {
+ panic("Calling WithNewClient on the fake runtime, but no factory has been set.")
+ }
+ client := factory(ctx, opts...)
+ return context.WithValue(ctx, clientKey, client), client, nil
}
func (r *Runtime) GetClient(ctx *context.T) rpc.Client {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
@@ -26,11 +35,6 @@
return c
}
-func (r *Runtime) WithNewStreamManager(ctx *context.T) (*context.T, error) {
- defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
- panic("unimplemented")
-}
-
func (r *Runtime) GetListenSpec(ctx *context.T) rpc.ListenSpec {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
ls, _ := ctx.Value(listenSpecKey).(rpc.ListenSpec)
@@ -43,21 +47,28 @@
return ctx
}
-func SetFlowManager(ctx *context.T, manager flow.Manager) *context.T {
- return context.WithValue(ctx, flowManagerKey, manager)
-}
-
-func (r *Runtime) ExperimentalGetFlowManager(ctx *context.T) flow.Manager {
- defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
- fm, _ := ctx.Value(flowManagerKey).(flow.Manager)
- return fm
-}
-
-func (r *Runtime) ExperimentalWithNewFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
+func (r *Runtime) WithNewStreamManager(ctx *context.T) (*context.T, error) {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
panic("unimplemented")
}
+// SetFlowManagerFactory can be used to inject a mock FlowManager
+// implementation into the context. When v23.NewFlowManager is called
+// passed function will be invoked.
+func SetFlowManagerFactory(ctx *context.T, factory func(ctx *context.T) flow.Manager) *context.T {
+ return context.WithValue(ctx, flowFactoryKey, factory)
+}
+
+func (r *Runtime) NewFlowManager(ctx *context.T) (flow.Manager, error) {
+ defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
+
+ factory, ok := ctx.Value(flowFactoryKey).(func(ctx *context.T) flow.Manager)
+ if !ok {
+ panic("Calling NewFlowManager on the fake runtime, but no factory has been set.")
+ }
+ return factory(ctx), nil
+}
+
func (r *Runtime) WithNewServer(ctx *context.T, name string, object interface{}, auth security.Authorizer, opts ...rpc.ServerOpt) (*context.T, rpc.Server, error) {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
panic("unimplemented")
diff --git a/runtime/factories/fake/runtime.go b/runtime/factories/fake/runtime.go
index 464ab11..ef7e102 100644
--- a/runtime/factories/fake/runtime.go
+++ b/runtime/factories/fake/runtime.go
@@ -24,7 +24,9 @@
loggerKey
backgroundKey
listenSpecKey
- flowManagerKey
+
+ clientFactoryKey
+ flowFactoryKey
)
type Runtime struct {
diff --git a/runtime/internal/flow/conn/auth.go b/runtime/internal/flow/conn/auth.go
index 16b31d1..40337ec 100644
--- a/runtime/internal/flow/conn/auth.go
+++ b/runtime/internal/flow/conn/auth.go
@@ -14,7 +14,6 @@
"golang.org/x/crypto/nacl/box"
"v.io/v23"
"v.io/v23/context"
- "v.io/v23/flow"
"v.io/v23/flow/message"
"v.io/v23/rpc/version"
"v.io/v23/security"
@@ -33,8 +32,9 @@
if err != nil {
return err
}
+
bflow := c.newFlowLocked(ctx, blessingsFlowID, 0, 0, true, true)
- bflow.worker.Release(ctx, defaultBufferSize)
+ bflow.worker.Release(ctx, DefaultBytesBufferedPerFlow)
c.blessingsFlow = newBlessingsFlow(ctx, &c.loopWG, bflow, true)
if err = c.readRemoteAuth(ctx, authAcceptorTag, binding); err != nil {
@@ -217,6 +217,7 @@
type blessingsFlow struct {
enc *vom.Encoder
dec *vom.Decoder
+ f *flw
mu sync.Mutex
cond *sync.Cond
@@ -226,8 +227,9 @@
byBKey map[uint64]*Blessings
}
-func newBlessingsFlow(ctx *context.T, loopWG *sync.WaitGroup, f flow.Flow, dialed bool) *blessingsFlow {
+func newBlessingsFlow(ctx *context.T, loopWG *sync.WaitGroup, f *flw, dialed bool) *blessingsFlow {
b := &blessingsFlow{
+ f: f,
enc: vom.NewEncoder(f),
dec: vom.NewDecoder(f),
nextKey: 1,
diff --git a/runtime/internal/flow/conn/auth_test.go b/runtime/internal/flow/conn/auth_test.go
index 422eb73..c64bc5b 100644
--- a/runtime/internal/flow/conn/auth_test.go
+++ b/runtime/internal/flow/conn/auth_test.go
@@ -95,14 +95,18 @@
func (fc *fakeDischargeClient) StartCall(*context.T, string, string, []interface{}, ...rpc.CallOpt) (rpc.ClientCall, error) {
return nil, nil
}
-func (fc *fakeDischargeClient) Close() {}
+func (fc *fakeDischargeClient) Close() {}
+func (fc *fakeDischargeClient) Closed() <-chan struct{} { return nil }
func TestUnidirectional(t *testing.T) {
defer goroutines.NoLeaks(t, leakWaitTime)()
ctx, shutdown := v23.Init()
defer shutdown()
- ctx = fake.SetClient(ctx, &fakeDischargeClient{v23.GetPrincipal(ctx)})
+ ctx = fake.SetClientFactory(ctx, func(ctx *context.T, opts ...rpc.ClientOpt) rpc.Client {
+ return &fakeDischargeClient{v23.GetPrincipal(ctx)}
+ })
+ ctx, _, _ = v23.WithNewClient(ctx)
dctx := NewPrincipalWithTPCaveat(t, ctx, "dialer")
actx := NewPrincipalWithTPCaveat(t, ctx, "acceptor")
@@ -136,7 +140,10 @@
ctx, shutdown := v23.Init()
defer shutdown()
- ctx = fake.SetClient(ctx, &fakeDischargeClient{v23.GetPrincipal(ctx)})
+ ctx = fake.SetClientFactory(ctx, func(ctx *context.T, opts ...rpc.ClientOpt) rpc.Client {
+ return &fakeDischargeClient{v23.GetPrincipal(ctx)}
+ })
+ ctx, _, _ = v23.WithNewClient(ctx)
dctx := NewPrincipalWithTPCaveat(t, ctx, "dialer")
actx := NewPrincipalWithTPCaveat(t, ctx, "acceptor")
diff --git a/runtime/internal/flow/conn/conn.go b/runtime/internal/flow/conn/conn.go
index 17475a5..846fce3 100644
--- a/runtime/internal/flow/conn/conn.go
+++ b/runtime/internal/flow/conn/conn.go
@@ -29,7 +29,7 @@
)
const mtu = 1 << 16
-const defaultBufferSize = 1 << 20
+const DefaultBytesBufferedPerFlow = 1 << 20
const (
expressPriority = iota
@@ -95,10 +95,11 @@
conn flow.MsgReadWriteCloser,
local, remote naming.Endpoint,
versions version.RPCVersionRange,
+ handshakeTimeout time.Duration,
handler FlowHandler,
events chan<- StatusUpdate) (*Conn, error) {
c := &Conn{
- fc: flowcontrol.New(defaultBufferSize, mtu),
+ fc: flowcontrol.New(DefaultBytesBufferedPerFlow, mtu),
mp: newMessagePipe(conn),
handler: handler,
lBlessings: v23.GetPrincipal(ctx).BlessingStore().Default(),
@@ -112,7 +113,14 @@
borrowing: map[uint64]bool{},
events: events,
}
- if err := c.dialHandshake(ctx, versions); err != nil {
+ // TODO(mattr): This scheme for deadlines is nice, but it doesn't
+ // provide for cancellation when ctx is canceled.
+ t := time.AfterFunc(handshakeTimeout, func() { conn.Close() })
+ err := c.dialHandshake(ctx, versions)
+ if stopped := t.Stop(); !stopped {
+ err = verror.NewErrTimeout(ctx)
+ }
+ if err != nil {
c.Close(ctx, err)
return nil, err
}
@@ -127,10 +135,11 @@
conn flow.MsgReadWriteCloser,
local naming.Endpoint,
versions version.RPCVersionRange,
+ handshakeTimeout time.Duration,
handler FlowHandler,
events chan<- StatusUpdate) (*Conn, error) {
c := &Conn{
- fc: flowcontrol.New(defaultBufferSize, mtu),
+ fc: flowcontrol.New(DefaultBytesBufferedPerFlow, mtu),
mp: newMessagePipe(conn),
handler: handler,
lBlessings: v23.GetPrincipal(ctx).BlessingStore().Default(),
@@ -143,7 +152,14 @@
borrowing: map[uint64]bool{},
events: events,
}
- if err := c.acceptHandshake(ctx, versions); err != nil {
+ // FIXME(mattr): This scheme for deadlines is nice, but it doesn't
+ // provide for cancellation when ctx is canceled.
+ t := time.AfterFunc(handshakeTimeout, func() { conn.Close() })
+ err := c.acceptHandshake(ctx, versions)
+ if stopped := t.Stop(); !stopped {
+ err = verror.NewErrTimeout(ctx)
+ }
+ if err != nil {
c.Close(ctx, err)
return nil, err
}
@@ -250,17 +266,28 @@
if err != nil {
msg = err.Error()
}
- cerr := c.fc.Run(ctx, "close", expressPriority, func(_ int) (int, bool, error) {
+ // We use a root context here to ensure that the message get's sent even if
+ // the context is canceled.
+ // TODO(mattr): Consider other options like a special mode in the flow controller
+ // that doesn't care about the context. Or perhaps 'stop' the flow controller
+ // and then just write the message directly.
+ rootctx, cancel := context.WithRootCancel(ctx)
+ cerr := c.fc.Run(rootctx, "close", expressPriority, func(_ int) (int, bool, error) {
return 0, true, c.mp.writeMsg(ctx, &message.TearDown{Message: msg})
})
if cerr != nil {
ctx.Errorf("Error sending tearDown on connection to %s: %v", c.remote, cerr)
}
+ cancel()
}
+ flowErr := NewErrConnectionClosed(ctx)
for _, f := range flows {
- f.close(ctx, NewErrConnectionClosed(ctx))
+ f.close(ctx, flowErr)
}
- if cerr := c.mp.close(); cerr != nil {
+ if c.blessingsFlow != nil {
+ c.blessingsFlow.f.close(ctx, flowErr)
+ }
+ if cerr := c.mp.rw.Close(); cerr != nil {
ctx.Errorf("Error closing underlying connection for %s: %v", c.remote, cerr)
}
c.loopWG.Wait()
@@ -282,9 +309,9 @@
c.toRelease[fid] += count
if c.borrowing[fid] {
c.toRelease[invalidFlowID] += count
- release = c.toRelease[invalidFlowID] > defaultBufferSize/2
+ release = c.toRelease[invalidFlowID] > DefaultBytesBufferedPerFlow/2
} else {
- release = c.toRelease[fid] > defaultBufferSize/2
+ release = c.toRelease[fid] > DefaultBytesBufferedPerFlow/2
}
if release {
toRelease = c.toRelease
@@ -358,7 +385,7 @@
handler := c.handler
f := c.newFlowLocked(ctx, msg.ID, msg.BlessingsKey, msg.DischargeKey, false, true)
f.worker.Release(ctx, int(msg.InitialCounters))
- c.toRelease[msg.ID] = defaultBufferSize
+ c.toRelease[msg.ID] = DefaultBytesBufferedPerFlow
c.borrowing[msg.ID] = true
c.mu.Unlock()
@@ -403,7 +430,8 @@
f := c.flows[msg.ID]
c.mu.Unlock()
if f == nil {
- ctx.Infof("Ignoring data message for unknown flow on connection to %s: %d", c.remote, msg.ID)
+ ctx.Infof("Ignoring data message for unknown flow on connection to %s: %d",
+ c.remote, msg.ID)
return nil
}
if err := f.q.put(ctx, msg.Payload); err != nil {
diff --git a/runtime/internal/flow/conn/conn_test.go b/runtime/internal/flow/conn/conn_test.go
index 6837486..ce1961b 100644
--- a/runtime/internal/flow/conn/conn_test.go
+++ b/runtime/internal/flow/conn/conn_test.go
@@ -27,7 +27,7 @@
func init() {
test.Init()
- randData = make([]byte, 2*defaultBufferSize)
+ randData = make([]byte, 2*DefaultBytesBufferedPerFlow)
if _, err := rand.Read(randData); err != nil {
panic("Could not read random data.")
}
@@ -100,14 +100,14 @@
q1, q2 := make(chan flow.Flow, 1), make(chan flow.Flow, 1)
fh1, fh2 := fh(q1), fh(q2)
go func() {
- d, err := NewDialed(ctx, dmrw, ep, ep, versions, nil, nil)
+ d, err := NewDialed(ctx, dmrw, ep, ep, versions, time.Minute, nil, nil)
if err != nil {
panic(err)
}
dch <- d
}()
go func() {
- a, err := NewAccepted(ctx, amrw, ep, versions, fh1, nil)
+ a, err := NewAccepted(ctx, amrw, ep, versions, time.Minute, fh1, nil)
if err != nil {
panic(err)
}
diff --git a/runtime/internal/flow/conn/flow.go b/runtime/internal/flow/conn/flow.go
index 04fdfcf..72929d1 100644
--- a/runtime/internal/flow/conn/flow.go
+++ b/runtime/internal/flow/conn/flow.go
@@ -6,6 +6,7 @@
import (
"strconv"
+ "time"
"v.io/v23/context"
"v.io/v23/flow"
@@ -42,7 +43,7 @@
dkey: dkey,
opened: preopen,
}
- f.SetContext(ctx)
+ f.ctx, f.cancel = context.WithCancel(ctx)
if !f.opened {
c.unopenedFlows.Add(1)
}
@@ -132,7 +133,7 @@
} else {
err = f.conn.mp.writeMsg(f.ctx, &message.OpenFlow{
ID: f.id,
- InitialCounters: defaultBufferSize,
+ InitialCounters: DefaultBytesBufferedPerFlow,
BlessingsKey: f.bkey,
DischargeKey: f.dkey,
Flags: d.Flags,
@@ -177,12 +178,16 @@
//
// TODO(mattr): update v23/flow documentation.
// SetContext may not be called concurrently with other methods.
-func (f *flw) SetContext(ctx *context.T) error {
+func (f *flw) SetDeadlineContext(ctx *context.T, deadline time.Time) *context.T {
if f.cancel != nil {
f.cancel()
}
- f.ctx, f.cancel = context.WithCancel(ctx)
- return nil
+ if !deadline.IsZero() {
+ f.ctx, f.cancel = context.WithDeadline(ctx, deadline)
+ } else {
+ f.ctx, f.cancel = context.WithCancel(ctx)
+ }
+ return f.ctx
}
// LocalBlessings returns the blessings presented by the local end of the flow
@@ -264,36 +269,38 @@
}
func (f *flw) close(ctx *context.T, err error) {
- f.q.close(ctx)
- f.cancel()
- // We want to try to send this message even if ctx is already canceled.
- ctx, cancel := context.WithRootCancel(ctx)
- serr := f.worker.Run(ctx, func(tokens int) (int, bool, error) {
- f.conn.mu.Lock()
- delete(f.conn.flows, f.id)
- connClosed := f.conn.flows == nil
- f.conn.mu.Unlock()
+ if f.q.close(ctx) {
+ f.cancel()
+ // We want to try to send this message even if ctx is already canceled.
+ ctx, cancel := context.WithRootCancel(ctx)
+ serr := f.worker.Run(ctx, func(tokens int) (int, bool, error) {
+ f.conn.mu.Lock()
+ delete(f.conn.flows, f.id)
+ connClosed := f.conn.flows == nil
+ f.conn.mu.Unlock()
- if !f.opened {
- // Closing a flow that was never opened.
- f.conn.unopenedFlows.Done()
- return 0, true, nil
- } else if eid := verror.ErrorID(err); eid == ErrFlowClosedRemotely.ID || connClosed {
- // Note: If the conn is closed there is no point in trying to send
- // the flow close message as it will fail. This is racy with the connection
- // closing, but there are no ill-effects other than spamming the logs a little
- // so it's OK.
- return 0, true, nil
- }
- return 0, true, f.conn.mp.writeMsg(ctx, &message.Data{
- ID: f.id,
- Flags: message.CloseFlag,
+ eid := verror.ErrorID(err)
+ if !f.opened {
+ // Closing a flow that was never opened.
+ f.conn.unopenedFlows.Done()
+ return 0, true, nil
+ } else if eid == ErrFlowClosedRemotely.ID || connClosed {
+ // Note: If the conn is closed there is no point in trying to send
+ // the flow close message as it will fail. This is racy with the connection
+ // closing, but there are no ill-effects other than spamming the logs a little
+ // so it's OK.
+ return 0, true, nil
+ }
+ return 0, true, f.conn.mp.writeMsg(ctx, &message.Data{
+ ID: f.id,
+ Flags: message.CloseFlag,
+ })
})
- })
- if serr != nil {
- ctx.Errorf("Could not send close flow message: %v", err)
+ if serr != nil {
+ ctx.Errorf("Could not send close flow message: %v", err)
+ }
+ cancel()
}
- cancel()
}
// Close marks the flow as closed. After Close is called, new data cannot be
diff --git a/runtime/internal/flow/conn/message.go b/runtime/internal/flow/conn/message.go
index 9435354..cf51f8f 100644
--- a/runtime/internal/flow/conn/message.go
+++ b/runtime/internal/flow/conn/message.go
@@ -35,10 +35,6 @@
return p.cipher.ChannelBinding()
}
-func (p *messagePipe) close() error {
- return p.rw.Close()
-}
-
func (p *messagePipe) writeMsg(ctx *context.T, m message.Message) (err error) {
// TODO(mattr): Because of the API of the underlying crypto library,
// an enormous amount of copying happens here.
diff --git a/runtime/internal/flow/conn/readq.go b/runtime/internal/flow/conn/readq.go
index 54db134..89abc97 100644
--- a/runtime/internal/flow/conn/readq.go
+++ b/runtime/internal/flow/conn/readq.go
@@ -50,7 +50,7 @@
return nil
}
newSize := l + r.size
- if newSize > defaultBufferSize {
+ if newSize > DefaultBytesBufferedPerFlow {
return NewErrCounterOverflow(ctx)
}
newBufs := r.nbufs + len(bufs)
@@ -124,13 +124,16 @@
return
}
-func (r *readq) close(ctx *context.T) {
+func (r *readq) close(ctx *context.T) bool {
r.mu.Lock()
+ closed := false
if r.e != -1 {
+ closed = true
r.e = -1
close(r.notify)
}
r.mu.Unlock()
+ return closed
}
func (r *readq) reserveLocked(n int) {
diff --git a/runtime/internal/flow/conn/util_test.go b/runtime/internal/flow/conn/util_test.go
index b68247b..eeb7207 100644
--- a/runtime/internal/flow/conn/util_test.go
+++ b/runtime/internal/flow/conn/util_test.go
@@ -51,7 +51,7 @@
if dflows != nil {
handler = fh(dflows)
}
- d, err := NewDialed(dctx, dmrw, ep, ep, versions, handler, events)
+ d, err := NewDialed(dctx, dmrw, ep, ep, versions, time.Minute, handler, events)
if err != nil {
panic(err)
}
@@ -62,7 +62,7 @@
if aflows != nil {
handler = fh(aflows)
}
- a, err := NewAccepted(actx, amrw, ep, versions, handler, events)
+ a, err := NewAccepted(actx, amrw, ep, versions, time.Minute, handler, events)
if err != nil {
panic(err)
}
diff --git a/runtime/internal/flow/manager/conncache_test.go b/runtime/internal/flow/manager/conncache_test.go
index 4c31df8..4dcc990 100644
--- a/runtime/internal/flow/manager/conncache_test.go
+++ b/runtime/internal/flow/manager/conncache_test.go
@@ -7,13 +7,13 @@
import (
"strconv"
"testing"
+ "time"
"v.io/v23"
"v.io/v23/context"
"v.io/v23/flow"
"v.io/v23/naming"
"v.io/v23/rpc/version"
-
connpackage "v.io/x/ref/runtime/internal/flow/conn"
"v.io/x/ref/runtime/internal/flow/flowtest"
inaming "v.io/x/ref/runtime/internal/naming"
@@ -296,7 +296,7 @@
ach := make(chan *connpackage.Conn)
go func() {
d, err := connpackage.NewDialed(ctx, dmrw, ep, ep,
- version.RPCVersionRange{Min: 1, Max: 5}, nil, nil)
+ version.RPCVersionRange{Min: 1, Max: 5}, time.Minute, nil, nil)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
@@ -305,7 +305,7 @@
fh := fh{t, make(chan struct{})}
go func() {
a, err := connpackage.NewAccepted(ctx, amrw, ep,
- version.RPCVersionRange{Min: 1, Max: 5}, fh, nil)
+ version.RPCVersionRange{Min: 1, Max: 5}, time.Minute, fh, nil)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
diff --git a/runtime/internal/flow/manager/manager.go b/runtime/internal/flow/manager/manager.go
index d8c8fee..0bc1ea2 100644
--- a/runtime/internal/flow/manager/manager.go
+++ b/runtime/internal/flow/manager/manager.go
@@ -20,7 +20,6 @@
"v.io/v23/security"
"v.io/v23/verror"
- iflow "v.io/x/ref/runtime/internal/flow"
"v.io/x/ref/runtime/internal/flow/conn"
"v.io/x/ref/runtime/internal/lib/upcqueue"
inaming "v.io/x/ref/runtime/internal/naming"
@@ -30,6 +29,7 @@
const (
reconnectDelay = 50 * time.Millisecond
reapCacheInterval = 5 * time.Minute
+ handshakeTimeout = time.Minute
)
type manager struct {
@@ -37,6 +37,7 @@
closed chan struct{}
cache *ConnCache
ls *listenState
+ ctx *context.T
}
func New(ctx *context.T, rid naming.RoutingID) flow.Manager {
@@ -44,6 +45,7 @@
rid: rid,
closed: make(chan struct{}),
cache: NewConnCache(),
+ ctx: ctx,
}
var events chan conn.StatusUpdate
if rid != naming.NullRoutingID {
@@ -122,17 +124,13 @@
case <-m.closed:
case <-done:
}
+ // Now nobody should send any more flows, so close the queue.
+ m.ls.q.Close()
}
// Listen causes the Manager to accept flows from the provided protocol and address.
// Listen may be called muliple times.
-//
-// The flow.Manager associated with ctx must be the receiver of the method,
-// otherwise an error is returned.
func (m *manager) Listen(ctx *context.T, protocol, address string) error {
- if err := m.validateContext(ctx); err != nil {
- return err
- }
if m.ls == nil {
return NewErrListeningWithNullRid(ctx)
}
@@ -165,13 +163,7 @@
//
// update gets passed the complete set of endpoints for the proxy every time it
// is called.
-//
-// The flow.Manager associated with ctx must be the receiver of the method,
-// otherwise an error is returned.
func (m *manager) ProxyListen(ctx *context.T, ep naming.Endpoint, update func([]naming.Endpoint)) error {
- if err := m.validateContext(ctx); err != nil {
- return err
- }
if m.ls == nil {
return NewErrListeningWithNullRid(ctx)
}
@@ -263,16 +255,22 @@
flowConn, err = ln.Accept(ctx)
}
if err != nil {
- ctx.Errorf("ln.Accept on localEP %v failed: %v", local, err)
+ m.ls.mu.Lock()
+ closed := m.ls.listeners == nil
+ m.ls.mu.Unlock()
+ if !closed {
+ ctx.Errorf("ln.Accept on localEP %v failed: %v", local, err)
+ }
return
}
fh := &flowHandler{m, make(chan struct{})}
m.ls.activeConns.Add(1)
c, err := conn.NewAccepted(
- ctx,
+ m.ctx,
flowConn,
local,
version.Supported,
+ handshakeTimeout,
fh,
m.ls.events)
if err != nil {
@@ -311,6 +309,7 @@
f,
f.Conn().LocalEndpoint(),
version.Supported,
+ handshakeTimeout,
fh,
h.m.ls.events)
if err != nil {
@@ -355,13 +354,7 @@
// }
//
// can be used to accept Flows initiated by remote processes.
-//
-// The flow.Manager associated with ctx must be the receiver of the method,
-// otherwise an error is returned.
func (m *manager) Accept(ctx *context.T) (flow.Flow, error) {
- if err := m.validateContext(ctx); err != nil {
- return nil, err
- }
if m.ls == nil {
return nil, NewErrListeningWithNullRid(ctx)
}
@@ -381,13 +374,7 @@
//
// To maximize re-use of connections, the Manager will also Listen on Dialed
// connections for the lifetime of the connection.
-//
-// The flow.Manager associated with ctx must be the receiver of the method,
-// otherwise an error is returned.
func (m *manager) Dial(ctx *context.T, remote naming.Endpoint, fn flow.BlessingsForPeer) (flow.Flow, error) {
- if err := m.validateContext(ctx); err != nil {
- return nil, err
- }
f, _, err := m.internalDial(ctx, remote, fn)
return f, err
}
@@ -443,11 +430,12 @@
m.ls.activeConns.Add(1)
}
c, err = conn.NewDialed(
- ctx,
+ m.ctx,
flowConn,
localEndpoint(flowConn, m.rid),
remote,
version.Supported,
+ handshakeTimeout,
fh,
events,
)
@@ -474,11 +462,12 @@
m.ls.activeConns.Add(1)
}
c, err = conn.NewDialed(
- ctx,
+ m.ctx,
f,
proxyConn.LocalEndpoint(),
remote,
version.Supported,
+ handshakeTimeout,
fh,
events)
if err != nil {
@@ -512,13 +501,6 @@
return m.closed
}
-func (m *manager) validateContext(ctx *context.T) error {
- if v23.ExperimentalGetFlowManager(ctx) != m {
- return flow.NewErrBadArg(ctx, iflow.NewErrWrongObjectInContext(ctx, "manager"))
- }
- return nil
-}
-
func dial(ctx *context.T, p flow.Protocol, protocol, address string) (flow.Conn, error) {
if p != nil {
var timeout time.Duration
diff --git a/runtime/internal/flow/manager/manager_test.go b/runtime/internal/flow/manager/manager_test.go
index 4babdd0..c151586 100644
--- a/runtime/internal/flow/manager/manager_test.go
+++ b/runtime/internal/flow/manager/manager_test.go
@@ -14,8 +14,7 @@
"v.io/v23/context"
"v.io/v23/flow"
"v.io/v23/naming"
-
- "v.io/x/ref/runtime/factories/fake"
+ _ "v.io/x/ref/runtime/factories/fake"
"v.io/x/ref/runtime/internal/flow/conn"
"v.io/x/ref/runtime/internal/flow/flowtest"
"v.io/x/ref/test"
@@ -33,14 +32,12 @@
ctx, shutdown := v23.Init()
am := New(ctx, naming.FixedRoutingID(0x5555))
- actx := fake.SetFlowManager(ctx, am)
- if err := am.Listen(actx, "tcp", "127.0.0.1:0"); err != nil {
+ if err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil {
t.Fatal(err)
}
dm := New(ctx, naming.FixedRoutingID(0x1111))
- dctx := fake.SetFlowManager(ctx, dm)
- testFlows(t, dctx, actx, flowtest.BlessingsForPeer)
+ testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
shutdown()
<-am.Closed()
@@ -52,26 +49,24 @@
ctx, shutdown := v23.Init()
am := New(ctx, naming.FixedRoutingID(0x5555))
- actx := fake.SetFlowManager(ctx, am)
- if err := am.Listen(actx, "tcp", "127.0.0.1:0"); err != nil {
+ if err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil {
t.Fatal(err)
}
dm := New(ctx, naming.FixedRoutingID(0x1111))
- dctx := fake.SetFlowManager(ctx, dm)
// At first the cache should be empty.
if got, want := len(dm.(*manager).cache.addrCache), 0; got != want {
t.Fatalf("got cache size %v, want %v", got, want)
}
// After dialing a connection the cache should hold one connection.
- testFlows(t, dctx, actx, flowtest.BlessingsForPeer)
+ testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
if got, want := len(dm.(*manager).cache.addrCache), 1; got != want {
t.Fatalf("got cache size %v, want %v", got, want)
}
old := dm.(*manager).cache.ridCache[am.RoutingID()]
// After dialing another connection the cache should still hold one connection
// because the connections should be reused.
- testFlows(t, dctx, actx, flowtest.BlessingsForPeer)
+ testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
if got, want := len(dm.(*manager).cache.addrCache), 1; got != want {
t.Errorf("got cache size %v, want %v", got, want)
}
@@ -89,16 +84,14 @@
ctx, shutdown := v23.Init()
am := New(ctx, naming.FixedRoutingID(0x5555))
- actx := fake.SetFlowManager(ctx, am)
- if err := am.Listen(actx, "tcp", "127.0.0.1:0"); err != nil {
+ if err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil {
t.Fatal(err)
}
dm := New(ctx, naming.FixedRoutingID(0x1111))
- dctx := fake.SetFlowManager(ctx, dm)
- testFlows(t, dctx, actx, flowtest.BlessingsForPeer)
+ testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
// Now am should be able to make a flow to dm even though dm is not listening.
- testFlows(t, actx, dctx, flowtest.BlessingsForPeer)
+ testFlows(t, ctx, am, dm, flowtest.BlessingsForPeer)
shutdown()
<-am.Closed()
@@ -110,20 +103,17 @@
ctx, shutdown := v23.Init()
am := New(ctx, naming.FixedRoutingID(0x5555))
- actx := fake.SetFlowManager(ctx, am)
- if err := am.Listen(actx, "tcp", "127.0.0.1:0"); err != nil {
+ if err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil {
t.Fatal(err)
}
nulldm := New(ctx, naming.NullRoutingID)
- nctx := fake.SetFlowManager(ctx, nulldm)
- _, af := testFlows(t, nctx, actx, flowtest.BlessingsForPeer)
+ _, af := testFlows(t, ctx, nulldm, am, flowtest.BlessingsForPeer)
// Ensure that the remote blessings of the underlying conn of the accepted flow are zero.
if rBlessings := af.Conn().(*conn.Conn).RemoteBlessings(); !rBlessings.IsZero() {
t.Errorf("got %v, want zero-value blessings", rBlessings)
}
dm := New(ctx, naming.FixedRoutingID(0x1111))
- dctx := fake.SetFlowManager(ctx, dm)
- _, af = testFlows(t, dctx, actx, flowtest.BlessingsForPeer)
+ _, af = testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
// Ensure that the remote blessings of the underlying conn of the accepted flow are
// non-zero if we did specify a RoutingID.
if rBlessings := af.Conn().(*conn.Conn).RemoteBlessings(); rBlessings.IsZero() {
@@ -141,19 +131,15 @@
ctx, shutdown := v23.Init()
am := New(ctx, naming.FixedRoutingID(0x5555))
- actx := fake.SetFlowManager(ctx, am)
- if err := am.Listen(actx, "tcp", "127.0.0.1:0"); err != nil {
+ if err := am.Listen(ctx, "tcp", "127.0.0.1:0"); err != nil {
t.Fatal(err)
}
-
dm := New(ctx, naming.FixedRoutingID(0x1111))
- dctx := fake.SetFlowManager(ctx, dm)
+ testFlows(t, ctx, dm, am, flowtest.BlessingsForPeer)
- testFlows(t, dctx, actx, flowtest.BlessingsForPeer)
+ am.StopListening(ctx)
- am.StopListening(actx)
-
- if f, err := dm.Dial(dctx, am.ListeningEndpoints()[0], flowtest.BlessingsForPeer); err == nil {
+ if f, err := dm.Dial(ctx, am.ListeningEndpoints()[0], flowtest.BlessingsForPeer); err == nil {
t.Errorf("dialing a lame duck should fail, but didn't %#v.", f)
}
@@ -162,17 +148,16 @@
<-dm.Closed()
}
-func testFlows(t *testing.T, dctx, actx *context.T, bFn flow.BlessingsForPeer) (df, af flow.Flow) {
- am := v23.ExperimentalGetFlowManager(actx)
+func testFlows(t *testing.T, ctx *context.T, dm, am flow.Manager, bFn flow.BlessingsForPeer) (df, af flow.Flow) {
ep := am.ListeningEndpoints()[0]
var err error
- df, err = v23.ExperimentalGetFlowManager(dctx).Dial(dctx, ep, bFn)
+ df, err = dm.Dial(ctx, ep, bFn)
if err != nil {
t.Fatal(err)
}
want := "do you read me?"
writeLine(df, want)
- af, err = am.Accept(actx)
+ af, err = am.Accept(ctx)
if err != nil {
t.Fatal(err)
}
diff --git a/runtime/internal/rpc/client.go b/runtime/internal/rpc/client.go
index a7f427f..33db6e7 100644
--- a/runtime/internal/rpc/client.go
+++ b/runtime/internal/rpc/client.go
@@ -75,35 +75,24 @@
ns namespace.T
vcOpts []stream.VCOpt // vc opts passed to dial
preferredProtocols []string
+ vcCache *vc.VCCache
+ wg sync.WaitGroup
+ dc vc.DischargeClient
- // We cache the IP networks on the device since it is not that cheap to read
- // network interfaces through os syscall.
- // TODO(toddw): this can be removed since netstate now implements caching
- // directly.
- ipNets []*net.IPNet
-
- vcCache *vc.VCCache
-
- wg sync.WaitGroup
- mu sync.Mutex
- closed bool
-
- dc vc.DischargeClient
+ mu sync.Mutex
+ closed bool
+ closech chan struct{}
}
var _ rpc.Client = (*client)(nil)
-func DeprecatedNewClient(streamMgr stream.Manager, ns namespace.T, opts ...rpc.ClientOpt) (rpc.Client, error) {
+func DeprecatedNewClient(streamMgr stream.Manager, ns namespace.T, opts ...rpc.ClientOpt) rpc.Client {
c := &client{
streamMgr: streamMgr,
ns: ns,
vcCache: vc.NewVCCache(),
+ closech: make(chan struct{}),
}
- ipNets, err := ipNetworks()
- if err != nil {
- return nil, err
- }
- c.ipNets = ipNets
c.dc = InternalNewDischargeClient(nil, c, 0)
for _, opt := range opts {
// Collect all client opts that are also vc opts.
@@ -114,8 +103,7 @@
c.preferredProtocols = v
}
}
-
- return c, nil
+ return c
}
func (c *client) createFlow(ctx *context.T, principal security.Principal, ep naming.Endpoint, vcOpts []stream.VCOpt) (stream.Flow, *verror.SubErr) {
@@ -446,7 +434,7 @@
return nil, verror.NoRetry, true, verror.New(verror.ErrInternal, ctx, name)
}
// An empty set of protocols means all protocols...
- if resolved.Servers, err = filterAndOrderServers(resolved.Servers, c.preferredProtocols, c.ipNets); err != nil {
+ if resolved.Servers, err = filterAndOrderServers(resolved.Servers, c.preferredProtocols); err != nil {
return nil, verror.RetryRefetch, true, verror.New(verror.ErrNoServers, ctx, name, err)
}
}
@@ -752,7 +740,10 @@
func (c *client) Close() {
defer apilog.LogCall(nil)(nil) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
c.mu.Lock()
- c.closed = true
+ if !c.closed {
+ c.closed = true
+ close(c.closech)
+ }
c.mu.Unlock()
for _, v := range c.vcCache.Close() {
c.streamMgr.ShutdownEndpoint(v.RemoteEndpoint())
@@ -760,6 +751,10 @@
c.wg.Wait()
}
+func (c *client) Closed() <-chan struct{} {
+ return c.closech
+}
+
// flowClient implements the RPC client-side protocol for a single RPC, over a
// flow that's already connected to the server.
type flowClient struct {
diff --git a/runtime/internal/rpc/errors.vdl b/runtime/internal/rpc/errors.vdl
index 8f24707..a57dafc 100644
--- a/runtime/internal/rpc/errors.vdl
+++ b/runtime/internal/rpc/errors.vdl
@@ -27,4 +27,7 @@
badAuth(suffix, method string, err error) {
"en": "not authorized to call {suffix}.{method}: {err}",
}
+ typeFlowFailure(err error) {
+ "en": "type flow could not be constructed{:err}",
+ }
)
diff --git a/runtime/internal/rpc/errors.vdl.go b/runtime/internal/rpc/errors.vdl.go
index 2c543d7..949ba3b 100644
--- a/runtime/internal/rpc/errors.vdl.go
+++ b/runtime/internal/rpc/errors.vdl.go
@@ -23,6 +23,7 @@
errBadBlessingsCache = verror.Register("v.io/x/ref/runtime/internal/rpc.badBlessingsCache", verror.NoRetry, "{1:}{2:} failed to find blessings in cache: {3}")
errBadDischarge = verror.Register("v.io/x/ref/runtime/internal/rpc.badDischarge", verror.NoRetry, "{1:}{2:} failed to decode discharge #{3}: {4}")
errBadAuth = verror.Register("v.io/x/ref/runtime/internal/rpc.badAuth", verror.NoRetry, "{1:}{2:} not authorized to call {3}.{4}: {5}")
+ errTypeFlowFailure = verror.Register("v.io/x/ref/runtime/internal/rpc.typeFlowFailure", verror.NoRetry, "{1:}{2:} type flow could not be constructed{:3}")
)
func init() {
@@ -33,6 +34,7 @@
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(errBadBlessingsCache.ID), "{1:}{2:} failed to find blessings in cache: {3}")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(errBadDischarge.ID), "{1:}{2:} failed to decode discharge #{3}: {4}")
i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(errBadAuth.ID), "{1:}{2:} not authorized to call {3}.{4}: {5}")
+ i18n.Cat().SetWithBase(i18n.LangID("en"), i18n.MsgID(errTypeFlowFailure.ID), "{1:}{2:} type flow could not be constructed{:3}")
}
// newErrBadRequest returns an error with the errBadRequest ID.
@@ -69,3 +71,8 @@
func newErrBadAuth(ctx *context.T, suffix string, method string, err error) error {
return verror.New(errBadAuth, ctx, suffix, method, err)
}
+
+// newErrTypeFlowFailure returns an error with the errTypeFlowFailure ID.
+func newErrTypeFlowFailure(ctx *context.T, err error) error {
+ return verror.New(errTypeFlowFailure, ctx, err)
+}
diff --git a/runtime/internal/rpc/full_test.go b/runtime/internal/rpc/full_test.go
index 495c048..6dfe486 100644
--- a/runtime/internal/rpc/full_test.go
+++ b/runtime/internal/rpc/full_test.go
@@ -70,10 +70,7 @@
}
func testInternalNewServerWithPubsub(ctx *context.T, streamMgr stream.Manager, ns namespace.T, settingsPublisher *pubsub.Publisher, settingsStreamName string, opts ...rpc.ServerOpt) (DeprecatedServer, error) {
- client, err := DeprecatedNewClient(streamMgr, ns)
- if err != nil {
- return nil, err
- }
+ client := DeprecatedNewClient(streamMgr, ns)
return DeprecatedNewServer(ctx, streamMgr, ns, settingsPublisher, settingsStreamName, client, opts...)
}
@@ -366,10 +363,7 @@
if server != nil {
b.ep, b.server = startServerWS(t, ctx, server, b.sm, b.ns, b.name, testServerDisp{ts}, shouldUseWebsocket)
}
- var err error
- if b.client, err = DeprecatedNewClient(b.sm, b.ns); err != nil {
- t.Fatalf("DeprecatedNewClient failed: %v", err)
- }
+ b.client = DeprecatedNewClient(b.sm, b.ns)
return
}
@@ -502,10 +496,7 @@
if err := server.ServeDispatcher("mp/server", disp); err != nil {
t.Fatalf("server.Serve failed: %v", err)
}
- client, err := DeprecatedNewClient(sm, ns)
- if err != nil {
- t.Fatalf("DeprecatedNewClient failed: %v", err)
- }
+ client := DeprecatedNewClient(sm, ns)
// When using SecurityNone, all authorization checks should be skipped, so
// unauthorized methods should be callable.
var got string
@@ -535,10 +526,7 @@
if err := server.ServeDispatcher("mp/server", disp); err != nil {
t.Fatalf("server.Serve failed: %v", err)
}
- client, err := DeprecatedNewClient(sm, ns)
- if err != nil {
- t.Fatalf("DeprecatedNewClient failed: %v", err)
- }
+ client := DeprecatedNewClient(sm, ns)
// A call should fail if the principal in the ctx is nil and SecurityNone is not specified.
ctx, err = v23.WithPrincipal(ctx, nil)
@@ -583,12 +571,7 @@
runClient := func(server string) ([]string, error) {
smc := imanager.InternalNew(ctx, naming.FixedRoutingID(0xc))
defer smc.Shutdown()
- client, err := DeprecatedNewClient(
- smc,
- ns)
- if err != nil {
- return nil, err
- }
+ client := DeprecatedNewClient(smc, ns)
defer client.Close()
ctx, _ = v23.WithPrincipal(cctx, pclient)
call, err := client.StartCall(cctx, server, "Closure", nil)
@@ -653,10 +636,7 @@
}
smc := imanager.InternalNew(ctx, rid)
defer smc.Shutdown()
- client, err := DeprecatedNewClient(smc, ns)
- if err != nil {
- t.Fatalf("failed to create client: %v", err)
- }
+ client := DeprecatedNewClient(smc, ns)
defer client.Close()
var opts []rpc.CallOpt
if noDischarges {
@@ -718,10 +698,7 @@
}
sm := imanager.InternalNew(ctx, rid)
- c, err := DeprecatedNewClient(sm, ns)
- if err != nil {
- t.Fatalf("failed to create client: %v", err)
- }
+ c := DeprecatedNewClient(sm, ns)
dc := c.(*client).dc
tpcav2, err := security.NewPublicKeyCaveat(pdischarger2.PublicKey(), "mountpoint/discharger2", security.ThirdPartyRequirements{}, mkCaveat(security.NewExpiryCaveat(time.Now().Add(time.Hour))))
if err != nil {
@@ -768,11 +745,7 @@
}
smc := imanager.InternalNew(sctx, rid)
defer smc.Shutdown()
- client, err := DeprecatedNewClient(smc, ns)
- if err != nil {
- t.Fatalf("failed to create client: %v", err)
- }
- return client
+ return DeprecatedNewClient(smc, ns)
}
runClient := func(client rpc.Client) {
@@ -847,14 +820,12 @@
defer runServer(t, sctx, ns, mountName, &testServer{}).Shutdown()
smc := imanager.InternalNew(sctx, naming.FixedRoutingID(0xc))
- client, err := DeprecatedNewClient(smc, ns)
- if err != nil {
- t.Fatal(err)
- }
+ client := DeprecatedNewClient(smc, ns)
defer smc.Shutdown()
defer client.Close()
// The call should succeed when the server presents the same public as the opt...
+ var err error
if _, err = client.StartCall(cctx, mountName, "Closure", nil, options.SkipServerEndpointAuthorization{}, options.ServerPublicKey{
PublicKey: pserver.PublicKey(),
}); err != nil {
@@ -918,10 +889,7 @@
}
smc := imanager.InternalNew(ctx, rid)
defer smc.Shutdown()
- client, err := DeprecatedNewClient(smc, ns)
- if err != nil {
- t.Fatalf("failed to create client: %v", err)
- }
+ client := DeprecatedNewClient(smc, ns)
defer client.Close()
dc := InternalNewDischargeClient(ctx, client, 0)
diff --git a/runtime/internal/rpc/resolve_internal_test.go b/runtime/internal/rpc/resolve_internal_test.go
index 98e9dba..56ebd6f 100644
--- a/runtime/internal/rpc/resolve_internal_test.go
+++ b/runtime/internal/rpc/resolve_internal_test.go
@@ -5,13 +5,14 @@
package rpc
import (
+ "v.io/v23/context"
"v.io/v23/rpc"
"v.io/x/ref"
)
-func InternalServerResolveToEndpoint(s rpc.Server, name string) (string, error) {
+func InternalServerResolveToEndpoint(ctx *context.T, s rpc.Server, name string) (string, error) {
if ref.RPCTransitionState() == ref.XServers {
- ep, err := s.(*xserver).resolveToEndpoint(name)
+ ep, err := s.(*xserver).resolveToEndpoint(ctx, name)
if err != nil {
return "", err
}
diff --git a/runtime/internal/rpc/resolve_test.go b/runtime/internal/rpc/resolve_test.go
index be38047..455334b 100644
--- a/runtime/internal/rpc/resolve_test.go
+++ b/runtime/internal/rpc/resolve_test.go
@@ -141,7 +141,7 @@
{"unknown", "", notfound},
}
for _, tc := range testcases {
- result, err := irpc.InternalServerResolveToEndpoint(server, tc.address)
+ result, err := irpc.InternalServerResolveToEndpoint(ctx, server, tc.address)
if (err == nil) != (tc.err == nil) {
t.Errorf("Unexpected err for %q. Got %v, expected %v", tc.address, err, tc.err)
}
diff --git a/runtime/internal/rpc/server.go b/runtime/internal/rpc/server.go
index 846ac32..ed23e5c 100644
--- a/runtime/internal/rpc/server.go
+++ b/runtime/internal/rpc/server.go
@@ -185,7 +185,6 @@
// We cache the IP networks on the device since it is not that cheap to read
// network interfaces through os syscall.
// TODO(jhahn): Add monitoring the network interface changes.
- ipNets []*net.IPNet
ns namespace.T
servesMountTable bool
isLeaf bool
@@ -273,11 +272,6 @@
dischargeExpiryBuffer = vc.DefaultServerDischargeExpiryBuffer
securityLevel options.SecurityLevel
)
- ipNets, err := ipNetworks()
- if err != nil {
- return nil, err
- }
- s.ipNets = ipNets
for _, opt := range opts {
switch opt := opt.(type) {
@@ -399,7 +393,7 @@
}}
}
// An empty set of protocols means all protocols...
- if resolved.Servers, err = filterAndOrderServers(resolved.Servers, s.preferredProtocols, s.ipNets); err != nil {
+ if resolved.Servers, err = filterAndOrderServers(resolved.Servers, s.preferredProtocols); err != nil {
return "", err
}
for _, n := range resolved.Names() {
@@ -1023,6 +1017,10 @@
return nil
}
+func (s *server) Closed() <-chan struct{} {
+ return s.ctx.Done()
+}
+
// flowServer implements the RPC server-side protocol for a single RPC, over a
// flow that's already connected to the client.
type flowServer struct {
diff --git a/runtime/internal/rpc/sort_endpoints.go b/runtime/internal/rpc/sort_endpoints.go
index 3b5b896..c587027 100644
--- a/runtime/internal/rpc/sort_endpoints.go
+++ b/runtime/internal/rpc/sort_endpoints.go
@@ -82,7 +82,13 @@
// will be used, but unlike the previous case, any servers that don't support
// these protocols will be returned also, but following the default
// preferences.
-func filterAndOrderServers(servers []naming.MountedServer, protocols []string, ipnets []*net.IPNet) ([]naming.MountedServer, error) {
+func filterAndOrderServers(servers []naming.MountedServer, protocols []string, ipnets ...*net.IPNet) ([]naming.MountedServer, error) {
+ if ipnets == nil {
+ var err error
+ if ipnets, err = ipNetworks(); err != nil {
+ return nil, err
+ }
+ }
var (
errs = verror.SubErrs{}
list = make(sortableServerList, 0, len(servers))
@@ -94,6 +100,7 @@
adderr := func(name string, err error) {
errs = append(errs, verror.SubErr{Name: "server=" + name, Err: err, Options: verror.Print})
}
+
for _, server := range servers {
name := server.Server
ep, err := name2endpoint(name)
diff --git a/runtime/internal/rpc/sort_internal_test.go b/runtime/internal/rpc/sort_internal_test.go
index 7600d2b..59edbcd 100644
--- a/runtime/internal/rpc/sort_internal_test.go
+++ b/runtime/internal/rpc/sort_internal_test.go
@@ -21,7 +21,7 @@
func TestIncompatible(t *testing.T) {
servers := []naming.MountedServer{}
- _, err := filterAndOrderServers(servers, []string{"tcp"}, nil)
+ _, err := filterAndOrderServers(servers, []string{"tcp"})
if err == nil || err.Error() != "failed to find any compatible servers" {
t.Errorf("expected a different error: %v", err)
}
@@ -31,7 +31,7 @@
servers = append(servers, naming.MountedServer{Server: name})
}
- _, err = filterAndOrderServers(servers, []string{"foobar"}, nil)
+ _, err = filterAndOrderServers(servers, []string{"foobar"})
if err == nil || !strings.HasSuffix(err.Error(), "undesired protocol: tcp]") {
t.Errorf("expected a different error to: %v", err)
}
@@ -58,7 +58,7 @@
name := naming.JoinAddressName(naming.FormatEndpoint("tcp6", a), "")
servers = append(servers, naming.MountedServer{Server: name})
}
- if _, err := filterAndOrderServers(servers, []string{"batman"}, ipnets); err == nil {
+ if _, err := filterAndOrderServers(servers, []string{"batman"}, ipnets...); err == nil {
t.Fatalf("expected an error")
}
@@ -77,7 +77,7 @@
"/@6@tcp4@127.0.0.2@@@@@@",
"/127.0.0.12:14141",
}
- result, err := filterAndOrderServers(servers, []string{"foobar", "tcp4"}, ipnets)
+ result, err := filterAndOrderServers(servers, []string{"foobar", "tcp4"}, ipnets...)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -101,14 +101,14 @@
"/@6@foobar@127.0.0.11@@@@@@",
"/127.0.0.12:14141",
}
- if result, err = filterAndOrderServers(servers, nil, ipnets); err != nil {
+ if result, err = filterAndOrderServers(servers, nil, ipnets...); err != nil {
t.Fatalf("unexpected error: %s", err)
}
if got := servers2names(result); !reflect.DeepEqual(got, want) {
t.Errorf("got: %v, want %v", got, want)
}
- if result, err = filterAndOrderServers(servers, []string{}, ipnets); err != nil {
+ if result, err = filterAndOrderServers(servers, []string{}, ipnets...); err != nil {
t.Fatalf("unexpected error: %s", err)
}
if got := servers2names(result); !reflect.DeepEqual(got, want) {
@@ -125,7 +125,7 @@
"/@6@tcp6@127.0.0.8@@@@@@",
"/127.0.0.12:14141",
}
- if result, err = filterAndOrderServers(servers, []string{"tcp"}, ipnets); err != nil {
+ if result, err = filterAndOrderServers(servers, []string{"tcp"}, ipnets...); err != nil {
t.Fatalf("unexpected error: %s", err)
}
if got := servers2names(result); !reflect.DeepEqual(got, want) {
@@ -154,7 +154,7 @@
name := naming.JoinAddressName(naming.FormatEndpoint("foobar", a), "")
servers = append(servers, naming.MountedServer{Server: name})
}
- if result, err = filterAndOrderServers(servers, []string{}, ipnets); err != nil {
+ if result, err = filterAndOrderServers(servers, []string{}, ipnets...); err != nil {
t.Fatalf("unexpected error: %s", err)
}
if got := servers2names(result); !reflect.DeepEqual(got, want) {
@@ -171,7 +171,7 @@
name := naming.JoinAddressName(naming.FormatEndpoint("tcp", a), "")
servers = append(servers, naming.MountedServer{Server: name})
}
- result, err := filterAndOrderServers(servers, []string{"tcp"}, ipnets)
+ result, err := filterAndOrderServers(servers, []string{"tcp"}, ipnets...)
if err != nil {
t.Fatalf("unexpected error: %s", err)
}
@@ -190,7 +190,7 @@
servers = append(servers, naming.MountedServer{Server: name})
}
- if result, err = filterAndOrderServers(servers, []string{"ws", "tcp"}, ipnets); err != nil {
+ if result, err = filterAndOrderServers(servers, []string{"ws", "tcp"}, ipnets...); err != nil {
t.Fatalf("unexpected error: %s", err)
}
want = []string{
diff --git a/runtime/internal/rpc/test/cancel_test.go b/runtime/internal/rpc/test/cancel_test.go
index b363a33..42ef4c2 100644
--- a/runtime/internal/rpc/test/cancel_test.go
+++ b/runtime/internal/rpc/test/cancel_test.go
@@ -13,6 +13,8 @@
"v.io/v23/rpc"
"v.io/v23/security"
"v.io/v23/verror"
+ "v.io/x/ref"
+ "v.io/x/ref/runtime/internal/flow/conn"
"v.io/x/ref/runtime/internal/rpc/stream/vc"
"v.io/x/ref/test"
)
@@ -27,14 +29,18 @@
func (c *canceld) Run(ctx *context.T, _ rpc.ServerCall) error {
close(c.started)
client := v23.GetClient(ctx)
- ctx.Infof("Run: %s", c.child)
+ var done chan struct{}
if c.child != "" {
- if _, err := client.StartCall(ctx, c.child, "Run", []interface{}{}); err != nil {
- ctx.Error(err)
- return err
- }
+ done = make(chan struct{})
+ go func() {
+ client.Call(ctx, c.child, "Run", nil, nil)
+ close(done)
+ }()
}
<-ctx.Done()
+ if done != nil {
+ <-done
+ }
close(c.canceled)
return nil
}
@@ -50,7 +56,6 @@
if err != nil {
return nil, err
}
- ctx.Infof("Serving: %q", name)
return c, nil
}
@@ -70,20 +75,18 @@
}
ctx, cancel := context.WithCancel(ctx)
- _, err = v23.GetClient(ctx).StartCall(ctx, "c1", "Run", []interface{}{})
- if err != nil {
- t.Fatalf("can't call: ", err)
- }
+ done := make(chan struct{})
+ go func() {
+ v23.GetClient(ctx).Call(ctx, "c1", "Run", nil, nil)
+ close(done)
+ }()
<-c1.started
<-c2.started
-
- ctx.Info("cancelling initial call")
cancel()
-
- ctx.Info("waiting for children to be canceled")
<-c1.canceled
<-c2.canceled
+ <-done
}
type cancelTestServer struct {
@@ -141,11 +144,13 @@
t.Fatal(err)
}
cctx, cancel := context.WithCancel(cctx)
- _, err = v23.GetClient(cctx).StartCall(cctx, "cancel", "CancelStreamReader", []interface{}{})
- if err != nil {
- t.Fatalf("Start call failed: %v", err)
- }
+ done := make(chan struct{})
+ go func() {
+ v23.GetClient(cctx).Call(cctx, "cancel", "CancelStreamReader", nil, nil)
+ close(done)
+ }()
waitForCancel(t, ts, cancel)
+ <-done
}
// TestCancelWithFullBuffers tests that even if the writer has filled the buffers and
@@ -163,16 +168,25 @@
t.Fatal(err)
}
cctx, cancel := context.WithCancel(cctx)
- call, err := v23.GetClient(cctx).StartCall(cctx, "cancel", "CancelStreamIgnorer", []interface{}{})
+ call, err := v23.GetClient(cctx).StartCall(cctx, "cancel", "CancelStreamIgnorer", nil)
if err != nil {
t.Fatalf("Start call failed: %v", err)
}
// Fill up all the write buffers to ensure that cancelling works even when the stream
// is blocked.
- // TODO(mattr): Update for new RPC system.
- call.Send(make([]byte, vc.MaxSharedBytes))
- call.Send(make([]byte, vc.DefaultBytesBufferedPerFlow))
+ if ref.RPCTransitionState() >= ref.XServers {
+ call.Send(conn.DefaultBytesBufferedPerFlow)
+ } else {
+ call.Send(make([]byte, vc.MaxSharedBytes))
+ call.Send(make([]byte, vc.DefaultBytesBufferedPerFlow))
+ }
+ done := make(chan struct{})
+ go func() {
+ call.Finish()
+ close(done)
+ }()
waitForCancel(t, ts, cancel)
+ <-done
}
diff --git a/runtime/internal/rpc/transitionclient.go b/runtime/internal/rpc/transitionclient.go
index ade2e3a..2516a43 100644
--- a/runtime/internal/rpc/transitionclient.go
+++ b/runtime/internal/rpc/transitionclient.go
@@ -6,7 +6,6 @@
import (
"v.io/v23/context"
- "v.io/v23/flow"
"v.io/v23/flow/message"
"v.io/v23/namespace"
"v.io/v23/rpc"
@@ -15,22 +14,24 @@
)
type transitionClient struct {
- c, xc rpc.Client
+ c, xc rpc.Client
+ closech chan struct{}
}
var _ = rpc.Client((*transitionClient)(nil))
-func NewTransitionClient(ctx *context.T, streamMgr stream.Manager, flowMgr flow.Manager, ns namespace.T, opts ...rpc.ClientOpt) (rpc.Client, error) {
- var err error
- ret := &transitionClient{}
- if ret.xc, err = NewXClient(ctx, flowMgr, ns, opts...); err != nil {
- return nil, err
+func NewTransitionClient(ctx *context.T, streamMgr stream.Manager, ns namespace.T, opts ...rpc.ClientOpt) rpc.Client {
+ tc := &transitionClient{
+ xc: NewXClient(ctx, ns, opts...),
+ c: DeprecatedNewClient(streamMgr, ns, opts...),
+ closech: make(chan struct{}),
}
- if ret.c, err = DeprecatedNewClient(streamMgr, ns, opts...); err != nil {
- ret.xc.Close()
- return nil, err
- }
- return ret, nil
+ go func() {
+ <-tc.xc.Closed()
+ <-tc.c.Closed()
+ close(tc.closech)
+ }()
+ return tc
}
func (t *transitionClient) StartCall(ctx *context.T, name, method string, args []interface{}, opts ...rpc.CallOpt) (rpc.ClientCall, error) {
@@ -50,6 +51,10 @@
}
func (t *transitionClient) Close() {
- t.xc.Close()
t.c.Close()
+ <-t.xc.Closed()
+}
+
+func (t *transitionClient) Closed() <-chan struct{} {
+ return t.closech
}
diff --git a/runtime/internal/rpc/typecache.go b/runtime/internal/rpc/typecache.go
index c734913..2fcb6bc 100644
--- a/runtime/internal/rpc/typecache.go
+++ b/runtime/internal/rpc/typecache.go
@@ -7,6 +7,7 @@
import (
"sync"
+ "v.io/v23/context"
"v.io/v23/flow"
"v.io/v23/vom"
)
@@ -54,7 +55,7 @@
return
}
-func (tc *typeCache) get(c flow.ManagedConn) (*vom.TypeEncoder, *vom.TypeDecoder) {
+func (tc *typeCache) get(ctx *context.T, c flow.ManagedConn) (*vom.TypeEncoder, *vom.TypeDecoder, error) {
tc.mu.Lock()
tce := tc.flows[c]
if tce == nil {
@@ -62,8 +63,14 @@
tc.flows[c] = tce
}
tc.mu.Unlock()
- <-tce.ready
- return tce.enc, tce.dec
+ select {
+ case <-c.Closed():
+ return nil, nil, newErrTypeFlowFailure(ctx, nil)
+ case <-ctx.Done():
+ return nil, nil, ctx.Err()
+ case <-tce.ready:
+ }
+ return tce.enc, tce.dec, nil
}
func (tc *typeCache) collect() {
diff --git a/runtime/internal/rpc/x_test.go b/runtime/internal/rpc/x_test.go
index 0c4bf18..5568371 100644
--- a/runtime/internal/rpc/x_test.go
+++ b/runtime/internal/rpc/x_test.go
@@ -9,6 +9,7 @@
"v.io/v23"
"v.io/v23/context"
+ "v.io/v23/flow"
"v.io/v23/naming"
"v.io/v23/rpc"
"v.io/v23/security"
@@ -26,21 +27,24 @@
func TestXClientServer(t *testing.T) {
ctx, shutdown := v23.Init()
defer shutdown()
- ctx = fake.SetFlowManager(ctx, manager.New(ctx, naming.FixedRoutingID(0x1)))
+ var i uint64 = 1
+ ctx = fake.SetFlowManagerFactory(ctx, func(ctx *context.T) flow.Manager {
+ i++
+ return manager.New(ctx, naming.FixedRoutingID(i))
+ })
+ ctx = fake.SetClientFactory(ctx, func(ctx *context.T, o ...rpc.ClientOpt) rpc.Client {
+ return NewXClient(ctx, v23.GetNamespace(ctx), o...)
+ })
+
ctx = v23.WithListenSpec(ctx, rpc.ListenSpec{
Addrs: rpc.ListenAddrs{{Protocol: "tcp", Address: "127.0.0.1:0"}},
})
- _, err := NewServer(ctx, "server", &testService{}, nil, nil, "")
- if err != nil {
- t.Fatal(verror.DebugString(err))
- }
- ctx = fake.SetFlowManager(ctx, manager.New(ctx, naming.FixedRoutingID(0x2)))
- client, err := NewXClient(ctx, v23.ExperimentalGetFlowManager(ctx), v23.GetNamespace(ctx))
+ _, _, err := WithNewServer(ctx, "server", &testService{}, nil, nil, "")
if err != nil {
t.Fatal(verror.DebugString(err))
}
var result string
- if err = client.Call(ctx, "server", "Echo", []interface{}{"hello"}, []interface{}{&result}); err != nil {
+ if err = v23.GetClient(ctx).Call(ctx, "server", "Echo", []interface{}{"hello"}, []interface{}{&result}); err != nil {
t.Fatal(verror.DebugString(err))
}
if want := "response:hello"; result != want {
@@ -57,21 +61,24 @@
func TestXClientDispatchingServer(t *testing.T) {
ctx, shutdown := v23.Init()
defer shutdown()
- ctx = fake.SetFlowManager(ctx, manager.New(ctx, naming.FixedRoutingID(0x1)))
+ var i uint64 = 1
+ ctx = fake.SetFlowManagerFactory(ctx, func(ctx *context.T) flow.Manager {
+ i++
+ return manager.New(ctx, naming.FixedRoutingID(i))
+ })
+ ctx = fake.SetClientFactory(ctx, func(ctx *context.T, o ...rpc.ClientOpt) rpc.Client {
+ return NewXClient(ctx, v23.GetNamespace(ctx), o...)
+ })
+
ctx = v23.WithListenSpec(ctx, rpc.ListenSpec{
Addrs: rpc.ListenAddrs{{Protocol: "tcp", Address: "127.0.0.1:0"}},
})
- _, err := NewDispatchingServer(ctx, "server", &testDispatcher{}, nil, "")
- if err != nil {
- t.Fatal(verror.DebugString(err))
- }
- ctx = fake.SetFlowManager(ctx, manager.New(ctx, naming.FixedRoutingID(0x2)))
- client, err := NewXClient(ctx, v23.ExperimentalGetFlowManager(ctx), v23.GetNamespace(ctx))
+ _, _, err := WithNewDispatchingServer(ctx, "server", &testDispatcher{}, nil, "")
if err != nil {
t.Fatal(verror.DebugString(err))
}
var result string
- if err = client.Call(ctx, "server", "Echo", []interface{}{"hello"}, []interface{}{&result}); err != nil {
+ if err = v23.GetClient(ctx).Call(ctx, "server", "Echo", []interface{}{"hello"}, []interface{}{&result}); err != nil {
t.Fatal(verror.DebugString(err))
}
if want := "response:hello"; result != want {
diff --git a/runtime/internal/rpc/xclient.go b/runtime/internal/rpc/xclient.go
index 7e6c618..c96d14a 100644
--- a/runtime/internal/rpc/xclient.go
+++ b/runtime/internal/rpc/xclient.go
@@ -7,7 +7,6 @@
import (
"fmt"
"io"
- "net"
"sync"
"time"
@@ -27,6 +26,7 @@
"v.io/x/ref/lib/apilog"
slib "v.io/x/ref/lib/security"
+ "v.io/x/ref/runtime/internal/flow/manager"
inaming "v.io/x/ref/runtime/internal/naming"
)
@@ -37,16 +37,17 @@
dischargeBuffer = time.Minute
)
+type clientFlowManagerOpt struct {
+ mgr flow.Manager
+}
+
+func (clientFlowManagerOpt) RPCClientOpt() {}
+
type xclient struct {
flowMgr flow.Manager
ns namespace.T
preferredProtocols []string
-
- // We cache the IP networks on the device since it is not that cheap to read
- // network interfaces through os syscall.
- // TODO(toddw): this can be removed since netstate now implements caching
- // directly.
- ipNets []*net.IPNet
+ ctx *context.T
// typeCache maintains a cache of type encoders and decoders.
typeCache *typeCache
@@ -58,24 +59,37 @@
var _ rpc.Client = (*xclient)(nil)
-func NewXClient(ctx *context.T, fm flow.Manager, ns namespace.T, opts ...rpc.ClientOpt) (rpc.Client, error) {
+func NewXClient(ctx *context.T, ns namespace.T, opts ...rpc.ClientOpt) rpc.Client {
c := &xclient{
- flowMgr: fm,
+ ctx: ctx,
ns: ns,
typeCache: newTypeCache(),
}
- ipNets, err := ipNetworks()
- if err != nil {
- return nil, err
- }
- c.ipNets = ipNets
+
for _, opt := range opts {
switch v := opt.(type) {
case PreferredProtocols:
c.preferredProtocols = v
+ case clientFlowManagerOpt:
+ c.flowMgr = v.mgr
}
}
- return c, nil
+ if c.flowMgr == nil {
+ c.flowMgr = manager.New(ctx, naming.NullRoutingID)
+ }
+
+ go func() {
+ <-ctx.Done()
+ c.mu.Lock()
+ // TODO(mattr): Do we really need c.closed?
+ c.closed = true
+ c.mu.Unlock()
+
+ <-c.flowMgr.Closed()
+ c.wg.Wait()
+ }()
+
+ return c
}
func (c *xclient) StartCall(ctx *context.T, name, method string, args []interface{}, opts ...rpc.CallOpt) (rpc.ClientCall, error) {
@@ -173,26 +187,43 @@
return
}
bfp := blessingsForPeer{auth, method, suffix, args}.run
- if status.flow, err = c.flowMgr.Dial(ctx, ep, bfp); err != nil {
+ flow, err := c.flowMgr.Dial(ctx, ep, bfp)
+ if err != nil {
ctx.VI(2).Infof("rpc: failed to create Flow with %v: %v", server, err)
status.serverErr = suberr(err)
return
}
- if write := c.typeCache.writer(status.flow.Conn()); write != nil {
- if tflow, err := c.flowMgr.Dial(ctx, ep, bfp); err != nil {
- ctx.VI(2).Infof("rpc: failed to create type Flow with %v: %v", server, err)
- status.serverErr = suberr(err)
+ if write := c.typeCache.writer(flow.Conn()); write != nil {
+ // Create a type flow, note that we use c.ctx instead of ctx.
+ // This is because type flows have a longer lifetime than the
+ // main flow being constructed.
+ tflow, err := c.flowMgr.Dial(c.ctx, ep, bfp)
+ if err != nil {
+ status.serverErr = suberr(newErrTypeFlowFailure(ctx, err))
+ flow.Close()
return
- } else if _, err = tflow.Write([]byte{typeFlow}); err != nil {
- ctx.VI(2).Infof("rpc: Failed to write type byte. %v: %v", server, err)
- tflow.Close()
- status.serverErr = suberr(err)
- return
- } else {
- write(tflow)
}
+ if tflow.Conn() != flow.Conn() {
+ status.serverErr = suberr(newErrTypeFlowFailure(ctx, nil))
+ flow.Close()
+ tflow.Close()
+ return
+ }
+ if _, err = tflow.Write([]byte{typeFlow}); err != nil {
+ status.serverErr = suberr(newErrTypeFlowFailure(ctx, nil))
+ flow.Close()
+ tflow.Close()
+ return
+ }
+ write(tflow)
}
- status.typeEnc, status.typeDec = c.typeCache.get(status.flow.Conn())
+ status.typeEnc, status.typeDec, err = c.typeCache.get(ctx, flow.Conn())
+ if err != nil {
+ status.serverErr = suberr(newErrTypeFlowFailure(ctx, err))
+ flow.Close()
+ return
+ }
+ status.flow = flow
}
type blessingsForPeer struct {
@@ -259,7 +290,7 @@
// This should never happen.
return nil, verror.NoRetry, true, verror.New(verror.ErrInternal, ctx, name)
}
- if resolved.Servers, err = filterAndOrderServers(resolved.Servers, c.preferredProtocols, c.ipNets); err != nil {
+ if resolved.Servers, err = filterAndOrderServers(resolved.Servers, c.preferredProtocols); err != nil {
return nil, verror.RetryRefetch, true, verror.New(verror.ErrNoServers, ctx, name, err)
}
@@ -306,7 +337,6 @@
}
}
case <-ctx.Done():
- ctx.VI(2).Infof("rpc: timeout on connection to server %v ", name)
_, _, _, err := c.failedTryCall(ctx, name, method, responses, ch)
if verror.ErrorID(err) != verror.ErrTimeout.ID {
return nil, verror.NoRetry, false, verror.New(verror.ErrTimeout, ctx, err)
@@ -446,12 +476,11 @@
}
func (c *xclient) Close() {
- defer apilog.LogCall(nil)(nil) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
- c.mu.Lock()
- c.closed = true
- c.mu.Unlock()
- // TODO(toddw): Implement this!
- c.wg.Wait()
+ panic("this method is deprecated.")
+}
+
+func (c *xclient) Closed() <-chan struct{} {
+ return c.flowMgr.Closed()
}
// flowXClient implements the RPC client-side protocol for a single RPC, over a
@@ -705,6 +734,7 @@
berr := verror.New(id, fc.ctx, verror.New(errResponseDecoding, fc.ctx, verr))
return fc.close(berr)
}
+
// The response header must indicate the streaming results have ended.
if fc.response.Error == nil && !fc.response.EndStreamResults {
berr := verror.New(errRemainingStreamResults, fc.ctx)
diff --git a/runtime/internal/rpc/xserver.go b/runtime/internal/rpc/xserver.go
index 7266ab0..f2bdbac 100644
--- a/runtime/internal/rpc/xserver.go
+++ b/runtime/internal/rpc/xserver.go
@@ -19,7 +19,6 @@
"v.io/v23/context"
"v.io/v23/flow"
"v.io/v23/i18n"
- "v.io/v23/namespace"
"v.io/v23/naming"
"v.io/v23/options"
"v.io/v23/rpc"
@@ -63,50 +62,46 @@
disp rpc.Dispatcher // dispatcher to serve RPCs
dispReserved rpc.Dispatcher // dispatcher for reserved methods
active sync.WaitGroup // active goroutines we've spawned.
- stoppedChan chan struct{} // closed when the server has been stopped.
preferredProtocols []string // protocols to use when resolving proxy name to endpoint.
- // We cache the IP networks on the device since it is not that cheap to read
- // network interfaces through os syscall.
- // TODO(jhahn): Add monitoring the network interface changes.
- ipNets []*net.IPNet
- ns namespace.T
- servesMountTable bool
- isLeaf bool
+ servesMountTable bool
+ isLeaf bool
// TODO(cnicolaou): add roaming stats to rpcStats
stats *rpcStats // stats for this server.
}
-func NewServer(ctx *context.T, name string, object interface{}, authorizer security.Authorizer, settingsPublisher *pubsub.Publisher, settingsName string, opts ...rpc.ServerOpt) (rpc.Server, error) {
+func WithNewServer(ctx *context.T,
+ name string, object interface{}, authorizer security.Authorizer,
+ settingsPublisher *pubsub.Publisher, settingsName string,
+ opts ...rpc.ServerOpt) (*context.T, rpc.Server, error) {
if object == nil {
- return nil, verror.New(verror.ErrBadArg, ctx, "nil object")
+ return ctx, nil, verror.New(verror.ErrBadArg, ctx, "nil object")
}
invoker, err := objectToInvoker(object)
if err != nil {
- return nil, verror.New(verror.ErrBadArg, ctx, fmt.Sprintf("bad object: %v", err))
+ return ctx, nil, verror.New(verror.ErrBadArg, ctx, fmt.Sprintf("bad object: %v", err))
}
d := &leafDispatcher{invoker, authorizer}
opts = append([]rpc.ServerOpt{options.IsLeaf(true)}, opts...)
- return NewDispatchingServer(ctx, name, d, settingsPublisher, settingsName, opts...)
+ return WithNewDispatchingServer(ctx, name, d, settingsPublisher, settingsName, opts...)
}
-func NewDispatchingServer(ctx *context.T, name string, dispatcher rpc.Dispatcher, settingsPublisher *pubsub.Publisher, settingsName string, opts ...rpc.ServerOpt) (rpc.Server, error) {
+func WithNewDispatchingServer(ctx *context.T,
+ name string, dispatcher rpc.Dispatcher,
+ settingsPublisher *pubsub.Publisher, settingsName string,
+ opts ...rpc.ServerOpt) (*context.T, rpc.Server, error) {
if dispatcher == nil {
- return nil, verror.New(verror.ErrBadArg, ctx, "nil dispatcher")
+ return ctx, nil, verror.New(verror.ErrBadArg, ctx, "nil dispatcher")
}
- ctx, cancel := context.WithRootCancel(ctx)
- flowMgr := v23.ExperimentalGetFlowManager(ctx)
- ns, principal := v23.GetNamespace(ctx), v23.GetPrincipal(ctx)
- statsPrefix := naming.Join("rpc", "server", "routing-id", flowMgr.RoutingID().String())
+
+ rootCtx, cancel := context.WithRootCancel(ctx)
+ fm, err := v23.NewFlowManager(rootCtx)
+
+ statsPrefix := naming.Join("rpc", "server", "routing-id", fm.RoutingID().String())
s := &xserver{
- ctx: ctx,
cancel: cancel,
- flowMgr: flowMgr,
- principal: principal,
- blessings: principal.BlessingStore().Default(),
- publisher: publisher.New(ctx, ns, publishPeriod),
- stoppedChan: make(chan struct{}),
- ns: ns,
+ flowMgr: fm,
+ blessings: v23.GetPrincipal(rootCtx).BlessingStore().Default(),
stats: newRPCStats(statsPrefix),
settingsPublisher: settingsPublisher,
settingsName: settingsName,
@@ -114,12 +109,6 @@
typeCache: newTypeCache(),
proxyEndpoints: make(map[string]map[string]*inaming.Endpoint),
}
- ipNets, err := ipNetworks()
- if err != nil {
- return nil, err
- }
- s.ipNets = ipNets
-
for _, opt := range opts {
switch opt := opt.(type) {
case options.ServesMountTable:
@@ -132,19 +121,25 @@
s.preferredProtocols = []string(opt)
}
}
-
- blessingsStatsName := naming.Join(statsPrefix, "security", "blessings")
- // TODO(caprita): revist printing the blessings with %s, and
- // instead expose them as a list.
- stats.NewString(blessingsStatsName).Set(fmt.Sprintf("%s", s.blessings))
- stats.NewStringFunc(blessingsStatsName, func() string {
- return fmt.Sprintf("%s (default)", s.principal.BlessingStore().Default())
- })
- if err = s.listen(ctx, v23.GetListenSpec(ctx)); err != nil {
- s.Stop()
- return nil, err
+ rootCtx, _, err = v23.WithNewClient(rootCtx,
+ clientFlowManagerOpt{fm},
+ PreferredProtocols(s.preferredProtocols))
+ if err != nil {
+ cancel()
+ return ctx, nil, err
}
+ s.ctx = rootCtx
+ s.publisher = publisher.New(rootCtx, v23.GetNamespace(rootCtx), publishPeriod)
+
+ // TODO(caprita): revist printing the blessings with string, and
+ // instead expose them as a list.
+ blessingsStatsName := naming.Join(statsPrefix, "security", "blessings")
+ stats.NewString(blessingsStatsName).Set(s.blessings.String())
+
+ s.listen(rootCtx, v23.GetListenSpec(rootCtx))
if len(name) > 0 {
+ // TODO(mattr): We only call AddServer here, but if someone calls AddName
+ // later there will be no servers?
s.mu.Lock()
for k, _ := range s.chosenEndpoints {
s.publisher.AddServer(k)
@@ -153,7 +148,66 @@
s.publisher.AddName(name, s.servesMountTable, s.isLeaf)
vtrace.GetSpan(s.ctx).Annotate("Serving under name: " + name)
}
- return s, nil
+
+ go func() {
+ <-ctx.Done()
+ serverDebug := fmt.Sprintf("Dispatcher: %T, Status:[%v]", s.disp, s.Status())
+ s.ctx.VI(1).Infof("Stop: %s", serverDebug)
+ defer s.ctx.VI(1).Infof("Stop done: %s", serverDebug)
+
+ s.stats.stop()
+ s.publisher.Stop()
+
+ done := make(chan struct{})
+ go func() {
+ s.flowMgr.StopListening(ctx)
+ s.publisher.WaitForStop()
+ // At this point no new flows should arrive. Wait for existing calls
+ // to complete.
+ s.active.Wait()
+ close(done)
+ }()
+
+ s.Lock()
+ // TODO(mattr): I don't understand what this is.
+ if dhcp := s.dhcpState; dhcp != nil {
+ // TODO(cnicolaou,caprita): investigate not having to close and drain
+ // the channel here. It's a little awkward right now since we have to
+ // be careful to not close the channel in two places, i.e. here and
+ // and from the publisher's Shutdown method.
+ if err := dhcp.publisher.CloseFork(dhcp.name, dhcp.ch); err == nil {
+ drain:
+ for {
+ select {
+ case v := <-dhcp.ch:
+ if v == nil {
+ break drain
+ }
+ default:
+ close(dhcp.ch)
+ break drain
+ }
+ }
+ }
+ }
+ s.Unlock()
+
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second): // TODO(mattr): This should be configurable.
+ s.ctx.Errorf("%s: Timedout waiting for active requests to complete", serverDebug)
+ }
+ // Now we cancel the root context which closes all the connections
+ // in the flow manager and cancels all the contexts used by
+ // ongoing requests. Hopefully this will bring all outstanding
+ // operations to a close.
+ s.cancel()
+ <-s.flowMgr.Closed()
+ // Now we really will wait forever. If this doesn't exit, there's something
+ // wrong with the users code.
+ <-done
+ }()
+ return rootCtx, s, nil
}
func (s *xserver) Status() rpc.ServerStatus {
@@ -185,11 +239,12 @@
}
// resolveToEndpoint resolves an object name or address to an endpoint.
-func (s *xserver) resolveToEndpoint(address string) (naming.Endpoint, error) {
+func (s *xserver) resolveToEndpoint(ctx *context.T, address string) (naming.Endpoint, error) {
var resolved *naming.MountEntry
var err error
- if s.ns != nil {
- if resolved, err = s.ns.Resolve(s.ctx, address); err != nil {
+ // TODO(mattr): Why should ns be nil?
+ if ns := v23.GetNamespace(ctx); ns != nil {
+ if resolved, err = ns.Resolve(ctx, address); err != nil {
return nil, err
}
} else {
@@ -199,7 +254,7 @@
}}
}
// An empty set of protocols means all protocols...
- if resolved.Servers, err = filterAndOrderServers(resolved.Servers, s.preferredProtocols, s.ipNets); err != nil {
+ if resolved.Servers, err = filterAndOrderServers(resolved.Servers, s.preferredProtocols); err != nil {
return nil, err
}
for _, n := range resolved.Names() {
@@ -295,13 +350,14 @@
return ret
}
-func (s *xserver) listen(ctx *context.T, listenSpec rpc.ListenSpec) error {
+func (s *xserver) listen(ctx *context.T, listenSpec rpc.ListenSpec) {
s.Lock()
defer s.Unlock()
var lastErr error
+ var ep naming.Endpoint
if len(listenSpec.Proxy) > 0 {
- var ep naming.Endpoint
- if ep, lastErr = s.resolveToEndpoint(listenSpec.Proxy); lastErr != nil {
+ ep, lastErr = s.resolveToEndpoint(ctx, listenSpec.Proxy)
+ if lastErr != nil {
s.ctx.VI(2).Infof("resolveToEndpoint(%q) failed: %v", listenSpec.Proxy, lastErr)
} else {
lastErr = s.flowMgr.ProxyListen(ctx, ep, s.update(ep))
@@ -320,7 +376,6 @@
}
leps := s.flowMgr.ListeningEndpoints()
-
s.addressChooser = listenSpec.AddressChooser
roaming := false
chosenEps := make(map[string]*inaming.Endpoint)
@@ -346,7 +401,6 @@
s.active.Add(1)
go s.acceptLoop(ctx)
- return nil
}
func (s *xserver) acceptLoop(ctx *context.T) error {
@@ -389,12 +443,9 @@
}
}
case typeFlow:
- write := s.typeCache.writer(fl.Conn())
- if write == nil {
- s.ctx.VI(1).Infof("ignoring duplicate type flow.")
- return
+ if write := s.typeCache.writer(fl.Conn()); write != nil {
+ write(fl)
}
- write(fl)
}
}(fl)
}
@@ -422,105 +473,23 @@
func (s *xserver) Stop() error {
defer apilog.LogCall(nil)(nil) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
+ panic("unimplemented")
+}
- serverDebug := fmt.Sprintf("Dispatcher: %T, Status:[%v]", s.disp, s.Status())
- s.ctx.VI(1).Infof("Stop: %s", serverDebug)
- defer s.ctx.VI(1).Infof("Stop done: %s", serverDebug)
-
- s.Lock()
- if s.disp == nil {
- s.Unlock()
- return nil
- }
- s.disp = nil
- close(s.stoppedChan)
- s.Unlock()
-
- // Delete the stats object.
- s.stats.stop()
-
- // Note, It's safe to Stop/WaitForStop on the publisher outside of the
- // server lock, since publisher is safe for concurrent access.
- // Stop the publisher, which triggers unmounting of published names.
- s.publisher.Stop()
-
- // Wait for the publisher to be done unmounting before we can proceed to
- // close the listeners (to minimize the number of mounted names pointing
- // to endpoint that are no longer serving).
- //
- // TODO(caprita): See if make sense to fail fast on rejecting
- // connections once listeners are closed, and parallelize the publisher
- // and listener shutdown.
- s.publisher.WaitForStop()
-
- s.Lock()
-
- // TODO(mattr): What should we do when we stop a server now? We need to
- // interrupt Accept at some point, but it's weird to stop the flowmanager.
- // Close all listeners. No new flows will be accepted, while in-flight
- // flows will continue until they terminate naturally.
- // nListeners := len(s.listeners)
- // errCh := make(chan error, nListeners)
- // for ln, _ := range s.listeners {
- // go func(ln stream.Listener) {
- // errCh <- ln.Close()
- // }(ln)
- // }
-
- if dhcp := s.dhcpState; dhcp != nil {
- // TODO(cnicolaou,caprita): investigate not having to close and drain
- // the channel here. It's a little awkward right now since we have to
- // be careful to not close the channel in two places, i.e. here and
- // and from the publisher's Shutdown method.
- if err := dhcp.publisher.CloseFork(dhcp.name, dhcp.ch); err == nil {
- drain:
- for {
- select {
- case v := <-dhcp.ch:
- if v == nil {
- break drain
- }
- default:
- close(dhcp.ch)
- break drain
- }
- }
- }
- }
-
- s.Unlock()
-
- // At this point, we are guaranteed that no new requests are going to be
- // accepted.
-
- // Wait for the publisher and active listener + flows to finish.
- done := make(chan struct{}, 1)
- go func() { s.active.Wait(); done <- struct{}{} }()
-
- select {
- case <-done:
- case <-time.After(5 * time.Second):
- s.ctx.Errorf("%s: Timedout waiting for goroutines to stop", serverDebug)
- // TODO(mattr): This doesn't make sense, shouldn't we not wait after timing out?
- <-done
- s.ctx.Infof("%s: Done waiting.", serverDebug)
- }
-
- s.cancel()
- return nil
+func (s *xserver) Closed() <-chan struct{} {
+ return s.ctx.Done()
}
// flowServer implements the RPC server-side protocol for a single RPC, over a
// flow that's already connected to the client.
type xflowServer struct {
- ctx *context.T // context associated with the RPC
server *xserver // rpc.Server that this flow server belongs to
disp rpc.Dispatcher // rpc.Dispatcher that will serve RPCs on this flow
- dec *vom.Decoder // to decode requests and args from the client
- enc *vom.Encoder // to encode responses and results to the client
flow flow.Flow // underlying flow
// Fields filled in during the server invocation.
+ dec *vom.Decoder // to decode requests and args from the client
+ enc *vom.Encoder // to encode responses and results to the client
grantedBlessings security.Blessings
method, suffix string
tags []*vdl.Value
@@ -535,17 +504,10 @@
)
func newXFlowServer(flow flow.Flow, server *xserver) (*xflowServer, error) {
- server.Lock()
- disp := server.disp
- server.Unlock()
- typeEnc, typeDec := server.typeCache.get(flow.Conn())
fs := &xflowServer{
- ctx: server.ctx,
server: server,
- disp: disp,
+ disp: server.disp,
flow: flow,
- enc: vom.NewEncoderWithTypeEncoder(flow, typeEnc),
- dec: vom.NewDecoderWithTypeDecoder(flow, typeDec),
discharges: make(map[string]security.Discharge),
}
return fs, nil
@@ -566,20 +528,19 @@
if fs.server.dispReserved != nil {
_, auth, _ = fs.server.dispReserved.Lookup(ctx, params.Suffix)
}
- return authorize(fs.ctx, security.NewCall(params), auth)
+ return authorize(ctx, security.NewCall(params), auth)
}
func (fs *xflowServer) serve() error {
defer fs.flow.Close()
- results, err := fs.processRequest()
-
- vtrace.GetSpan(fs.ctx).Finish()
+ ctx, results, err := fs.processRequest()
+ vtrace.GetSpan(ctx).Finish()
var traceResponse vtrace.Response
// Check if the caller is permitted to view vtrace data.
- if fs.authorizeVtrace(fs.ctx) == nil {
- traceResponse = vtrace.GetResponse(fs.ctx)
+ if fs.authorizeVtrace(ctx) == nil {
+ traceResponse = vtrace.GetResponse(ctx)
}
// Respond to the client with the response header and positional results.
@@ -593,7 +554,7 @@
if err == io.EOF {
return err
}
- return verror.New(errResponseEncoding, fs.ctx, fs.LocalEndpoint().String(), fs.RemoteEndpoint().String(), err)
+ return verror.New(errResponseEncoding, ctx, fs.LocalEndpoint().String(), fs.RemoteEndpoint().String(), err)
}
if response.Error != nil {
return response.Error
@@ -603,7 +564,7 @@
if err == io.EOF {
return err
}
- return verror.New(errResultEncoding, fs.ctx, ix, fmt.Sprintf("%T=%v", res, res), err)
+ return verror.New(errResultEncoding, ctx, ix, fmt.Sprintf("%T=%v", res, res), err)
}
}
// TODO(ashankar): Should unread data from the flow be drained?
@@ -622,69 +583,67 @@
return nil
}
-func (fs *xflowServer) readRPCRequest() (*rpc.Request, error) {
- // TODO(toddw): How do we set the initial timeout? It might be shorter than
- // the timeout we set later, which we learn after we've decoded the request.
- /*
- // Set a default timeout before reading from the flow. Without this timeout,
- // a client that sends no request or a partial request will retain the flow
- // indefinitely (and lock up server resources).
- initTimer := newTimer(defaultCallTimeout)
- defer initTimer.Stop()
- fs.flow.SetDeadline(initTimer.C)
- */
-
+func (fs *xflowServer) readRPCRequest(ctx *context.T) (*rpc.Request, error) {
// Decode the initial request.
var req rpc.Request
if err := fs.dec.Decode(&req); err != nil {
- return nil, verror.New(verror.ErrBadProtocol, fs.ctx, newErrBadRequest(fs.ctx, err))
+ return nil, verror.New(verror.ErrBadProtocol, ctx, newErrBadRequest(ctx, err))
}
return &req, nil
}
-func (fs *xflowServer) processRequest() ([]interface{}, error) {
+func (fs *xflowServer) processRequest() (*context.T, []interface{}, error) {
fs.starttime = time.Now()
- req, err := fs.readRPCRequest()
+
+ // Set an initial deadline on the flow to ensure that we don't wait forever
+ // for the initial read.
+ ctx := fs.flow.SetDeadlineContext(fs.server.ctx, time.Now().Add(defaultCallTimeout))
+
+ typeEnc, typeDec, err := fs.server.typeCache.get(ctx, fs.flow.Conn())
+ if err != nil {
+ return ctx, nil, err
+ }
+ fs.enc = vom.NewEncoderWithTypeEncoder(fs.flow, typeEnc)
+ fs.dec = vom.NewDecoderWithTypeDecoder(fs.flow, typeDec)
+
+ req, err := fs.readRPCRequest(ctx)
if err != nil {
// We don't know what the rpc call was supposed to be, but we'll create
// a placeholder span so we can capture annotations.
- fs.ctx, _ = vtrace.WithNewSpan(fs.ctx, fmt.Sprintf("\"%s\".UNKNOWN", fs.suffix))
- return nil, err
+ // TODO(mattr): I'm not sure this makes sense anymore, but I'll revisit it
+ // when I'm doing another round of vtrace next quarter.
+ ctx, _ = vtrace.WithNewSpan(ctx, fmt.Sprintf("\"%s\".UNKNOWN", fs.suffix))
+ return ctx, nil, err
}
+
+ // Start building up a new context for the request now that we know
+ // the header information.
+ ctx = fs.server.ctx
+
// We must call fs.drainDecoderArgs for any error that occurs
// after this point, and before we actually decode the arguments.
fs.method = req.Method
fs.suffix = strings.TrimLeft(req.Suffix, "/")
-
if req.Language != "" {
- fs.ctx = i18n.WithLangID(fs.ctx, i18n.LangID(req.Language))
+ ctx = i18n.WithLangID(ctx, i18n.LangID(req.Language))
}
// TODO(mattr): Currently this allows users to trigger trace collection
// on the server even if they will not be allowed to collect the
// results later. This might be considered a DOS vector.
spanName := fmt.Sprintf("\"%s\".%s", fs.suffix, fs.method)
- fs.ctx, _ = vtrace.WithContinuedTrace(fs.ctx, spanName, req.TraceRequest)
+ ctx, _ = vtrace.WithContinuedTrace(ctx, spanName, req.TraceRequest)
+ ctx = fs.flow.SetDeadlineContext(ctx, req.Deadline.Time)
- var cancel context.CancelFunc
- if !req.Deadline.IsZero() {
- fs.ctx, cancel = context.WithDeadline(fs.ctx, req.Deadline.Time)
- } else {
- fs.ctx, cancel = context.WithCancel(fs.ctx)
- }
- fs.flow.SetContext(fs.ctx)
- // TODO(toddw): Explicitly cancel the context when the flow is done.
- _ = cancel
-
- if err := fs.readGrantedBlessings(req); err != nil {
+ if err := fs.readGrantedBlessings(ctx, req); err != nil {
fs.drainDecoderArgs(int(req.NumPosArgs))
- return nil, err
+ return ctx, nil, err
}
// Lookup the invoker.
- invoker, auth, err := fs.lookup(fs.suffix, fs.method)
+ invoker, auth, err := fs.lookup(ctx, fs.suffix, fs.method)
if err != nil {
fs.drainDecoderArgs(int(req.NumPosArgs))
- return nil, err
+ return ctx, nil, err
}
// Note that we strip the reserved prefix when calling the invoker so
@@ -694,31 +653,31 @@
// Prepare invoker and decode args.
numArgs := int(req.NumPosArgs)
- argptrs, tags, err := invoker.Prepare(fs.ctx, strippedMethod, numArgs)
+ argptrs, tags, err := invoker.Prepare(ctx, strippedMethod, numArgs)
fs.tags = tags
if err != nil {
fs.drainDecoderArgs(numArgs)
- return nil, err
+ return ctx, nil, err
}
if called, want := req.NumPosArgs, uint64(len(argptrs)); called != want {
fs.drainDecoderArgs(numArgs)
- return nil, newErrBadNumInputArgs(fs.ctx, fs.suffix, fs.method, called, want)
+ return ctx, nil, newErrBadNumInputArgs(ctx, fs.suffix, fs.method, called, want)
}
for ix, argptr := range argptrs {
if err := fs.dec.Decode(argptr); err != nil {
- return nil, newErrBadInputArg(fs.ctx, fs.suffix, fs.method, uint64(ix), err)
+ return ctx, nil, newErrBadInputArg(ctx, fs.suffix, fs.method, uint64(ix), err)
}
}
// Check application's authorization policy.
- if err := authorize(fs.ctx, fs, auth); err != nil {
- return nil, err
+ if err := authorize(ctx, fs, auth); err != nil {
+ return ctx, nil, err
}
// Invoke the method.
- results, err := invoker.Invoke(fs.ctx, fs, strippedMethod, argptrs)
+ results, err := invoker.Invoke(ctx, fs, strippedMethod, argptrs)
fs.server.stats.record(fs.method, time.Since(fs.starttime))
- return results, err
+ return ctx, results, err
}
// drainDecoderArgs drains the next n arguments encoded onto the flows decoder.
@@ -741,7 +700,7 @@
// with rpc.DebugKeyword, we use the internal debug dispatcher to look up the
// invoker. Otherwise, and we use the server's dispatcher. The suffix and method
// value may be modified to match the actual suffix and method to use.
-func (fs *xflowServer) lookup(suffix string, method string) (rpc.Invoker, security.Authorizer, error) {
+func (fs *xflowServer) lookup(ctx *context.T, suffix string, method string) (rpc.Invoker, security.Authorizer, error) {
if naming.IsReserved(method) {
return reservedInvoker(fs.disp, fs.server.dispReserved), security.AllowEveryone(), nil
}
@@ -749,26 +708,26 @@
if naming.IsReserved(suffix) {
disp = fs.server.dispReserved
} else if fs.server.isLeaf && suffix != "" {
- innerErr := verror.New(errUnexpectedSuffix, fs.ctx, suffix)
- return nil, nil, verror.New(verror.ErrUnknownSuffix, fs.ctx, suffix, innerErr)
+ innerErr := verror.New(errUnexpectedSuffix, ctx, suffix)
+ return nil, nil, verror.New(verror.ErrUnknownSuffix, ctx, suffix, innerErr)
}
if disp != nil {
- obj, auth, err := disp.Lookup(fs.ctx, suffix)
+ obj, auth, err := disp.Lookup(ctx, suffix)
switch {
case err != nil:
return nil, nil, err
case obj != nil:
invoker, err := objectToInvoker(obj)
if err != nil {
- return nil, nil, verror.New(verror.ErrInternal, fs.ctx, "invalid received object", err)
+ return nil, nil, verror.New(verror.ErrInternal, ctx, "invalid received object", err)
}
return invoker, auth, nil
}
}
- return nil, nil, verror.New(verror.ErrUnknownSuffix, fs.ctx, suffix)
+ return nil, nil, verror.New(verror.ErrUnknownSuffix, ctx, suffix)
}
-func (fs *xflowServer) readGrantedBlessings(req *rpc.Request) error {
+func (fs *xflowServer) readGrantedBlessings(ctx *context.T, req *rpc.Request) error {
if req.GrantedBlessings.IsZero() {
return nil
}
@@ -781,7 +740,8 @@
// this - should servers be able to assume that a blessing is something that
// does not have the authorizations that the server's own identity has?
if got, want := req.GrantedBlessings.PublicKey(), fs.LocalPrincipal().PublicKey(); got != nil && !reflect.DeepEqual(got, want) {
- return verror.New(verror.ErrNoAccess, fs.ctx, verror.New(errBlessingsNotBound, fs.ctx, got, want))
+ return verror.New(verror.ErrNoAccess, ctx,
+ verror.New(errBlessingsNotBound, ctx, got, want))
}
fs.grantedBlessings = req.GrantedBlessings
return nil
diff --git a/runtime/internal/rt/runtime.go b/runtime/internal/rt/runtime.go
index 77f07f9..fd3f075 100644
--- a/runtime/internal/rt/runtime.go
+++ b/runtime/internal/rt/runtime.go
@@ -54,7 +54,6 @@
backgroundKey
reservedNameKey
listenKey
- flowManagerKey
// initKey is used to store values that are only set at init time.
initKey
@@ -169,12 +168,6 @@
return nil, nil, nil, err
}
- // Add the flow.Manager to the context.
- // This initial Flow Manager can only be used as a client.
- ctx, _, err = r.setNewClientFlowManager(ctx)
- if err != nil {
- return nil, nil, nil, err
- }
// Add the Client to the context.
ctx, _, err = r.WithNewClient(ctx)
if err != nil {
@@ -306,27 +299,6 @@
return sm, nil
}
-func (r *Runtime) setNewClientFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
- return r.setNewFlowManager(ctx, naming.NullRoutingID)
-}
-
-func (r *Runtime) setNewBidiFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
- rid, err := naming.NewRoutingID()
- if err != nil {
- return nil, nil, err
- }
- return r.setNewFlowManager(ctx, rid)
-}
-
-func (r *Runtime) setNewFlowManager(ctx *context.T, rid naming.RoutingID) (*context.T, flow.Manager, error) {
- fm := manager.New(ctx, rid)
- if err := r.addChild(ctx, fm, func() { <-fm.Closed() }); err != nil {
- return ctx, nil, err
- }
- newctx := context.WithValue(ctx, flowManagerKey, fm)
- return newctx, fm, nil
-}
-
func (r *Runtime) setNewStreamManager(ctx *context.T) (*context.T, error) {
sm, err := newStreamManager(ctx)
if err != nil {
@@ -381,14 +353,6 @@
if newctx, err = r.setNewStreamManager(newctx); err != nil {
return ctx, err
}
- if rid := r.ExperimentalGetFlowManager(newctx).RoutingID(); rid == naming.NullRoutingID {
- newctx, _, err = r.setNewClientFlowManager(newctx)
- } else {
- newctx, _, err = r.setNewBidiFlowManager(newctx)
- }
- if err != nil {
- return ctx, err
- }
if newctx, _, err = r.setNewNamespace(newctx, r.GetNamespace(ctx).Roots()...); err != nil {
return ctx, err
}
@@ -412,35 +376,28 @@
p, _ := ctx.Value(principalKey).(security.Principal)
sm, _ := ctx.Value(streamManagerKey).(stream.Manager)
ns, _ := ctx.Value(namespaceKey).(namespace.T)
- fm, _ := ctx.Value(flowManagerKey).(flow.Manager)
otherOpts = append(otherOpts, imanager.DialTimeout(5*time.Minute))
-
if id, _ := ctx.Value(initKey).(*initData); id.protocols != nil {
otherOpts = append(otherOpts, irpc.PreferredProtocols(id.protocols))
}
var client rpc.Client
- var err error
deps := []interface{}{vtraceDependency{}}
- if fm != nil && ref.RPCTransitionState() >= ref.XClients {
- client, err = irpc.NewTransitionClient(ctx, sm, fm, ns, otherOpts...)
- deps = append(deps, fm, sm)
- } else {
- client, err = irpc.DeprecatedNewClient(sm, ns, otherOpts...)
+ if ref.RPCTransitionState() >= ref.XClients {
+ client = irpc.NewTransitionClient(ctx, sm, ns, otherOpts...)
deps = append(deps, sm)
- }
-
- if err != nil {
- return ctx, nil, err
+ } else {
+ client = irpc.DeprecatedNewClient(sm, ns, otherOpts...)
+ deps = append(deps, sm)
}
newctx := context.WithValue(ctx, clientKey, client)
if p != nil {
deps = append(deps, p)
}
- if err = r.addChild(ctx, client, client.Close, deps...); err != nil {
+ if err := r.addChild(ctx, client, client.Close, deps...); err != nil {
return ctx, nil, err
}
- return newctx, client, err
+ return newctx, client, nil
}
func (*Runtime) GetClient(ctx *context.T) rpc.Client {
@@ -539,33 +496,16 @@
return nil
}
-func (*Runtime) ExperimentalGetFlowManager(ctx *context.T) flow.Manager {
- // nologcall
- if d, ok := ctx.Value(flowManagerKey).(flow.Manager); ok {
- return d
- }
- return nil
-}
-
-func (r *Runtime) ExperimentalWithNewFlowManager(ctx *context.T) (*context.T, flow.Manager, error) {
+func (r *Runtime) NewFlowManager(ctx *context.T) (flow.Manager, error) {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
- newctx, m, err := r.setNewBidiFlowManager(ctx)
+ rid, err := naming.NewRoutingID()
if err != nil {
- return ctx, nil, err
+ return nil, err
}
- // Create a new client since it depends on the flow manager.
- newctx, _, err = r.WithNewClient(newctx)
- if err != nil {
- return ctx, nil, err
- }
- return newctx, m, nil
+ return manager.New(ctx, rid), nil
}
-func (r *Runtime) commonServerInit(ctx *context.T, opts ...rpc.ServerOpt) (*context.T, *pubsub.Publisher, string, []rpc.ServerOpt, error) {
- newctx, _, err := r.ExperimentalWithNewFlowManager(ctx)
- if err != nil {
- return ctx, nil, "", nil, err
- }
+func (r *Runtime) commonServerInit(ctx *context.T, opts ...rpc.ServerOpt) (*pubsub.Publisher, string, []rpc.ServerOpt, error) {
otherOpts := append([]rpc.ServerOpt{}, opts...)
if reservedDispatcher := r.GetReservedNameDispatcher(ctx); reservedDispatcher != nil {
otherOpts = append(otherOpts, irpc.ReservedNameDispatcher{
@@ -576,20 +516,21 @@
if id.protocols != nil {
otherOpts = append(otherOpts, irpc.PreferredServerResolveProtocols(id.protocols))
}
- return newctx, id.settingsPublisher, id.settingsName, otherOpts, nil
+ return id.settingsPublisher, id.settingsName, otherOpts, nil
}
func (r *Runtime) WithNewServer(ctx *context.T, name string, object interface{}, auth security.Authorizer, opts ...rpc.ServerOpt) (*context.T, rpc.Server, error) {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
if ref.RPCTransitionState() >= ref.XServers {
- // TODO(mattr): Deal with shutdown deps.
- newctx, spub, sname, opts, err := r.commonServerInit(ctx, opts...)
+ spub, sname, opts, err := r.commonServerInit(ctx, opts...)
if err != nil {
return ctx, nil, err
}
- s, err := irpc.NewServer(newctx, name, object, auth, spub, sname, opts...)
+ newctx, s, err := irpc.WithNewServer(ctx, name, object, auth, spub, sname, opts...)
if err != nil {
- // TODO(mattr): Stop the flow manager.
+ return ctx, nil, err
+ }
+ if err = r.addChild(ctx, s, func() { <-s.Closed() }); err != nil {
return ctx, nil, err
}
return newctx, s, nil
@@ -612,14 +553,15 @@
func (r *Runtime) WithNewDispatchingServer(ctx *context.T, name string, disp rpc.Dispatcher, opts ...rpc.ServerOpt) (*context.T, rpc.Server, error) {
defer apilog.LogCall(ctx)(ctx) // gologcop: DO NOT EDIT, MUST BE FIRST STATEMENT
if ref.RPCTransitionState() >= ref.XServers {
- // TODO(mattr): Deal with shutdown deps.
- newctx, spub, sname, opts, err := r.commonServerInit(ctx, opts...)
+ spub, sname, opts, err := r.commonServerInit(ctx, opts...)
if err != nil {
return ctx, nil, err
}
- s, err := irpc.NewDispatchingServer(newctx, name, disp, spub, sname, opts...)
+ newctx, s, err := irpc.WithNewDispatchingServer(ctx, name, disp, spub, sname, opts...)
if err != nil {
- // TODO(mattr): Stop the flow manager.
+ return ctx, nil, err
+ }
+ if err = r.addChild(ctx, s, func() { <-s.Closed() }); err != nil {
return ctx, nil, err
}
return newctx, s, nil
diff --git a/runtime/internal/rt/runtime_test.go b/runtime/internal/rt/runtime_test.go
index da50a92..0a424e7 100644
--- a/runtime/internal/rt/runtime_test.go
+++ b/runtime/internal/rt/runtime_test.go
@@ -141,25 +141,12 @@
ctx, shutdown := test.V23Init()
defer shutdown()
- oldman := v23.ExperimentalGetFlowManager(ctx)
- if oldman == nil {
- t.Error("ExperimentalGetFlowManager should have returned a non-nil value")
+ oldman, err := v23.NewFlowManager(ctx)
+ if err != nil || oldman == nil {
+ t.Error("NewFlowManager failed: %v, %v", oldman, err)
}
- if rid := oldman.RoutingID(); rid != naming.NullRoutingID {
- t.Errorf("Initial flow.Manager should have NullRoutingID, got %v", rid)
- }
- newctx, newman, err := v23.ExperimentalWithNewFlowManager(ctx)
+ newman, err := v23.NewFlowManager(ctx)
if err != nil || newman == nil || newman == oldman {
- t.Fatalf("Could not create flow manager: %v", err)
- }
- if !newctx.Initialized() {
- t.Fatal("Got uninitialized context.")
- }
- man := v23.ExperimentalGetFlowManager(newctx)
- if man != newman || man == oldman {
- t.Error("ExperimentalWithNewFlowManager didn't update the context properly")
- }
- if man.RoutingID() == naming.NullRoutingID {
- t.Error("Newly created flow.Manager should not have NullRoutingID")
+ t.Fatalf("NewFlowManager failed: %v, %v", newman, err)
}
}
diff --git a/services/wspr/internal/lib/signature_manager_test.go b/services/wspr/internal/lib/signature_manager_test.go
index 689df80..97240ab 100644
--- a/services/wspr/internal/lib/signature_manager_test.go
+++ b/services/wspr/internal/lib/signature_manager_test.go
@@ -11,6 +11,7 @@
"v.io/v23"
"v.io/v23/context"
+ "v.io/v23/rpc"
"v.io/v23/vdl"
"v.io/v23/vdlroot/signature"
"v.io/x/ref/runtime/factories/fake"
@@ -38,7 +39,9 @@
"__Signature": []interface{}{initialSig},
},
)
- ctx = fake.SetClient(ctx, client)
+ ctx = fake.SetClientFactory(ctx, func(ctx *context.T, opts ...rpc.ClientOpt) rpc.Client {
+ return client
+ })
return ctx, client, shutdown
}
diff --git a/services/wspr/internal/lib/simple_client.go b/services/wspr/internal/lib/simple_client.go
index a46fd82..17d18ab 100644
--- a/services/wspr/internal/lib/simple_client.go
+++ b/services/wspr/internal/lib/simple_client.go
@@ -90,6 +90,10 @@
func (*simpleMockClient) Close() {
}
+func (*simpleMockClient) Closed() <-chan struct{} {
+ return nil
+}
+
// mockCall implements rpc.ClientCall
type mockCall struct {
mockStream
diff --git a/services/xproxyd/proxy_test.go b/services/xproxyd/proxy_test.go
index 4a08676..5343d26 100644
--- a/services/xproxyd/proxy_test.go
+++ b/services/xproxyd/proxy_test.go
@@ -89,11 +89,6 @@
if err != nil {
t.Fatal(err)
}
- // Create a new flow manager for the client.
- cctx, _, err := v23.ExperimentalWithNewFlowManager(ctx)
- if err != nil {
- t.Fatal(err)
- }
// Wait for the server to finish listening through the proxy.
eps := s.Status().Endpoints
for ; len(eps) == 0 || eps[0].Addr().Network() == ""; eps = s.Status().Endpoints {
@@ -101,7 +96,7 @@
}
var got string
- if err := v23.GetClient(cctx).Call(ctx, eps[0].Name(), "Echo", []interface{}{"hello"}, []interface{}{&got}); err != nil {
+ if err := v23.GetClient(ctx).Call(ctx, eps[0].Name(), "Echo", []interface{}{"hello"}, []interface{}{&got}); err != nil {
t.Fatal(err)
}
if want := "response:hello"; got != want {
@@ -114,30 +109,29 @@
defer goroutines.NoLeaks(t, leakWaitTime)()
kp := newKillProtocol()
flow.RegisterProtocol("kill", kp)
- pctx, shutdown := v23.Init()
+ ctx, shutdown := v23.Init()
defer shutdown()
- actx, am, err := v23.ExperimentalWithNewFlowManager(pctx)
+ am, err := v23.NewFlowManager(ctx)
if err != nil {
t.Fatal(err)
}
- dctx, dm, err := v23.ExperimentalWithNewFlowManager(pctx)
+ dm, err := v23.NewFlowManager(ctx)
if err != nil {
t.Fatal(err)
}
- pep := startProxy(t, pctx, address{"kill", "127.0.0.1:0"})
+ pep := startProxy(t, ctx, address{"kill", "127.0.0.1:0"})
done := make(chan struct{})
update := func(eps []naming.Endpoint) {
if len(eps) > 0 {
- if err := testEndToEndConnection(t, dctx, actx, dm, am, eps[0]); err != nil {
+ if err := testEndToEndConnection(t, ctx, dm, am, eps[0]); err != nil {
t.Error(err)
}
close(done)
}
}
-
- if err := am.ProxyListen(actx, pep, update); err != nil {
+ if err := am.ProxyListen(ctx, pep, update); err != nil {
t.Fatal(err)
}
<-done
@@ -147,22 +141,22 @@
defer goroutines.NoLeaks(t, leakWaitTime)()
kp := newKillProtocol()
flow.RegisterProtocol("kill", kp)
- pctx, shutdown := v23.Init()
+ ctx, shutdown := v23.Init()
defer shutdown()
- actx, am, err := v23.ExperimentalWithNewFlowManager(pctx)
+ am, err := v23.NewFlowManager(ctx)
if err != nil {
t.Fatal(err)
}
- dctx, dm, err := v23.ExperimentalWithNewFlowManager(pctx)
+ dm, err := v23.NewFlowManager(ctx)
if err != nil {
t.Fatal(err)
}
- pep := startProxy(t, pctx, address{"kill", "127.0.0.1:0"})
+ pep := startProxy(t, ctx, address{"kill", "127.0.0.1:0"})
- p2ep := startProxy(t, pctx, address{"v23", pep.String()}, address{"kill", "127.0.0.1:0"})
+ p2ep := startProxy(t, ctx, address{"v23", pep.String()}, address{"kill", "127.0.0.1:0"})
- p3ep := startProxy(t, pctx, address{"v23", p2ep.String()}, address{"kill", "127.0.0.1:0"})
+ p3ep := startProxy(t, ctx, address{"v23", p2ep.String()}, address{"kill", "127.0.0.1:0"})
done := make(chan struct{})
update := func(eps []naming.Endpoint) {
@@ -173,24 +167,24 @@
// 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, eps[rand.Int()%3]); err != nil {
+ if err := testEndToEndConnection(t, ctx, dm, am, eps[rand.Int()%3]); err != nil {
t.Error(err)
}
close(done)
}
}
- if err := am.ProxyListen(actx, p3ep, update); err != nil {
+ if err := am.ProxyListen(ctx, p3ep, update); err != nil {
t.Fatal(err)
}
<-done
}
-func testEndToEndConnection(t *testing.T, dctx, actx *context.T, dm, am flow.Manager, aep naming.Endpoint) error {
+func testEndToEndConnection(t *testing.T, ctx *context.T, dm, am flow.Manager, aep naming.Endpoint) error {
// The dialing flow.Manager dials a flow to the accepting flow.Manager.
want := "Do you read me?"
- df, err := dm.Dial(dctx, aep, bfp)
+ df, err := dm.Dial(ctx, aep, bfp)
if err != nil {
return err
}
@@ -198,7 +192,7 @@
if err := writeLine(df, want); err != nil {
return err
}
- af, err := am.Accept(actx)
+ af, err := am.Accept(ctx)
if err != nil {
return err
}
diff --git a/services/xproxyd/proxyd.go b/services/xproxyd/proxyd.go
index 85d66f7..1ead875 100644
--- a/services/xproxyd/proxyd.go
+++ b/services/xproxyd/proxyd.go
@@ -25,7 +25,7 @@
}
func New(ctx *context.T) (*proxy, *context.T, error) {
- ctx, mgr, err := v23.ExperimentalWithNewFlowManager(ctx)
+ mgr, err := v23.NewFlowManager(ctx)
if err != nil {
return nil, nil, err
}
