veyron/runtimes/google/rt: Each ipc.Server has its own RoutingID (and stream.Manager).
MOTIVATION:
Prior to this change, all ipc.Servers in a Runtime shared the same RoutingID.
This made demultiplexing impossible - for example, if two ipc.Server objects
(with the same RoutingID) were "Listening" through a proxy, the proxy did
not have enough information to decide which of the two a packet destined
for that RoutingID should go to.
With this change, each Server gets its own RoutingID and thus RoutingID
identifies a "Server" rather than a "Runtime".
That said, it is expected that an overwhelming majority of Veyron
applications would use a single ipc.Server, and for this common
case the ipc.Server and ipc.Client objects will share a stream.Manager
and RoutingID.
The only reason to have multiple ipc.Server objects in a single
Runtime is if the Servers are supposed to Listen on different addresses.
For example, a process has some services that it only wants to export
only via a bluetooth interface and other services that it wants to export
on an IP network.
While at it, updated the HTTP debug server so that it lists out all
stream.Mangers (and RoutingIDs) created for a runtime.
Change-Id: If61126ea11c265fe509208139cf9d78230ede0c5
diff --git a/runtimes/google/rt/http.go b/runtimes/google/rt/http.go
index d6f8588..c05e8cd 100644
--- a/runtimes/google/rt/http.go
+++ b/runtimes/google/rt/http.go
@@ -1,6 +1,8 @@
package rt
import (
+ "fmt"
+ "html/template"
"net"
"net/http"
"sync"
@@ -9,34 +11,47 @@
"net/http/pprof"
"veyron/runtimes/google/ipc/stream/manager"
-
"veyron2/ipc/stream"
+ "veyron2/naming"
"veyron2/vlog"
)
-var httpOnce sync.Once
+type debugServer struct {
+ init sync.Once
+ addr string
+ mux *http.ServeMux
-func (rt *vrt) startHTTPDebugServerOnce() {
- httpOnce.Do(func() {
- if len(rt.httpServer) > 0 {
- rt.startHTTPDebugServer(rt.httpServer, rt.sm)
- }
- })
+ mu sync.RWMutex
+ rids []naming.RoutingID // GUARDED_BY(mu)
}
-func (_ *vrt) startHTTPDebugServer(addr string, sm stream.Manager) {
- ln, err := net.Listen("tcp", addr)
- if err != nil {
- vlog.Errorf("Failed to setup debugging HTTP server. net.Listen(%q, %q): %v", "tcp", addr, err)
+func (rt *vrt) initHTTPDebugServer() {
+ // TODO(ashankar,cnicolaou): Change the default debug address to
+ // the empty string.
+ // In March 2014 this was temporarily set to "127.0.0.1:0" so that the debugging
+ // HTTP server always runs, which was useful during initial veyron
+ // development. We restrict it to localhost to avoid annoying firewall
+ // warnings and to provide a modicum of security.
+ rt.debug.addr = "127.0.0.1:0"
+ rt.debug.mux = http.NewServeMux()
+}
+
+func (rt *vrt) startHTTPDebugServerOnce() {
+ rt.debug.init.Do(func() { startHTTPDebugServer(&rt.debug) })
+}
+
+func startHTTPDebugServer(info *debugServer) {
+ if len(info.addr) == 0 {
return
}
- vlog.Infof("Starting HTTP debug server. See http://%v/debug/pprof and http://%v/debug/veyron", ln.Addr(), ln.Addr())
- mux := http.NewServeMux()
- if h, err := manager.HTTPHandler(sm); err != nil {
- vlog.Errorf("Failed to setup handler for ipc/stream/Manager implementation: %v", err)
- } else {
- mux.Handle("/debug/veyron/", h)
+ ln, err := net.Listen("tcp", info.addr)
+ if err != nil {
+ vlog.Errorf("Failed to setup debugging HTTP server. net.Listen(%q, %q): %v", "tcp", info.addr, err)
+ return
}
+ vlog.Infof("Starting HTTP debug server. See http://%v/debug", ln.Addr())
+ mux := info.mux
+ mux.Handle("/debug/", info)
// Since a custom ServeMux is used, net/http/pprof.init needs to be replicated
mux.HandleFunc("/debug/pprof/", pprof.Index)
mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
@@ -45,7 +60,45 @@
go func() {
server := &http.Server{Addr: ln.Addr().String(), Handler: mux}
if err := server.Serve(ln); err != nil {
- vlog.Infof("Debug HTTP server serve exited: %v", err)
+ vlog.Infof("Debug HTTP server exited: %v", err)
}
}()
}
+
+func (s *debugServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+ if err := tmplMain.Execute(w, s.rids); err != nil {
+ vlog.Errorf("Error executing HTTP template: %v", err)
+ }
+}
+
+func (s *debugServer) RegisterStreamManager(rid naming.RoutingID, sm stream.Manager) error {
+ h, err := manager.HTTPHandler(sm)
+ if err != nil {
+ return fmt.Errorf("failed to setup HTTP handler for ipc/stream/Manager implementation: %v", err)
+ }
+ s.mu.Lock()
+ s.rids = append(s.rids, rid)
+ s.mu.Unlock()
+ s.mux.Handle(fmt.Sprintf("/debug/veyron/%v", rid), h)
+ return nil
+}
+
+var (
+ tmplMain = template.Must(template.New("Debug").Parse(`<!doctype html>
+<html>
+<head>
+<meta charset="UTF-8">
+<title>Veyron Debug Server</title>
+</head>
+<body>
+<ul>
+<li><a href="/debug/pprof/">profiling</a></li>
+{{range .}}
+<li>ipc/stream/Manager for RoutingID <a href="/debug/veyron/{{.}}">{{.}}</a></li>
+{{end}}
+</ul>
+</body>
+</html>`))
+)
diff --git a/runtimes/google/rt/ipc.go b/runtimes/google/rt/ipc.go
index aaa4d11..2b54fad 100644
--- a/runtimes/google/rt/ipc.go
+++ b/runtimes/google/rt/ipc.go
@@ -1,6 +1,8 @@
package rt
import (
+ "fmt"
+
iipc "veyron/runtimes/google/ipc"
imanager "veyron/runtimes/google/ipc/stream/manager"
@@ -47,16 +49,28 @@
}
func (rt *vrt) NewServer(opts ...ipc.ServerOpt) (ipc.Server, error) {
- // Start the http debug server exactly once for this process
- rt.startHTTPDebugServerOnce()
+ var err error
+ // Create a new RoutingID (and StreamManager) for each server.
+ // Except, in the common case of a process having a single Server,
+ // use the same RoutingID (and StreamManager) that is used for Clients.
+ rt.mu.Lock()
sm := rt.sm
+ rt.nServers++
+ if rt.nServers > 1 {
+ sm, err = rt.NewStreamManager()
+ }
+ rt.mu.Unlock()
+
+ if err != nil {
+ return nil, fmt.Errorf("failed to create ipc/stream/Manager: %v", err)
+ }
+ // Start the http debug server exactly once for this runtime.
+ rt.startHTTPDebugServerOnce()
ns := rt.ns
var otherOpts []ipc.ServerOpt
for _, opt := range opts {
switch topt := opt.(type) {
- case veyron2.StreamManagerOpt:
- sm = topt
case veyron2.NamespaceOpt:
ns = topt
default:
@@ -75,10 +89,7 @@
if err != nil {
return nil, err
}
- rt.rid = rid
- return imanager.InternalNew(rt.rid), nil
-}
-
-func (rt *vrt) StreamManager() stream.Manager {
- return rt.sm
+ sm := imanager.InternalNew(rid)
+ rt.debug.RegisterStreamManager(rid, sm)
+ return sm, nil
}
diff --git a/runtimes/google/rt/rt.go b/runtimes/google/rt/rt.go
index cbffa02..a27b496 100644
--- a/runtimes/google/rt/rt.go
+++ b/runtimes/google/rt/rt.go
@@ -19,15 +19,17 @@
)
type vrt struct {
- product product.T
- httpServer string
- sm stream.Manager
- ns naming.Namespace
- rid naming.RoutingID
- signals chan os.Signal
- id *currentIDOpt
- client ipc.Client
- mgmt *mgmtImpl
+ product product.T
+ sm stream.Manager
+ ns naming.Namespace
+ signals chan os.Signal
+ id *currentIDOpt
+ client ipc.Client
+ mgmt *mgmtImpl
+ debug debugServer
+
+ mu sync.Mutex
+ nServers int // GUARDED_BY(mu)
}
var (
@@ -58,13 +60,7 @@
// init initalizes the runtime instance it is invoked on.
func (rt *vrt) init(opts ...veyron2.ROpt) error {
flag.Parse()
- // TODO(ashankar,cnicolaou): Change the default debug address to
- // the empty string.
- // In March 2014 this was temporarily set to "127.0.0.1:0" so that the debugging
- // HTTP server always runs, which was useful during initial veyron
- // development. We restrict it to localhost to avoid annoying firewall
- // warnings and to provide a modicum of security.
- rt.httpServer = "127.0.0.1:0"
+ rt.initHTTPDebugServer()
rt.id = ¤tIDOpt{}
nsRoots := []string{}
@@ -77,7 +73,7 @@
case veyron2.NamespaceRoots:
nsRoots = v
case veyron2.HTTPDebugOpt:
- rt.httpServer = string(v)
+ rt.debug.addr = string(v)
default:
return fmt.Errorf("option has wrong type %T", o)
}
diff --git a/services/wspr/wsprd/lib/server.go b/services/wspr/wsprd/lib/server.go
index 058ea42..0497ae6 100644
--- a/services/wspr/wsprd/lib/server.go
+++ b/services/wspr/wsprd/lib/server.go
@@ -58,16 +58,8 @@
veyronProxy: veyronProxy,
outstandingServerRequests: make(map[int64]chan *serverRPCReply),
}
-
- manager, err := helper.rt().NewStreamManager()
-
- if err != nil {
- return nil, err
- }
-
- server.server, err = helper.rt().NewServer(veyron2.StreamManager(manager))
-
- if err != nil {
+ var err error
+ if server.server, err = helper.rt().NewServer(); err != nil {
return nil, err
}
return server, nil
